返回 CodeWhale
automation_manager.rs
根目录 / crates / tui / src / automation_manager.rs
1 //! Durable automation records and scheduler support.
2 //!
3 //! Automations are local-first recurring jobs that enqueue standard background
4 //! tasks. This module stores automation definitions and run history under
5 //! `~/.codewhale/automations` (or `DEEPSEEK_AUTOMATIONS_DIR` override).
6
7 use std::collections::BTreeMap;
8 use std::fs;
9 use std::future::Future;
10 use std::path::{Path, PathBuf};
11 use std::sync::Arc;
12
13 use anyhow::{Context, Result, bail};
14 use chrono::{
15 DateTime, Datelike, Duration, Local, NaiveDateTime, TimeZone, Timelike, Utc, Weekday,
16 };
17 use serde::{Deserialize, Serialize};
18 use tokio::sync::Mutex;
19 use tokio::time::sleep;
20 use tokio_util::sync::CancellationToken;
21 use uuid::Uuid;
22
23 use crate::task_manager::{NewTaskRequest, SharedTaskManager, TaskStatus};
24 use crate::utils::spawn_supervised;
25
26 const CURRENT_AUTOMATION_SCHEMA_VERSION: u32 = 1;
27 const CURRENT_RUN_SCHEMA_VERSION: u32 = 1;
28 const CURRENT_TRIGGER_SCHEMA_VERSION: u32 = 1;
29 const DEFAULT_AUTOMATION_MODE: &str = "agent";
30 const DEFAULT_AUTOMATION_ALLOW_SHELL: bool = false;
31 const DEFAULT_AUTOMATION_TRUST_MODE: bool = false;
32 const DEFAULT_AUTOMATION_AUTO_APPROVE: bool = false;
33 const DEFAULT_AUTOMATION_DELIVERY_MODE: AutomationDeliveryMode = AutomationDeliveryMode::Task;
34 pub const AUTOMATION_WATCHER_NO_REPORT_SENTINEL: &str = "NOTHING_TO_REPORT";
35 const MAX_HOURLY_SEARCH_STEPS: usize = 24 * 21;
36 const MAX_CRON_SEARCH_MINUTES: usize = 60 * 24 * 366 * 5;
37
38 const fn default_automation_schema_version() -> u32 {
39 CURRENT_AUTOMATION_SCHEMA_VERSION
40 }
41
42 const fn default_run_schema_version() -> u32 {
43 CURRENT_RUN_SCHEMA_VERSION
44 }
45
46 const fn default_trigger_schema_version() -> u32 {
47 CURRENT_TRIGGER_SCHEMA_VERSION
48 }
49
50 // ── Delayed-trigger types ──────────────────────────────────────────────────
51
52 /// Status of a one-shot delayed trigger.
53 #[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
54 #[serde(rename_all = "snake_case")]
55 pub enum DelayedTriggerStatus {
56 /// Waiting to fire.
57 Pending,
58 /// The trigger was fired and a task was enqueued.
59 Fired,
60 /// The trigger was explicitly canceled before it fired.
61 Canceled,
62 /// The trigger fired but failed to enqueue a task.
63 Failed,
64 }
65
66 /// A durable one-shot delayed continuation record.
67 ///
68 /// Stored under `~/.codewhale/automations/triggers/{trigger_id}.json`.
69 #[derive(Debug, Clone, Serialize, Deserialize)]
70 pub struct DelayedTriggerRecord {
71 #[serde(default = "default_trigger_schema_version")]
72 pub schema_version: u32,
73 pub trigger_id: String,
74 /// Absolute UTC time at which the trigger should fire.
75 pub fire_at: DateTime<Utc>,
76 /// The message that will be submitted as a new task when the trigger fires.
77 pub message: String,
78 /// Working directory for the task that fires when the trigger trips.
79 #[serde(skip_serializing_if = "Option::is_none")]
80 pub workspace: Option<PathBuf>,
81 pub status: DelayedTriggerStatus,
82 pub created_at: DateTime<Utc>,
83 #[serde(skip_serializing_if = "Option::is_none")]
84 pub fired_at: Option<DateTime<Utc>>,
85 #[serde(skip_serializing_if = "Option::is_none")]
86 pub task_id: Option<String>,
87 #[serde(skip_serializing_if = "Option::is_none")]
88 pub thread_id: Option<String>,
89 #[serde(skip_serializing_if = "Option::is_none")]
90 pub error: Option<String>,
91 /// Optional lineage: the trigger id that scheduled this one (for re-arm chains).
92 #[serde(skip_serializing_if = "Option::is_none")]
93 pub parent_trigger_id: Option<String>,
94 }
95
96 /// Input for creating a new delayed trigger.
97 #[derive(Debug, Clone)]
98 pub struct CreateDelayedTriggerRequest {
99 /// Absolute fire time. Callers must resolve `delay_minutes` → `fire_at`
100 /// before calling this function.
101 pub fire_at: DateTime<Utc>,
102 /// Message to submit as a new task when the trigger fires.
103 pub message: String,
104 /// Optional workspace directory for the fired task.
105 pub workspace: Option<PathBuf>,
106 /// Optional parent trigger id for re-arm lineage tracking.
107 pub parent_trigger_id: Option<String>,
108 }
109
110 // ── End delayed-trigger types ──────────────────────────────────────────────
111
112 #[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
113 #[serde(rename_all = "snake_case")]
114 pub enum AutomationStatus {
115 Active,
116 Paused,
117 }
118
119 #[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
120 #[serde(rename_all = "snake_case")]
121 pub enum AutomationRunStatus {
122 Queued,
123 Running,
124 Completed,
125 Failed,
126 Canceled,
127 }
128
129 #[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, Default)]
130 #[serde(rename_all = "snake_case")]
131 pub enum AutomationDeliveryMode {
132 #[default]
133 Task,
134 Watcher,
135 }
136
137 #[derive(Debug, Clone, Serialize, Deserialize)]
138 pub struct AutomationRecord {
139 #[serde(default = "default_automation_schema_version")]
140 pub schema_version: u32,
141 pub id: String,
142 pub name: String,
143 pub prompt: String,
144 pub rrule: String,
145 #[serde(default)]
146 pub cwds: Vec<PathBuf>,
147 #[serde(default, skip_serializing_if = "Option::is_none")]
148 pub mode: Option<String>,
149 #[serde(default, skip_serializing_if = "Option::is_none")]
150 pub allow_shell: Option<bool>,
151 #[serde(default, skip_serializing_if = "Option::is_none")]
152 pub trust_mode: Option<bool>,
153 #[serde(default, skip_serializing_if = "Option::is_none")]
154 pub auto_approve: Option<bool>,
155 #[serde(default, skip_serializing_if = "Option::is_none")]
156 pub delivery_mode: Option<AutomationDeliveryMode>,
157 pub status: AutomationStatus,
158 pub created_at: DateTime<Utc>,
159 pub updated_at: DateTime<Utc>,
160 #[serde(skip_serializing_if = "Option::is_none")]
161 pub next_run_at: Option<DateTime<Utc>>,
162 #[serde(skip_serializing_if = "Option::is_none")]
163 pub last_run_at: Option<DateTime<Utc>>,
164 }
165
166 impl AutomationRecord {
167 fn task_mode(&self) -> String {
168 self.mode
169 .as_deref()
170 .map(str::trim)
171 .filter(|mode| !mode.is_empty())
172 .unwrap_or(DEFAULT_AUTOMATION_MODE)
173 .to_string()
174 }
175
176 fn task_allow_shell(&self) -> bool {
177 self.allow_shell.unwrap_or(DEFAULT_AUTOMATION_ALLOW_SHELL)
178 }
179
180 fn task_trust_mode(&self) -> bool {
181 self.trust_mode.unwrap_or(DEFAULT_AUTOMATION_TRUST_MODE)
182 }
183
184 fn task_auto_approve(&self) -> bool {
185 self.auto_approve.unwrap_or(DEFAULT_AUTOMATION_AUTO_APPROVE)
186 }
187
188 fn delivery_mode(&self) -> AutomationDeliveryMode {
189 self.delivery_mode
190 .unwrap_or(DEFAULT_AUTOMATION_DELIVERY_MODE)
191 }
192 }
193
194 #[derive(Debug, Clone, Serialize, Deserialize)]
195 pub struct AutomationRunRecord {
196 #[serde(default = "default_run_schema_version")]
197 pub schema_version: u32,
198 pub id: String,
199 pub automation_id: String,
200 pub scheduled_for: DateTime<Utc>,
201 pub status: AutomationRunStatus,
202 pub created_at: DateTime<Utc>,
203 #[serde(skip_serializing_if = "Option::is_none")]
204 pub started_at: Option<DateTime<Utc>>,
205 #[serde(skip_serializing_if = "Option::is_none")]
206 pub ended_at: Option<DateTime<Utc>>,
207 #[serde(skip_serializing_if = "Option::is_none")]
208 pub task_id: Option<String>,
209 #[serde(skip_serializing_if = "Option::is_none")]
210 pub thread_id: Option<String>,
211 #[serde(skip_serializing_if = "Option::is_none")]
212 pub turn_id: Option<String>,
213 #[serde(skip_serializing_if = "Option::is_none")]
214 pub error: Option<String>,
215 }
216
217 #[derive(Debug, Clone, Serialize, Deserialize)]
218 pub struct CreateAutomationRequest {
219 pub name: String,
220 pub prompt: String,
221 pub rrule: String,
222 #[serde(default)]
223 pub cwds: Vec<PathBuf>,
224 #[serde(default)]
225 pub mode: Option<String>,
226 #[serde(default)]
227 pub allow_shell: Option<bool>,
228 #[serde(default)]
229 pub trust_mode: Option<bool>,
230 #[serde(default)]
231 pub auto_approve: Option<bool>,
232 #[serde(default)]
233 pub delivery_mode: Option<AutomationDeliveryMode>,
234 #[serde(default)]
235 pub status: Option<AutomationStatus>,
236 }
237
238 #[derive(Debug, Clone, Serialize, Deserialize, Default)]
239 pub struct UpdateAutomationRequest {
240 pub name: Option<String>,
241 pub prompt: Option<String>,
242 pub rrule: Option<String>,
243 pub cwds: Option<Vec<PathBuf>>,
244 pub mode: Option<String>,
245 pub allow_shell: Option<bool>,
246 pub trust_mode: Option<bool>,
247 pub auto_approve: Option<bool>,
248 pub delivery_mode: Option<AutomationDeliveryMode>,
249 pub status: Option<AutomationStatus>,
250 }
251
252 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
253 enum AutomationFrequency {
254 Hourly,
255 Weekly,
256 }
257
258 #[derive(Debug, Clone)]
259 pub enum AutomationSchedule {
260 Once {
261 at: DateTime<Utc>,
262 },
263 Hourly {
264 interval_hours: u32,
265 byday: Option<Vec<Weekday>>,
266 anchor_hour: Option<u32>,
267 anchor_minute: Option<u32>,
268 },
269 Weekly {
270 byday: Vec<Weekday>,
271 byhour: u32,
272 byminute: u32,
273 },
274 Cron {
275 expr: String,
276 },
277 }
278
279 impl AutomationSchedule {
280 pub fn parse_rrule(rrule: &str) -> Result<Self> {
281 let mut parts: BTreeMap<String, String> = BTreeMap::new();
282 for raw in rrule.split(';') {
283 let item = raw.trim();
284 if item.is_empty() {
285 continue;
286 }
287 let Some((k, v)) = item.split_once('=') else {
288 bail!("Invalid RRULE segment '{item}'");
289 };
290 parts.insert(k.trim().to_ascii_uppercase(), v.trim().to_string());
291 }
292
293 let freq = match parts
294 .get("FREQ")
295 .map(|value| value.trim().to_ascii_uppercase())
296 .as_deref()
297 {
298 Some("ONCE") => return parse_once_schedule(&parts),
299 Some("HOURLY") => AutomationFrequency::Hourly,
300 Some("WEEKLY") => AutomationFrequency::Weekly,
301 Some("CRON") => return parse_cron_schedule(&parts),
302 Some(other) => {
303 bail!("Unsupported RRULE FREQ '{other}'. Supported: ONCE, HOURLY, WEEKLY, and CRON")
304 }
305 None => bail!("RRULE must include FREQ"),
306 };
307
308 match freq {
309 AutomationFrequency::Hourly => {
310 for key in parts.keys() {
311 if key != "FREQ"
312 && key != "INTERVAL"
313 && key != "BYDAY"
314 && key != "BYHOUR"
315 && key != "BYMINUTE"
316 {
317 bail!(
318 "Unsupported RRULE field '{key}' for HOURLY. Allowed: FREQ,INTERVAL,BYDAY,BYHOUR,BYMINUTE"
319 );
320 }
321 }
322 let interval_hours = parts
323 .get("INTERVAL")
324 .map(|v| v.parse::<u32>())
325 .transpose()
326 .context("Failed to parse INTERVAL")?
327 .unwrap_or(1);
328 if interval_hours == 0 {
329 bail!("INTERVAL must be >= 1 for HOURLY schedules");
330 }
331 let byday = parts
332 .get("BYDAY")
333 .map(|value| parse_byday(&value.to_ascii_uppercase()))
334 .transpose()?;
335 let anchor_hour = parts
336 .get("BYHOUR")
337 .map(|value| value.parse::<u32>())
338 .transpose()
339 .context("Failed to parse BYHOUR")?;
340 let anchor_minute = parts
341 .get("BYMINUTE")
342 .map(|value| value.parse::<u32>())
343 .transpose()
344 .context("Failed to parse BYMINUTE")?;
345 if anchor_hour.is_some_and(|hour| hour > 23) {
346 bail!("BYHOUR must be between 0 and 23");
347 }
348 if anchor_minute.is_some_and(|minute| minute > 59) {
349 bail!("BYMINUTE must be between 0 and 59");
350 }
351 Ok(Self::Hourly {
352 interval_hours,
353 byday,
354 anchor_hour,
355 anchor_minute,
356 })
357 }
358 AutomationFrequency::Weekly => {
359 for key in parts.keys() {
360 if key != "FREQ" && key != "BYDAY" && key != "BYHOUR" && key != "BYMINUTE" {
361 bail!(
362 "Unsupported RRULE field '{key}' for WEEKLY. Allowed: FREQ,BYDAY,BYHOUR,BYMINUTE"
363 );
364 }
365 }
366 let byday_raw = parts
367 .get("BYDAY")
368 .ok_or_else(|| anyhow::anyhow!("WEEKLY schedules require BYDAY"))?;
369 let byday = parse_byday(&byday_raw.to_ascii_uppercase())?;
370 if byday.is_empty() {
371 bail!("BYDAY cannot be empty for WEEKLY schedules");
372 }
373 let byhour = parts
374 .get("BYHOUR")
375 .ok_or_else(|| anyhow::anyhow!("WEEKLY schedules require BYHOUR"))?
376 .parse::<u32>()
377 .context("Failed to parse BYHOUR")?;
378 let byminute = parts
379 .get("BYMINUTE")
380 .ok_or_else(|| anyhow::anyhow!("WEEKLY schedules require BYMINUTE"))?
381 .parse::<u32>()
382 .context("Failed to parse BYMINUTE")?;
383
384 if byhour > 23 {
385 bail!("BYHOUR must be between 0 and 23");
386 }
387 if byminute > 59 {
388 bail!("BYMINUTE must be between 0 and 59");
389 }
390
391 Ok(Self::Weekly {
392 byday,
393 byhour,
394 byminute,
395 })
396 }
397 }
398 }
399
400 fn next_after_with_anchor(
401 &self,
402 after: DateTime<Utc>,
403 anchor_reference: DateTime<Utc>,
404 ) -> Result<DateTime<Utc>> {
405 self.next_after_in_timezone(after, anchor_reference, &Local)
406 }
407
408 fn next_after_in_timezone<Tz: TimeZone>(
409 &self,
410 after: DateTime<Utc>,
411 anchor_reference: DateTime<Utc>,
412 timezone: &Tz,
413 ) -> Result<DateTime<Utc>> {
414 let local_after = after.with_timezone(timezone);
415 match self {
416 Self::Once { at } => {
417 if *at > after {
418 Ok(*at)
419 } else {
420 bail!(
421 "Once schedule has no future run after {}",
422 after.to_rfc3339()
423 )
424 }
425 }
426 Self::Hourly {
427 interval_hours,
428 byday,
429 anchor_hour,
430 anchor_minute,
431 } => {
432 if anchor_hour.is_some() || anchor_minute.is_some() {
433 let local_anchor_reference = anchor_reference.with_timezone(timezone);
434 let hour = anchor_hour.unwrap_or(local_anchor_reference.hour());
435 let minute = anchor_minute.unwrap_or(0);
436 let anchor_naive = local_anchor_reference
437 .date_naive()
438 .and_hms_opt(hour, minute, 0)
439 .ok_or_else(|| anyhow::anyhow!("Unable to construct HOURLY anchor"))?;
440 let interval_seconds = i64::from(*interval_hours) * 60 * 60;
441 let elapsed_seconds = local_after
442 .naive_local()
443 .signed_duration_since(anchor_naive)
444 .num_seconds();
445 let mut steps = if elapsed_seconds < 0 {
446 0
447 } else {
448 elapsed_seconds / interval_seconds + 1
449 };
450
451 for _ in 0..MAX_HOURLY_SEARCH_STEPS {
452 let hours = i64::from(*interval_hours)
453 .checked_mul(steps)
454 .ok_or_else(|| anyhow::anyhow!("HOURLY schedule exceeded its range"))?;
455 let delta = Duration::try_hours(hours)
456 .ok_or_else(|| anyhow::anyhow!("HOURLY schedule exceeded its range"))?;
457 let candidate_naive = anchor_naive
458 .checked_add_signed(delta)
459 .ok_or_else(|| anyhow::anyhow!("HOURLY schedule exceeded its range"))?;
460
461 if byday
462 .as_ref()
463 .is_none_or(|days| days.contains(&candidate_naive.weekday()))
464 && let Some(candidate) =
465 resolve_local_datetime(timezone, candidate_naive)
466 {
467 let candidate = candidate.with_timezone(&Utc);
468 if candidate > after {
469 return Ok(candidate);
470 }
471 }
472
473 steps = steps
474 .checked_add(1)
475 .ok_or_else(|| anyhow::anyhow!("HOURLY schedule exceeded its range"))?;
476 }
477 bail!("Unable to compute next anchored HOURLY run");
478 }
479
480 let after_second = local_after.second();
481 let after_nanosecond = local_after.nanosecond();
482 let mut candidate = local_after + Duration::hours(i64::from(*interval_hours))
483 - Duration::seconds(i64::from(after_second))
484 - Duration::nanoseconds(i64::from(after_nanosecond));
485
486 if let Some(days) = byday {
487 for _ in 0..(24 * 21) {
488 if days.contains(&candidate.weekday()) {
489 return Ok(candidate.with_timezone(&Utc));
490 }
491 candidate += Duration::hours(i64::from(*interval_hours));
492 }
493 bail!("Unable to compute next HOURLY run for BYDAY filter");
494 }
495
496 Ok(candidate.with_timezone(&Utc))
497 }
498 Self::Weekly {
499 byday,
500 byhour,
501 byminute,
502 } => {
503 for day_offset in 0..15 {
504 let date = local_after.date_naive() + Duration::days(i64::from(day_offset));
505 if !byday.contains(&date.weekday()) {
506 continue;
507 }
508 let Some(candidate_naive) = date.and_hms_opt(*byhour, *byminute, 0) else {
509 continue;
510 };
511 if let Some(candidate) = resolve_local_datetime(timezone, candidate_naive)
512 && candidate.with_timezone(&Utc) > after
513 {
514 return Ok(candidate.with_timezone(&Utc));
515 }
516 }
517 bail!("Unable to compute next WEEKLY run");
518 }
519 Self::Cron { expr } => {
520 let cron = ParsedCronExpr::parse(expr)?;
521 let mut candidate_naive = local_after
522 .naive_local()
523 .with_second(0)
524 .and_then(|dt| dt.with_nanosecond(0))
525 .ok_or_else(|| anyhow::anyhow!("Unable to round CRON search start"))?
526 .checked_add_signed(Duration::minutes(1))
527 .ok_or_else(|| anyhow::anyhow!("CRON schedule exceeded its range"))?;
528
529 for _ in 0..MAX_CRON_SEARCH_MINUTES {
530 if cron.matches(candidate_naive)
531 && let Some(candidate) = resolve_local_datetime(timezone, candidate_naive)
532 {
533 let candidate = candidate.with_timezone(&Utc);
534 if candidate > after {
535 return Ok(candidate);
536 }
537 }
538 candidate_naive = candidate_naive
539 .checked_add_signed(Duration::minutes(1))
540 .ok_or_else(|| anyhow::anyhow!("CRON schedule exceeded its range"))?;
541 }
542 bail!("Unable to compute next CRON run within 5 years");
543 }
544 }
545 }
546
547 fn next_after_slot(
548 &self,
549 slot: DateTime<Utc>,
550 anchor_reference: DateTime<Utc>,
551 ) -> Result<Option<DateTime<Utc>>> {
552 match self {
553 Self::Once { .. } => Ok(None),
554 _ => self
555 .next_after_with_anchor(slot, anchor_reference)
556 .map(Some),
557 }
558 }
559 }
560
561 /// Resolve one calendar-local schedule slot.
562 ///
563 /// Nonexistent wall times in a forward clock change are skipped rather than
564 /// shifted to a different clock time. Ambiguous wall times in a backward clock
565 /// change use the first occurrence only, preventing a recurring automation from
566 /// running twice for one calendar slot.
567 fn resolve_local_datetime<Tz: TimeZone>(
568 timezone: &Tz,
569 naive: NaiveDateTime,
570 ) -> Option<DateTime<Tz>> {
571 timezone.from_local_datetime(&naive).earliest()
572 }
573
574 fn parse_byday(value: &str) -> Result<Vec<Weekday>> {
575 let mut days = Vec::new();
576 for token in value.split(',') {
577 let day = match token.trim().to_ascii_uppercase().as_str() {
578 "MO" => Weekday::Mon,
579 "TU" => Weekday::Tue,
580 "WE" => Weekday::Wed,
581 "TH" => Weekday::Thu,
582 "FR" => Weekday::Fri,
583 "SA" => Weekday::Sat,
584 "SU" => Weekday::Sun,
585 other => bail!("Invalid BYDAY value '{other}'"),
586 };
587 if !days.contains(&day) {
588 days.push(day);
589 }
590 }
591 Ok(days)
592 }
593
594 fn parse_once_schedule(parts: &BTreeMap<String, String>) -> Result<AutomationSchedule> {
595 for key in parts.keys() {
596 if key != "FREQ" && key != "AT" {
597 bail!("Unsupported RRULE field '{key}' for ONCE. Allowed: FREQ,AT");
598 }
599 }
600 let raw_at = parts
601 .get("AT")
602 .ok_or_else(|| anyhow::anyhow!("ONCE schedules require AT"))?;
603 let at = parse_once_at(raw_at)?;
604 Ok(AutomationSchedule::Once { at })
605 }
606
607 fn parse_cron_schedule(parts: &BTreeMap<String, String>) -> Result<AutomationSchedule> {
608 for key in parts.keys() {
609 if key != "FREQ" && key != "EXPR" {
610 bail!("Unsupported RRULE field '{key}' for CRON. Allowed: FREQ,EXPR");
611 }
612 }
613 let expr = parts
614 .get("EXPR")
615 .map(|value| value.trim().to_string())
616 .filter(|value| !value.is_empty())
617 .ok_or_else(|| anyhow::anyhow!("CRON schedules require EXPR"))?;
618 ParsedCronExpr::parse(&expr)?;
619 Ok(AutomationSchedule::Cron { expr })
620 }
621
622 fn parse_once_at(raw: &str) -> Result<DateTime<Utc>> {
623 let trimmed = raw.trim();
624 if let Ok(at) = DateTime::parse_from_rfc3339(trimmed) {
625 return Ok(at.with_timezone(&Utc));
626 }
627 for format in ["%Y-%m-%dT%H:%M:%S", "%Y-%m-%dT%H:%M"] {
628 if let Ok(naive) = NaiveDateTime::parse_from_str(trimmed, format) {
629 return resolve_local_datetime(&Local, naive)
630 .map(|value| value.with_timezone(&Utc))
631 .ok_or_else(|| anyhow::anyhow!("ONCE local time does not exist: {trimmed}"));
632 }
633 }
634 bail!("Failed to parse ONCE AT '{trimmed}'. Use local YYYY-MM-DDTHH:MM[:SS] or RFC3339")
635 }
636
637 #[derive(Debug, Clone)]
638 struct ParsedCronExpr {
639 minute: CronField,
640 hour: CronField,
641 day_of_month: CronField,
642 month: CronField,
643 day_of_week: CronField,
644 }
645
646 impl ParsedCronExpr {
647 fn parse(expr: &str) -> Result<Self> {
648 let fields: Vec<&str> = expr.split_whitespace().collect();
649 if fields.len() != 5 {
650 bail!(
651 "CRON EXPR must have exactly 5 fields: minute hour day-of-month month day-of-week"
652 );
653 }
654 let parsed = Self {
655 minute: CronField::parse(fields[0], 0, 59, CronNameMap::none(), "minute")?,
656 hour: CronField::parse(fields[1], 0, 23, CronNameMap::none(), "hour")?,
657 day_of_month: CronField::parse(fields[2], 1, 31, CronNameMap::none(), "day-of-month")?,
658 month: CronField::parse(fields[3], 1, 12, CronNameMap::month(), "month")?,
659 day_of_week: CronField::parse(fields[4], 0, 7, CronNameMap::weekday(), "day-of-week")?
660 .normalized_day_of_week(),
661 };
662 parsed.validate_date_space()?;
663 Ok(parsed)
664 }
665
666 fn matches(&self, candidate: NaiveDateTime) -> bool {
667 if !self.minute.contains(candidate.minute())
668 || !self.hour.contains(candidate.hour())
669 || !self.month.contains(candidate.month())
670 {
671 return false;
672 }
673
674 let day_of_month = self.day_of_month.contains(candidate.day());
675 let weekday = self
676 .day_of_week
677 .contains(weekday_to_cron(candidate.weekday()));
678 if self.day_of_month.is_wildcard && self.day_of_week.is_wildcard {
679 true
680 } else if self.day_of_month.is_wildcard {
681 weekday
682 } else if self.day_of_week.is_wildcard {
683 day_of_month
684 } else {
685 day_of_month || weekday
686 }
687 }
688
689 fn validate_date_space(&self) -> Result<()> {
690 if self.day_of_month.is_wildcard {
691 return Ok(());
692 }
693 let months = self.month.values();
694 let days = self.day_of_month.values();
695 let valid = months.iter().copied().any(|month| {
696 let common = days_in_month(2025, month);
697 let leap = days_in_month(2024, month);
698 days.iter().copied().any(|day| day <= common || day <= leap)
699 });
700 if valid {
701 Ok(())
702 } else {
703 bail!("CRON EXPR day-of-month/month combination can never occur")
704 }
705 }
706 }
707
708 #[derive(Debug, Clone)]
709 struct CronField {
710 values: Vec<u32>,
711 is_wildcard: bool,
712 }
713
714 impl CronField {
715 fn parse(raw: &str, min: u32, max: u32, names: CronNameMap, field_name: &str) -> Result<Self> {
716 let trimmed = raw.trim();
717 if trimmed.is_empty() {
718 bail!("CRON {field_name} field must not be empty");
719 }
720 let mut values = Vec::new();
721 let is_wildcard = trimmed == "*";
722 for part in trimmed.split(',') {
723 let part = part.trim();
724 if part.is_empty() {
725 bail!("CRON {field_name} field contains an empty list item");
726 }
727 let (base, step) = if let Some((base, step)) = part.split_once('/') {
728 let step = step
729 .trim()
730 .parse::<u32>()
731 .with_context(|| format!("Failed to parse CRON {field_name} step"))?;
732 if step == 0 {
733 bail!("CRON {field_name} step must be >= 1");
734 }
735 (base.trim(), step)
736 } else {
737 (part, 1)
738 };
739
740 let range = if base == "*" {
741 (min, max)
742 } else if let Some((start, end)) = base.split_once('-') {
743 let start = parse_cron_atom(start.trim(), min, max, names, field_name)?;
744 let end = parse_cron_atom(end.trim(), min, max, names, field_name)?;
745 if start > end {
746 bail!("CRON {field_name} range start must be <= end");
747 }
748 (start, end)
749 } else {
750 let start = parse_cron_atom(base, min, max, names, field_name)?;
751 if part.contains('/') {
752 (start, max)
753 } else {
754 (start, start)
755 }
756 };
757
758 let mut current = range.0;
759 while current <= range.1 {
760 if !values.contains(&current) {
761 values.push(current);
762 }
763 let Some(next) = current.checked_add(step) else {
764 break;
765 };
766 if next <= current {
767 break;
768 }
769 current = next;
770 }
771 }
772 values.sort_unstable();
773 Ok(Self {
774 values,
775 is_wildcard,
776 })
777 }
778
779 fn normalized_day_of_week(mut self) -> Self {
780 for value in &mut self.values {
781 if *value == 7 {
782 *value = 0;
783 }
784 }
785 self.values.sort_unstable();
786 self.values.dedup();
787 self
788 }
789
790 fn contains(&self, value: u32) -> bool {
791 self.values.binary_search(&value).is_ok()
792 }
793
794 fn values(&self) -> &[u32] {
795 &self.values
796 }
797 }
798
799 #[derive(Debug, Clone, Copy)]
800 struct CronNameMap(&'static [(&'static str, u32)]);
801
802 impl CronNameMap {
803 const fn none() -> Self {
804 Self(&[])
805 }
806
807 const fn month() -> Self {
808 Self(&[
809 ("JAN", 1),
810 ("FEB", 2),
811 ("MAR", 3),
812 ("APR", 4),
813 ("MAY", 5),
814 ("JUN", 6),
815 ("JUL", 7),
816 ("AUG", 8),
817 ("SEP", 9),
818 ("OCT", 10),
819 ("NOV", 11),
820 ("DEC", 12),
821 ])
822 }
823
824 const fn weekday() -> Self {
825 Self(&[
826 ("SUN", 0),
827 ("MON", 1),
828 ("TUE", 2),
829 ("WED", 3),
830 ("THU", 4),
831 ("FRI", 5),
832 ("SAT", 6),
833 ])
834 }
835
836 fn lookup(self, token: &str) -> Option<u32> {
837 let needle = token.trim().to_ascii_uppercase();
838 self.0
839 .iter()
840 .find_map(|(name, value)| (*name == needle).then_some(*value))
841 }
842 }
843
844 fn parse_cron_atom(
845 raw: &str,
846 min: u32,
847 max: u32,
848 names: CronNameMap,
849 field_name: &str,
850 ) -> Result<u32> {
851 let value = names
852 .lookup(raw)
853 .or_else(|| raw.parse::<u32>().ok())
854 .ok_or_else(|| anyhow::anyhow!("Invalid CRON {field_name} value '{raw}'"))?;
855 if !(min..=max).contains(&value) {
856 bail!("CRON {field_name} value {value} is out of range {min}-{max}");
857 }
858 Ok(value)
859 }
860
861 fn weekday_to_cron(day: Weekday) -> u32 {
862 match day {
863 Weekday::Sun => 0,
864 Weekday::Mon => 1,
865 Weekday::Tue => 2,
866 Weekday::Wed => 3,
867 Weekday::Thu => 4,
868 Weekday::Fri => 5,
869 Weekday::Sat => 6,
870 }
871 }
872
873 fn days_in_month(year: i32, month: u32) -> u32 {
874 match month {
875 1 | 3 | 5 | 7 | 8 | 10 | 12 => 31,
876 4 | 6 | 9 | 11 => 30,
877 2 => {
878 let leap = (year % 4 == 0 && year % 100 != 0) || year % 400 == 0;
879 if leap { 29 } else { 28 }
880 }
881 _ => 0,
882 }
883 }
884
885 #[derive(Debug, Clone)]
886 pub struct AutomationManager {
887 automations_dir: PathBuf,
888 runs_dir: PathBuf,
889 triggers_dir: PathBuf,
890 }
891
892 impl AutomationManager {
893 pub fn open(root: PathBuf) -> Result<Self> {
894 let automations_dir = root.join("automations");
895 let runs_dir = root.join("runs");
896 let triggers_dir = root.join("triggers");
897 fs::create_dir_all(&automations_dir)
898 .with_context(|| format!("Failed to create {}", automations_dir.display()))?;
899 fs::create_dir_all(&runs_dir)
900 .with_context(|| format!("Failed to create {}", runs_dir.display()))?;
901 fs::create_dir_all(&triggers_dir)
902 .with_context(|| format!("Failed to create {}", triggers_dir.display()))?;
903 Ok(Self {
904 automations_dir,
905 runs_dir,
906 triggers_dir,
907 })
908 }
909
910 pub fn default_location() -> Result<Self> {
911 Self::open(default_automations_dir())
912 }
913
914 fn automation_path(&self, id: &str) -> Result<PathBuf> {
915 ensure_safe_storage_id("automation id", id)?;
916 Ok(self.automations_dir.join(format!("{id}.json")))
917 }
918
919 fn runs_dir_for(&self, automation_id: &str) -> Result<PathBuf> {
920 ensure_safe_storage_id("automation id", automation_id)?;
921 Ok(self.runs_dir.join(automation_id))
922 }
923
924 fn trigger_path(&self, trigger_id: &str) -> Result<PathBuf> {
925 ensure_safe_storage_id("trigger id", trigger_id)?;
926 Ok(self.triggers_dir.join(format!("{trigger_id}.json")))
927 }
928
929 /// Current run file name: `{sortable-created-at}-{run_id}.json`. The
930 /// fixed-width timestamp prefix makes directory listings sort
931 /// chronologically without reading file contents (see [`Self::list_runs`]).
932 fn run_path(&self, run: &AutomationRunRecord) -> Result<PathBuf> {
933 ensure_safe_storage_id("run id", &run.id)?;
934 Ok(self.runs_dir_for(&run.automation_id)?.join(format!(
935 "{}-{}.json",
936 run_file_stamp(run.created_at),
937 run.id
938 )))
939 }
940
941 /// Pre-sortable-name run file: `{run_id}.json` (run ids are UUIDs, so
942 /// these carry no ordering hint and must be read to learn `created_at`).
943 fn legacy_run_path(&self, automation_id: &str, run_id: &str) -> Result<PathBuf> {
944 ensure_safe_storage_id("run id", run_id)?;
945 Ok(self
946 .runs_dir_for(automation_id)?
947 .join(format!("{run_id}.json")))
948 }
949
950 pub fn create_automation(&self, req: CreateAutomationRequest) -> Result<AutomationRecord> {
951 validate_name_and_prompt(&req.name, &req.prompt)?;
952 let schedule = AutomationSchedule::parse_rrule(&req.rrule)?;
953 let now = Utc::now();
954 let status = req.status.unwrap_or(AutomationStatus::Active);
955 let next_run_at = if matches!(status, AutomationStatus::Active) {
956 Some(schedule.next_after_with_anchor(now, now)?)
957 } else {
958 None
959 };
960
961 let record = AutomationRecord {
962 schema_version: CURRENT_AUTOMATION_SCHEMA_VERSION,
963 id: Uuid::new_v4().to_string(),
964 name: req.name.trim().to_string(),
965 prompt: req.prompt.trim().to_string(),
966 rrule: req.rrule.trim().to_ascii_uppercase(),
967 cwds: req.cwds,
968 mode: normalize_optional_string(req.mode),
969 allow_shell: req.allow_shell,
970 trust_mode: req.trust_mode,
971 auto_approve: req.auto_approve,
972 delivery_mode: req.delivery_mode,
973 status,
974 created_at: now,
975 updated_at: now,
976 next_run_at,
977 last_run_at: None,
978 };
979
980 self.save_automation(&record)?;
981 Ok(record)
982 }
983
984 pub fn get_automation(&self, id: &str) -> Result<AutomationRecord> {
985 let path = self.automation_path(id)?;
986 let raw = fs::read_to_string(&path)
987 .with_context(|| format!("Failed to read automation {}", path.display()))?;
988 let record: AutomationRecord = serde_json::from_str(&raw)
989 .with_context(|| format!("Failed to parse automation {}", path.display()))?;
990 if record.schema_version > CURRENT_AUTOMATION_SCHEMA_VERSION {
991 bail!(
992 "Automation schema v{} is newer than supported v{}",
993 record.schema_version,
994 CURRENT_AUTOMATION_SCHEMA_VERSION
995 );
996 }
997 Ok(record)
998 }
999
1000 pub fn save_automation(&self, record: &AutomationRecord) -> Result<()> {
1001 write_json_atomic(&self.automation_path(&record.id)?, record)
1002 }
1003
1004 pub fn list_automations(&self) -> Result<Vec<AutomationRecord>> {
1005 let mut out = Vec::new();
1006 for entry in fs::read_dir(&self.automations_dir)
1007 .with_context(|| format!("Failed to read {}", self.automations_dir.display()))?
1008 {
1009 let entry = entry?;
1010 let path = entry.path();
1011 if path.extension().is_none_or(|ext| ext != "json") {
1012 continue;
1013 }
1014 let raw = fs::read_to_string(&path)
1015 .with_context(|| format!("Failed to read {}", path.display()))?;
1016 let record: AutomationRecord = serde_json::from_str(&raw)
1017 .with_context(|| format!("Failed to parse {}", path.display()))?;
1018 if record.schema_version > CURRENT_AUTOMATION_SCHEMA_VERSION {
1019 bail!(
1020 "Automation schema v{} is newer than supported v{}",
1021 record.schema_version,
1022 CURRENT_AUTOMATION_SCHEMA_VERSION
1023 );
1024 }
1025 out.push(record);
1026 }
1027 out.sort_by_key(|r| std::cmp::Reverse(r.updated_at));
1028 Ok(out)
1029 }
1030
1031 pub fn update_automation(
1032 &self,
1033 id: &str,
1034 req: UpdateAutomationRequest,
1035 ) -> Result<AutomationRecord> {
1036 let mut existing = self.get_automation(id)?;
1037
1038 if let Some(name) = req.name {
1039 if name.trim().is_empty() {
1040 bail!("Automation name cannot be empty");
1041 }
1042 existing.name = name.trim().to_string();
1043 }
1044 if let Some(prompt) = req.prompt {
1045 if prompt.trim().is_empty() {
1046 bail!("Automation prompt cannot be empty");
1047 }
1048 existing.prompt = prompt.trim().to_string();
1049 }
1050 if let Some(rrule) = req.rrule {
1051 let normalized = rrule.trim().to_ascii_uppercase();
1052 AutomationSchedule::parse_rrule(&normalized)?;
1053 existing.rrule = normalized;
1054 if matches!(existing.status, AutomationStatus::Active) {
1055 let schedule = AutomationSchedule::parse_rrule(&existing.rrule)?;
1056 existing.next_run_at =
1057 Some(schedule.next_after_with_anchor(Utc::now(), existing.created_at)?);
1058 }
1059 }
1060 if let Some(cwds) = req.cwds {
1061 existing.cwds = cwds;
1062 }
1063 if let Some(mode) = req.mode {
1064 existing.mode = normalize_optional_string(Some(mode));
1065 }
1066 if let Some(allow_shell) = req.allow_shell {
1067 existing.allow_shell = Some(allow_shell);
1068 }
1069 if let Some(trust_mode) = req.trust_mode {
1070 existing.trust_mode = Some(trust_mode);
1071 }
1072 if let Some(auto_approve) = req.auto_approve {
1073 existing.auto_approve = Some(auto_approve);
1074 }
1075 if let Some(delivery_mode) = req.delivery_mode {
1076 existing.delivery_mode = Some(delivery_mode);
1077 }
1078 if let Some(status) = req.status {
1079 existing.status = status;
1080 if matches!(status, AutomationStatus::Paused) {
1081 existing.next_run_at = None;
1082 } else {
1083 let schedule = AutomationSchedule::parse_rrule(&existing.rrule)?;
1084 existing.next_run_at =
1085 Some(schedule.next_after_with_anchor(Utc::now(), existing.created_at)?);
1086 }
1087 }
1088
1089 existing.updated_at = Utc::now();
1090 self.save_automation(&existing)?;
1091 Ok(existing)
1092 }
1093
1094 pub fn pause_automation(&self, id: &str) -> Result<AutomationRecord> {
1095 self.update_automation(
1096 id,
1097 UpdateAutomationRequest {
1098 status: Some(AutomationStatus::Paused),
1099 ..UpdateAutomationRequest::default()
1100 },
1101 )
1102 }
1103
1104 pub fn resume_automation(&self, id: &str) -> Result<AutomationRecord> {
1105 self.update_automation(
1106 id,
1107 UpdateAutomationRequest {
1108 status: Some(AutomationStatus::Active),
1109 ..UpdateAutomationRequest::default()
1110 },
1111 )
1112 }
1113
1114 pub fn delete_automation(&self, id: &str) -> Result<AutomationRecord> {
1115 let existing = self.get_automation(id)?;
1116 let path = self.automation_path(id)?;
1117 fs::remove_file(&path)
1118 .with_context(|| format!("Failed to delete automation {}", path.display()))?;
1119
1120 let runs_dir = self.runs_dir_for(id)?;
1121 if runs_dir.exists() {
1122 fs::remove_dir_all(&runs_dir).with_context(|| {
1123 format!("Failed to delete automation runs {}", runs_dir.display())
1124 })?;
1125 }
1126
1127 Ok(existing)
1128 }
1129
1130 pub fn list_runs(
1131 &self,
1132 automation_id: &str,
1133 limit: Option<usize>,
1134 ) -> Result<Vec<AutomationRunRecord>> {
1135 let dir = self.runs_dir_for(automation_id)?;
1136 if !dir.exists() {
1137 return Ok(Vec::new());
1138 }
1139
1140 // Split the listing into sortable-name files (newest-first by file
1141 // name alone, so reads stop after the newest `limit`) and legacy
1142 // `{uuid}.json` files, which must all be read to learn `created_at`.
1143 let mut sortable = Vec::new();
1144 let mut legacy = Vec::new();
1145 for entry in
1146 fs::read_dir(&dir).with_context(|| format!("Failed to read {}", dir.display()))?
1147 {
1148 let entry = entry?;
1149 let path = entry.path();
1150 if path.extension().is_none_or(|ext| ext != "json") {
1151 continue;
1152 }
1153 if path
1154 .file_stem()
1155 .and_then(|stem| stem.to_str())
1156 .is_some_and(has_sortable_run_stem)
1157 {
1158 sortable.push(path);
1159 } else {
1160 legacy.push(path);
1161 }
1162 }
1163
1164 sortable.sort_by(|a, b| b.file_name().cmp(&a.file_name()));
1165 if let Some(limit) = limit {
1166 // Any sortable file dropped here is older than the `limit` newest
1167 // sortable files, so it can never make the merged top `limit`.
1168 sortable.truncate(limit);
1169 }
1170
1171 let mut out = Vec::new();
1172 for path in sortable.into_iter().chain(legacy) {
1173 out.push(read_run_file(&path)?);
1174 }
1175
1176 out.sort_by_key(|r| std::cmp::Reverse(r.created_at));
1177 // A crash between the sortable-name write and the legacy-file removal
1178 // in `save_run` can leave one run under both names; keep the sortable
1179 // copy (chained first above, so it survives the stable sort).
1180 out.dedup_by(|a, b| a.id == b.id);
1181 if let Some(limit) = limit {
1182 out.truncate(limit);
1183 }
1184 Ok(out)
1185 }
1186
1187 fn save_run(&self, run: &AutomationRunRecord) -> Result<()> {
1188 let dir = self.runs_dir_for(&run.automation_id)?;
1189 fs::create_dir_all(&dir).with_context(|| format!("Failed to create {}", dir.display()))?;
1190 let path = self.run_path(run)?;
1191 write_json_atomic(&path, run)?;
1192 // Rewrites of a legacy-named run migrate it to the sortable name; drop
1193 // the old file so the run never exists twice.
1194 let legacy = self.legacy_run_path(&run.automation_id, &run.id)?;
1195 if legacy != path && legacy.exists() {
1196 fs::remove_file(&legacy)
1197 .with_context(|| format!("Failed to remove legacy run {}", legacy.display()))?;
1198 }
1199 Ok(())
1200 }
1201
1202 fn delete_run(&self, run: &AutomationRunRecord) -> Result<()> {
1203 let sortable = self.run_path(run)?;
1204 if sortable.exists() {
1205 fs::remove_file(&sortable)
1206 .with_context(|| format!("Failed to delete run {}", sortable.display()))?;
1207 }
1208 let legacy = self.legacy_run_path(&run.automation_id, &run.id)?;
1209 if legacy.exists() {
1210 fs::remove_file(&legacy)
1211 .with_context(|| format!("Failed to delete run {}", legacy.display()))?;
1212 }
1213 Ok(())
1214 }
1215
1216 /// Sweep all automations under one lock hold: initialize/advance schedule
1217 /// bookkeeping and return the (automation, run) pairs that must be
1218 /// enqueued. `next_run_at` for returned pairs is only advanced after the
1219 /// run is persisted (see [`scheduler_tick_shared`]) so a crash mid-enqueue
1220 /// retries the slot; the run-per-slot check keeps that idempotent.
1221 fn collect_due_runs(
1222 &self,
1223 now: DateTime<Utc>,
1224 ) -> Result<Vec<(AutomationRecord, AutomationRunRecord)>> {
1225 let mut due = Vec::new();
1226 for mut automation in self.list_automations()? {
1227 if !matches!(automation.status, AutomationStatus::Active) {
1228 continue;
1229 }
1230
1231 let schedule = AutomationSchedule::parse_rrule(&automation.rrule)?;
1232 let Some(due_at) = automation.next_run_at else {
1233 automation.next_run_at =
1234 match schedule.next_after_with_anchor(now, automation.created_at) {
1235 Ok(next) => Some(next),
1236 Err(err)
1237 if matches!(schedule, AutomationSchedule::Once { .. })
1238 && err.to_string().contains("Once schedule has no future run") =>
1239 {
1240 automation.status = AutomationStatus::Paused;
1241 None
1242 }
1243 Err(err) => return Err(err),
1244 };
1245 automation.updated_at = now;
1246 self.save_automation(&automation)?;
1247 continue;
1248 };
1249 if due_at > now {
1250 continue;
1251 }
1252
1253 // Idempotency: if a run already exists for this schedule slot, skip enqueue and
1254 // advance next_run_at.
1255 let existing_for_slot = self
1256 .list_runs(&automation.id, Some(25))?
1257 .into_iter()
1258 .any(|run| run.scheduled_for == due_at);
1259
1260 if existing_for_slot {
1261 self.advance_automation_after_slot(&mut automation, &schedule, due_at, now)?;
1262 continue;
1263 }
1264
1265 let run = new_run_record(&automation.id, due_at, now);
1266 due.push((automation, run));
1267 }
1268 Ok(due)
1269 }
1270
1271 /// Persist a completed enqueue attempt and advance the schedule slot.
1272 /// The run record is saved unconditionally: `enqueue_run_task` already
1273 /// created a real task before this is called, so even when the automation
1274 /// was deleted while the enqueue await ran outside the lock, the run must
1275 /// be persisted (not orphaned) — only the schedule advance is skipped.
1276 fn finish_scheduled_run(&self, run: &AutomationRunRecord, now: DateTime<Utc>) -> Result<()> {
1277 self.save_run(run)?;
1278 let Ok(mut automation) = self.get_automation(&run.automation_id) else {
1279 return Ok(());
1280 };
1281 let schedule = AutomationSchedule::parse_rrule(&automation.rrule)?;
1282 self.advance_automation_after_slot(&mut automation, &schedule, run.scheduled_for, now)
1283 }
1284
1285 fn advance_automation_after_slot(
1286 &self,
1287 automation: &mut AutomationRecord,
1288 schedule: &AutomationSchedule,
1289 slot: DateTime<Utc>,
1290 now: DateTime<Utc>,
1291 ) -> Result<()> {
1292 automation.updated_at = now;
1293 automation.next_run_at = schedule.next_after_slot(slot, automation.created_at)?;
1294 if automation.next_run_at.is_none() {
1295 automation.status = AutomationStatus::Paused;
1296 }
1297 self.save_automation(automation)
1298 }
1299
1300 /// Snapshot runs still waiting on task-manager state, for reconciliation
1301 /// outside the lock.
1302 fn collect_pending_runs(&self) -> Result<Vec<AutomationRunRecord>> {
1303 let mut pending = Vec::new();
1304 for automation in self.list_automations()? {
1305 for run in self.list_runs(&automation.id, Some(100))? {
1306 if matches!(
1307 run.status,
1308 AutomationRunStatus::Queued | AutomationRunStatus::Running
1309 ) && run.task_id.is_some()
1310 {
1311 pending.push(run);
1312 }
1313 }
1314 }
1315 Ok(pending)
1316 }
1317
1318 // ── Delayed-trigger storage methods ──────────────────────────────────
1319
1320 /// Persist a new delayed trigger and return the record.
1321 pub fn create_trigger(&self, req: CreateDelayedTriggerRequest) -> Result<DelayedTriggerRecord> {
1322 let now = Utc::now();
1323 if req.fire_at <= now {
1324 bail!(
1325 "fire_at must be in the future (got {}, now is {})",
1326 req.fire_at.to_rfc3339(),
1327 now.to_rfc3339()
1328 );
1329 }
1330 if req.message.trim().is_empty() {
1331 bail!("Trigger message must not be empty");
1332 }
1333 let record = DelayedTriggerRecord {
1334 schema_version: CURRENT_TRIGGER_SCHEMA_VERSION,
1335 trigger_id: format!("trig_{}", Uuid::new_v4().simple()),
1336 fire_at: req.fire_at,
1337 message: req.message.trim().to_string(),
1338 workspace: req.workspace,
1339 status: DelayedTriggerStatus::Pending,
1340 created_at: now,
1341 fired_at: None,
1342 task_id: None,
1343 thread_id: None,
1344 error: None,
1345 parent_trigger_id: req.parent_trigger_id,
1346 };
1347 self.save_trigger(&record)?;
1348 Ok(record)
1349 }
1350
1351 /// Load a trigger by id.
1352 pub fn get_trigger(&self, trigger_id: &str) -> Result<DelayedTriggerRecord> {
1353 let path = self.trigger_path(trigger_id)?;
1354 let raw = fs::read_to_string(&path)
1355 .with_context(|| format!("Trigger '{trigger_id}' not found"))?;
1356 let record: DelayedTriggerRecord = serde_json::from_str(&raw)
1357 .with_context(|| format!("Failed to parse trigger '{trigger_id}'"))?;
1358 if record.schema_version > CURRENT_TRIGGER_SCHEMA_VERSION {
1359 bail!(
1360 "Trigger schema v{} is newer than supported v{}",
1361 record.schema_version,
1362 CURRENT_TRIGGER_SCHEMA_VERSION
1363 );
1364 }
1365 Ok(record)
1366 }
1367
1368 /// Atomically persist a trigger record.
1369 pub fn save_trigger(&self, record: &DelayedTriggerRecord) -> Result<()> {
1370 let path = self.trigger_path(&record.trigger_id)?;
1371 write_json_atomic(&path, record)
1372 }
1373
1374 /// List triggers, newest first. Pass `status_filter` to restrict results.
1375 pub fn list_triggers(
1376 &self,
1377 status_filter: Option<DelayedTriggerStatus>,
1378 limit: Option<usize>,
1379 ) -> Result<Vec<DelayedTriggerRecord>> {
1380 let mut out = Vec::new();
1381 if !self.triggers_dir.exists() {
1382 return Ok(out);
1383 }
1384 for entry in fs::read_dir(&self.triggers_dir)
1385 .with_context(|| format!("Failed to read {}", self.triggers_dir.display()))?
1386 {
1387 let entry = entry?;
1388 let path = entry.path();
1389 if path.extension().is_none_or(|ext| ext != "json") {
1390 continue;
1391 }
1392 match fs::read_to_string(&path)
1393 .ok()
1394 .and_then(|raw| serde_json::from_str::<DelayedTriggerRecord>(&raw).ok())
1395 {
1396 Some(record) => {
1397 if let Some(filter) = status_filter
1398 && record.status != filter
1399 {
1400 continue;
1401 }
1402 out.push(record);
1403 }
1404 None => {
1405 tracing::warn!("Skipping unreadable trigger file {}", path.display());
1406 }
1407 }
1408 }
1409 out.sort_by_key(|r| std::cmp::Reverse(r.created_at));
1410 if let Some(limit) = limit {
1411 out.truncate(limit);
1412 }
1413 Ok(out)
1414 }
1415
1416 /// Cancel a pending trigger. Returns the updated record.
1417 pub fn cancel_trigger(&self, trigger_id: &str) -> Result<DelayedTriggerRecord> {
1418 let mut record = self.get_trigger(trigger_id)?;
1419 if !matches!(record.status, DelayedTriggerStatus::Pending) {
1420 bail!(
1421 "Trigger '{trigger_id}' cannot be canceled (status: {:?})",
1422 record.status
1423 );
1424 }
1425 record.status = DelayedTriggerStatus::Canceled;
1426 self.save_trigger(&record)?;
1427 Ok(record)
1428 }
1429
1430 /// Return all pending triggers whose `fire_at` is at or before `now`.
1431 pub fn collect_due_triggers(&self, now: DateTime<Utc>) -> Result<Vec<DelayedTriggerRecord>> {
1432 let pending = self.list_triggers(Some(DelayedTriggerStatus::Pending), None)?;
1433 Ok(pending.into_iter().filter(|t| t.fire_at <= now).collect())
1434 }
1435 }
1436
1437 fn new_run_record(
1438 automation_id: &str,
1439 scheduled_for: DateTime<Utc>,
1440 created_at: DateTime<Utc>,
1441 ) -> AutomationRunRecord {
1442 AutomationRunRecord {
1443 schema_version: CURRENT_RUN_SCHEMA_VERSION,
1444 id: Uuid::new_v4().to_string(),
1445 automation_id: automation_id.to_string(),
1446 scheduled_for,
1447 status: AutomationRunStatus::Queued,
1448 created_at,
1449 started_at: None,
1450 ended_at: None,
1451 task_id: None,
1452 thread_id: None,
1453 turn_id: None,
1454 error: None,
1455 }
1456 }
1457
1458 /// Enqueue the automation's durable task, folding the outcome into `run`.
1459 /// Free function (no `AutomationManager` receiver) so callers can await
1460 /// task-manager latency without holding the shared manager mutex.
1461 async fn enqueue_run_task(
1462 automation: &AutomationRecord,
1463 run: &mut AutomationRunRecord,
1464 task_manager: &SharedTaskManager,
1465 ) {
1466 let workspace = automation.cwds.first().cloned();
1467
1468 let new_task = NewTaskRequest {
1469 prompt: automation.prompt.clone(),
1470 model: None,
1471 workspace,
1472 mode: Some(automation.task_mode()),
1473 allow_shell: Some(automation.task_allow_shell()),
1474 trust_mode: Some(automation.task_trust_mode()),
1475 auto_approve: Some(automation.task_auto_approve()),
1476 owner_session_id: None,
1477 };
1478
1479 match task_manager.add_task(new_task).await {
1480 Ok(task) => {
1481 run.status = AutomationRunStatus::Running;
1482 run.started_at = Some(Utc::now());
1483 run.task_id = Some(task.id.clone());
1484 run.thread_id = task.thread_id.clone();
1485 run.turn_id = task.turn_id.clone();
1486 run.error = None;
1487 }
1488 Err(err) => {
1489 run.status = AutomationRunStatus::Failed;
1490 run.ended_at = Some(Utc::now());
1491 run.error = Some(format!("Failed to enqueue task: {err}"));
1492 }
1493 }
1494 }
1495
1496 /// Run an automation immediately. The shared manager mutex is held only for
1497 /// the read and persist phases, never across the task-manager await, so
1498 /// listing/pausing/resuming stay responsive behind a slow enqueue.
1499 pub async fn run_now_shared(
1500 automations: &SharedAutomationManager,
1501 automation_id: &str,
1502 task_manager: &SharedTaskManager,
1503 ) -> Result<AutomationRunRecord> {
1504 let task_manager = Arc::clone(task_manager);
1505 run_now_with(
1506 automations,
1507 automation_id,
1508 move |automation, mut run| async move {
1509 enqueue_run_task(&automation, &mut run, &task_manager).await;
1510 run
1511 },
1512 )
1513 .await
1514 }
1515
1516 /// Lock-phased core of [`run_now_shared`], generic over the enqueue await so
1517 /// tests can stub task-manager latency.
1518 async fn run_now_with<F, Fut>(
1519 automations: &SharedAutomationManager,
1520 automation_id: &str,
1521 enqueue: F,
1522 ) -> Result<AutomationRunRecord>
1523 where
1524 F: FnOnce(AutomationRecord, AutomationRunRecord) -> Fut,
1525 Fut: Future<Output = AutomationRunRecord>,
1526 {
1527 // Phase 1: read state under the lock.
1528 let automation = {
1529 let manager = automations.lock().await;
1530 manager.get_automation(automation_id)?
1531 };
1532 let now = Utc::now();
1533 let run = new_run_record(&automation.id, now, now);
1534
1535 // Phase 2: await the task manager without the lock.
1536 let run = enqueue(automation, run).await;
1537
1538 // Phase 3: reacquire to persist the final run state.
1539 let manager = automations.lock().await;
1540 manager.save_run(&run)?;
1541 // Re-read: the record may have changed (or been deleted) while unlocked.
1542 if let Ok(mut automation) = manager.get_automation(automation_id) {
1543 automation.updated_at = Utc::now();
1544 if matches!(
1545 run.status,
1546 AutomationRunStatus::Completed
1547 | AutomationRunStatus::Failed
1548 | AutomationRunStatus::Canceled
1549 ) {
1550 automation.last_run_at = run.ended_at.or(Some(Utc::now()));
1551 }
1552 manager.save_automation(&automation)?;
1553 }
1554
1555 Ok(run)
1556 }
1557
1558 async fn scheduler_tick_shared(
1559 automations: &SharedAutomationManager,
1560 task_manager: &SharedTaskManager,
1561 ) -> Result<()> {
1562 let now = Utc::now();
1563 // Phase 1: compute due runs and schedule bookkeeping under the lock.
1564 let due_runs = {
1565 let manager = automations.lock().await;
1566 manager.collect_due_runs(now)?
1567 };
1568
1569 for (automation, mut run) in due_runs {
1570 // Phase 2: enqueue without the lock.
1571 enqueue_run_task(&automation, &mut run, task_manager).await;
1572
1573 // Phase 3: reacquire to persist the run and advance the slot.
1574 let manager = automations.lock().await;
1575 manager.finish_scheduled_run(&run, now)?;
1576 }
1577
1578 Ok(())
1579 }
1580
1581 /// Enqueue a fired delayed trigger as a durable task and persist the updated
1582 /// trigger record. The manager mutex is never held across the task-manager
1583 /// await.
1584 async fn fire_due_triggers_shared(
1585 automations: &SharedAutomationManager,
1586 task_manager: &SharedTaskManager,
1587 ) -> Result<()> {
1588 let now = Utc::now();
1589
1590 // Phase 1: collect due triggers under the lock.
1591 let due_triggers = {
1592 let manager = automations.lock().await;
1593 manager.collect_due_triggers(now)?
1594 };
1595
1596 for mut trigger in due_triggers {
1597 // Phase 2: enqueue without holding the lock.
1598 let workspace = trigger.workspace.clone();
1599 let new_task = NewTaskRequest {
1600 prompt: trigger.message.clone(),
1601 model: None,
1602 workspace,
1603 mode: Some("agent".to_string()),
1604 allow_shell: Some(false),
1605 trust_mode: Some(false),
1606 auto_approve: Some(false),
1607 owner_session_id: None,
1608 };
1609
1610 match task_manager.add_task(new_task).await {
1611 Ok(task) => {
1612 trigger.status = DelayedTriggerStatus::Fired;
1613 trigger.fired_at = Some(Utc::now());
1614 trigger.task_id = Some(task.id.clone());
1615 trigger.thread_id = task.thread_id.clone();
1616 trigger.error = None;
1617 }
1618 Err(err) => {
1619 trigger.status = DelayedTriggerStatus::Failed;
1620 trigger.fired_at = Some(Utc::now());
1621 trigger.error = Some(format!("Failed to enqueue task: {err}"));
1622 }
1623 }
1624
1625 // Phase 3: persist the outcome under the lock.
1626 let manager = automations.lock().await;
1627 manager.save_trigger(&trigger)?;
1628 }
1629
1630 Ok(())
1631 }
1632
1633 /// Fold a durable task's state back into its automation run. Returns whether
1634 /// the run changed and needs persisting.
1635 fn apply_task_status(
1636 run: &mut AutomationRunRecord,
1637 task: &crate::task_manager::TaskRecord,
1638 ) -> bool {
1639 run.thread_id = task.thread_id.clone();
1640 run.turn_id = task.turn_id.clone();
1641
1642 let mut changed = false;
1643 match task.status {
1644 TaskStatus::Queued => {
1645 if !matches!(run.status, AutomationRunStatus::Queued) {
1646 run.status = AutomationRunStatus::Queued;
1647 changed = true;
1648 }
1649 }
1650 TaskStatus::Running => {
1651 if !matches!(run.status, AutomationRunStatus::Running) {
1652 run.status = AutomationRunStatus::Running;
1653 changed = true;
1654 }
1655 if run.started_at.is_none() {
1656 run.started_at = Some(task.started_at.unwrap_or_else(Utc::now));
1657 changed = true;
1658 }
1659 }
1660 TaskStatus::Completed => {
1661 run.status = AutomationRunStatus::Completed;
1662 run.started_at = run.started_at.or(task.started_at);
1663 run.ended_at = task.ended_at.or(Some(Utc::now()));
1664 run.error = None;
1665 changed = true;
1666 }
1667 TaskStatus::Failed => {
1668 run.status = AutomationRunStatus::Failed;
1669 run.started_at = run.started_at.or(task.started_at);
1670 run.ended_at = task.ended_at.or(Some(Utc::now()));
1671 run.error = task.error.clone();
1672 changed = true;
1673 }
1674 TaskStatus::Canceled => {
1675 run.status = AutomationRunStatus::Canceled;
1676 run.started_at = run.started_at.or(task.started_at);
1677 run.ended_at = task.ended_at.or(Some(Utc::now()));
1678 changed = true;
1679 }
1680 }
1681 changed
1682 }
1683
1684 async fn reconcile_run_statuses_shared(
1685 automations: &SharedAutomationManager,
1686 task_manager: &SharedTaskManager,
1687 ) -> Result<()> {
1688 // Phase 1: snapshot pending runs under the lock.
1689 let pending = {
1690 let manager = automations.lock().await;
1691 manager.collect_pending_runs()?
1692 };
1693
1694 for mut run in pending {
1695 let Some(task_id) = run.task_id.clone() else {
1696 continue;
1697 };
1698 // Phase 2: task lookups happen without the lock.
1699 let task = match task_manager.get_task(&task_id).await {
1700 Ok(task) => task,
1701 Err(_) => continue,
1702 };
1703
1704 let watcher_noop = {
1705 let manager = automations.lock().await;
1706 manager
1707 .get_automation(&run.automation_id)
1708 .ok()
1709 .is_some_and(|automation| {
1710 automation.delivery_mode() == AutomationDeliveryMode::Watcher
1711 && task.status == TaskStatus::Completed
1712 && task.result_summary.as_deref().is_some_and(|summary| {
1713 summary.trim() == AUTOMATION_WATCHER_NO_REPORT_SENTINEL
1714 })
1715 })
1716 };
1717 if watcher_noop {
1718 let manager = automations.lock().await;
1719 manager.delete_run(&run)?;
1720 continue;
1721 }
1722
1723 if !apply_task_status(&mut run, &task) {
1724 continue;
1725 }
1726
1727 // Phase 3: reacquire to persist the reconciled state.
1728 let manager = automations.lock().await;
1729 manager.save_run(&run)?;
1730 if matches!(
1731 run.status,
1732 AutomationRunStatus::Completed
1733 | AutomationRunStatus::Failed
1734 | AutomationRunStatus::Canceled
1735 ) && let Ok(mut updated_automation) = manager.get_automation(&run.automation_id)
1736 {
1737 updated_automation.last_run_at = run.ended_at.or(Some(Utc::now()));
1738 updated_automation.updated_at = Utc::now();
1739 manager.save_automation(&updated_automation)?;
1740 }
1741 }
1742
1743 Ok(())
1744 }
1745
1746 /// Fixed-width, lexically-sortable UTC stamp for run file names, e.g.
1747 /// `20260705T142530123Z` (millisecond precision; the run id suffix breaks
1748 /// same-millisecond ties deterministically).
1749 const RUN_STAMP_FORMAT: &str = "%Y%m%dT%H%M%S%3fZ";
1750 const RUN_STAMP_LEN: usize = "20260705T142530123Z".len();
1751
1752 fn run_file_stamp(created_at: DateTime<Utc>) -> String {
1753 created_at.format(RUN_STAMP_FORMAT).to_string()
1754 }
1755
1756 /// Shape check for `{stamp}-{run_id}` file stems. Ordering trusts the file
1757 /// name only for pruning; the parsed record's `created_at` stays
1758 /// authoritative for the final sort.
1759 fn has_sortable_run_stem(stem: &str) -> bool {
1760 let Some((stamp, rest)) = stem.split_at_checked(RUN_STAMP_LEN) else {
1761 return false;
1762 };
1763 if !rest.starts_with('-') || rest.len() < 2 {
1764 return false;
1765 }
1766 stamp.char_indices().all(|(idx, ch)| match idx {
1767 8 => ch == 'T',
1768 18 => ch == 'Z',
1769 _ => ch.is_ascii_digit(),
1770 })
1771 }
1772
1773 fn read_run_file(path: &Path) -> Result<AutomationRunRecord> {
1774 let raw =
1775 fs::read_to_string(path).with_context(|| format!("Failed to read {}", path.display()))?;
1776 let run: AutomationRunRecord = serde_json::from_str(&raw)
1777 .with_context(|| format!("Failed to parse {}", path.display()))?;
1778 if run.schema_version > CURRENT_RUN_SCHEMA_VERSION {
1779 bail!(
1780 "Automation run schema v{} is newer than supported v{}",
1781 run.schema_version,
1782 CURRENT_RUN_SCHEMA_VERSION
1783 );
1784 }
1785 Ok(run)
1786 }
1787
1788 fn ensure_safe_storage_id(kind: &str, value: &str) -> Result<()> {
1789 let mut components = Path::new(value).components();
1790 let Some(component) = components.next() else {
1791 bail!("{kind} must not be empty");
1792 };
1793 if components.next().is_some() || !matches!(component, std::path::Component::Normal(_)) {
1794 bail!("{kind} must be a single path component");
1795 }
1796 Ok(())
1797 }
1798
1799 fn validate_name_and_prompt(name: &str, prompt: &str) -> Result<()> {
1800 if name.trim().is_empty() {
1801 bail!("Automation name is required");
1802 }
1803 if prompt.trim().is_empty() {
1804 bail!("Automation prompt is required");
1805 }
1806 Ok(())
1807 }
1808
1809 fn normalize_optional_string(value: Option<String>) -> Option<String> {
1810 value
1811 .map(|value| value.trim().to_string())
1812 .filter(|value| !value.is_empty())
1813 }
1814
1815 fn write_json_atomic<T: Serialize>(path: &Path, value: &T) -> Result<()> {
1816 if let Some(parent) = path.parent() {
1817 fs::create_dir_all(parent)
1818 .with_context(|| format!("Failed to create {}", parent.display()))?;
1819 }
1820 let content = serde_json::to_string_pretty(value)?;
1821 let tmp = path.with_extension("json.tmp");
1822 fs::write(&tmp, content).with_context(|| format!("Failed to write {}", tmp.display()))?;
1823 fs::rename(&tmp, path).with_context(|| {
1824 format!(
1825 "Failed to move temporary file {} to {}",
1826 tmp.display(),
1827 path.display()
1828 )
1829 })?;
1830 Ok(())
1831 }
1832
1833 pub fn default_automations_dir() -> PathBuf {
1834 // Most-specific override: an explicit automations dir.
1835 for var in ["CODEWHALE_AUTOMATIONS_DIR", "DEEPSEEK_AUTOMATIONS_DIR"] {
1836 if let Ok(path) = std::env::var(var) {
1837 let trimmed = path.trim();
1838 if !trimmed.is_empty() {
1839 return PathBuf::from(trimmed);
1840 }
1841 }
1842 }
1843 // $CODEWHALE_HOME is a hard override of the base data directory
1844 // (docs/CONFIGURATION.md): when SET, automations live under it and we do
1845 // NOT fall back to the legacy ~/.deepseek path — silent fallback would
1846 // defeat the isolation the override promises. Check the env var directly
1847 // (not codewhale_home()'s Ok/Err, which succeeds for the default home too).
1848 if let Some(home) = codewhale_paths::codewhale_home_override().ok().flatten() {
1849 return home.join("automations");
1850 }
1851 codewhale_paths::user_home()
1852 .map(|home| {
1853 let primary = home.join(".codewhale").join("automations");
1854 let legacy = home.join(".deepseek").join("automations");
1855 if primary.exists() || !legacy.exists() {
1856 return primary;
1857 }
1858 legacy
1859 })
1860 .unwrap_or_else(|| PathBuf::from(".codewhale").join("automations"))
1861 }
1862
1863 pub type SharedAutomationManager = Arc<Mutex<AutomationManager>>;
1864
1865 #[derive(Debug, Clone)]
1866 pub struct AutomationSchedulerConfig {
1867 pub tick_interval_secs: u64,
1868 }
1869
1870 impl Default for AutomationSchedulerConfig {
1871 fn default() -> Self {
1872 Self {
1873 tick_interval_secs: 15,
1874 }
1875 }
1876 }
1877
1878 pub fn spawn_scheduler(
1879 automations: SharedAutomationManager,
1880 task_manager: SharedTaskManager,
1881 cancel: CancellationToken,
1882 config: AutomationSchedulerConfig,
1883 ) -> tokio::task::JoinHandle<()> {
1884 spawn_supervised(
1885 "automation-scheduler",
1886 std::panic::Location::caller(),
1887 async move {
1888 let interval = config.tick_interval_secs.max(5);
1889 loop {
1890 if cancel.is_cancelled() {
1891 break;
1892 }
1893
1894 // Lock scope lives inside the shared helpers: the manager
1895 // mutex is dropped across every task-manager await so API and
1896 // tool callers are never queued behind enqueue/status latency.
1897 if let Err(err) = scheduler_tick_shared(&automations, &task_manager).await {
1898 tracing::warn!("automation scheduler tick failed: {err}");
1899 }
1900 if let Err(err) = reconcile_run_statuses_shared(&automations, &task_manager).await {
1901 tracing::warn!("automation reconcile failed: {err}");
1902 }
1903 if let Err(err) = fire_due_triggers_shared(&automations, &task_manager).await {
1904 tracing::warn!("delayed trigger tick failed: {err}");
1905 }
1906
1907 tokio::select! {
1908 _ = cancel.cancelled() => break,
1909 _ = sleep(std::time::Duration::from_secs(interval)) => {}
1910 }
1911 }
1912 },
1913 )
1914 }
1915
1916 #[cfg(test)]
1917 mod tests {
1918 use super::*;
1919 use async_trait::async_trait;
1920 use chrono::{FixedOffset, LocalResult, NaiveDate};
1921 use tokio::sync::mpsc;
1922
1923 use crate::task_manager::{
1924 ExecutionTask, TaskExecutionEvent, TaskExecutionResult, TaskExecutor, TaskManager,
1925 TaskManagerConfig,
1926 };
1927
1928 struct AutomationNoopExecutor;
1929 struct AutomationWatcherNoopExecutor;
1930
1931 /// A deterministic America/New_York-compatible zone for the 2026 DST
1932 /// boundary tests. Keeping the transition table local avoids mutating the
1933 /// process-wide `TZ` setting while the test binary runs in parallel.
1934 #[derive(Debug, Clone, Copy)]
1935 struct Eastern2026;
1936
1937 impl Eastern2026 {
1938 fn standard_offset() -> FixedOffset {
1939 FixedOffset::west_opt(5 * 60 * 60).expect("valid standard offset")
1940 }
1941
1942 fn daylight_offset() -> FixedOffset {
1943 FixedOffset::west_opt(4 * 60 * 60).expect("valid daylight offset")
1944 }
1945
1946 fn time(month: u32, day: u32, hour: u32) -> NaiveDateTime {
1947 NaiveDate::from_ymd_opt(2026, month, day)
1948 .expect("valid transition date")
1949 .and_hms_opt(hour, 0, 0)
1950 .expect("valid transition time")
1951 }
1952 }
1953
1954 impl TimeZone for Eastern2026 {
1955 type Offset = FixedOffset;
1956
1957 fn from_offset(_offset: &Self::Offset) -> Self {
1958 Self
1959 }
1960
1961 fn offset_from_local_date(&self, local: &NaiveDate) -> LocalResult<Self::Offset> {
1962 self.offset_from_local_datetime(
1963 &local
1964 .and_hms_opt(12, 0, 0)
1965 .expect("valid local date midpoint"),
1966 )
1967 }
1968
1969 fn offset_from_local_datetime(&self, local: &NaiveDateTime) -> LocalResult<Self::Offset> {
1970 let gap_start = Self::time(3, 8, 2);
1971 let gap_end = Self::time(3, 8, 3);
1972 let fold_start = Self::time(11, 1, 1);
1973 let fold_end = Self::time(11, 1, 2);
1974
1975 if *local >= gap_start && *local < gap_end {
1976 LocalResult::None
1977 } else if *local >= fold_start && *local < fold_end {
1978 LocalResult::Ambiguous(Self::daylight_offset(), Self::standard_offset())
1979 } else if *local >= gap_end && *local < fold_start {
1980 LocalResult::Single(Self::daylight_offset())
1981 } else {
1982 LocalResult::Single(Self::standard_offset())
1983 }
1984 }
1985
1986 fn offset_from_utc_date(&self, utc: &NaiveDate) -> Self::Offset {
1987 self.offset_from_utc_datetime(
1988 &utc.and_hms_opt(12, 0, 0).expect("valid UTC date midpoint"),
1989 )
1990 }
1991
1992 fn offset_from_utc_datetime(&self, utc: &NaiveDateTime) -> Self::Offset {
1993 let daylight_start = Self::time(3, 8, 7);
1994 let daylight_end = Self::time(11, 1, 6);
1995 if *utc >= daylight_start && *utc < daylight_end {
1996 Self::daylight_offset()
1997 } else {
1998 Self::standard_offset()
1999 }
2000 }
2001 }
2002
2003 #[async_trait]
2004 impl TaskExecutor for AutomationNoopExecutor {
2005 async fn execute(
2006 &self,
2007 _task: ExecutionTask,
2008 _events: mpsc::UnboundedSender<TaskExecutionEvent>,
2009 _cancel: CancellationToken,
2010 ) -> TaskExecutionResult {
2011 TaskExecutionResult {
2012 status: TaskStatus::Completed,
2013 result_text: Some("done".to_string()),
2014 error: None,
2015 }
2016 }
2017 }
2018
2019 #[async_trait]
2020 impl TaskExecutor for AutomationWatcherNoopExecutor {
2021 async fn execute(
2022 &self,
2023 _task: ExecutionTask,
2024 _events: mpsc::UnboundedSender<TaskExecutionEvent>,
2025 _cancel: CancellationToken,
2026 ) -> TaskExecutionResult {
2027 TaskExecutionResult {
2028 status: TaskStatus::Completed,
2029 result_text: Some(AUTOMATION_WATCHER_NO_REPORT_SENTINEL.to_string()),
2030 error: None,
2031 }
2032 }
2033 }
2034
2035 fn automation_task_config(root: PathBuf) -> TaskManagerConfig {
2036 TaskManagerConfig {
2037 data_dir: root,
2038 worker_count: 1,
2039 default_workspace: PathBuf::from("."),
2040 default_model: "deepseek-v4-flash".to_string(),
2041 default_mode: "plan".to_string(),
2042 allow_shell: true,
2043 trust_mode: true,
2044 }
2045 }
2046
2047 fn automation_record_with_settings(
2048 mode: Option<&str>,
2049 allow_shell: Option<bool>,
2050 trust_mode: Option<bool>,
2051 auto_approve: Option<bool>,
2052 ) -> AutomationRecord {
2053 let now = Utc::now();
2054 AutomationRecord {
2055 schema_version: CURRENT_AUTOMATION_SCHEMA_VERSION,
2056 id: Uuid::new_v4().to_string(),
2057 name: "Test automation".to_string(),
2058 prompt: "Run the automation".to_string(),
2059 rrule: "FREQ=HOURLY;INTERVAL=1".to_string(),
2060 cwds: Vec::new(),
2061 mode: mode.map(ToString::to_string),
2062 allow_shell,
2063 trust_mode,
2064 auto_approve,
2065 delivery_mode: None,
2066 status: AutomationStatus::Active,
2067 created_at: now,
2068 updated_at: now,
2069 next_run_at: None,
2070 last_run_at: None,
2071 }
2072 }
2073
2074 fn queued_run_for(automation: &AutomationRecord) -> AutomationRunRecord {
2075 let now = Utc::now();
2076 AutomationRunRecord {
2077 schema_version: CURRENT_RUN_SCHEMA_VERSION,
2078 id: Uuid::new_v4().to_string(),
2079 automation_id: automation.id.clone(),
2080 scheduled_for: now,
2081 status: AutomationRunStatus::Queued,
2082 created_at: now,
2083 started_at: None,
2084 ended_at: None,
2085 task_id: None,
2086 thread_id: None,
2087 turn_id: None,
2088 error: None,
2089 }
2090 }
2091
2092 fn eastern_datetime(year: i32, month: u32, day: u32, hour: u32, minute: u32) -> DateTime<Utc> {
2093 Eastern2026
2094 .with_ymd_and_hms(year, month, day, hour, minute, 0)
2095 .single()
2096 .expect("unambiguous Eastern wall time")
2097 .with_timezone(&Utc)
2098 }
2099
2100 fn anchored_automation(
2101 created_at: DateTime<Utc>,
2102 status: AutomationStatus,
2103 ) -> AutomationRecord {
2104 let mut record = automation_record_with_settings(None, None, None, None);
2105 record.rrule = "FREQ=HOURLY;INTERVAL=7;BYMINUTE=17".to_string();
2106 record.status = status;
2107 record.created_at = created_at;
2108 record.updated_at = created_at;
2109 record.next_run_at = None;
2110 record
2111 }
2112
2113 fn local_naive_to_utc(naive: NaiveDateTime) -> DateTime<Utc> {
2114 Local
2115 .from_local_datetime(&naive)
2116 .earliest()
2117 .expect("valid unambiguous local time")
2118 .with_timezone(&Utc)
2119 }
2120
2121 #[test]
2122 fn parses_hourly_rrule() {
2123 let parsed =
2124 AutomationSchedule::parse_rrule("FREQ=HOURLY;INTERVAL=2;BYDAY=MO,TU").expect("parse");
2125 match parsed {
2126 AutomationSchedule::Hourly {
2127 interval_hours,
2128 byday,
2129 ..
2130 } => {
2131 assert_eq!(interval_hours, 2);
2132 assert_eq!(byday.expect("byday").len(), 2);
2133 }
2134 _ => panic!("expected hourly"),
2135 }
2136 }
2137
2138 #[test]
2139 fn parses_once_rrule() {
2140 let parsed =
2141 AutomationSchedule::parse_rrule("FREQ=ONCE;AT=2026-08-03T14:30").expect("parse");
2142 match parsed {
2143 AutomationSchedule::Once { at } => {
2144 assert_eq!(
2145 at,
2146 local_naive_to_utc(
2147 NaiveDateTime::parse_from_str("2026-08-03T14:30", "%Y-%m-%dT%H:%M")
2148 .expect("naive")
2149 )
2150 );
2151 }
2152 _ => panic!("expected once"),
2153 }
2154 }
2155
2156 #[test]
2157 fn parses_hourly_clock_anchor() {
2158 let parsed =
2159 AutomationSchedule::parse_rrule("FREQ=HOURLY;INTERVAL=24;BYHOUR=8;BYMINUTE=30")
2160 .expect("parse anchored hourly schedule");
2161
2162 assert!(matches!(
2163 parsed,
2164 AutomationSchedule::Hourly {
2165 anchor_hour: Some(8),
2166 anchor_minute: Some(30),
2167 ..
2168 }
2169 ));
2170
2171 let minute_only = AutomationSchedule::parse_rrule("FREQ=HOURLY;INTERVAL=1;BYMINUTE=15")
2172 .expect("parse minute-only anchor");
2173 assert!(matches!(
2174 minute_only,
2175 AutomationSchedule::Hourly {
2176 anchor_hour: None,
2177 anchor_minute: Some(15),
2178 ..
2179 }
2180 ));
2181 }
2182
2183 #[test]
2184 fn anchored_hourly_schedule_keeps_wall_time_across_spring_forward() {
2185 let schedule =
2186 AutomationSchedule::parse_rrule("FREQ=HOURLY;INTERVAL=24;BYHOUR=8;BYMINUTE=30")
2187 .expect("parse");
2188 let created_at = eastern_datetime(2026, 3, 6, 7, 0);
2189 let after = eastern_datetime(2026, 3, 7, 9, 0);
2190
2191 let next = schedule
2192 .next_after_in_timezone(after, created_at, &Eastern2026)
2193 .expect("next run");
2194
2195 assert_eq!(next, eastern_datetime(2026, 3, 8, 8, 30));
2196 }
2197
2198 #[test]
2199 fn anchored_hourly_schedule_keeps_wall_time_across_fall_back() {
2200 let schedule =
2201 AutomationSchedule::parse_rrule("FREQ=HOURLY;INTERVAL=24;BYHOUR=8;BYMINUTE=30")
2202 .expect("parse");
2203 let created_at = eastern_datetime(2026, 10, 30, 7, 0);
2204 let after = eastern_datetime(2026, 10, 31, 9, 0);
2205
2206 let next = schedule
2207 .next_after_in_timezone(after, created_at, &Eastern2026)
2208 .expect("next run");
2209
2210 assert_eq!(next, eastern_datetime(2026, 11, 1, 8, 30));
2211 }
2212
2213 #[test]
2214 fn anchored_hourly_schedule_skips_nonexistent_wall_time() {
2215 let schedule =
2216 AutomationSchedule::parse_rrule("FREQ=HOURLY;INTERVAL=24;BYHOUR=2;BYMINUTE=30")
2217 .expect("parse");
2218 let created_at = eastern_datetime(2026, 3, 7, 1, 0);
2219 let after = eastern_datetime(2026, 3, 7, 3, 0);
2220
2221 let next = schedule
2222 .next_after_in_timezone(after, created_at, &Eastern2026)
2223 .expect("next run after spring-forward gap");
2224
2225 assert_eq!(next, eastern_datetime(2026, 3, 9, 2, 30));
2226 }
2227
2228 #[test]
2229 fn anchored_hourly_schedule_uses_first_ambiguous_wall_time_once() {
2230 let schedule =
2231 AutomationSchedule::parse_rrule("FREQ=HOURLY;INTERVAL=24;BYHOUR=1;BYMINUTE=30")
2232 .expect("parse");
2233 let created_at = eastern_datetime(2026, 10, 31, 0, 0);
2234 let after = eastern_datetime(2026, 10, 31, 2, 0);
2235 let first_fold_occurrence = Eastern2026
2236 .with_ymd_and_hms(2026, 11, 1, 1, 30, 0)
2237 .earliest()
2238 .expect("first fold occurrence")
2239 .with_timezone(&Utc);
2240
2241 let next = schedule
2242 .next_after_in_timezone(after, created_at, &Eastern2026)
2243 .expect("next run at fall-back fold");
2244 assert_eq!(next, first_fold_occurrence);
2245
2246 let during_second_fold = Eastern2026
2247 .with_ymd_and_hms(2026, 11, 1, 1, 15, 0)
2248 .latest()
2249 .expect("second fold occurrence")
2250 .with_timezone(&Utc);
2251 let after_fold = schedule
2252 .next_after_in_timezone(during_second_fold, created_at, &Eastern2026)
2253 .expect("next run after fold");
2254 assert_eq!(after_fold, eastern_datetime(2026, 11, 2, 1, 30));
2255 }
2256
2257 #[test]
2258 fn anchored_hourly_schedule_reuses_persisted_anchor_after_restart_and_resume() {
2259 let rrule = "FREQ=HOURLY;INTERVAL=24;BYHOUR=8;BYMINUTE=30";
2260 let created_at = eastern_datetime(2026, 3, 6, 7, 0);
2261 let schedule = AutomationSchedule::parse_rrule(rrule).expect("parse");
2262 let before_restart = schedule
2263 .next_after_in_timezone(
2264 eastern_datetime(2026, 3, 7, 12, 0),
2265 created_at,
2266 &Eastern2026,
2267 )
2268 .expect("next before restart");
2269 assert_eq!(before_restart, eastern_datetime(2026, 3, 8, 8, 30));
2270
2271 // Reparsing models a process restart; the persisted creation timestamp
2272 // remains the recurrence anchor when the record is loaded or resumed.
2273 let restarted = AutomationSchedule::parse_rrule(rrule).expect("reparse after restart");
2274 let after_restart = restarted
2275 .next_after_in_timezone(
2276 eastern_datetime(2026, 3, 8, 10, 0),
2277 created_at,
2278 &Eastern2026,
2279 )
2280 .expect("next after restart");
2281 assert_eq!(after_restart, eastern_datetime(2026, 3, 9, 8, 30));
2282
2283 let after_resume = restarted
2284 .next_after_in_timezone(
2285 eastern_datetime(2026, 3, 10, 12, 0),
2286 created_at,
2287 &Eastern2026,
2288 )
2289 .expect("next after resume");
2290 assert_eq!(after_resume, eastern_datetime(2026, 3, 11, 8, 30));
2291 }
2292
2293 #[test]
2294 fn scheduler_restart_uses_persisted_creation_anchor() {
2295 let tempdir = tempfile::tempdir().expect("tempdir");
2296 let now = Utc::now();
2297 let created_at = now - Duration::hours(51);
2298 let automation = anchored_automation(created_at, AutomationStatus::Active);
2299 let schedule = AutomationSchedule::parse_rrule(&automation.rrule).expect("parse");
2300 let expected = schedule
2301 .next_after_with_anchor(now, created_at)
2302 .expect("persisted-anchor schedule");
2303 let reset_anchor = schedule
2304 .next_after_with_anchor(now, now)
2305 .expect("reset-anchor schedule");
2306 assert_ne!(expected, reset_anchor, "fixture must detect anchor resets");
2307
2308 let manager = AutomationManager::open(tempdir.path().to_path_buf()).expect("manager");
2309 manager.save_automation(&automation).expect("save");
2310 drop(manager);
2311
2312 let restarted = AutomationManager::open(tempdir.path().to_path_buf()).expect("reopen");
2313 assert!(
2314 restarted
2315 .collect_due_runs(now)
2316 .expect("restart tick")
2317 .is_empty(),
2318 "an uninitialized future slot must not enqueue immediately"
2319 );
2320 let reloaded = restarted
2321 .get_automation(&automation.id)
2322 .expect("reloaded automation");
2323 assert_eq!(reloaded.next_run_at, Some(expected));
2324 }
2325
2326 #[test]
2327 fn resume_uses_persisted_creation_anchor() {
2328 let tempdir = tempfile::tempdir().expect("tempdir");
2329 let manager = AutomationManager::open(tempdir.path().to_path_buf()).expect("manager");
2330 let before = Utc::now();
2331 let created_at = before - Duration::hours(51);
2332 let automation = anchored_automation(created_at, AutomationStatus::Paused);
2333 let schedule = AutomationSchedule::parse_rrule(&automation.rrule).expect("parse");
2334 manager.save_automation(&automation).expect("save");
2335
2336 let expected_before = schedule
2337 .next_after_with_anchor(before, created_at)
2338 .expect("next before resume");
2339 let reset_anchor = schedule
2340 .next_after_with_anchor(before, before)
2341 .expect("reset-anchor schedule");
2342 assert_ne!(
2343 expected_before, reset_anchor,
2344 "fixture must detect anchor resets"
2345 );
2346
2347 let resumed = manager
2348 .resume_automation(&automation.id)
2349 .expect("resume automation");
2350 let after = Utc::now();
2351 let expected_after = schedule
2352 .next_after_with_anchor(after, created_at)
2353 .expect("next after resume");
2354 let actual = resumed.next_run_at.expect("resumed next run");
2355 assert!(
2356 actual == expected_before || actual == expected_after,
2357 "resume must keep the persisted creation anchor"
2358 );
2359 }
2360
2361 #[test]
2362 fn anchored_hourly_schedule_applies_byday_on_calendar_slots() {
2363 let schedule = AutomationSchedule::parse_rrule(
2364 "FREQ=HOURLY;INTERVAL=24;BYDAY=MO,TU,WE,TH,FR;BYHOUR=8;BYMINUTE=30",
2365 )
2366 .expect("parse");
2367 let created_at = eastern_datetime(2026, 3, 6, 7, 0);
2368
2369 let next = schedule
2370 .next_after_in_timezone(eastern_datetime(2026, 3, 6, 9, 0), created_at, &Eastern2026)
2371 .expect("next weekday run");
2372
2373 assert_eq!(next, eastern_datetime(2026, 3, 9, 8, 30));
2374 }
2375
2376 #[test]
2377 fn parses_weekly_rrule() {
2378 let parsed =
2379 AutomationSchedule::parse_rrule("FREQ=WEEKLY;BYDAY=MO,WE;BYHOUR=9;BYMINUTE=30")
2380 .expect("parse");
2381 match parsed {
2382 AutomationSchedule::Weekly {
2383 byday,
2384 byhour,
2385 byminute,
2386 } => {
2387 assert_eq!(byday.len(), 2);
2388 assert_eq!(byhour, 9);
2389 assert_eq!(byminute, 30);
2390 }
2391 _ => panic!("expected weekly"),
2392 }
2393 }
2394
2395 #[test]
2396 fn parses_cron_rrule_and_computes_next_minute_slot() {
2397 let schedule =
2398 AutomationSchedule::parse_rrule("FREQ=CRON;EXPR=*/17 * * * *").expect("parse");
2399 let after = Utc
2400 .with_ymd_and_hms(2026, 8, 3, 9, 17, 1)
2401 .single()
2402 .expect("after");
2403 let next = schedule
2404 .next_after_in_timezone(after, after, &Utc)
2405 .expect("next cron run");
2406 assert_eq!(
2407 next,
2408 Utc.with_ymd_and_hms(2026, 8, 3, 9, 34, 0)
2409 .single()
2410 .expect("next")
2411 );
2412 }
2413
2414 #[test]
2415 fn cron_weekday_schedule_uses_standard_five_field_local_time() {
2416 let schedule =
2417 AutomationSchedule::parse_rrule("FREQ=CRON;EXPR=3 9 * * MON-FRI").expect("parse");
2418 let after = Utc
2419 .with_ymd_and_hms(2026, 8, 7, 9, 4, 0)
2420 .single()
2421 .expect("after");
2422 let next = schedule
2423 .next_after_in_timezone(after, after, &Utc)
2424 .expect("next weekday cron run");
2425 assert_eq!(
2426 next,
2427 Utc.with_ymd_and_hms(2026, 8, 10, 9, 3, 0)
2428 .single()
2429 .expect("next")
2430 );
2431 }
2432
2433 #[test]
2434 fn cron_rejects_impossible_date() {
2435 let err = AutomationSchedule::parse_rrule("FREQ=CRON;EXPR=0 9 31 2 *")
2436 .expect_err("impossible february date must fail");
2437 assert!(err.to_string().contains("can never occur"));
2438 }
2439
2440 #[test]
2441 fn rejects_invalid_rrule_fields() {
2442 let err =
2443 AutomationSchedule::parse_rrule("FREQ=WEEKLY;BYSECOND=5").expect_err("should fail");
2444 assert!(err.to_string().contains("Unsupported RRULE field"));
2445 }
2446
2447 #[test]
2448 fn deletes_automation_and_runs() {
2449 let tempdir = tempfile::tempdir().expect("tempdir");
2450 let manager = AutomationManager::open(tempdir.path().to_path_buf()).expect("manager");
2451
2452 let created = manager
2453 .create_automation(CreateAutomationRequest {
2454 name: "Delete me".to_string(),
2455 prompt: "prompt".to_string(),
2456 rrule: "FREQ=HOURLY;INTERVAL=1".to_string(),
2457 cwds: Vec::new(),
2458 mode: None,
2459 allow_shell: None,
2460 trust_mode: None,
2461 auto_approve: None,
2462 delivery_mode: None,
2463 status: Some(AutomationStatus::Active),
2464 })
2465 .expect("create");
2466
2467 let run = AutomationRunRecord {
2468 schema_version: CURRENT_RUN_SCHEMA_VERSION,
2469 id: Uuid::new_v4().to_string(),
2470 automation_id: created.id.clone(),
2471 scheduled_for: Utc::now(),
2472 status: AutomationRunStatus::Queued,
2473 created_at: Utc::now(),
2474 started_at: None,
2475 ended_at: None,
2476 task_id: None,
2477 thread_id: None,
2478 turn_id: None,
2479 error: None,
2480 };
2481 manager.save_run(&run).expect("save run");
2482 assert!(
2483 manager
2484 .runs_dir_for(&created.id)
2485 .expect("runs dir")
2486 .exists()
2487 );
2488
2489 manager
2490 .delete_automation(&created.id)
2491 .expect("delete automation");
2492
2493 assert!(manager.get_automation(&created.id).is_err());
2494 assert!(
2495 !manager
2496 .runs_dir_for(&created.id)
2497 .expect("runs dir")
2498 .exists()
2499 );
2500 }
2501
2502 #[test]
2503 fn automation_storage_rejects_traversal_ids() {
2504 let tempdir = tempfile::tempdir().expect("tempdir");
2505 let manager = AutomationManager::open(tempdir.path().join("root")).expect("manager");
2506 let escaped_file = tempdir.path().join("escape.json");
2507 let escaped_runs = tempdir.path().join("escape-runs");
2508
2509 let err = manager
2510 .get_automation("../escape")
2511 .expect_err("traversal automation ids must be rejected");
2512 assert!(err.to_string().contains("single path component"));
2513 assert!(!escaped_file.exists());
2514
2515 let err = manager
2516 .list_runs("../escape-runs", None)
2517 .expect_err("traversal run dirs must be rejected");
2518 assert!(err.to_string().contains("single path component"));
2519 assert!(!escaped_runs.exists());
2520
2521 let run = AutomationRunRecord {
2522 schema_version: CURRENT_RUN_SCHEMA_VERSION,
2523 id: "../escape-run".to_string(),
2524 automation_id: Uuid::new_v4().to_string(),
2525 scheduled_for: Utc::now(),
2526 status: AutomationRunStatus::Queued,
2527 created_at: Utc::now(),
2528 started_at: None,
2529 ended_at: None,
2530 task_id: None,
2531 thread_id: None,
2532 turn_id: None,
2533 error: None,
2534 };
2535 let err = manager
2536 .save_run(&run)
2537 .expect_err("traversal run ids must be rejected");
2538 assert!(err.to_string().contains("single path component"));
2539 assert!(!tempdir.path().join("escape-run.json").exists());
2540 }
2541
2542 #[test]
2543 fn automation_task_settings_default_for_legacy_records() {
2544 let now = Utc::now().to_rfc3339();
2545 let record: AutomationRecord = serde_json::from_value(serde_json::json!({
2546 "schema_version": CURRENT_AUTOMATION_SCHEMA_VERSION,
2547 "id": Uuid::new_v4().to_string(),
2548 "name": "Legacy automation",
2549 "prompt": "Run legacy automation",
2550 "rrule": "FREQ=HOURLY;INTERVAL=1",
2551 "cwds": [],
2552 "status": "active",
2553 "created_at": now,
2554 "updated_at": now
2555 }))
2556 .expect("legacy automation record should deserialize");
2557
2558 assert_eq!(record.mode, None);
2559 assert_eq!(record.task_mode(), "agent");
2560 assert!(!record.task_allow_shell());
2561 assert!(!record.task_trust_mode());
2562 assert!(!record.task_auto_approve());
2563 assert_eq!(record.delivery_mode(), AutomationDeliveryMode::Task);
2564 }
2565
2566 #[tokio::test]
2567 async fn automation_enqueue_uses_default_and_explicit_task_settings() -> Result<()> {
2568 let tempdir = tempfile::tempdir().expect("tempdir");
2569 let task_manager = TaskManager::start_with_executor(
2570 automation_task_config(tempdir.path().join("tasks")),
2571 std::sync::Arc::new(AutomationNoopExecutor),
2572 )
2573 .await?;
2574
2575 let default_automation = automation_record_with_settings(None, None, None, None);
2576 let mut default_run = queued_run_for(&default_automation);
2577 enqueue_run_task(&default_automation, &mut default_run, &task_manager).await;
2578 let default_task = task_manager
2579 .get_task(default_run.task_id.as_deref().expect("task id"))
2580 .await?;
2581 assert_eq!(default_task.mode, "agent");
2582 assert!(!default_task.allow_shell);
2583 assert!(!default_task.trust_mode);
2584 assert!(!default_task.auto_approve);
2585
2586 let explicit_automation =
2587 automation_record_with_settings(Some("plan"), Some(true), Some(true), Some(true));
2588 let mut explicit_run = queued_run_for(&explicit_automation);
2589 enqueue_run_task(&explicit_automation, &mut explicit_run, &task_manager).await;
2590 let explicit_task = task_manager
2591 .get_task(explicit_run.task_id.as_deref().expect("task id"))
2592 .await?;
2593 assert_eq!(explicit_task.mode, "plan");
2594 assert!(explicit_task.allow_shell);
2595 assert!(explicit_task.trust_mode);
2596 assert!(explicit_task.auto_approve);
2597
2598 task_manager.shutdown();
2599 Ok(())
2600 }
2601
2602 fn write_legacy_run_file(manager: &AutomationManager, run: &AutomationRunRecord) {
2603 let dir = manager.runs_dir_for(&run.automation_id).expect("runs dir");
2604 fs::create_dir_all(&dir).expect("create runs dir");
2605 fs::write(
2606 dir.join(format!("{}.json", run.id)),
2607 serde_json::to_string_pretty(run).expect("serialize run"),
2608 )
2609 .expect("write legacy run");
2610 }
2611
2612 fn run_created_at(
2613 automation: &AutomationRecord,
2614 created_at: DateTime<Utc>,
2615 ) -> AutomationRunRecord {
2616 let mut run = queued_run_for(automation);
2617 run.created_at = created_at;
2618 run.scheduled_for = created_at;
2619 run
2620 }
2621
2622 #[test]
2623 fn save_run_uses_sortable_names_and_migrates_legacy_files() {
2624 let tempdir = tempfile::tempdir().expect("tempdir");
2625 let manager = AutomationManager::open(tempdir.path().to_path_buf()).expect("manager");
2626 let automation = automation_record_with_settings(None, None, None, None);
2627 let run = queued_run_for(&automation);
2628
2629 write_legacy_run_file(&manager, &run);
2630 manager.save_run(&run).expect("save run");
2631
2632 let dir = manager.runs_dir_for(&automation.id).expect("runs dir");
2633 let names: Vec<String> = fs::read_dir(&dir)
2634 .expect("read dir")
2635 .map(|entry| {
2636 entry
2637 .expect("entry")
2638 .file_name()
2639 .to_string_lossy()
2640 .into_owned()
2641 })
2642 .collect();
2643 let expected = format!("{}-{}.json", run_file_stamp(run.created_at), run.id);
2644 assert_eq!(names, vec![expected.clone()]);
2645 assert!(has_sortable_run_stem(expected.trim_end_matches(".json")));
2646 // Legacy uuid stems are not mistaken for sortable names.
2647 assert!(!has_sortable_run_stem(&run.id));
2648 }
2649
2650 #[test]
2651 fn finish_scheduled_run_persists_run_when_automation_deleted_mid_enqueue() {
2652 let tempdir = tempfile::tempdir().expect("tempdir");
2653 let manager = AutomationManager::open(tempdir.path().to_path_buf()).expect("manager");
2654 let automation = automation_record_with_settings(None, None, None, None);
2655 manager.save_automation(&automation).expect("save");
2656 let run = queued_run_for(&automation);
2657
2658 // Simulate the automation being deleted while the enqueue await ran
2659 // outside the lock. The task already exists in the task manager at
2660 // this point, so the run record must still be persisted — an early
2661 // return here orphans a real running task.
2662 manager.delete_automation(&automation.id).expect("delete");
2663 manager
2664 .finish_scheduled_run(&run, Utc::now())
2665 .expect("finish");
2666
2667 let runs = manager.list_runs(&automation.id, None).expect("list runs");
2668 assert_eq!(
2669 runs.iter().map(|r| r.id.as_str()).collect::<Vec<_>>(),
2670 vec![run.id.as_str()],
2671 "run must be persisted even though its automation was deleted"
2672 );
2673 assert!(
2674 manager.get_automation(&automation.id).is_err(),
2675 "the deleted automation must not be resurrected"
2676 );
2677 }
2678
2679 #[test]
2680 fn once_schedule_fires_once_and_auto_completes() {
2681 let tempdir = tempfile::tempdir().expect("tempdir");
2682 let manager = AutomationManager::open(tempdir.path().to_path_buf()).expect("manager");
2683 let due_at = Utc::now() - Duration::minutes(1);
2684 let automation = AutomationRecord {
2685 rrule: due_at
2686 .format("FREQ=ONCE;AT=%Y-%m-%dT%H:%M:%S+00:00")
2687 .to_string(),
2688 next_run_at: Some(due_at),
2689 created_at: due_at - Duration::minutes(5),
2690 updated_at: due_at - Duration::minutes(5),
2691 ..automation_record_with_settings(None, None, None, None)
2692 };
2693 manager
2694 .save_automation(&automation)
2695 .expect("save automation");
2696
2697 let due = manager
2698 .collect_due_runs(Utc::now())
2699 .expect("collect due runs");
2700 assert_eq!(due.len(), 1);
2701 let (_automation, run) = &due[0];
2702 assert_eq!(run.scheduled_for, due_at);
2703
2704 manager
2705 .finish_scheduled_run(run, Utc::now())
2706 .expect("finish one-shot run");
2707 let updated = manager
2708 .get_automation(&automation.id)
2709 .expect("updated automation");
2710 assert_eq!(updated.status, AutomationStatus::Paused);
2711 assert_eq!(updated.next_run_at, None);
2712 assert!(
2713 manager
2714 .collect_due_runs(Utc::now() + Duration::hours(1))
2715 .expect("later tick")
2716 .is_empty()
2717 );
2718 }
2719
2720 #[test]
2721 fn list_runs_merges_legacy_and_sortable_files_newest_first() {
2722 let tempdir = tempfile::tempdir().expect("tempdir");
2723 let manager = AutomationManager::open(tempdir.path().to_path_buf()).expect("manager");
2724 let automation = automation_record_with_settings(None, None, None, None);
2725 let base = Utc::now();
2726
2727 // Legacy files sit at both ends of the timeline to prove the merge is
2728 // by created_at, not by file-name era.
2729 let legacy_oldest = run_created_at(&automation, base - Duration::minutes(30));
2730 let legacy_newest = run_created_at(&automation, base + Duration::minutes(30));
2731 write_legacy_run_file(&manager, &legacy_oldest);
2732 write_legacy_run_file(&manager, &legacy_newest);
2733
2734 let sortable_old = run_created_at(&automation, base - Duration::minutes(20));
2735 let sortable_new = run_created_at(&automation, base + Duration::minutes(20));
2736 manager.save_run(&sortable_old).expect("save old");
2737 manager.save_run(&sortable_new).expect("save new");
2738
2739 let all = manager.list_runs(&automation.id, None).expect("list all");
2740 let ids: Vec<&str> = all.iter().map(|run| run.id.as_str()).collect();
2741 assert_eq!(
2742 ids,
2743 vec![
2744 legacy_newest.id.as_str(),
2745 sortable_new.id.as_str(),
2746 sortable_old.id.as_str(),
2747 legacy_oldest.id.as_str(),
2748 ]
2749 );
2750
2751 let top_two = manager.list_runs(&automation.id, Some(2)).expect("list 2");
2752 let top_ids: Vec<&str> = top_two.iter().map(|run| run.id.as_str()).collect();
2753 assert_eq!(
2754 top_ids,
2755 vec![legacy_newest.id.as_str(), sortable_new.id.as_str()]
2756 );
2757 }
2758
2759 #[test]
2760 fn list_runs_with_limit_skips_older_sortable_files_entirely() {
2761 let tempdir = tempfile::tempdir().expect("tempdir");
2762 let manager = AutomationManager::open(tempdir.path().to_path_buf()).expect("manager");
2763 let automation = automation_record_with_settings(None, None, None, None);
2764 let base = Utc::now();
2765
2766 let newest = run_created_at(&automation, base);
2767 manager.save_run(&newest).expect("save newest");
2768
2769 // A corrupt sortable-named file older than the newest run: bounded
2770 // listing must never open it, while an unbounded listing fails.
2771 let dir = manager.runs_dir_for(&automation.id).expect("runs dir");
2772 let stale_stamp = run_file_stamp(base - Duration::minutes(5));
2773 fs::write(
2774 dir.join(format!("{stale_stamp}-{}.json", Uuid::new_v4())),
2775 "{ not json",
2776 )
2777 .expect("write corrupt run");
2778
2779 let bounded = manager
2780 .list_runs(&automation.id, Some(1))
2781 .expect("bounded list must not read files beyond the limit");
2782 assert_eq!(bounded.len(), 1);
2783 assert_eq!(bounded[0].id, newest.id);
2784
2785 assert!(manager.list_runs(&automation.id, None).is_err());
2786 }
2787
2788 #[tokio::test]
2789 async fn list_automations_completes_during_slow_enqueue() {
2790 let tempdir = tempfile::tempdir().expect("tempdir");
2791 let manager = AutomationManager::open(tempdir.path().to_path_buf()).expect("manager");
2792 let created = manager
2793 .create_automation(CreateAutomationRequest {
2794 name: "Slow enqueue".to_string(),
2795 prompt: "prompt".to_string(),
2796 rrule: "FREQ=HOURLY;INTERVAL=1".to_string(),
2797 cwds: Vec::new(),
2798 mode: None,
2799 allow_shell: None,
2800 trust_mode: None,
2801 auto_approve: None,
2802 delivery_mode: None,
2803 status: Some(AutomationStatus::Active),
2804 })
2805 .expect("create");
2806 let shared: SharedAutomationManager = Arc::new(Mutex::new(manager));
2807
2808 let (entered_tx, entered_rx) = tokio::sync::oneshot::channel::<()>();
2809 let (release_tx, release_rx) = tokio::sync::oneshot::channel::<()>();
2810
2811 let run_task = tokio::spawn({
2812 let shared = Arc::clone(&shared);
2813 let automation_id = created.id.clone();
2814 async move {
2815 run_now_with(&shared, &automation_id, move |_, mut run| async move {
2816 // Delayed task-manager stub: stall the enqueue await until
2817 // the test has proven the manager mutex is free.
2818 let _ = entered_tx.send(());
2819 let _ = release_rx.await;
2820 run.status = AutomationRunStatus::Failed;
2821 run.ended_at = Some(Utc::now());
2822 run.error = Some("stubbed enqueue".to_string());
2823 run
2824 })
2825 .await
2826 }
2827 });
2828
2829 entered_rx.await.expect("enqueue phase entered");
2830
2831 let listed = tokio::time::timeout(std::time::Duration::from_secs(2), async {
2832 shared.lock().await.list_automations()
2833 })
2834 .await
2835 .expect("list_automations must not block behind a slow enqueue")
2836 .expect("list automations");
2837 assert_eq!(listed.len(), 1);
2838
2839 release_tx.send(()).expect("release stub");
2840 let run = run_task.await.expect("join").expect("run now");
2841 assert!(matches!(run.status, AutomationRunStatus::Failed));
2842
2843 // The final run state was persisted after the lock was reacquired.
2844 let manager = shared.lock().await;
2845 let runs = manager.list_runs(&created.id, None).expect("list runs");
2846 assert_eq!(runs.len(), 1);
2847 assert_eq!(runs[0].id, run.id);
2848 assert!(matches!(runs[0].status, AutomationRunStatus::Failed));
2849 let automation = manager.get_automation(&created.id).expect("automation");
2850 assert!(automation.last_run_at.is_some());
2851 }
2852
2853 #[tokio::test]
2854 async fn watcher_noop_completion_removes_run_row() -> Result<()> {
2855 let tempdir = tempfile::tempdir().expect("tempdir");
2856 let task_manager = TaskManager::start_with_executor(
2857 automation_task_config(tempdir.path().join("tasks")),
2858 std::sync::Arc::new(AutomationWatcherNoopExecutor),
2859 )
2860 .await?;
2861 let mut automation = automation_record_with_settings(None, None, None, None);
2862 automation.delivery_mode = Some(AutomationDeliveryMode::Watcher);
2863 automation.next_run_at = Some(Utc::now() - Duration::seconds(1));
2864 let manager = AutomationManager::open(tempdir.path().join("automations")).expect("manager");
2865 manager
2866 .save_automation(&automation)
2867 .expect("save automation");
2868 let shared: SharedAutomationManager = Arc::new(Mutex::new(manager));
2869
2870 scheduler_tick_shared(&shared, &task_manager).await?;
2871 tokio::time::sleep(std::time::Duration::from_millis(100)).await;
2872 reconcile_run_statuses_shared(&shared, &task_manager).await?;
2873
2874 let manager = shared.lock().await;
2875 assert!(
2876 manager.list_runs(&automation.id, None)?.is_empty(),
2877 "watcher no-op must not leave a phantom run row"
2878 );
2879 let updated = manager.get_automation(&automation.id)?;
2880 assert!(
2881 updated.next_run_at.is_some(),
2882 "watcher should keep scheduling"
2883 );
2884 assert_eq!(
2885 updated.last_run_at, None,
2886 "no-op checks are not reportable runs"
2887 );
2888 drop(manager);
2889 task_manager.shutdown();
2890 Ok(())
2891 }
2892
2893 #[test]
2894 fn default_automations_dir_honors_codewhale_home_as_hard_override() {
2895 let _lock = crate::test_support::lock_test_env();
2896 let tmp = tempfile::TempDir::new().unwrap();
2897 // SAFETY: serialised by lock_test_env.
2898 unsafe {
2899 std::env::remove_var("DEEPSEEK_AUTOMATIONS_DIR");
2900 std::env::set_var("CODEWHALE_HOME", tmp.path());
2901 }
2902 // $CODEWHALE_HOME IS the home dir (no ".codewhale" appended); the
2903 // legacy ~/.deepseek fallback is bypassed entirely.
2904 assert_eq!(default_automations_dir(), tmp.path().join("automations"));
2905 // SAFETY: cleanup under the same lock.
2906 unsafe {
2907 std::env::remove_var("CODEWHALE_HOME");
2908 }
2909 }
2910
2911 #[test]
2912 fn default_automations_dir_prefers_deepseek_automations_dir_over_codewhale_home() {
2913 let _lock = crate::test_support::lock_test_env();
2914 let tmp = tempfile::TempDir::new().unwrap();
2915 // SAFETY: serialised by lock_test_env.
2916 unsafe {
2917 std::env::set_var("DEEPSEEK_AUTOMATIONS_DIR", tmp.path());
2918 std::env::set_var("CODEWHALE_HOME", "/should/not/be/used");
2919 }
2920 // The most-specific override wins over the base-data-dir override.
2921 assert_eq!(default_automations_dir(), tmp.path());
2922 // SAFETY: cleanup under the same lock.
2923 unsafe {
2924 std::env::remove_var("DEEPSEEK_AUTOMATIONS_DIR");
2925 std::env::remove_var("CODEWHALE_HOME");
2926 }
2927 }
2928 }
2929
2929 lines RUST