Vakyartha documentation

1//! Scheduled-task store (docs/design/29-personal-os.md P2). Same
2//! `tasks.json` shape the server surface wrote — legacy files load
3//! unchanged — plus additive optional fields: 5-field cron schedules,
4//! watchdog shell one-liners (XOR with prompt), and per-task model pins.
5//! The store is plain JSON under `<sessions-home>/tasks.json`; sessions
6//! themselves stay append-only ledgers.
7
8use std::collections::{BTreeSet, HashMap};
9use std::io::Write;
10use std::path::{Path, PathBuf};
11
12use chrono::{DateTime, Datelike, TimeZone, Timelike, Utc};
13
14/// How far ahead [`cron_next_after`] searches before giving up. Cron can
15/// legitimately sleep ~4 years (Feb 29); anything beyond five is a bug in
16/// the expression, not a schedule.
17const CRON_HORIZON_DAYS: i64 = 366 * 5;
18
19#[derive(Debug, Clone, PartialEq)]
20struct FieldSet {
21 values: BTreeSet<u32>,
22 /// True only for a literal `*` field — vixie-cron's dom/dow OR rule
23 /// keys on restriction, not on which values ended up selected.
24 starred: bool,
25}
26
27impl FieldSet {
28 fn contains(&self, v: u32) -> bool {
29 self.starred || self.values.contains(&v)
30 }
31}
32
33#[derive(Debug, Clone, PartialEq)]
34pub struct CronExpr {
35 minutes: FieldSet,
36 hours: FieldSet,
37 days_of_month: FieldSet,
38 months: FieldSet,
39 days_of_week: FieldSet,
40}
41
42impl CronExpr {
43 pub fn parse(expr: &str) -> Result<Self, String> {
44 let fields: Vec<&str> = expr.split_whitespace().collect();
45 if fields.len() != 5 {
46 return Err(format!(
47 "expected 5 fields (min hour dom mon dow), got {}",
48 fields.len()
49 ));
50 }
51 Ok(CronExpr {
52 minutes: parse_field(fields[0], 0, 59, false)?,
53 hours: parse_field(fields[1], 0, 23, false)?,
54 days_of_month: parse_field(fields[2], 1, 31, false)?,
55 months: parse_field(fields[3], 1, 12, false)?,
56 days_of_week: parse_field(fields[4], 0, 7, true)?,
57 })
58 }
59}
60
61fn parse_field(spec: &str, min: u32, max: u32, dow_wrap: bool) -> Result<FieldSet, String> {
62 let mut values = BTreeSet::new();
63 let mut starred = false;
64 // Vixie convention: 0 and 7 are both Sunday; mapping happens AFTER
65 // range expansion so `5-7` means Fri–Sun rather than a reversed range.
66 let wrap = |v: u32| if dow_wrap && v == 7 { 0 } else { v };
67 for term in spec.split(',') {
68 let term = term.trim();
69 if term.is_empty() {
70 return Err(format!("empty list element in '{spec}'"));
71 }
72 let (base, step) = match term.split_once('/') {
73 Some((b, s)) => {
74 let step: u32 = s.parse().map_err(|_| format!("bad step '{s}'"))?;
75 if step == 0 {
76 return Err(format!("step must be >= 1 in '{term}'"));
77 }
78 (b, Some(step))
79 }
80 None => (term, None),
81 };
82 let range: Vec<u32> = match base {
83 "*" => {
84 if term == "*" && step.is_none() {
85 starred = true;
86 }
87 (min..=max).collect()
88 }
89 b if b.contains('-') => {
90 let (lo, hi) = b
91 .split_once('-')
92 .ok_or_else(|| format!("bad range '{b}'"))?;
93 let lo = parse_number(lo, min, max)?;
94 let hi = parse_number(hi, min, max)?;
95 if hi < lo {
96 return Err(format!("reversed range '{b}'"));
97 }
98 (lo..=hi).map(wrap).collect()
99 }
100 b => {
101 if step.is_some() {
102 return Err(format!("step requires '*' or a range in '{term}'"));
103 }
104 vec![wrap(parse_number(b, min, max)?)]
105 }
106 };
107 match step {
108 None => values.extend(range),
109 Some(s) => values.extend(range.into_iter().step_by(s as usize)),
110 }
111 }
112 Ok(FieldSet { values, starred })
113}
114
115fn parse_number(raw: &str, min: u32, max: u32) -> Result<u32, String> {
116 let v: u32 = raw
117 .trim()
118 .parse()
119 .map_err(|_| format!("bad number '{raw}'"))?;
120 if !(min..=max).contains(&v) {
121 return Err(format!("value {v} out of range {min}-{max}"));
122 }
123 Ok(v)
124}
125
126/// Next local time strictly after `after` where the expression matches.
127///
128/// DST semantics are deterministic by construction: candidate wall-clock
129/// times that do not exist (spring-forward gap) never fire — scanning
130/// continues to the next matching time that DOES exist; ambiguous times
131/// (fall-back fold) resolve to the earliest instant. Pure: no clock reads.
132pub fn cron_next_after(
133 expr: &str,
134 after: chrono::DateTime<chrono::Local>,
135) -> Result<DateTime<chrono::Local>, String> {
136 let parsed = CronExpr::parse(expr)?;
137 next_fire(&parsed, after)
138 .ok_or_else(|| format!("expression '{expr}' never fires within {CRON_HORIZON_DAYS} days"))
139}
140
141/// Named-IANA equivalent of [`cron_next_after`].
142pub fn cron_next_after_timezone(
143 expr: &str,
144 after: DateTime<Utc>,
145 timezone: &str,
146) -> Result<DateTime<Utc>, String> {
147 let tz: chrono_tz::Tz = timezone
148 .parse()
149 .map_err(|_| format!("unknown IANA timezone '{timezone}'"))?;
150 let local_after = after.with_timezone(&tz);
151 next_fire(&CronExpr::parse(expr)?, local_after)
152 .map(|value| value.with_timezone(&Utc))
153 .ok_or_else(|| format!("expression '{expr}' never fires within {CRON_HORIZON_DAYS} days"))
154}
155
156fn next_fire<Tz: TimeZone>(expr: &CronExpr, after: DateTime<Tz>) -> Option<DateTime<Tz>> {
157 let tz = after.timezone();
158 let naive = after.naive_local();
159 // Strictly-after: start from the minute boundary following `after`.
160 let start_minute =
161 naive.date().and_hms_opt(naive.hour(), naive.minute(), 0)? + chrono::Duration::minutes(1);
162
163 let dom_restricted = !expr.days_of_month.starred;
164 let dow_restricted = !expr.days_of_week.starred;
165
166 for day_offset in 0..CRON_HORIZON_DAYS {
167 let date = start_minute.date() + chrono::Duration::days(day_offset);
168 if !expr.months.contains(date.month()) {
169 continue;
170 }
171 let dom_match = expr.days_of_month.contains(date.day());
172 // 0=Sunday … 6=Saturday, matching the cron numbering.
173 let dow_match = expr
174 .days_of_week
175 .contains(date.weekday().num_days_from_sunday());
176 let day_matches = match (dom_restricted, dow_restricted) {
177 // Vixie rule: when both day fields are restricted, EITHER may
178 // fire; otherwise every listed field must match.
179 (true, true) => dom_match || dow_match,
180 _ => dom_match && dow_match,
181 };
182 if !day_matches {
183 continue;
184 }
185 for h in &expr.hours.values {
186 for m in &expr.minutes.values {
187 let candidate = date.and_hms_opt(*h, *m, 0)?;
188 if candidate < start_minute {
189 continue;
190 }
191 // Gap → None → skip forward to the next existing match;
192 // fold → earliest instant wins.
193 if let Some(dt) = tz.from_local_datetime(&candidate).earliest() {
194 return Some(dt);
195 }
196 }
197 }
198 }
199 None
200}
201
202pub fn tasks_file(sessions_home: &Path) -> PathBuf {
203 sessions_home.join("tasks.json")
204}
205
206/// Worktree metadata recorded with the last run. Shape-compatible twin of
207/// the server's WtMeta so old tasks.json round-trips without data loss.
208#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
209pub struct WtMeta {
210 pub path: PathBuf,
211 pub branch: String,
212}
213
214#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
215pub struct TaskDef {
216 pub id: String,
217 pub name: String,
218 #[serde(default)]
219 pub prompt: String,
220 #[serde(default = "default_interval")]
221 pub interval_secs: u64,
222 pub enabled: bool,
223 pub cwd: PathBuf,
224 pub created_at: DateTime<Utc>,
225 pub last_run_at: Option<DateTime<Utc>>,
226 pub last_session_id: Option<String>,
227 pub last_summary: Option<String>,
228 #[serde(default)]
229 pub last_result_id: Option<String>,
230 #[serde(default)]
231 pub last_run_status: Option<String>,
232 #[serde(default)]
233 pub last_delivery_state: Option<String>,
234 #[serde(default)]
235 pub last_wt: Option<WtMeta>,
236 #[serde(default)]
237 pub deliver_to: Option<String>,
238 /// 5-field cron (`m h dom mon dow`, local time) replacing interval
239 /// ticking when present. Validated against the cron grammar.
240 #[serde(default)]
241 pub schedule: Option<String>,
242 /// Optional IANA timezone name for recurring schedules. When absent,
243 /// legacy schedules retain system-local behavior.
244 #[serde(default)]
245 pub timezone: Option<String>,
246 /// Optional one-shot instant. When present, it takes precedence over
247 /// interval/cron scheduling and is consumed after the first run.
248 #[serde(default)]
249 pub due_at: Option<DateTime<Utc>>,
250 /// Watchdog shell one-liner. XOR with `prompt`: script tasks run
251 /// brokered bash and cost zero tokens when stdout stays empty.
252 #[serde(default)]
253 pub script: Option<String>,
254 /// Pinned model id; a pinned task dispatches only this model and
255 /// never escalates.
256 #[serde(default)]
257 pub model_pin: Option<String>,
258 /// Agent identity selected when this task was created.
259 #[serde(default)]
260 pub agent_id: Option<String>,
261 #[serde(default)]
262 pub agent_revision: Option<u64>,
263}
264
265fn default_interval() -> u64 {
266 3600
267}
268
269#[derive(Debug, thiserror::Error)]
270pub enum TaskError {
271 #[error("task '{name}': exactly one of `prompt` or `script` is required")]
272 PromptScriptXor { name: String },
273 #[error("invalid schedule '{expr}': {reason}")]
274 BadSchedule { expr: String, reason: String },
275 #[error("io error on {path}: {source}")]
276 Io {
277 path: PathBuf,
278 source: std::io::Error,
279 },
280 #[error("corrupt tasks file {path}: {source}")]
281 Json {
282 path: PathBuf,
283 source: serde_json::Error,
284 },
285}
286
287impl TaskDef {
288 /// Structural validation: prompt XOR script required, cron parsed when
289 /// scheduled. Pure; never touches the store.
290 pub fn validate(&self) -> Result<(), TaskError> {
291 let has_prompt = !self.prompt.trim().is_empty();
292 let has_script = self.script.as_deref().is_some_and(|s| !s.trim().is_empty());
293 if has_prompt == has_script {
294 return Err(TaskError::PromptScriptXor {
295 name: self.name.clone(),
296 });
297 }
298 if let Some(expr) = &self.schedule {
299 CronExpr::parse(expr).map_err(|reason| TaskError::BadSchedule {
300 expr: expr.clone(),
301 reason,
302 })?;
303 }
304 if self.due_at.is_some() && self.schedule.is_some() {
305 return Err(TaskError::BadSchedule {
306 expr: self.schedule.clone().unwrap_or_default(),
307 reason: "one-shot due_at cannot be combined with cron schedule".into(),
308 });
309 }
310 if let Some(zone) = self.timezone.as_deref() {
311 if zone.trim().is_empty() {
312 return Err(TaskError::BadSchedule {
313 expr: zone.into(),
314 reason: "timezone must be a named IANA zone".into(),
315 });
316 }
317 if zone.parse::<chrono_tz::Tz>().is_err() {
318 return Err(TaskError::BadSchedule {
319 expr: zone.into(),
320 reason: "unknown IANA timezone".into(),
321 });
322 }
323 }
324 Ok(())
325 }
326}
327
328/// fsyncs a directory so a prior rename into it is durable across a crash.
329/// On non-unix platforms directory fsync isn't a thing; the rename itself
330/// is still atomic there, so this is a no-op rather than an error.
331#[cfg(unix)]
332fn sync_directory(path: &Path) -> std::io::Result<()> {
333 std::fs::File::open(path).and_then(|dir| dir.sync_all())
334}
335
336#[cfg(not(unix))]
337fn sync_directory(_path: &Path) -> std::io::Result<()> {
338 Ok(())
339}
340
341/// In-process mirror of the persisted tasks.json array, keyed by id.
342pub struct TaskStore {
343 path: PathBuf,
344 tasks: HashMap<String, TaskDef>,
345}
346
347impl TaskStore {
348 /// Missing file loads as an empty store; a corrupt file is a typed
349 /// error, never silently dropped work.
350 pub fn load(sessions_home: &Path) -> Result<Self, TaskError> {
351 let path = tasks_file(sessions_home);
352 let raw = match std::fs::read_to_string(&path) {
353 Ok(raw) => raw,
354 Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
355 return Ok(TaskStore {
356 path,
357 tasks: HashMap::new(),
358 });
359 }
360 Err(source) => {
361 return Err(TaskError::Io {
362 path: path.clone(),
363 source,
364 });
365 }
366 };
367 let list: Vec<TaskDef> = serde_json::from_str(&raw).map_err(|source| TaskError::Json {
368 path: path.clone(),
369 source,
370 })?;
371 Ok(TaskStore {
372 path,
373 tasks: list.into_iter().map(|t| (t.id.clone(), t)).collect(),
374 })
375 }
376
377 pub fn save(&self) -> Result<(), TaskError> {
378 if let Some(parent) = self.path.parent() {
379 std::fs::create_dir_all(parent).map_err(|source| TaskError::Io {
380 path: parent.to_path_buf(),
381 source,
382 })?;
383 }
384 let mut list: Vec<TaskDef> = self.tasks.values().cloned().collect();
385 list.sort_by_key(|t| t.created_at);
386 let json = serde_json::to_string_pretty(&list).map_err(|source| TaskError::Json {
387 path: self.path.clone(),
388 source,
389 })?;
390 // Atomic + durable write: a plain `fs::write` truncates the file
391 // before the new bytes land, so a crash or power loss mid-write
392 // leaves `tasks.json` corrupt and unrecoverable. Write to a sibling
393 // temp file, fsync its contents, rename over the real path, then
394 // fsync the containing directory — on all platforms this crate
395 // targets, `rename` onto an existing path is atomic, so readers
396 // (this store, the server, the desktop app) only ever see the
397 // fully-old or fully-new content, never a partial write; the fsyncs
398 // ensure that content and the rename itself survive a crash right
399 // after this call returns, not just torn-write-free while running.
400 let tmp_path = self.path.with_extension("json.tmp");
401 {
402 let mut file = std::fs::File::create(&tmp_path).map_err(|source| TaskError::Io {
403 path: tmp_path.clone(),
404 source,
405 })?;
406 file.write_all(json.as_bytes())
407 .map_err(|source| TaskError::Io {
408 path: tmp_path.clone(),
409 source,
410 })?;
411 file.sync_all().map_err(|source| TaskError::Io {
412 path: tmp_path.clone(),
413 source,
414 })?;
415 }
416 std::fs::rename(&tmp_path, &self.path).map_err(|source| TaskError::Io {
417 path: self.path.clone(),
418 source,
419 })?;
420 if let Some(parent) = self.path.parent() {
421 sync_directory(parent).map_err(|source| TaskError::Io {
422 path: parent.to_path_buf(),
423 source,
424 })?;
425 }
426 Ok(())
427 }
428
429 pub fn get(&self, id: &str) -> Option<&TaskDef> {
430 self.tasks.get(id)
431 }
432
433 /// Inserts or replaces; returns the previous definition when replacing.
434 pub fn put(&mut self, task: TaskDef) -> Option<TaskDef> {
435 self.tasks.insert(task.id.clone(), task)
436 }
437
438 pub fn remove(&mut self, id: &str) -> bool {
439 self.tasks.remove(id).is_some()
440 }
441
442 pub fn len(&self) -> usize {
443 self.tasks.len()
444 }
445
446 #[must_use]
447 pub fn is_empty(&self) -> bool {
448 self.tasks.is_empty()
449 }
450
451 /// Every task, oldest first (created_at order, like the server API).
452 pub fn all(&self) -> Vec<TaskDef> {
453 let mut list: Vec<TaskDef> = self.tasks.values().cloned().collect();
454 list.sort_by_key(|t| t.created_at);
455 list
456 }
457
458 pub fn for_cwd(&self, cwd: &Path) -> Vec<TaskDef> {
459 self.all().into_iter().filter(|t| t.cwd == cwd).collect()
460 }
461}
462
463#[cfg(test)]
464mod tests {
465 #![allow(clippy::unwrap_used, clippy::expect_used)]
466 use super::*;
467 use chrono::{FixedOffset, LocalResult, NaiveDate, NaiveDateTime, Offset, TimeZone};
468
469 fn local(y: i32, mo: u32, d: u32, h: u32, mi: u32) -> DateTime<chrono::Local> {
470 chrono::Local
471 .with_ymd_and_hms(y, mo, d, h, mi, 0)
472 .earliest()
473 .unwrap()
474 }
475
476 fn next(expr: &str, after: DateTime<chrono::Local>) -> DateTime<chrono::Local> {
477 cron_next_after(expr, after).unwrap()
478 }
479
480 // ---- cron grammar matrix -------------------------------------------
481
482 #[test]
483 fn wildcard_every_step_minutes() {
484 assert_eq!(
485 next("*/15 * * * *", local(2026, 8, 24, 0, 0)),
486 local(2026, 8, 24, 0, 15)
487 );
488 assert_eq!(
489 next("*/15 * * * *", local(2026, 8, 24, 0, 16)),
490 local(2026, 8, 24, 0, 30)
491 );
492 }
493
494 #[test]
495 fn wildcard_every_minute_is_strictly_after() {
496 assert_eq!(
497 next("* * * * *", local(2026, 8, 24, 12, 34)),
498 local(2026, 8, 24, 12, 35)
499 );
500 }
501
502 #[test]
503 fn ranges_and_lists_combine() {
504 // 04:05 on any weekday, Mon-Fri.
505 assert_eq!(
506 next("5 4 * * 1-5", local(2026, 8, 22, 10, 0)),
507 local(2026, 8, 24, 4, 5)
508 );
509 assert_eq!(
510 next("0 9-17 * * *", local(2026, 8, 24, 18, 0)),
511 local(2026, 8, 25, 9, 0)
512 );
513 assert_eq!(
514 next("0 0 1,15 * *", local(2026, 1, 16, 0, 0)),
515 local(2026, 2, 1, 0, 0)
516 );
517 // Range with step.
518 assert_eq!(
519 next("0 0-23/6 * * *", local(2026, 8, 24, 7, 30)),
520 local(2026, 8, 24, 12, 0)
521 );
522 }
523
524 #[test]
525 fn month_rollover_includes_feb_29() {
526 assert_eq!(
527 next("0 0 29 2 *", local(2026, 3, 1, 0, 0)),
528 local(2028, 2, 29, 0, 0)
529 );
530 assert_eq!(
531 next("0 0 29 2 *", local(2028, 3, 1, 0, 0)),
532 local(2032, 2, 29, 0, 0)
533 );
534 }
535
536 #[test]
537 fn dom_and_dow_use_vixie_or_semantics_when_both_restricted() {
538 // Friday the 13th OR "any Friday" — either restricted field fires.
539 // 2026-01-14 is a Wednesday; the first Friday after is Jan 16.
540 assert_eq!(
541 next("0 12 13 * 5", local(2026, 1, 14, 0, 0)),
542 local(2026, 1, 16, 12, 0)
543 );
544 // Fridays keep firing even when the 13th doesn't match…
545 assert_eq!(
546 next("0 12 13 * 5", local(2026, 1, 17, 0, 0)),
547 local(2026, 1, 23, 12, 0)
548 );
549 // …and Friday Feb 13 matches BOTH restricted fields at once.
550 assert_eq!(
551 next("0 12 13 * 5", local(2026, 2, 12, 0, 0)),
552 local(2026, 2, 13, 12, 0)
553 );
554 }
555
556 #[test]
557 fn dom_alone_is_conjunctive_with_unrestricted_fields() {
558 assert_eq!(
559 next("0 12 13 * *", local(2026, 1, 14, 0, 0)),
560 local(2026, 2, 13, 12, 0)
561 );
562 }
563
564 #[test]
565 fn dow_seven_maps_to_sunday() {
566 // 2026-08-22 is a Saturday.
567 assert_eq!(
568 next("0 12 * * 7", local(2026, 8, 22, 0, 0)),
569 local(2026, 8, 23, 12, 0)
570 );
571 assert_eq!(
572 next("0 12 * * 0", local(2026, 8, 22, 0, 0)),
573 local(2026, 8, 23, 12, 0)
574 );
575 }
576
577 #[test]
578 fn parse_errors_are_typed_strings() {
579 for bad in [
580 "* * * *",
581 "* * * * * *",
582 "",
583 "60 * * * *",
584 "-1 * * * *",
585 "* 24 * * *",
586 "* * 0 * *",
587 "* * * 13 *",
588 "* * * * 8",
589 "*/0 * * * *",
590 "5-1 * * * *",
591 "1,,2 * * * *",
592 "abc * * * *",
593 "5/x * * * *",
594 "5/10 * * * *",
595 ] {
596 assert!(
597 CronExpr::parse(bad).is_err(),
598 "expected '{bad}' to be rejected"
599 );
600 }
601 }
602
603 #[test]
604 fn unreachable_date_hits_horizon_error() {
605 // Feb 31 does not exist in any year.
606 let err = cron_next_after("0 0 31 2 *", local(2026, 1, 1, 0, 0)).unwrap_err();
607 assert!(err.contains("never fires"), "{err}");
608 }
609
610 /// Synthetic zone modeling a spring-forward gap (local 02:00–02:59 of
611 /// 2026-03-08 do not exist) and a fall-back fold (local 01:00–01:59 of
612 /// 2026-11-01 occur twice), so gap/fold behavior is testable without
613 /// touching process TZ state.
614 #[derive(Debug, Clone, Copy)]
615 struct GapTz;
616
617 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
618 struct GapOffset(FixedOffset);
619
620 const EAST: FixedOffset = FixedOffset::east_opt(3600).unwrap();
621 const WEST: FixedOffset = FixedOffset::east_opt(0).unwrap();
622
623 impl TimeZone for GapTz {
624 type Offset = GapOffset;
625 fn from_offset(_offset: &GapOffset) -> Self {
626 GapTz
627 }
628 fn offset_from_utc_date(&self, _utc: &chrono::NaiveDate) -> GapOffset {
629 GapOffset(EAST)
630 }
631 fn offset_from_utc_datetime(&self, _utc: &NaiveDateTime) -> GapOffset {
632 GapOffset(EAST)
633 }
634 fn offset_from_local_date(&self, local: &chrono::NaiveDate) -> LocalResult<GapOffset> {
635 // A date inherits its gap/fold status from some instant on it.
636 let probe = local.and_hms_opt(12, 0, 0).unwrap();
637 self.offset_from_local_datetime(&probe)
638 }
639 fn offset_from_local_datetime(&self, local: &NaiveDateTime) -> LocalResult<GapOffset> {
640 use chrono::NaiveDate;
641 let gap_day = NaiveDate::from_ymd_opt(2026, 3, 8).unwrap();
642 let fold_day = NaiveDate::from_ymd_opt(2026, 11, 1).unwrap();
643 let t = local.time();
644 if local.date() == gap_day && t.hour() == 2 {
645 return LocalResult::None;
646 }
647 if local.date() == fold_day && t.hour() == 1 {
648 // (earliest, latest): the larger offset yields the earlier
649 // instant, so the pre-fold EAST pass comes first.
650 return LocalResult::Ambiguous(GapOffset(EAST), GapOffset(WEST));
651 }
652 LocalResult::Single(GapOffset(EAST))
653 }
654 }
655
656 impl chrono::Offset for GapOffset {
657 fn fix(&self) -> FixedOffset {
658 self.0
659 }
660 }
661
662 fn gap_next(expr: &str, after: NaiveDateTime) -> Option<DateTime<GapTz>> {
663 next_fire(
664 &CronExpr::parse(expr).unwrap(),
665 GapTz.from_utc_datetime(&after),
666 )
667 }
668
669 #[test]
670 fn dst_gap_skips_forward_to_first_existing_match() {
671 let before_gap = NaiveDate::from_ymd_opt(2026, 3, 8)
672 .unwrap()
673 .and_hms_opt(1, 30, 0)
674 .unwrap();
675 // Hourly job: 02:00 does not exist → fires at 03:00.
676 let fired = gap_next("0 * * * *", before_gap).unwrap();
677 assert_eq!(
678 fired.naive_local(),
679 NaiveDate::from_ymd_opt(2026, 3, 8)
680 .unwrap()
681 .and_hms_opt(3, 0, 0)
682 .unwrap()
683 );
684 // Daily-at-02:30 job: today's 02:30 doesn't exist → tomorrow's.
685 let daily = gap_next("30 2 * * *", before_gap).unwrap();
686 assert_eq!(
687 daily.naive_local(),
688 NaiveDate::from_ymd_opt(2026, 3, 9)
689 .unwrap()
690 .and_hms_opt(2, 30, 0)
691 .unwrap()
692 );
693 }
694
695 #[test]
696 fn dst_fold_resolves_to_earliest_instant() {
697 let before_fold = NaiveDate::from_ymd_opt(2026, 11, 1)
698 .unwrap()
699 .and_hms_opt(0, 30, 0)
700 .unwrap();
701 let fired = gap_next("45 1 * * *", before_fold).unwrap();
702 assert_eq!(
703 fired.naive_local(),
704 NaiveDate::from_ymd_opt(2026, 11, 1)
705 .unwrap()
706 .and_hms_opt(1, 45, 0)
707 .unwrap()
708 );
709 // Earliest pass carries the pre-fold offset (EAST here).
710 assert_eq!(fired.offset().fix(), EAST);
711 }
712
713 // ---- task validation + store ----------------------------------------
714
715 fn base_task() -> TaskDef {
716 TaskDef {
717 id: uuid::Uuid::now_v7().to_string(),
718 name: "nightly".into(),
719 prompt: "tidy the repo".into(),
720 interval_secs: 3600,
721 enabled: true,
722 cwd: PathBuf::from("/tmp/ws"),
723 created_at: Utc::now(),
724 last_run_at: None,
725 last_session_id: None,
726 last_summary: None,
727 last_result_id: None,
728 last_run_status: None,
729 last_delivery_state: None,
730 last_wt: None,
731 deliver_to: None,
732 schedule: None,
733 timezone: None,
734 due_at: None,
735 script: None,
736 model_pin: None,
737 agent_id: None,
738 agent_revision: None,
739 }
740 }
741
742 #[test]
743 fn prompt_xor_script_validation_matrix() {
744 let mut both = base_task();
745 both.script = Some("echo tick".into());
746 assert!(matches!(
747 both.validate(),
748 Err(TaskError::PromptScriptXor { .. })
749 ));
750
751 let neither = TaskDef {
752 prompt: " ".into(),
753 ..base_task()
754 };
755 assert!(matches!(
756 neither.validate(),
757 Err(TaskError::PromptScriptXor { .. })
758 ));
759
760 assert!(base_task().validate().is_ok());
761 let watchdog = TaskDef {
762 prompt: String::new(),
763 script: Some("curl -sf http://x/health".into()),
764 ..base_task()
765 };
766 assert!(watchdog.validate().is_ok());
767 }
768
769 #[test]
770 fn one_shot_and_timezone_validation_is_explicit() {
771 let mut task = base_task();
772 task.due_at = Some(Utc::now());
773 task.schedule = Some("0 9 * * *".into());
774 assert!(matches!(
775 task.validate(),
776 Err(TaskError::BadSchedule { .. })
777 ));
778 task.schedule = None;
779 task.timezone = Some(" ".into());
780 assert!(matches!(
781 task.validate(),
782 Err(TaskError::BadSchedule { .. })
783 ));
784 }
785
786 #[test]
787 fn named_timezone_cron_returns_utc_instant() {
788 let after = chrono::DateTime::parse_from_rfc3339("2026-01-01T12:00:00Z")
789 .expect("timestamp")
790 .with_timezone(&Utc);
791 let next = cron_next_after_timezone("0 9 * * *", after, "America/New_York")
792 .expect("next named-zone fire");
793 assert_eq!(next.to_rfc3339(), "2026-01-01T14:00:00+00:00");
794 }
795
796 #[test]
797 fn bad_cron_rejected_by_validate() {
798 let mut t = base_task();
799 t.schedule = Some("99 * * * *".into());
800 let err = t.validate().unwrap_err();
801 assert!(
802 matches!(err, TaskError::BadSchedule { ref expr, .. } if expr == "99 * * * *"),
803 "{err}"
804 );
805 t.schedule = Some("*/15 * * * *".into());
806 assert!(t.validate().is_ok());
807 }
808
809 #[test]
810 fn a_tasks_file_written_before_added_fields_loads_and_roundtrips() {
811 let dir = tempfile::tempdir().unwrap();
812 let raw = r#"[
813 {
814 "id": "018f0000-0000-7000-8000-000000000001",
815 "name": "standup-notes",
816 "prompt": "summarize yesterday",
817 "interval_secs": 86400,
818 "enabled": true,
819 "cwd": "/Users/me/proj",
820 "created_at": "2026-07-01T09:00:00Z",
821 "last_run_at": "2026-08-01T09:00:00Z",
822 "last_session_id": "abc",
823 "last_summary": "done",
824 "last_result_id": "result-1",
825 "last_wt": { "path": "/tmp/wt", "branch": "vak/abc" },
826 "deliver_to": "log:ops"
827 }
828 ]"#;
829 std::fs::create_dir_all(dir.path()).unwrap();
830 std::fs::write(tasks_file(dir.path()), raw).unwrap();
831
832 let store = TaskStore::load(dir.path()).unwrap();
833 assert_eq!(store.len(), 1);
834 let t = store.get("018f0000-0000-7000-8000-000000000001").unwrap();
835 assert_eq!(t.prompt, "summarize yesterday");
836 assert_eq!(t.interval_secs, 86400);
837 assert_eq!(t.last_wt.as_ref().unwrap().branch, "vak/abc");
838 assert_eq!(t.deliver_to.as_deref(), Some("log:ops"));
839 assert_eq!(t.schedule, None);
840 assert_eq!(t.script, None);
841 assert_eq!(t.model_pin, None);
842
843 store.save().unwrap();
844 let reloaded = TaskStore::load(dir.path()).unwrap();
845 assert_eq!(reloaded.all().len(), 1);
846 assert_eq!(
847 reloaded.get(t.id.as_str()).unwrap().last_wt,
848 t.last_wt,
849 "legacy last_wt must survive a load/save cycle"
850 );
851 }
852
853 #[test]
854 fn new_fields_roundtrip_through_store() {
855 let dir = tempfile::tempdir().unwrap();
856 let mut store = TaskStore::load(dir.path()).unwrap();
857 assert!(store.is_empty());
858
859 let mut t = base_task();
860 t.prompt = String::new();
861 t.script = Some("systemctl is-active nginx".into());
862 t.schedule = Some("0 7 * * 1-5".into());
863 t.model_pin = Some("haiku-fast".into());
864 store.put(t.clone());
865 store.save().unwrap();
866
867 let back = TaskStore::load(dir.path()).unwrap();
868 let got = back.get(&t.id).unwrap();
869 assert_eq!(got.script.as_deref(), Some("systemctl is-active nginx"));
870 assert_eq!(got.schedule.as_deref(), Some("0 7 * * 1-5"));
871 assert_eq!(got.model_pin.as_deref(), Some("haiku-fast"));
872 assert!(got.validate().is_ok());
873 }
874
875 #[test]
876 fn missing_file_is_empty_store_and_save_creates_it() {
877 let dir = tempfile::tempdir().unwrap();
878 let mut store = TaskStore::load(dir.path()).unwrap();
879 assert!(store.is_empty());
880 store.put(base_task());
881 store.save().unwrap();
882 assert!(tasks_file(dir.path()).is_file());
883 }
884
885 #[test]
886 fn save_is_atomic_and_leaves_no_tmp_file_behind() {
887 let dir = tempfile::tempdir().unwrap();
888 let mut store = TaskStore::load(dir.path()).unwrap();
889 store.put(base_task());
890 store.save().unwrap();
891 assert!(tasks_file(dir.path()).is_file());
892 assert!(
893 !dir.path().join("tasks.json.tmp").exists(),
894 "the temp file used for the atomic rename must not survive a successful save"
895 );
896
897 // A second save (overwrite path) must round-trip cleanly too, and
898 // still leave no tmp file — this is the path a real crash-mid-write
899 // would otherwise corrupt with a plain `fs::write`.
900 store.put(base_task());
901 store.save().unwrap();
902 let reloaded = TaskStore::load(dir.path()).unwrap();
903 assert_eq!(reloaded.tasks.len(), 2);
904 assert!(!dir.path().join("tasks.json.tmp").exists());
905 }
906
907 #[test]
908 fn corrupt_file_is_a_typed_error_not_silence() {
909 let dir = tempfile::tempdir().unwrap();
910 std::fs::write(tasks_file(dir.path()), "{not json").unwrap();
911 assert!(matches!(
912 TaskStore::load(dir.path()),
913 Err(TaskError::Json { .. })
914 ));
915 }
916
917 #[test]
918 fn for_cwd_filters_and_sorts_by_created_at() {
919 let ws_a = PathBuf::from("/ws/a");
920 let ws_b = PathBuf::from("/ws/b");
921 let mut early = base_task();
922 early.id = "early".into();
923 early.cwd = ws_a.clone();
924 early.created_at = Utc::now() - chrono::Duration::hours(2);
925 let mut late = early.clone();
926 late.id = "late".into();
927 late.created_at = Utc::now();
928 let mut other = base_task();
929 other.id = "other".into();
930 other.cwd = ws_b;
931
932 let mut map = HashMap::new();
933 for t in [early, late, other] {
934 map.insert(t.id.clone(), t);
935 }
936 let store = TaskStore {
937 path: PathBuf::from("unused"),
938 tasks: map,
939 };
940 let mine = store.for_cwd(&ws_a);
941 assert_eq!(mine.len(), 2);
942 assert_eq!(mine[0].id, "early", "oldest first");
943 assert_eq!(mine[1].id, "late");
944 }
945}
946

