| 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 | //! `~/.deepseek/automations` (or `DEEPSEEK_AUTOMATIONS_DIR` override). |
| 6 | |
| 7 | use std::collections::BTreeMap; |
| 8 | use std::fs; |
| 9 | use std::path::{Path, PathBuf}; |
| 10 | use std::sync::Arc; |
| 11 | |
| 12 | use anyhow::{Context, Result, bail}; |
| 13 | use chrono::{DateTime, Datelike, Duration, Local, TimeZone, Timelike, Utc, Weekday}; |
| 14 | use serde::{Deserialize, Serialize}; |
| 15 | use tokio::sync::Mutex; |
| 16 | use tokio::time::sleep; |
| 17 | use tokio_util::sync::CancellationToken; |
| 18 | use uuid::Uuid; |
| 19 | |
| 20 | use crate::task_manager::{NewTaskRequest, SharedTaskManager, TaskStatus}; |
| 21 | use crate::utils::spawn_supervised; |
| 22 | |
| 23 | const CURRENT_AUTOMATION_SCHEMA_VERSION: u32 = 1; |
| 24 | const CURRENT_RUN_SCHEMA_VERSION: u32 = 1; |
| 25 | |
| 26 | const fn default_automation_schema_version() -> u32 { |
| 27 | CURRENT_AUTOMATION_SCHEMA_VERSION |
| 28 | } |
| 29 | |
| 30 | const fn default_run_schema_version() -> u32 { |
| 31 | CURRENT_RUN_SCHEMA_VERSION |
| 32 | } |
| 33 | |
| 34 | #[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] |
| 35 | #[serde(rename_all = "snake_case")] |
| 36 | pub enum AutomationStatus { |
| 37 | Active, |
| 38 | Paused, |
| 39 | } |
| 40 | |
| 41 | #[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] |
| 42 | #[serde(rename_all = "snake_case")] |
| 43 | pub enum AutomationRunStatus { |
| 44 | Queued, |
| 45 | Running, |
| 46 | Completed, |
| 47 | Failed, |
| 48 | Canceled, |
| 49 | } |
| 50 | |
| 51 | #[derive(Debug, Clone, Serialize, Deserialize)] |
| 52 | pub struct AutomationRecord { |
| 53 | #[serde(default = "default_automation_schema_version")] |
| 54 | pub schema_version: u32, |
| 55 | pub id: String, |
| 56 | pub name: String, |
| 57 | pub prompt: String, |
| 58 | pub rrule: String, |
| 59 | #[serde(default)] |
| 60 | pub cwds: Vec<PathBuf>, |
| 61 | pub status: AutomationStatus, |
| 62 | pub created_at: DateTime<Utc>, |
| 63 | pub updated_at: DateTime<Utc>, |
| 64 | #[serde(skip_serializing_if = "Option::is_none")] |
| 65 | pub next_run_at: Option<DateTime<Utc>>, |
| 66 | #[serde(skip_serializing_if = "Option::is_none")] |
| 67 | pub last_run_at: Option<DateTime<Utc>>, |
| 68 | } |
| 69 | |
| 70 | #[derive(Debug, Clone, Serialize, Deserialize)] |
| 71 | pub struct AutomationRunRecord { |
| 72 | #[serde(default = "default_run_schema_version")] |
| 73 | pub schema_version: u32, |
| 74 | pub id: String, |
| 75 | pub automation_id: String, |
| 76 | pub scheduled_for: DateTime<Utc>, |
| 77 | pub status: AutomationRunStatus, |
| 78 | pub created_at: DateTime<Utc>, |
| 79 | #[serde(skip_serializing_if = "Option::is_none")] |
| 80 | pub started_at: Option<DateTime<Utc>>, |
| 81 | #[serde(skip_serializing_if = "Option::is_none")] |
| 82 | pub ended_at: Option<DateTime<Utc>>, |
| 83 | #[serde(skip_serializing_if = "Option::is_none")] |
| 84 | pub task_id: Option<String>, |
| 85 | #[serde(skip_serializing_if = "Option::is_none")] |
| 86 | pub thread_id: Option<String>, |
| 87 | #[serde(skip_serializing_if = "Option::is_none")] |
| 88 | pub turn_id: Option<String>, |
| 89 | #[serde(skip_serializing_if = "Option::is_none")] |
| 90 | pub error: Option<String>, |
| 91 | } |
| 92 | |
| 93 | #[derive(Debug, Clone, Serialize, Deserialize)] |
| 94 | pub struct CreateAutomationRequest { |
| 95 | pub name: String, |
| 96 | pub prompt: String, |
| 97 | pub rrule: String, |
| 98 | #[serde(default)] |
| 99 | pub cwds: Vec<PathBuf>, |
| 100 | #[serde(default)] |
| 101 | pub status: Option<AutomationStatus>, |
| 102 | } |
| 103 | |
| 104 | #[derive(Debug, Clone, Serialize, Deserialize, Default)] |
| 105 | pub struct UpdateAutomationRequest { |
| 106 | pub name: Option<String>, |
| 107 | pub prompt: Option<String>, |
| 108 | pub rrule: Option<String>, |
| 109 | pub cwds: Option<Vec<PathBuf>>, |
| 110 | pub status: Option<AutomationStatus>, |
| 111 | } |
| 112 | |
| 113 | #[derive(Debug, Clone, Copy, PartialEq, Eq)] |
| 114 | enum AutomationFrequency { |
| 115 | Hourly, |
| 116 | Weekly, |
| 117 | } |
| 118 | |
| 119 | #[derive(Debug, Clone)] |
| 120 | pub enum AutomationSchedule { |
| 121 | Hourly { |
| 122 | interval_hours: u32, |
| 123 | byday: Option<Vec<Weekday>>, |
| 124 | }, |
| 125 | Weekly { |
| 126 | byday: Vec<Weekday>, |
| 127 | byhour: u32, |
| 128 | byminute: u32, |
| 129 | }, |
| 130 | } |
| 131 | |
| 132 | impl AutomationSchedule { |
| 133 | pub fn parse_rrule(rrule: &str) -> Result<Self> { |
| 134 | let mut parts: BTreeMap<String, String> = BTreeMap::new(); |
| 135 | for raw in rrule.split(';') { |
| 136 | let item = raw.trim(); |
| 137 | if item.is_empty() { |
| 138 | continue; |
| 139 | } |
| 140 | let Some((k, v)) = item.split_once('=') else { |
| 141 | bail!("Invalid RRULE segment '{item}'"); |
| 142 | }; |
| 143 | parts.insert(k.trim().to_ascii_uppercase(), v.trim().to_ascii_uppercase()); |
| 144 | } |
| 145 | |
| 146 | let freq = match parts.get("FREQ").map(String::as_str) { |
| 147 | Some("HOURLY") => AutomationFrequency::Hourly, |
| 148 | Some("WEEKLY") => AutomationFrequency::Weekly, |
| 149 | Some(other) => bail!("Unsupported RRULE FREQ '{other}'. Supported: HOURLY and WEEKLY"), |
| 150 | None => bail!("RRULE must include FREQ"), |
| 151 | }; |
| 152 | |
| 153 | match freq { |
| 154 | AutomationFrequency::Hourly => { |
| 155 | for key in parts.keys() { |
| 156 | if key != "FREQ" && key != "INTERVAL" && key != "BYDAY" { |
| 157 | bail!( |
| 158 | "Unsupported RRULE field '{key}' for HOURLY. Allowed: FREQ,INTERVAL,BYDAY" |
| 159 | ); |
| 160 | } |
| 161 | } |
| 162 | let interval_hours = parts |
| 163 | .get("INTERVAL") |
| 164 | .map(|v| v.parse::<u32>()) |
| 165 | .transpose() |
| 166 | .context("Failed to parse INTERVAL")? |
| 167 | .unwrap_or(1); |
| 168 | if interval_hours == 0 { |
| 169 | bail!("INTERVAL must be >= 1 for HOURLY schedules"); |
| 170 | } |
| 171 | let byday = parts |
| 172 | .get("BYDAY") |
| 173 | .map(|value| parse_byday(value)) |
| 174 | .transpose()?; |
| 175 | Ok(Self::Hourly { |
| 176 | interval_hours, |
| 177 | byday, |
| 178 | }) |
| 179 | } |
| 180 | AutomationFrequency::Weekly => { |
| 181 | for key in parts.keys() { |
| 182 | if key != "FREQ" && key != "BYDAY" && key != "BYHOUR" && key != "BYMINUTE" { |
| 183 | bail!( |
| 184 | "Unsupported RRULE field '{key}' for WEEKLY. Allowed: FREQ,BYDAY,BYHOUR,BYMINUTE" |
| 185 | ); |
| 186 | } |
| 187 | } |
| 188 | let byday_raw = parts |
| 189 | .get("BYDAY") |
| 190 | .ok_or_else(|| anyhow::anyhow!("WEEKLY schedules require BYDAY"))?; |
| 191 | let byday = parse_byday(byday_raw)?; |
| 192 | if byday.is_empty() { |
| 193 | bail!("BYDAY cannot be empty for WEEKLY schedules"); |
| 194 | } |
| 195 | let byhour = parts |
| 196 | .get("BYHOUR") |
| 197 | .ok_or_else(|| anyhow::anyhow!("WEEKLY schedules require BYHOUR"))? |
| 198 | .parse::<u32>() |
| 199 | .context("Failed to parse BYHOUR")?; |
| 200 | let byminute = parts |
| 201 | .get("BYMINUTE") |
| 202 | .ok_or_else(|| anyhow::anyhow!("WEEKLY schedules require BYMINUTE"))? |
| 203 | .parse::<u32>() |
| 204 | .context("Failed to parse BYMINUTE")?; |
| 205 | |
| 206 | if byhour > 23 { |
| 207 | bail!("BYHOUR must be between 0 and 23"); |
| 208 | } |
| 209 | if byminute > 59 { |
| 210 | bail!("BYMINUTE must be between 0 and 59"); |
| 211 | } |
| 212 | |
| 213 | Ok(Self::Weekly { |
| 214 | byday, |
| 215 | byhour, |
| 216 | byminute, |
| 217 | }) |
| 218 | } |
| 219 | } |
| 220 | } |
| 221 | |
| 222 | pub fn next_after(&self, after: DateTime<Utc>) -> Result<DateTime<Utc>> { |
| 223 | let local_after = after.with_timezone(&Local); |
| 224 | match self { |
| 225 | Self::Hourly { |
| 226 | interval_hours, |
| 227 | byday, |
| 228 | } => { |
| 229 | let mut candidate = local_after + Duration::hours(i64::from(*interval_hours)) |
| 230 | - Duration::seconds(i64::from(local_after.second())) |
| 231 | - Duration::nanoseconds(i64::from(local_after.nanosecond())); |
| 232 | |
| 233 | if let Some(days) = byday { |
| 234 | for _ in 0..(24 * 21) { |
| 235 | if days.contains(&candidate.weekday()) { |
| 236 | return Ok(candidate.with_timezone(&Utc)); |
| 237 | } |
| 238 | candidate += Duration::hours(i64::from(*interval_hours)); |
| 239 | } |
| 240 | bail!("Unable to compute next HOURLY run for BYDAY filter"); |
| 241 | } |
| 242 | |
| 243 | Ok(candidate.with_timezone(&Utc)) |
| 244 | } |
| 245 | Self::Weekly { |
| 246 | byday, |
| 247 | byhour, |
| 248 | byminute, |
| 249 | } => { |
| 250 | for day_offset in 0..15 { |
| 251 | let date = local_after.date_naive() + Duration::days(i64::from(day_offset)); |
| 252 | if !byday.contains(&date.weekday()) { |
| 253 | continue; |
| 254 | } |
| 255 | let Some(candidate_naive) = date.and_hms_opt(*byhour, *byminute, 0) else { |
| 256 | continue; |
| 257 | }; |
| 258 | if let Some(candidate) = resolve_local_datetime(candidate_naive) |
| 259 | && candidate > local_after |
| 260 | { |
| 261 | return Ok(candidate.with_timezone(&Utc)); |
| 262 | } |
| 263 | } |
| 264 | bail!("Unable to compute next WEEKLY run"); |
| 265 | } |
| 266 | } |
| 267 | } |
| 268 | } |
| 269 | |
| 270 | fn resolve_local_datetime(naive: chrono::NaiveDateTime) -> Option<DateTime<Local>> { |
| 271 | Local |
| 272 | .from_local_datetime(&naive) |
| 273 | .single() |
| 274 | .or_else(|| Local.from_local_datetime(&naive).earliest()) |
| 275 | .or_else(|| Local.from_local_datetime(&naive).latest()) |
| 276 | } |
| 277 | |
| 278 | fn parse_byday(value: &str) -> Result<Vec<Weekday>> { |
| 279 | let mut days = Vec::new(); |
| 280 | for token in value.split(',') { |
| 281 | let day = match token.trim().to_ascii_uppercase().as_str() { |
| 282 | "MO" => Weekday::Mon, |
| 283 | "TU" => Weekday::Tue, |
| 284 | "WE" => Weekday::Wed, |
| 285 | "TH" => Weekday::Thu, |
| 286 | "FR" => Weekday::Fri, |
| 287 | "SA" => Weekday::Sat, |
| 288 | "SU" => Weekday::Sun, |
| 289 | other => bail!("Invalid BYDAY value '{other}'"), |
| 290 | }; |
| 291 | if !days.contains(&day) { |
| 292 | days.push(day); |
| 293 | } |
| 294 | } |
| 295 | Ok(days) |
| 296 | } |
| 297 | |
| 298 | #[derive(Debug, Clone)] |
| 299 | pub struct AutomationManager { |
| 300 | automations_dir: PathBuf, |
| 301 | runs_dir: PathBuf, |
| 302 | } |
| 303 | |
| 304 | impl AutomationManager { |
| 305 | pub fn open(root: PathBuf) -> Result<Self> { |
| 306 | let automations_dir = root.join("automations"); |
| 307 | let runs_dir = root.join("runs"); |
| 308 | fs::create_dir_all(&automations_dir) |
| 309 | .with_context(|| format!("Failed to create {}", automations_dir.display()))?; |
| 310 | fs::create_dir_all(&runs_dir) |
| 311 | .with_context(|| format!("Failed to create {}", runs_dir.display()))?; |
| 312 | Ok(Self { |
| 313 | automations_dir, |
| 314 | runs_dir, |
| 315 | }) |
| 316 | } |
| 317 | |
| 318 | pub fn default_location() -> Result<Self> { |
| 319 | Self::open(default_automations_dir()) |
| 320 | } |
| 321 | |
| 322 | fn automation_path(&self, id: &str) -> PathBuf { |
| 323 | self.automations_dir.join(format!("{id}.json")) |
| 324 | } |
| 325 | |
| 326 | fn runs_dir_for(&self, automation_id: &str) -> PathBuf { |
| 327 | self.runs_dir.join(automation_id) |
| 328 | } |
| 329 | |
| 330 | fn run_path(&self, automation_id: &str, run_id: &str) -> PathBuf { |
| 331 | self.runs_dir_for(automation_id) |
| 332 | .join(format!("{run_id}.json")) |
| 333 | } |
| 334 | |
| 335 | pub fn create_automation(&self, req: CreateAutomationRequest) -> Result<AutomationRecord> { |
| 336 | validate_name_and_prompt(&req.name, &req.prompt)?; |
| 337 | let schedule = AutomationSchedule::parse_rrule(&req.rrule)?; |
| 338 | let now = Utc::now(); |
| 339 | let status = req.status.unwrap_or(AutomationStatus::Active); |
| 340 | let next_run_at = if matches!(status, AutomationStatus::Active) { |
| 341 | Some(schedule.next_after(now)?) |
| 342 | } else { |
| 343 | None |
| 344 | }; |
| 345 | |
| 346 | let record = AutomationRecord { |
| 347 | schema_version: CURRENT_AUTOMATION_SCHEMA_VERSION, |
| 348 | id: Uuid::new_v4().to_string(), |
| 349 | name: req.name.trim().to_string(), |
| 350 | prompt: req.prompt.trim().to_string(), |
| 351 | rrule: req.rrule.trim().to_ascii_uppercase(), |
| 352 | cwds: req.cwds, |
| 353 | status, |
| 354 | created_at: now, |
| 355 | updated_at: now, |
| 356 | next_run_at, |
| 357 | last_run_at: None, |
| 358 | }; |
| 359 | |
| 360 | self.save_automation(&record)?; |
| 361 | Ok(record) |
| 362 | } |
| 363 | |
| 364 | pub fn get_automation(&self, id: &str) -> Result<AutomationRecord> { |
| 365 | let path = self.automation_path(id); |
| 366 | let raw = fs::read_to_string(&path) |
| 367 | .with_context(|| format!("Failed to read automation {}", path.display()))?; |
| 368 | let record: AutomationRecord = serde_json::from_str(&raw) |
| 369 | .with_context(|| format!("Failed to parse automation {}", path.display()))?; |
| 370 | if record.schema_version > CURRENT_AUTOMATION_SCHEMA_VERSION { |
| 371 | bail!( |
| 372 | "Automation schema v{} is newer than supported v{}", |
| 373 | record.schema_version, |
| 374 | CURRENT_AUTOMATION_SCHEMA_VERSION |
| 375 | ); |
| 376 | } |
| 377 | Ok(record) |
| 378 | } |
| 379 | |
| 380 | pub fn save_automation(&self, record: &AutomationRecord) -> Result<()> { |
| 381 | write_json_atomic(&self.automation_path(&record.id), record) |
| 382 | } |
| 383 | |
| 384 | pub fn list_automations(&self) -> Result<Vec<AutomationRecord>> { |
| 385 | let mut out = Vec::new(); |
| 386 | for entry in fs::read_dir(&self.automations_dir) |
| 387 | .with_context(|| format!("Failed to read {}", self.automations_dir.display()))? |
| 388 | { |
| 389 | let entry = entry?; |
| 390 | let path = entry.path(); |
| 391 | if path.extension().is_none_or(|ext| ext != "json") { |
| 392 | continue; |
| 393 | } |
| 394 | let raw = fs::read_to_string(&path) |
| 395 | .with_context(|| format!("Failed to read {}", path.display()))?; |
| 396 | let record: AutomationRecord = serde_json::from_str(&raw) |
| 397 | .with_context(|| format!("Failed to parse {}", path.display()))?; |
| 398 | if record.schema_version > CURRENT_AUTOMATION_SCHEMA_VERSION { |
| 399 | bail!( |
| 400 | "Automation schema v{} is newer than supported v{}", |
| 401 | record.schema_version, |
| 402 | CURRENT_AUTOMATION_SCHEMA_VERSION |
| 403 | ); |
| 404 | } |
| 405 | out.push(record); |
| 406 | } |
| 407 | out.sort_by_key(|r| std::cmp::Reverse(r.updated_at)); |
| 408 | Ok(out) |
| 409 | } |
| 410 | |
| 411 | pub fn update_automation( |
| 412 | &self, |
| 413 | id: &str, |
| 414 | req: UpdateAutomationRequest, |
| 415 | ) -> Result<AutomationRecord> { |
| 416 | let mut existing = self.get_automation(id)?; |
| 417 | |
| 418 | if let Some(name) = req.name { |
| 419 | if name.trim().is_empty() { |
| 420 | bail!("Automation name cannot be empty"); |
| 421 | } |
| 422 | existing.name = name.trim().to_string(); |
| 423 | } |
| 424 | if let Some(prompt) = req.prompt { |
| 425 | if prompt.trim().is_empty() { |
| 426 | bail!("Automation prompt cannot be empty"); |
| 427 | } |
| 428 | existing.prompt = prompt.trim().to_string(); |
| 429 | } |
| 430 | if let Some(rrule) = req.rrule { |
| 431 | let normalized = rrule.trim().to_ascii_uppercase(); |
| 432 | AutomationSchedule::parse_rrule(&normalized)?; |
| 433 | existing.rrule = normalized; |
| 434 | if matches!(existing.status, AutomationStatus::Active) { |
| 435 | let schedule = AutomationSchedule::parse_rrule(&existing.rrule)?; |
| 436 | existing.next_run_at = Some(schedule.next_after(Utc::now())?); |
| 437 | } |
| 438 | } |
| 439 | if let Some(cwds) = req.cwds { |
| 440 | existing.cwds = cwds; |
| 441 | } |
| 442 | if let Some(status) = req.status { |
| 443 | existing.status = status; |
| 444 | if matches!(status, AutomationStatus::Paused) { |
| 445 | existing.next_run_at = None; |
| 446 | } else { |
| 447 | let schedule = AutomationSchedule::parse_rrule(&existing.rrule)?; |
| 448 | existing.next_run_at = Some(schedule.next_after(Utc::now())?); |
| 449 | } |
| 450 | } |
| 451 | |
| 452 | existing.updated_at = Utc::now(); |
| 453 | self.save_automation(&existing)?; |
| 454 | Ok(existing) |
| 455 | } |
| 456 | |
| 457 | pub fn pause_automation(&self, id: &str) -> Result<AutomationRecord> { |
| 458 | self.update_automation( |
| 459 | id, |
| 460 | UpdateAutomationRequest { |
| 461 | status: Some(AutomationStatus::Paused), |
| 462 | ..UpdateAutomationRequest::default() |
| 463 | }, |
| 464 | ) |
| 465 | } |
| 466 | |
| 467 | pub fn resume_automation(&self, id: &str) -> Result<AutomationRecord> { |
| 468 | self.update_automation( |
| 469 | id, |
| 470 | UpdateAutomationRequest { |
| 471 | status: Some(AutomationStatus::Active), |
| 472 | ..UpdateAutomationRequest::default() |
| 473 | }, |
| 474 | ) |
| 475 | } |
| 476 | |
| 477 | pub fn delete_automation(&self, id: &str) -> Result<AutomationRecord> { |
| 478 | let existing = self.get_automation(id)?; |
| 479 | let path = self.automation_path(id); |
| 480 | fs::remove_file(&path) |
| 481 | .with_context(|| format!("Failed to delete automation {}", path.display()))?; |
| 482 | |
| 483 | let runs_dir = self.runs_dir_for(id); |
| 484 | if runs_dir.exists() { |
| 485 | fs::remove_dir_all(&runs_dir).with_context(|| { |
| 486 | format!("Failed to delete automation runs {}", runs_dir.display()) |
| 487 | })?; |
| 488 | } |
| 489 | |
| 490 | Ok(existing) |
| 491 | } |
| 492 | |
| 493 | pub fn list_runs( |
| 494 | &self, |
| 495 | automation_id: &str, |
| 496 | limit: Option<usize>, |
| 497 | ) -> Result<Vec<AutomationRunRecord>> { |
| 498 | let dir = self.runs_dir_for(automation_id); |
| 499 | if !dir.exists() { |
| 500 | return Ok(Vec::new()); |
| 501 | } |
| 502 | |
| 503 | let mut out = Vec::new(); |
| 504 | for entry in |
| 505 | fs::read_dir(&dir).with_context(|| format!("Failed to read {}", dir.display()))? |
| 506 | { |
| 507 | let entry = entry?; |
| 508 | let path = entry.path(); |
| 509 | if path.extension().is_none_or(|ext| ext != "json") { |
| 510 | continue; |
| 511 | } |
| 512 | let raw = fs::read_to_string(&path) |
| 513 | .with_context(|| format!("Failed to read {}", path.display()))?; |
| 514 | let run: AutomationRunRecord = serde_json::from_str(&raw) |
| 515 | .with_context(|| format!("Failed to parse {}", path.display()))?; |
| 516 | if run.schema_version > CURRENT_RUN_SCHEMA_VERSION { |
| 517 | bail!( |
| 518 | "Automation run schema v{} is newer than supported v{}", |
| 519 | run.schema_version, |
| 520 | CURRENT_RUN_SCHEMA_VERSION |
| 521 | ); |
| 522 | } |
| 523 | out.push(run); |
| 524 | } |
| 525 | |
| 526 | out.sort_by_key(|r| std::cmp::Reverse(r.created_at)); |
| 527 | if let Some(limit) = limit { |
| 528 | out.truncate(limit); |
| 529 | } |
| 530 | Ok(out) |
| 531 | } |
| 532 | |
| 533 | fn save_run(&self, run: &AutomationRunRecord) -> Result<()> { |
| 534 | let dir = self.runs_dir_for(&run.automation_id); |
| 535 | fs::create_dir_all(&dir).with_context(|| format!("Failed to create {}", dir.display()))?; |
| 536 | write_json_atomic(&self.run_path(&run.automation_id, &run.id), run) |
| 537 | } |
| 538 | |
| 539 | async fn enqueue_run_task( |
| 540 | &self, |
| 541 | automation: &AutomationRecord, |
| 542 | run: &mut AutomationRunRecord, |
| 543 | task_manager: &SharedTaskManager, |
| 544 | ) -> Result<()> { |
| 545 | let workspace = automation.cwds.first().cloned(); |
| 546 | |
| 547 | let new_task = NewTaskRequest { |
| 548 | prompt: automation.prompt.clone(), |
| 549 | model: None, |
| 550 | workspace, |
| 551 | mode: Some("agent".to_string()), |
| 552 | allow_shell: Some(false), |
| 553 | trust_mode: Some(false), |
| 554 | auto_approve: Some(true), |
| 555 | }; |
| 556 | |
| 557 | match task_manager.add_task(new_task).await { |
| 558 | Ok(task) => { |
| 559 | run.status = AutomationRunStatus::Running; |
| 560 | run.started_at = Some(Utc::now()); |
| 561 | run.task_id = Some(task.id.clone()); |
| 562 | run.thread_id = task.thread_id.clone(); |
| 563 | run.turn_id = task.turn_id.clone(); |
| 564 | run.error = None; |
| 565 | Ok(()) |
| 566 | } |
| 567 | Err(err) => { |
| 568 | run.status = AutomationRunStatus::Failed; |
| 569 | run.ended_at = Some(Utc::now()); |
| 570 | run.error = Some(format!("Failed to enqueue task: {err}")); |
| 571 | Ok(()) |
| 572 | } |
| 573 | } |
| 574 | } |
| 575 | |
| 576 | pub async fn run_now( |
| 577 | &self, |
| 578 | automation_id: &str, |
| 579 | task_manager: &SharedTaskManager, |
| 580 | ) -> Result<AutomationRunRecord> { |
| 581 | let mut automation = self.get_automation(automation_id)?; |
| 582 | let now = Utc::now(); |
| 583 | let mut run = AutomationRunRecord { |
| 584 | schema_version: CURRENT_RUN_SCHEMA_VERSION, |
| 585 | id: Uuid::new_v4().to_string(), |
| 586 | automation_id: automation.id.clone(), |
| 587 | scheduled_for: now, |
| 588 | status: AutomationRunStatus::Queued, |
| 589 | created_at: now, |
| 590 | started_at: None, |
| 591 | ended_at: None, |
| 592 | task_id: None, |
| 593 | thread_id: None, |
| 594 | turn_id: None, |
| 595 | error: None, |
| 596 | }; |
| 597 | |
| 598 | self.enqueue_run_task(&automation, &mut run, task_manager) |
| 599 | .await?; |
| 600 | self.save_run(&run)?; |
| 601 | |
| 602 | automation.updated_at = Utc::now(); |
| 603 | if matches!( |
| 604 | run.status, |
| 605 | AutomationRunStatus::Completed |
| 606 | | AutomationRunStatus::Failed |
| 607 | | AutomationRunStatus::Canceled |
| 608 | ) { |
| 609 | automation.last_run_at = run.ended_at.or(Some(Utc::now())); |
| 610 | } |
| 611 | self.save_automation(&automation)?; |
| 612 | |
| 613 | Ok(run) |
| 614 | } |
| 615 | |
| 616 | pub async fn scheduler_tick(&self, task_manager: &SharedTaskManager) -> Result<()> { |
| 617 | let now = Utc::now(); |
| 618 | let mut automations = self.list_automations()?; |
| 619 | |
| 620 | for automation in &mut automations { |
| 621 | if !matches!(automation.status, AutomationStatus::Active) { |
| 622 | continue; |
| 623 | } |
| 624 | |
| 625 | let schedule = AutomationSchedule::parse_rrule(&automation.rrule)?; |
| 626 | if automation.next_run_at.is_none() { |
| 627 | automation.next_run_at = Some(schedule.next_after(now)?); |
| 628 | automation.updated_at = now; |
| 629 | self.save_automation(automation)?; |
| 630 | continue; |
| 631 | } |
| 632 | |
| 633 | let due_at = automation.next_run_at.expect("checked above"); |
| 634 | if due_at > now { |
| 635 | continue; |
| 636 | } |
| 637 | |
| 638 | // Idempotency: if a run already exists for this schedule slot, skip enqueue and |
| 639 | // advance next_run_at. |
| 640 | let existing_for_slot = self |
| 641 | .list_runs(&automation.id, Some(25))? |
| 642 | .into_iter() |
| 643 | .any(|run| run.scheduled_for == due_at); |
| 644 | |
| 645 | if existing_for_slot { |
| 646 | automation.next_run_at = Some(schedule.next_after(due_at)?); |
| 647 | automation.updated_at = now; |
| 648 | self.save_automation(automation)?; |
| 649 | continue; |
| 650 | } |
| 651 | |
| 652 | let mut run = AutomationRunRecord { |
| 653 | schema_version: CURRENT_RUN_SCHEMA_VERSION, |
| 654 | id: Uuid::new_v4().to_string(), |
| 655 | automation_id: automation.id.clone(), |
| 656 | scheduled_for: due_at, |
| 657 | status: AutomationRunStatus::Queued, |
| 658 | created_at: now, |
| 659 | started_at: None, |
| 660 | ended_at: None, |
| 661 | task_id: None, |
| 662 | thread_id: None, |
| 663 | turn_id: None, |
| 664 | error: None, |
| 665 | }; |
| 666 | |
| 667 | self.enqueue_run_task(automation, &mut run, task_manager) |
| 668 | .await?; |
| 669 | self.save_run(&run)?; |
| 670 | |
| 671 | automation.updated_at = now; |
| 672 | automation.next_run_at = Some(schedule.next_after(due_at)?); |
| 673 | self.save_automation(automation)?; |
| 674 | } |
| 675 | |
| 676 | Ok(()) |
| 677 | } |
| 678 | |
| 679 | pub async fn reconcile_run_statuses(&self, task_manager: &SharedTaskManager) -> Result<()> { |
| 680 | let automations = self.list_automations()?; |
| 681 | for automation in automations { |
| 682 | let runs = self.list_runs(&automation.id, Some(100))?; |
| 683 | for mut run in runs { |
| 684 | if !matches!( |
| 685 | run.status, |
| 686 | AutomationRunStatus::Queued | AutomationRunStatus::Running |
| 687 | ) { |
| 688 | continue; |
| 689 | } |
| 690 | let Some(task_id) = run.task_id.clone() else { |
| 691 | continue; |
| 692 | }; |
| 693 | let task = match task_manager.get_task(&task_id).await { |
| 694 | Ok(task) => task, |
| 695 | Err(_) => continue, |
| 696 | }; |
| 697 | |
| 698 | run.thread_id = task.thread_id.clone(); |
| 699 | run.turn_id = task.turn_id.clone(); |
| 700 | |
| 701 | let mut changed = false; |
| 702 | match task.status { |
| 703 | TaskStatus::Queued => { |
| 704 | if !matches!(run.status, AutomationRunStatus::Queued) { |
| 705 | run.status = AutomationRunStatus::Queued; |
| 706 | changed = true; |
| 707 | } |
| 708 | } |
| 709 | TaskStatus::Running => { |
| 710 | if !matches!(run.status, AutomationRunStatus::Running) { |
| 711 | run.status = AutomationRunStatus::Running; |
| 712 | changed = true; |
| 713 | } |
| 714 | if run.started_at.is_none() { |
| 715 | run.started_at = Some(task.started_at.unwrap_or_else(Utc::now)); |
| 716 | changed = true; |
| 717 | } |
| 718 | } |
| 719 | TaskStatus::Completed => { |
| 720 | run.status = AutomationRunStatus::Completed; |
| 721 | run.started_at = run.started_at.or(task.started_at); |
| 722 | run.ended_at = task.ended_at.or(Some(Utc::now())); |
| 723 | run.error = None; |
| 724 | changed = true; |
| 725 | } |
| 726 | TaskStatus::Failed => { |
| 727 | run.status = AutomationRunStatus::Failed; |
| 728 | run.started_at = run.started_at.or(task.started_at); |
| 729 | run.ended_at = task.ended_at.or(Some(Utc::now())); |
| 730 | run.error = task.error.clone(); |
| 731 | changed = true; |
| 732 | } |
| 733 | TaskStatus::Canceled => { |
| 734 | run.status = AutomationRunStatus::Canceled; |
| 735 | run.started_at = run.started_at.or(task.started_at); |
| 736 | run.ended_at = task.ended_at.or(Some(Utc::now())); |
| 737 | changed = true; |
| 738 | } |
| 739 | } |
| 740 | |
| 741 | if changed { |
| 742 | self.save_run(&run)?; |
| 743 | if matches!( |
| 744 | run.status, |
| 745 | AutomationRunStatus::Completed |
| 746 | | AutomationRunStatus::Failed |
| 747 | | AutomationRunStatus::Canceled |
| 748 | ) { |
| 749 | let mut updated_automation = self.get_automation(&automation.id)?; |
| 750 | updated_automation.last_run_at = run.ended_at.or(Some(Utc::now())); |
| 751 | updated_automation.updated_at = Utc::now(); |
| 752 | self.save_automation(&updated_automation)?; |
| 753 | } |
| 754 | } |
| 755 | } |
| 756 | } |
| 757 | |
| 758 | Ok(()) |
| 759 | } |
| 760 | } |
| 761 | |
| 762 | fn validate_name_and_prompt(name: &str, prompt: &str) -> Result<()> { |
| 763 | if name.trim().is_empty() { |
| 764 | bail!("Automation name is required"); |
| 765 | } |
| 766 | if prompt.trim().is_empty() { |
| 767 | bail!("Automation prompt is required"); |
| 768 | } |
| 769 | Ok(()) |
| 770 | } |
| 771 | |
| 772 | fn write_json_atomic<T: Serialize>(path: &Path, value: &T) -> Result<()> { |
| 773 | if let Some(parent) = path.parent() { |
| 774 | fs::create_dir_all(parent) |
| 775 | .with_context(|| format!("Failed to create {}", parent.display()))?; |
| 776 | } |
| 777 | let content = serde_json::to_string_pretty(value)?; |
| 778 | let tmp = path.with_extension("json.tmp"); |
| 779 | fs::write(&tmp, content).with_context(|| format!("Failed to write {}", tmp.display()))?; |
| 780 | fs::rename(&tmp, path).with_context(|| { |
| 781 | format!( |
| 782 | "Failed to move temporary file {} to {}", |
| 783 | tmp.display(), |
| 784 | path.display() |
| 785 | ) |
| 786 | })?; |
| 787 | Ok(()) |
| 788 | } |
| 789 | |
| 790 | pub fn default_automations_dir() -> PathBuf { |
| 791 | if let Ok(path) = std::env::var("DEEPSEEK_AUTOMATIONS_DIR") { |
| 792 | let trimmed = path.trim(); |
| 793 | if !trimmed.is_empty() { |
| 794 | return PathBuf::from(trimmed); |
| 795 | } |
| 796 | } |
| 797 | dirs::home_dir() |
| 798 | .map(|home| home.join(".deepseek").join("automations")) |
| 799 | .unwrap_or_else(|| PathBuf::from(".deepseek").join("automations")) |
| 800 | } |
| 801 | |
| 802 | pub type SharedAutomationManager = Arc<Mutex<AutomationManager>>; |
| 803 | |
| 804 | #[derive(Debug, Clone)] |
| 805 | pub struct AutomationSchedulerConfig { |
| 806 | pub tick_interval_secs: u64, |
| 807 | } |
| 808 | |
| 809 | impl Default for AutomationSchedulerConfig { |
| 810 | fn default() -> Self { |
| 811 | Self { |
| 812 | tick_interval_secs: 15, |
| 813 | } |
| 814 | } |
| 815 | } |
| 816 | |
| 817 | pub fn spawn_scheduler( |
| 818 | automations: SharedAutomationManager, |
| 819 | task_manager: SharedTaskManager, |
| 820 | cancel: CancellationToken, |
| 821 | config: AutomationSchedulerConfig, |
| 822 | ) -> tokio::task::JoinHandle<()> { |
| 823 | spawn_supervised( |
| 824 | "automation-scheduler", |
| 825 | std::panic::Location::caller(), |
| 826 | async move { |
| 827 | let interval = config.tick_interval_secs.max(5); |
| 828 | loop { |
| 829 | if cancel.is_cancelled() { |
| 830 | break; |
| 831 | } |
| 832 | |
| 833 | { |
| 834 | let manager = automations.lock().await; |
| 835 | if let Err(err) = manager.scheduler_tick(&task_manager).await { |
| 836 | tracing::warn!("automation scheduler tick failed: {err}"); |
| 837 | } |
| 838 | if let Err(err) = manager.reconcile_run_statuses(&task_manager).await { |
| 839 | tracing::warn!("automation reconcile failed: {err}"); |
| 840 | } |
| 841 | } |
| 842 | |
| 843 | tokio::select! { |
| 844 | _ = cancel.cancelled() => break, |
| 845 | _ = sleep(std::time::Duration::from_secs(interval)) => {} |
| 846 | } |
| 847 | } |
| 848 | }, |
| 849 | ) |
| 850 | } |
| 851 | |
| 852 | #[cfg(test)] |
| 853 | mod tests { |
| 854 | use super::*; |
| 855 | |
| 856 | #[test] |
| 857 | fn parses_hourly_rrule() { |
| 858 | let parsed = |
| 859 | AutomationSchedule::parse_rrule("FREQ=HOURLY;INTERVAL=2;BYDAY=MO,TU").expect("parse"); |
| 860 | match parsed { |
| 861 | AutomationSchedule::Hourly { |
| 862 | interval_hours, |
| 863 | byday, |
| 864 | } => { |
| 865 | assert_eq!(interval_hours, 2); |
| 866 | assert_eq!(byday.expect("byday").len(), 2); |
| 867 | } |
| 868 | _ => panic!("expected hourly"), |
| 869 | } |
| 870 | } |
| 871 | |
| 872 | #[test] |
| 873 | fn parses_weekly_rrule() { |
| 874 | let parsed = |
| 875 | AutomationSchedule::parse_rrule("FREQ=WEEKLY;BYDAY=MO,WE;BYHOUR=9;BYMINUTE=30") |
| 876 | .expect("parse"); |
| 877 | match parsed { |
| 878 | AutomationSchedule::Weekly { |
| 879 | byday, |
| 880 | byhour, |
| 881 | byminute, |
| 882 | } => { |
| 883 | assert_eq!(byday.len(), 2); |
| 884 | assert_eq!(byhour, 9); |
| 885 | assert_eq!(byminute, 30); |
| 886 | } |
| 887 | _ => panic!("expected weekly"), |
| 888 | } |
| 889 | } |
| 890 | |
| 891 | #[test] |
| 892 | fn rejects_invalid_rrule_fields() { |
| 893 | let err = |
| 894 | AutomationSchedule::parse_rrule("FREQ=WEEKLY;BYSECOND=5").expect_err("should fail"); |
| 895 | assert!(err.to_string().contains("Unsupported RRULE field")); |
| 896 | } |
| 897 | |
| 898 | #[test] |
| 899 | fn deletes_automation_and_runs() { |
| 900 | let tempdir = tempfile::tempdir().expect("tempdir"); |
| 901 | let manager = AutomationManager::open(tempdir.path().to_path_buf()).expect("manager"); |
| 902 | |
| 903 | let created = manager |
| 904 | .create_automation(CreateAutomationRequest { |
| 905 | name: "Delete me".to_string(), |
| 906 | prompt: "prompt".to_string(), |
| 907 | rrule: "FREQ=HOURLY;INTERVAL=1".to_string(), |
| 908 | cwds: Vec::new(), |
| 909 | status: Some(AutomationStatus::Active), |
| 910 | }) |
| 911 | .expect("create"); |
| 912 | |
| 913 | let run = AutomationRunRecord { |
| 914 | schema_version: CURRENT_RUN_SCHEMA_VERSION, |
| 915 | id: Uuid::new_v4().to_string(), |
| 916 | automation_id: created.id.clone(), |
| 917 | scheduled_for: Utc::now(), |
| 918 | status: AutomationRunStatus::Queued, |
| 919 | created_at: Utc::now(), |
| 920 | started_at: None, |
| 921 | ended_at: None, |
| 922 | task_id: None, |
| 923 | thread_id: None, |
| 924 | turn_id: None, |
| 925 | error: None, |
| 926 | }; |
| 927 | manager.save_run(&run).expect("save run"); |
| 928 | assert!(manager.runs_dir_for(&created.id).exists()); |
| 929 | |
| 930 | manager |
| 931 | .delete_automation(&created.id) |
| 932 | .expect("delete automation"); |
| 933 | |
| 934 | assert!(manager.get_automation(&created.id).is_err()); |
| 935 | assert!(!manager.runs_dir_for(&created.id).exists()); |
| 936 | } |
| 937 | } |
| 938 |