1//! Scheduled-task store (docs/design/29-personal-os.md P2). Same
2//! `tasks.json` shape the server surface wrote — legacy files load
3//! unchanged — plus additive optional fields: 5-field cron schedules,
4//! watchdog shell one-liners (XOR with prompt), and per-task model pins.
5//! The store is plain JSON under `<sessions-home>/tasks.json`; sessions
6//! themselves stay append-only ledgers.
7
8use std::collections::{BTreeSet, HashMap};
9use std::io::Write;
10use std::path::{Path, PathBuf};
11
12use chrono::{DateTime, Datelike, TimeZone, Timelike, Utc};
13
14/// How far ahead [`cron_next_after`] searches before giving up. Cron can
15/// legitimately sleep ~4 years (Feb 29); anything beyond five is a bug in
16/// the expression, not a schedule.
17const CRON_HORIZON_DAYS: i64 = 366 * 5;
18
19#[derive(Debug, Clone, PartialEq)]
20struct FieldSet {
21 values: BTreeSet<u32>,
22 /// True only for a literal `*` field — vixie-cron's dom/dow OR rule
23 /// keys on restriction, not on which values ended up selected.
24 starred: bool,
25}
26
27impl FieldSet {
28 fn contains(&self, v: u32) -> bool {
29 self.starred || self.values.contains(&v)
30 }
31}
32
33#[derive(Debug, Clone, PartialEq)]
34pub struct CronExpr {
35 minutes: FieldSet,
36 hours: FieldSet,
37 days_of_month: FieldSet,
38 months: FieldSet,
39 days_of_week: FieldSet,
40}
41
42impl CronExpr {
43 pub fn parse(expr: &str) -> Result<Self, String> {
44 let fields: Vec<&str> = expr.split_whitespace().collect();
45 if fields.len() != 5 {
46 return Err(format!(
47 "expected 5 fields (min hour dom mon dow), got {}",
48 fields.len()
49 ));
50 }
51 Ok(CronExpr {
52 minutes: parse_field(fields[0], 0, 59, false)?,
53 hours: parse_field(fields[1], 0, 23, false)?,
54 days_of_month: parse_field(fields[2], 1, 31, false)?,
55 months: parse_field(fields[3], 1, 12, false)?,
56 days_of_week: parse_field(fields[4], 0, 7, true)?,
57 })
58 }
59}
60
61fn parse_field(spec: &str, min: u32, max: u32, dow_wrap: bool) -> Result<FieldSet, String> {
62 let mut values = BTreeSet::new();
63 let mut starred = false;
64 // Vixie convention: 0 and 7 are both Sunday; mapping happens AFTER
65 // range expansion so `5-7` means Fri–Sun rather than a reversed range.
66 let wrap = |v: u32| if dow_wrap && v == 7 { 0 } else { v };
67 for term in spec.split(',') {
68 let term = term.trim();
69 if term.is_empty() {
70 return Err(format!("empty list element in '{spec}'"));
71 }
72 let (base, step) = match term.split_once('/') {
73 Some((b, s)) => {
74 let step: u32 = s.parse().map_err(|_| format!("bad step '{s}'"))?;
75 if step == 0 {
76 return Err(format!("step must be >= 1 in '{term}'"));
77 }
78 (b, Some(step))
79 }
80 None => (term, None),
81 };
82 let range: Vec<u32> = match base {
83 "*" => {
84 if term == "*" && step.is_none() {
85 starred = true;
86 }
87 (min..=max).collect()
88 }
89 b if b.contains('-') => {
90 let (lo, hi) = b
91 .split_once('-')
92 .ok_or_else(|| format!("bad range '{b}'"))?;
93 let lo = parse_number(lo, min, max)?;
94 let hi = parse_number(hi, min, max)?;
95 if hi < lo {
96 return Err(format!("reversed range '{b}'"));
97 }
98 (lo..=hi).map(wrap).collect()
99 }
100 b => {
101 if step.is_some() {
102 return Err(format!("step requires '*' or a range in '{term}'"));
103 }
104 vec![wrap(parse_number(b, min, max)?)]
105 }
106 };
107 match step {
108 None => values.extend(range),
109 Some(s) => values.extend(range.into_iter().step_by(s as usize)),
110 }
111 }
112 Ok(FieldSet { values, starred })
113}
114
115fn parse_number(raw: &str, min: u32, max: u32) -> Result<u32, String> {
116 let v: u32 = raw
117 .trim()
118 .parse()
119 .map_err(|_| format!("bad number '{raw}'"))?;
120 if !(min..=max).contains(&v) {
121 return Err(format!("value {v} out of range {min}-{max}"));
122 }
123 Ok(v)
124}
125
126/// Next local time strictly after `after` where the expression matches.
127///
128/// DST semantics are deterministic by construction: candidate wall-clock
129/// times that do not exist (spring-forward gap) never fire — scanning
130/// continues to the next matching time that DOES exist; ambiguous times
131/// (fall-back fold) resolve to the earliest instant. Pure: no clock reads.
132pub fn cron_next_after(
133 expr: &str,
134 after: chrono::DateTime<chrono::Local>,
135) -> Result<DateTime<chrono::Local>, String> {
136 let parsed = CronExpr::parse(expr)?;
137 next_fire(&parsed, after)
138 .ok_or_else(|| format!("expression '{expr}' never fires within {CRON_HORIZON_DAYS} days"))
139}
140
141/// Named-IANA equivalent of [`cron_next_after`].
142pub fn cron_next_after_timezone(
143 expr: &str,
144 after: DateTime<Utc>,
145 timezone: &str,
146) -> Result<DateTime<Utc>, String> {
147 let tz: chrono_tz::Tz = timezone
148 .parse()
149 .map_err(|_| format!("unknown IANA timezone '{timezone}'"))?;
150 let local_after = after.with_timezone(&tz);
151 next_fire(&CronExpr::parse(expr)?, local_after)
152 .map(|value| value.with_timezone(&Utc))
153 .ok_or_else(|| format!("expression '{expr}' never fires within {CRON_HORIZON_DAYS} days"))
154}
155
156fn next_fire<Tz: TimeZone>(expr: &CronExpr, after: DateTime<Tz>) -> Option<DateTime<Tz>> {
157 let tz = after.timezone();
158 let naive = after.naive_local();
159 // Strictly-after: start from the minute boundary following `after`.
160 let start_minute =
161 naive.date().and_hms_opt(naive.hour(), naive.minute(), 0)? + chrono::Duration::minutes(1);
162
163 let dom_restricted = !expr.days_of_month.starred;
164 let dow_restricted = !expr.days_of_week.starred;
165
166 for day_offset in 0..CRON_HORIZON_DAYS {
167 let date = start_minute.date() + chrono::Duration::days(day_offset);
168 if !expr.months.contains(date.month()) {
169 continue;
170 }
171 let dom_match = expr.days_of_month.contains(date.day());
172 // 0=Sunday … 6=Saturday, matching the cron numbering.
173 let dow_match = expr
174 .days_of_week
175 .contains(date.weekday().num_days_from_sunday());
176 let day_matches = match (dom_restricted, dow_restricted) {
177 // Vixie rule: when both day fields are restricted, EITHER may
178 // fire; otherwise every listed field must match.
179 (true, true) => dom_match || dow_match,
180 _ => dom_match && dow_match,
181 };
182 if !day_matches {
183 continue;
184 }
185 for h in &expr.hours.values {
186 for m in &expr.minutes.values {
187 let candidate = date.and_hms_opt(*h, *m, 0)?;
188 if candidate < start_minute {
189 continue;
190 }
191 // Gap → None → skip forward to the next existing match;
192 // fold → earliest instant wins.
193 if let Some(dt) = tz.from_local_datetime(&candidate).earliest() {
194 return Some(dt);
195 }
196 }
197 }
198 }
199 None
200}
201
202pub fn tasks_file(sessions_home: &Path) -> PathBuf {
203 sessions_home.join("tasks.json")
204}
205
206/// Worktree metadata recorded with the last run. Shape-compatible twin of
207/// the server's WtMeta so old tasks.json round-trips without data loss.
208#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
209pub struct WtMeta {
210 pub path: PathBuf,
211 pub branch: String,
212}
213
214#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
215pub struct TaskDef {
216 pub id: String,
217 pub name: String,
218 #[serde(default)]
219 pub prompt: String,
220 #[serde(default = "default_interval")]
221 pub interval_secs: u64,
222 pub enabled: bool,
223 pub cwd: PathBuf,
224 pub created_at: DateTime<Utc>,
225 pub last_run_at: Option<DateTime<Utc>>,
226 pub last_session_id: Option<String>,
227 pub last_summary: Option<String>,
228 #[serde(default)]
229 pub last_result_id: Option<String>,
230 #[serde(default)]
231 pub last_run_status: Option<String>,
232 #[serde(default)]
233 pub last_delivery_state: Option<String>,
234 #[serde(default)]
235 pub last_wt: Option<WtMeta>,
236 #[serde(default)]
237 pub deliver_to: Option<String>,
238 /// 5-field cron (`m h dom mon dow`, local time) replacing interval
239 /// ticking when present. Validated against the cron grammar.
240 #[serde(default)]
241 pub schedule: Option<String>,
242 /// Optional IANA timezone name for recurring schedules. When absent,
243 /// legacy schedules retain system-local behavior.
244 #[serde(default)]
245 pub timezone: Option<String>,
246 /// Optional one-shot instant. When present, it takes precedence over
247 /// interval/cron scheduling and is consumed after the first run.
248 #[serde(default)]
249 pub due_at: Option<DateTime<Utc>>,
250 /// Watchdog shell one-liner. XOR with `prompt`: script tasks run
251 /// brokered bash and cost zero tokens when stdout stays empty.
252 #[serde(default)]
253 pub script: Option<String>,
254 /// Pinned model id; a pinned task dispatches only this model and
255 /// never escalates.
256 #[serde(default)]
257 pub model_pin: Option<String>,
258 /// Agent identity selected when this task was created.
259 #[serde(default)]
260 pub agent_id: Option<String>,
261 #[serde(default)]
262 pub agent_revision: Option<u64>,
263}
264
265fn default_interval() -> u64 {
266 3600
267}
268
269#[derive(Debug, thiserror::Error)]
270pub enum TaskError {
271 #[error("task '{name}': exactly one of `prompt` or `script` is required")]
272 PromptScriptXor { name: String },
273 #[error("invalid schedule '{expr}': {reason}")]
274 BadSchedule { expr: String, reason: String },
275 #[error("io error on {path}: {source}")]
276 Io {
277 path: PathBuf,
278 source: std::io::Error,
279 },
280 #[error("corrupt tasks file {path}: {source}")]
281 Json {
282 path: PathBuf,
283 source: serde_json::Error,
284 },
285}
286
287impl TaskDef {
288 /// Structural validation: prompt XOR script required, cron parsed when
289 /// scheduled. Pure; never touches the store.
290 pub fn validate(&self) -> Result<(), TaskError> {
291 let has_prompt = !self.prompt.trim().is_empty();
292 let has_script = self.script.as_deref().is_some_and(|s| !s.trim().is_empty());
293 if has_prompt == has_script {
294 return Err(TaskError::PromptScriptXor {
295 name: self.name.clone(),
296 });
297 }
298 if let Some(expr) = &self.schedule {
299 CronExpr::parse(expr).map_err(|reason| TaskError::BadSchedule {
300 expr: expr.clone(),
301 reason,
302 })?;
303 }
304 if self.due_at.is_some() && self.schedule.is_some() {
305 return Err(TaskError::BadSchedule {
306 expr: self.schedule.clone().unwrap_or_default(),
307 reason: "one-shot due_at cannot be combined with cron schedule".into(),
308 });
309 }
310 if let Some(zone) = self.timezone.as_deref() {
311 if zone.trim().is_empty() {
312 return Err(TaskError::BadSchedule {
313 expr: zone.into(),
314 reason: "timezone must be a named IANA zone".into(),
315 });
316 }
317 if zone.parse::<chrono_tz::Tz>().is_err() {
318 return Err(TaskError::BadSchedule {
319 expr: zone.into(),
320 reason: "unknown IANA timezone".into(),
321 });
322 }
323 }
324 Ok(())
325 }
326}
327
328/// fsyncs a directory so a prior rename into it is durable across a crash.
329/// On non-unix platforms directory fsync isn't a thing; the rename itself
330/// is still atomic there, so this is a no-op rather than an error.
331#[cfg(unix)]
332fn sync_directory(path: &Path) -> std::io::Result<()> {
333 std::fs::File::open(path).and_then(|dir| dir.sync_all())
334}
335
336#[cfg(not(unix))]
337fn sync_directory(_path: &Path) -> std::io::Result<()> {
338 Ok(())
339}
340
341/// In-process mirror of the persisted tasks.json array, keyed by id.
342pub struct TaskStore {
343 path: PathBuf,
344 tasks: HashMap<String, TaskDef>,
345}
346
347impl TaskStore {
348 /// Missing file loads as an empty store; a corrupt file is a typed
349 /// error, never silently dropped work.
350 pub fn load(sessions_home: &Path) -> Result<Self, TaskError> {
351 let path = tasks_file(sessions_home);
352 let raw = match std::fs::read_to_string(&path) {
353 Ok(raw) => raw,
354 Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
355 return Ok(TaskStore {
356 path,
357 tasks: HashMap::new(),
358 });
359 }
360 Err(source) => {
361 return Err(TaskError::Io {
362 path: path.clone(),
363 source,
364 });
365 }
366 };
367 let list: Vec<TaskDef> = serde_json::from_str(&raw).map_err(|source| TaskError::Json {
368 path: path.clone(),
369 source,
370 })?;
371 Ok(TaskStore {
372 path,
373 tasks: list.into_iter().map(|t| (t.id.clone(), t)).collect(),
374 })
375 }
376
377 pub fn save(&self) -> Result<(), TaskError> {
378 if let Some(parent) = self.path.parent() {
379 std::fs::create_dir_all(parent).map_err(|source| TaskError::Io {
380 path: parent.to_path_buf(),
381 source,
382 })?;
383 }
384 let mut list: Vec<TaskDef> = self.tasks.values().cloned().collect();
385 list.sort_by_key(|t| t.created_at);
386 let json = serde_json::to_string_pretty(&list).map_err(|source| TaskError::Json {
387 path: self.path.clone(),
388 source,
389 })?;
390 // Atomic + durable write: a plain `fs::write` truncates the file
391 // before the new bytes land, so a crash or power loss mid-write
392 // leaves `tasks.json` corrupt and unrecoverable. Write to a sibling
393 // temp file, fsync its contents, rename over the real path, then
394 // fsync the containing directory — on all platforms this crate
395 // targets, `rename` onto an existing path is atomic, so readers
396 // (this store, the server, the desktop app) only ever see the
397 // fully-old or fully-new content, never a partial write; the fsyncs
398 // ensure that content and the rename itself survive a crash right
399 // after this call returns, not just torn-write-free while running.
400 let tmp_path = self.path.with_extension("json.tmp");
401 {
402 let mut file = std::fs::File::create(&tmp_path).map_err(|source| TaskError::Io {
403 path: tmp_path.clone(),
404 source,
405 })?;
406 file.write_all(json.as_bytes())
407 .map_err(|source| TaskError::Io {
408 path: tmp_path.clone(),
409 source,
410 })?;
411 file.sync_all().map_err(|source| TaskError::Io {
412 path: tmp_path.clone(),
413 source,
414 })?;
415 }
416 std::fs::rename(&tmp_path, &self.path).map_err(|source| TaskError::Io {
417 path: self.path.clone(),
418 source,
419 })?;
420 if let Some(parent) = self.path.parent() {
421 sync_directory(parent).map_err(|source| TaskError::Io {
422 path: parent.to_path_buf(),
423 source,
424 })?;
425 }
426 Ok(())
427 }
428
429 pub fn get(&self, id: &str) -> Option<&TaskDef> {
430 self.tasks.get(id)
431 }
432
433 /// Inserts or replaces; returns the previous definition when replacing.
434 pub fn put(&mut self, task: TaskDef) -> Option<TaskDef> {
435 self.tasks.insert(task.id.clone(), task)
436 }
437
438 pub fn remove(&mut self, id: &str) -> bool {
439 self.tasks.remove(id).is_some()
440 }
441
442 pub fn len(&self) -> usize {
443 self.tasks.len()
444 }
445
446 #[must_use]
447 pub fn is_empty(&self) -> bool {
448 self.tasks.is_empty()
449 }
450
451 /// Every task, oldest first (created_at order, like the server API).
452 pub fn all(&self) -> Vec<TaskDef> {
453 let mut list: Vec<TaskDef> = self.tasks.values().cloned().collect();
454 list.sort_by_key(|t| t.created_at);
455 list
456 }
457
458 pub fn for_cwd(&self, cwd: &Path) -> Vec<TaskDef> {
459 self.all().into_iter().filter(|t| t.cwd == cwd).collect()
460 }
461}
462
463#[cfg(test)]
464mod tests {
465 #![allow(clippy::unwrap_used, clippy::expect_used)]
466 use super::*;
467 use chrono::{FixedOffset, LocalResult, NaiveDate, NaiveDateTime, Offset, TimeZone};
468
469 fn local(y: i32, mo: u32, d: u32, h: u32, mi: u32) -> DateTime<chrono::Local> {
470 chrono::Local
471 .with_ymd_and_hms(y, mo, d, h, mi, 0)
472 .earliest()
473 .unwrap()
474 }
475
476 fn next(expr: &str, after: DateTime<chrono::Local>) -> DateTime<chrono::Local> {
477 cron_next_after(expr, after).unwrap()
478 }
479
480 // ---- cron grammar matrix -------------------------------------------
481
482 #[test]
483 fn wildcard_every_step_minutes() {
484 assert_eq!(
485 next("*/15 * * * *", local(2026, 8, 24, 0, 0)),
486 local(2026, 8, 24, 0, 15)
487 );
488 assert_eq!(
489 next("*/15 * * * *", local(2026, 8, 24, 0, 16)),
490 local(2026, 8, 24, 0, 30)
491 );
492 }
493
494 #[test]
495 fn wildcard_every_minute_is_strictly_after() {
496 assert_eq!(
497 next("* * * * *", local(2026, 8, 24, 12, 34)),
498 local(2026, 8, 24, 12, 35)
499 );
500 }
501
502 #[test]
503 fn ranges_and_lists_combine() {
504 // 04:05 on any weekday, Mon-Fri.
505 assert_eq!(
506 next("5 4 * * 1-5", local(2026, 8, 22, 10, 0)),
507 local(2026, 8, 24, 4, 5)
508 );
509 assert_eq!(
510 next("0 9-17 * * *", local(2026, 8, 24, 18, 0)),
511 local(2026, 8, 25, 9, 0)
512 );
513 assert_eq!(
514 next("0 0 1,15 * *", local(2026, 1, 16, 0, 0)),
515 local(2026, 2, 1, 0, 0)
516 );
517 // Range with step.
518 assert_eq!(
519 next("0 0-23/6 * * *", local(2026, 8, 24, 7, 30)),
520 local(2026, 8, 24, 12, 0)
521 );
522 }
523
524 #[test]
525 fn month_rollover_includes_feb_29() {
526 assert_eq!(
527 next("0 0 29 2 *", local(2026, 3, 1, 0, 0)),
528 local(2028, 2, 29, 0, 0)
529 );
530 assert_eq!(
531 next("0 0 29 2 *", local(2028, 3, 1, 0, 0)),
532 local(2032, 2, 29, 0, 0)
533 );
534 }
535
536 #[test]
537 fn dom_and_dow_use_vixie_or_semantics_when_both_restricted() {
538 // Friday the 13th OR "any Friday" — either restricted field fires.
539 // 2026-01-14 is a Wednesday; the first Friday after is Jan 16.
540 assert_eq!(
541 next("0 12 13 * 5", local(2026, 1, 14, 0, 0)),
542 local(2026, 1, 16, 12, 0)
543 );
544 // Fridays keep firing even when the 13th doesn't match…
545 assert_eq!(
546 next("0 12 13 * 5", local(2026, 1, 17, 0, 0)),
547 local(2026, 1, 23, 12, 0)
548 );
549 // …and Friday Feb 13 matches BOTH restricted fields at once.
550 assert_eq!(
551 next("0 12 13 * 5", local(2026, 2, 12, 0, 0)),
552 local(2026, 2, 13, 12, 0)
553 );
554 }
555
556 #[test]
557 fn dom_alone_is_conjunctive_with_unrestricted_fields() {
558 assert_eq!(
559 next("0 12 13 * *", local(2026, 1, 14, 0, 0)),
560 local(2026, 2, 13, 12, 0)
561 );
562 }
563
564 #[test]
565 fn dow_seven_maps_to_sunday() {
566 // 2026-08-22 is a Saturday.
567 assert_eq!(
568 next("0 12 * * 7", local(2026, 8, 22, 0, 0)),
569 local(2026, 8, 23, 12, 0)
570 );
571 assert_eq!(
572 next("0 12 * * 0", local(2026, 8, 22, 0, 0)),
573 local(2026, 8, 23, 12, 0)
574 );
575 }
576
577 #[test]
578 fn parse_errors_are_typed_strings() {
579 for bad in [
580 "* * * *",
581 "* * * * * *",
582 "",
583 "60 * * * *",
584 "-1 * * * *",
585 "* 24 * * *",
586 "* * 0 * *",
587 "* * * 13 *",
588 "* * * * 8",
589 "*/0 * * * *",
590 "5-1 * * * *",
591 "1,,2 * * * *",
592 "abc * * * *",
593 "5/x * * * *",
594 "5/10 * * * *",
595 ] {
596 assert!(
597 CronExpr::parse(bad).is_err(),
598 "expected '{bad}' to be rejected"
599 );
600 }
601 }
602
603 #[test]
604 fn unreachable_date_hits_horizon_error() {
605 // Feb 31 does not exist in any year.
606 let err = cron_next_after("0 0 31 2 *", local(2026, 1, 1, 0, 0)).unwrap_err();
607 assert!(err.contains("never fires"), "{err}");
608 }
609
610 /// Synthetic zone modeling a spring-forward gap (local 02:00–02:59 of
611 /// 2026-03-08 do not exist) and a fall-back fold (local 01:00–01:59 of
612 /// 2026-11-01 occur twice), so gap/fold behavior is testable without
613 /// touching process TZ state.
614 #[derive(Debug, Clone, Copy)]
615 struct GapTz;
616
617 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
618 struct GapOffset(FixedOffset);
619
620 const EAST: FixedOffset = FixedOffset::east_opt(3600).unwrap();
621 const WEST: FixedOffset = FixedOffset::east_opt(0).unwrap();
622
623 impl TimeZone for GapTz {
624 type Offset = GapOffset;
625 fn from_offset(_offset: &GapOffset) -> Self {
626 GapTz
627 }
628 fn offset_from_utc_date(&self, _utc: &chrono::NaiveDate) -> GapOffset {
629 GapOffset(EAST)
630 }
631 fn offset_from_utc_datetime(&self, _utc: &NaiveDateTime) -> GapOffset {
632 GapOffset(EAST)
633 }
634 fn offset_from_local_date(&self, local: &chrono::NaiveDate) -> LocalResult<GapOffset> {
635 // A date inherits its gap/fold status from some instant on it.
636 let probe = local.and_hms_opt(12, 0, 0).unwrap();
637 self.offset_from_local_datetime(&probe)
638 }
639 fn offset_from_local_datetime(&self, local: &NaiveDateTime) -> LocalResult<GapOffset> {
640 use chrono::NaiveDate;
641 let gap_day = NaiveDate::from_ymd_opt(2026, 3, 8).unwrap();
642 let fold_day = NaiveDate::from_ymd_opt(2026, 11, 1).unwrap();
643 let t = local.time();
644 if local.date() == gap_day && t.hour() == 2 {
645 return LocalResult::None;
646 }
647 if local.date() == fold_day && t.hour() == 1 {
648 // (earliest, latest): the larger offset yields the earlier
649 // instant, so the pre-fold EAST pass comes first.
650 return LocalResult::Ambiguous(GapOffset(EAST), GapOffset(WEST));
651 }
652 LocalResult::Single(GapOffset(EAST))
653 }
654 }
655
656 impl chrono::Offset for GapOffset {
657 fn fix(&self) -> FixedOffset {
658 self.0
659 }
660 }
661
662 fn gap_next(expr: &str, after: NaiveDateTime) -> Option<DateTime<GapTz>> {
663 next_fire(
664 &CronExpr::parse(expr).unwrap(),
665 GapTz.from_utc_datetime(&after),
666 )
667 }
668
669 #[test]
670 fn dst_gap_skips_forward_to_first_existing_match() {
671 let before_gap = NaiveDate::from_ymd_opt(2026, 3, 8)
672 .unwrap()
673 .and_hms_opt(1, 30, 0)
674 .unwrap();
675 // Hourly job: 02:00 does not exist → fires at 03:00.
676 let fired = gap_next("0 * * * *", before_gap).unwrap();
677 assert_eq!(
678 fired.naive_local(),
679 NaiveDate::from_ymd_opt(2026, 3, 8)
680 .unwrap()
681 .and_hms_opt(3, 0, 0)
682 .unwrap()
683 );
684 // Daily-at-02:30 job: today's 02:30 doesn't exist → tomorrow's.
685 let daily = gap_next("30 2 * * *", before_gap).unwrap();
686 assert_eq!(
687 daily.naive_local(),
688 NaiveDate::from_ymd_opt(2026, 3, 9)
689 .unwrap()
690 .and_hms_opt(2, 30, 0)
691 .unwrap()
692 );
693 }
694
695 #[test]
696 fn dst_fold_resolves_to_earliest_instant() {
697 let before_fold = NaiveDate::from_ymd_opt(2026, 11, 1)
698 .unwrap()
699 .and_hms_opt(0, 30, 0)
700 .unwrap();
701 let fired = gap_next("45 1 * * *", before_fold).unwrap();
702 assert_eq!(
703 fired.naive_local(),
704 NaiveDate::from_ymd_opt(2026, 11, 1)
705 .unwrap()
706 .and_hms_opt(1, 45, 0)
707 .unwrap()
708 );
709 // Earliest pass carries the pre-fold offset (EAST here).
710 assert_eq!(fired.offset().fix(), EAST);
711 }
712
713 // ---- task validation + store ----------------------------------------
714
715 fn base_task() -> TaskDef {
716 TaskDef {
717 id: uuid::Uuid::now_v7().to_string(),
718 name: "nightly".into(),
719 prompt: "tidy the repo".into(),
720 interval_secs: 3600,
721 enabled: true,
722 cwd: PathBuf::from("/tmp/ws"),
723 created_at: Utc::now(),
724 last_run_at: None,
725 last_session_id: None,
726 last_summary: None,
727 last_result_id: None,
728 last_run_status: None,
729 last_delivery_state: None,
730 last_wt: None,
731 deliver_to: None,
732 schedule: None,
733 timezone: None,
734 due_at: None,
735 script: None,
736 model_pin: None,
737 agent_id: None,
738 agent_revision: None,
739 }
740 }
741
742 #[test]
743 fn prompt_xor_script_validation_matrix() {
744 let mut both = base_task();
745 both.script = Some("echo tick".into());
746 assert!(matches!(
747 both.validate(),
748 Err(TaskError::PromptScriptXor { .. })
749 ));
750
751 let neither = TaskDef {
752 prompt: " ".into(),
753 ..base_task()
754 };
755 assert!(matches!(
756 neither.validate(),
757 Err(TaskError::PromptScriptXor { .. })
758 ));
759
760 assert!(base_task().validate().is_ok());
761 let watchdog = TaskDef {
762 prompt: String::new(),
763 script: Some("curl -sf http://x/health".into()),
764 ..base_task()
765 };
766 assert!(watchdog.validate().is_ok());
767 }
768
769 #[test]
770 fn one_shot_and_timezone_validation_is_explicit() {
771 let mut task = base_task();
772 task.due_at = Some(Utc::now());
773 task.schedule = Some("0 9 * * *".into());
774 assert!(matches!(
775 task.validate(),
776 Err(TaskError::BadSchedule { .. })
777 ));
778 task.schedule = None;
779 task.timezone = Some(" ".into());
780 assert!(matches!(
781 task.validate(),
782 Err(TaskError::BadSchedule { .. })
783 ));
784 }
785
786 #[test]
787 fn named_timezone_cron_returns_utc_instant() {
788 let after = chrono::DateTime::parse_from_rfc3339("2026-01-01T12:00:00Z")
789 .expect("timestamp")
790 .with_timezone(&Utc);
791 let next = cron_next_after_timezone("0 9 * * *", after, "America/New_York")
792 .expect("next named-zone fire");
793 assert_eq!(next.to_rfc3339(), "2026-01-01T14:00:00+00:00");
794 }
795
796 #[test]
797 fn bad_cron_rejected_by_validate() {
798 let mut t = base_task();
799 t.schedule = Some("99 * * * *".into());
800 let err = t.validate().unwrap_err();
801 assert!(
802 matches!(err, TaskError::BadSchedule { ref expr, .. } if expr == "99 * * * *"),
803 "{err}"
804 );
805 t.schedule = Some("*/15 * * * *".into());
806 assert!(t.validate().is_ok());
807 }
808
809 #[test]
810 fn a_tasks_file_written_before_added_fields_loads_and_roundtrips() {
811 let dir = tempfile::tempdir().unwrap();
812 let raw = r#"[
813 {
814 "id": "018f0000-0000-7000-8000-000000000001",
815 "name": "standup-notes",
816 "prompt": "summarize yesterday",
817 "interval_secs": 86400,
818 "enabled": true,
819 "cwd": "/Users/me/proj",
820 "created_at": "2026-07-01T09:00:00Z",
821 "last_run_at": "2026-08-01T09:00:00Z",
822 "last_session_id": "abc",
823 "last_summary": "done",
824 "last_result_id": "result-1",
825 "last_wt": { "path": "/tmp/wt", "branch": "vak/abc" },
826 "deliver_to": "log:ops"
827 }
828 ]"#;
829 std::fs::create_dir_all(dir.path()).unwrap();
830 std::fs::write(tasks_file(dir.path()), raw).unwrap();
831
832 let store = TaskStore::load(dir.path()).unwrap();
833 assert_eq!(store.len(), 1);
834 let t = store.get("018f0000-0000-7000-8000-000000000001").unwrap();
835 assert_eq!(t.prompt, "summarize yesterday");
836 assert_eq!(t.interval_secs, 86400);
837 assert_eq!(t.last_wt.as_ref().unwrap().branch, "vak/abc");
838 assert_eq!(t.deliver_to.as_deref(), Some("log:ops"));
839 assert_eq!(t.schedule, None);
840 assert_eq!(t.script, None);
841 assert_eq!(t.model_pin, None);
842
843 store.save().unwrap();
844 let reloaded = TaskStore::load(dir.path()).unwrap();
845 assert_eq!(reloaded.all().len(), 1);
846 assert_eq!(
847 reloaded.get(t.id.as_str()).unwrap().last_wt,
848 t.last_wt,
849 "legacy last_wt must survive a load/save cycle"
850 );
851 }
852
853 #[test]
854 fn new_fields_roundtrip_through_store() {
855 let dir = tempfile::tempdir().unwrap();
856 let mut store = TaskStore::load(dir.path()).unwrap();
857 assert!(store.is_empty());
858
859 let mut t = base_task();
860 t.prompt = String::new();
861 t.script = Some("systemctl is-active nginx".into());
862 t.schedule = Some("0 7 * * 1-5".into());
863 t.model_pin = Some("haiku-fast".into());
864 store.put(t.clone());
865 store.save().unwrap();
866
867 let back = TaskStore::load(dir.path()).unwrap();
868 let got = back.get(&t.id).unwrap();
869 assert_eq!(got.script.as_deref(), Some("systemctl is-active nginx"));
870 assert_eq!(got.schedule.as_deref(), Some("0 7 * * 1-5"));
871 assert_eq!(got.model_pin.as_deref(), Some("haiku-fast"));
872 assert!(got.validate().is_ok());
873 }
874
875 #[test]
876 fn missing_file_is_empty_store_and_save_creates_it() {
877 let dir = tempfile::tempdir().unwrap();
878 let mut store = TaskStore::load(dir.path()).unwrap();
879 assert!(store.is_empty());
880 store.put(base_task());
881 store.save().unwrap();
882 assert!(tasks_file(dir.path()).is_file());
883 }
884
885 #[test]
886 fn save_is_atomic_and_leaves_no_tmp_file_behind() {
887 let dir = tempfile::tempdir().unwrap();
888 let mut store = TaskStore::load(dir.path()).unwrap();
889 store.put(base_task());
890 store.save().unwrap();
891 assert!(tasks_file(dir.path()).is_file());
892 assert!(
893 !dir.path().join("tasks.json.tmp").exists(),
894 "the temp file used for the atomic rename must not survive a successful save"
895 );
896
897 // A second save (overwrite path) must round-trip cleanly too, and
898 // still leave no tmp file — this is the path a real crash-mid-write
899 // would otherwise corrupt with a plain `fs::write`.
900 store.put(base_task());
901 store.save().unwrap();
902 let reloaded = TaskStore::load(dir.path()).unwrap();
903 assert_eq!(reloaded.tasks.len(), 2);
904 assert!(!dir.path().join("tasks.json.tmp").exists());
905 }
906
907 #[test]
908 fn corrupt_file_is_a_typed_error_not_silence() {
909 let dir = tempfile::tempdir().unwrap();
910 std::fs::write(tasks_file(dir.path()), "{not json").unwrap();
911 assert!(matches!(
912 TaskStore::load(dir.path()),
913 Err(TaskError::Json { .. })
914 ));
915 }
916
917 #[test]
918 fn for_cwd_filters_and_sorts_by_created_at() {
919 let ws_a = PathBuf::from("/ws/a");
920 let ws_b = PathBuf::from("/ws/b");
921 let mut early = base_task();
922 early.id = "early".into();
923 early.cwd = ws_a.clone();
924 early.created_at = Utc::now() - chrono::Duration::hours(2);
925 let mut late = early.clone();
926 late.id = "late".into();
927 late.created_at = Utc::now();
928 let mut other = base_task();
929 other.id = "other".into();
930 other.cwd = ws_b;
931
932 let mut map = HashMap::new();
933 for t in [early, late, other] {
934 map.insert(t.id.clone(), t);
935 }
936 let store = TaskStore {
937 path: PathBuf::from("unused"),
938 tasks: map,
939 };
940 let mine = store.for_cwd(&ws_a);
941 assert_eq!(mine.len(), 2);
942 assert_eq!(mine[0].id, "early", "oldest first");
943 assert_eq!(mine[1].id, "late");
944 }
945}
946

Tasks

Indexing the workspace…

Tasks