1use std::collections::BTreeMap;
2use std::fs;
3use std::path::PathBuf;
4
5use anyhow::Result;
6use chrono::SecondsFormat;
7use serde::{Deserialize, Serialize};
8use sha2::{Digest, Sha256};
9use thiserror::Error;
10use uuid::Uuid;
11
12use crate::BeamPaths;
13
14#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
15#[serde(rename_all = "camelCase")]
16pub enum ParsedScheduleKind {
17 Once,
18 Interval,
19 Cron,
20}
21
22#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
23#[serde(rename_all = "camelCase")]
24pub struct ParsedSchedule {
25 pub kind: ParsedScheduleKind,
26 #[serde(default, skip_serializing_if = "Option::is_none")]
27 pub run_at: Option<String>,
28 #[serde(default, skip_serializing_if = "Option::is_none")]
29 pub minutes: Option<u64>,
30 #[serde(default, skip_serializing_if = "Option::is_none")]
31 pub expr: Option<String>,
32 pub display: String,
33}
34
35#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
36#[serde(rename_all = "camelCase")]
37pub struct ScheduleRepeat {
38 pub times: Option<u64>,
39 #[serde(default)]
40 pub completed: u64,
41}
42
43#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
44#[serde(rename_all = "camelCase")]
45pub enum ScheduleChatType {
46 Group,
47 P2p,
48 TopicGroup,
49}
50
51#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
52#[serde(rename_all = "camelCase")]
53pub enum ScheduleDeliver {
54 Origin,
55 Local,
56}
57
58impl Default for ScheduleDeliver {
59 fn default() -> Self {
60 Self::Origin
61 }
62}
63
64#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
65#[serde(rename_all = "camelCase")]
66pub struct ScheduledTask {
67 pub id: String,
68 pub name: String,
69 pub schedule: String,
70 pub parsed: ParsedSchedule,
71 pub prompt: String,
72 pub working_dir: String,
73 pub chat_id: String,
74 #[serde(default, skip_serializing_if = "Option::is_none")]
75 pub root_message_id: Option<String>,
76 #[serde(default, skip_serializing_if = "Option::is_none")]
77 pub scope: Option<String>,
78 #[serde(default, skip_serializing_if = "Option::is_none")]
79 pub chat_type: Option<ScheduleChatType>,
80 #[serde(default, skip_serializing_if = "Option::is_none")]
81 pub lark_app_id: Option<String>,
82 #[serde(default, skip_serializing_if = "Option::is_none")]
83 pub creator_chat_id: Option<String>,
84 #[serde(default, skip_serializing_if = "Option::is_none")]
85 pub creator_root_message_id: Option<String>,
86 #[serde(default, skip_serializing_if = "Option::is_none")]
87 pub creator_lark_app_id: Option<String>,
88 #[serde(default = "default_true")]
89 pub enabled: bool,
90 pub created_at: String,
91 #[serde(default, skip_serializing_if = "Option::is_none")]
92 pub last_run_at: Option<String>,
93 #[serde(default, skip_serializing_if = "Option::is_none")]
94 pub next_run_at: Option<String>,
95 #[serde(default, skip_serializing_if = "Option::is_none")]
96 pub last_status: Option<String>,
97 #[serde(default, skip_serializing_if = "Option::is_none")]
98 pub last_error: Option<String>,
99 #[serde(default, skip_serializing_if = "Option::is_none")]
100 pub last_delivery_error: Option<String>,
101 #[serde(default, skip_serializing_if = "Option::is_none")]
102 pub repeat: Option<ScheduleRepeat>,
103 #[serde(default)]
104 pub deliver: ScheduleDeliver,
105}
106
107#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
108#[serde(rename_all = "camelCase")]
109pub struct CreateTaskInput {
110 #[serde(default, skip_serializing_if = "Option::is_none")]
111 pub id: Option<String>,
112 pub name: String,
113 pub schedule: String,
114 pub parsed: ParsedSchedule,
115 pub prompt: String,
116 pub working_dir: String,
117 pub chat_id: String,
118 #[serde(default, skip_serializing_if = "Option::is_none")]
119 pub root_message_id: Option<String>,
120 #[serde(default, skip_serializing_if = "Option::is_none")]
121 pub scope: Option<String>,
122 #[serde(default, skip_serializing_if = "Option::is_none")]
123 pub chat_type: Option<ScheduleChatType>,
124 #[serde(default, skip_serializing_if = "Option::is_none")]
125 pub lark_app_id: Option<String>,
126 #[serde(default, skip_serializing_if = "Option::is_none")]
127 pub creator_chat_id: Option<String>,
128 #[serde(default, skip_serializing_if = "Option::is_none")]
129 pub creator_root_message_id: Option<String>,
130 #[serde(default, skip_serializing_if = "Option::is_none")]
131 pub creator_lark_app_id: Option<String>,
132 #[serde(default, skip_serializing_if = "Option::is_none")]
133 pub next_run_at: Option<String>,
134 #[serde(default, skip_serializing_if = "Option::is_none")]
135 pub repeat: Option<ScheduleRepeat>,
136 #[serde(default, skip_serializing_if = "Option::is_none")]
137 pub deliver: Option<ScheduleDeliver>,
138}
139
140#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
141#[serde(rename_all = "camelCase")]
142pub struct ScheduleTaskUpdate {
143 #[serde(default, skip_serializing_if = "Option::is_none")]
144 pub enabled: Option<bool>,
145 #[serde(default, skip_serializing_if = "Option::is_none")]
146 pub last_run_at: Option<String>,
147 #[serde(default, skip_serializing_if = "Option::is_none")]
148 pub next_run_at: Option<Option<String>>,
149 #[serde(default, skip_serializing_if = "Option::is_none")]
150 pub last_status: Option<Option<String>>,
151 #[serde(default, skip_serializing_if = "Option::is_none")]
152 pub last_error: Option<Option<String>>,
153 #[serde(default, skip_serializing_if = "Option::is_none")]
154 pub last_delivery_error: Option<Option<String>>,
155 #[serde(default, skip_serializing_if = "Option::is_none")]
156 pub repeat: Option<Option<ScheduleRepeat>>,
157 #[serde(default, skip_serializing_if = "Option::is_none")]
158 pub root_message_id: Option<Option<String>>,
159 #[serde(default, skip_serializing_if = "Option::is_none")]
160 pub chat_type: Option<Option<ScheduleChatType>>,
161}
162
163#[derive(Debug, Error)]
164pub enum ScheduleStoreError {
165 #[error(
166 "IdempotencyConflict: schedule task {task_id} exists with different canonical input (existing={existing_input_hash}…, incoming={incoming_input_hash}…)"
167 )]
168 IdempotencyConflict {
169 task_id: String,
170 existing_input_hash: String,
171 incoming_input_hash: String,
172 },
173 #[error(transparent)]
174 Io(#[from] std::io::Error),
175 #[error(transparent)]
176 Serde(#[from] serde_json::Error),
177}
178
179pub fn create_task(
180 paths: &BeamPaths,
181 input: CreateTaskInput,
182) -> Result<ScheduledTask, ScheduleStoreError> {
183 let mut tasks = load_tasks(paths)?;
184 if let Some(id) = &input.id
185 && let Some(existing) = tasks.get(id)
186 {
187 let existing_hash = compute_input_hash(&canonical_schedule_input(existing))?;
188 let incoming_hash = compute_input_hash(&canonical_schedule_input_input(&input))?;
189 if existing_hash == incoming_hash {
190 return Ok(existing.clone());
191 }
192 return Err(ScheduleStoreError::IdempotencyConflict {
193 task_id: id.clone(),
194 existing_input_hash: existing_hash,
195 incoming_input_hash: incoming_hash,
196 });
197 }
198
199 let id = input.id.unwrap_or_else(|| {
200 Uuid::new_v4()
201 .simple()
202 .to_string()
203 .chars()
204 .take(8)
205 .collect()
206 });
207 let task = ScheduledTask {
208 id: id.clone(),
209 name: input.name,
210 schedule: input.schedule,
211 parsed: input.parsed,
212 prompt: input.prompt,
213 working_dir: input.working_dir,
214 chat_id: input.chat_id,
215 root_message_id: input.root_message_id,
216 scope: input.scope,
217 chat_type: input.chat_type,
218 lark_app_id: input.lark_app_id,
219 creator_chat_id: input.creator_chat_id,
220 creator_root_message_id: input.creator_root_message_id,
221 creator_lark_app_id: input.creator_lark_app_id,
222 enabled: true,
223 created_at: UtcNow::now(),
224 last_run_at: None,
225 next_run_at: input.next_run_at,
226 last_status: None,
227 last_error: None,
228 last_delivery_error: None,
229 repeat: input.repeat,
230 deliver: input.deliver.unwrap_or_default(),
231 };
232 tasks.insert(id, task.clone());
233 save_tasks(paths, &tasks)?;
234 Ok(task)
235}
236
237pub fn get_task(paths: &BeamPaths, id: &str) -> Result<Option<ScheduledTask>, ScheduleStoreError> {
238 let tasks = load_tasks(paths)?;
239 Ok(tasks.get(id).cloned())
240}
241
242pub fn remove_task(paths: &BeamPaths, id: &str) -> Result<bool, ScheduleStoreError> {
243 let mut tasks = load_tasks(paths)?;
244 let existed = tasks.remove(id).is_some();
245 if existed {
246 save_tasks(paths, &tasks)?;
247 }
248 Ok(existed)
249}
250
251pub fn update_task(
252 paths: &BeamPaths,
253 id: &str,
254 updates: ScheduleTaskUpdate,
255) -> Result<(), ScheduleStoreError> {
256 let mut tasks = load_tasks(paths)?;
257 if let Some(task) = tasks.get_mut(id) {
258 if let Some(enabled) = updates.enabled {
259 task.enabled = enabled;
260 }
261 if let Some(last_run_at) = updates.last_run_at {
262 task.last_run_at = Some(last_run_at);
263 }
264 if let Some(next_run_at) = updates.next_run_at {
265 task.next_run_at = next_run_at;
266 }
267 if let Some(last_status) = updates.last_status {
268 task.last_status = last_status;
269 }
270 if let Some(last_error) = updates.last_error {
271 task.last_error = last_error;
272 }
273 if let Some(last_delivery_error) = updates.last_delivery_error {
274 task.last_delivery_error = last_delivery_error;
275 }
276 if let Some(repeat) = updates.repeat {
277 task.repeat = repeat;
278 }
279 if let Some(root_message_id) = updates.root_message_id {
280 task.root_message_id = root_message_id;
281 }
282 if let Some(chat_type) = updates.chat_type {
283 task.chat_type = chat_type;
284 }
285 save_tasks(paths, &tasks)?;
286 }
287 Ok(())
288}
289
290pub fn mark_run(
291 paths: &BeamPaths,
292 id: &str,
293 success: bool,
294 error: Option<&str>,
295 delivery_error: Option<&str>,
296) -> Result<(), ScheduleStoreError> {
297 let mut tasks = load_tasks(paths)?;
298 let Some(task) = tasks.get_mut(id) else {
299 return Ok(());
300 };
301
302 task.last_run_at = Some(UtcNow::now());
303 task.last_status = Some(if success {
304 "ok".to_string()
305 } else {
306 "error".to_string()
307 });
308 task.last_error = if success {
309 None
310 } else {
311 error.map(|s| s.to_string())
312 };
313 task.last_delivery_error = delivery_error.map(|s| s.to_string());
314
315 if let Some(repeat) = task.repeat.as_mut() {
316 repeat.completed = repeat.completed.saturating_add(1);
317 if matches!(repeat.times, Some(times) if times > 0 && repeat.completed >= times) {
318 tasks.remove(id);
319 save_tasks(paths, &tasks)?;
320 return Ok(());
321 }
322 }
323
324 let now = chrono::Utc::now();
326 match task.parsed.kind {
327 ParsedScheduleKind::Once => {
328 task.enabled = false;
329 task.next_run_at = None;
330 }
331 ParsedScheduleKind::Interval => {
332 if let Some(minutes) = task.parsed.minutes {
333 let next = now + chrono::Duration::minutes(minutes as i64);
334 task.next_run_at = Some(next.to_rfc3339_opts(SecondsFormat::Millis, true));
335 }
336 }
337 ParsedScheduleKind::Cron => {
338 task.enabled = false;
341 task.next_run_at = None;
342 task.last_error = Some(
343 "cron schedule auto-advance not supported; re-create the schedule to re-enable"
344 .to_string(),
345 );
346 }
347 }
348
349 save_tasks(paths, &tasks)?;
350 Ok(())
351}
352
353pub fn list_tasks(paths: &BeamPaths) -> Result<Vec<ScheduledTask>, ScheduleStoreError> {
354 let tasks = load_tasks(paths)?;
355 Ok(tasks.values().cloned().collect())
356}
357
358pub fn append_output_log(
359 paths: &BeamPaths,
360 task_id: &str,
361 content: &str,
362) -> Result<PathBuf, ScheduleStoreError> {
363 let dir = task_output_dir(paths, task_id);
364 fs::create_dir_all(&dir)?;
365 let fname = format!("{}.md", UtcNow::now().replace(':', "-").replace('.', "-"));
366 let path = dir.join(fname);
367 fs::write(&path, content)?;
368 Ok(path)
369}
370
371fn load_tasks(paths: &BeamPaths) -> Result<BTreeMap<String, ScheduledTask>, ScheduleStoreError> {
372 let path = paths.schedules_json();
373 if !path.exists() {
374 return Ok(BTreeMap::new());
375 }
376 let raw = fs::read_to_string(&path)?;
377 if raw.trim().is_empty() {
378 return Ok(BTreeMap::new());
379 }
380 let tasks = serde_json::from_str(&raw)?;
381 Ok(tasks)
382}
383
384fn save_tasks(
385 paths: &BeamPaths,
386 tasks: &BTreeMap<String, ScheduledTask>,
387) -> Result<(), ScheduleStoreError> {
388 let path = paths.schedules_json();
389 if let Some(parent) = path.parent() {
390 fs::create_dir_all(parent)?;
391 }
392 let tmp = path.with_extension("json.tmp");
393 fs::write(&tmp, serde_json::to_vec_pretty(tasks)?)?;
394 fs::rename(&tmp, &path)?;
395 Ok(())
396}
397
398fn canonical_schedule_input(task: &ScheduledTask) -> serde_json::Value {
399 serde_json::json!({
400 "name": task.name,
401 "schedule": task.schedule,
402 "parsed": {
403 "kind": task.parsed.kind,
404 "runAt": task.parsed.run_at,
405 "minutes": task.parsed.minutes,
406 "expr": task.parsed.expr,
407 },
408 "prompt": task.prompt,
409 "workingDir": task.working_dir,
410 "chatId": task.chat_id,
411 "rootMessageId": task.root_message_id,
412 "scope": task.scope,
413 "larkAppId": task.lark_app_id,
414 "repeat": task.repeat.as_ref().map(|repeat| serde_json::json!({ "times": repeat.times })),
415 "deliver": match task.deliver {
416 ScheduleDeliver::Origin => "origin",
417 ScheduleDeliver::Local => "local",
418 }
419 })
420}
421
422fn canonical_schedule_input_input(input: &CreateTaskInput) -> serde_json::Value {
423 serde_json::json!({
424 "name": input.name,
425 "schedule": input.schedule,
426 "parsed": {
427 "kind": input.parsed.kind,
428 "runAt": input.parsed.run_at,
429 "minutes": input.parsed.minutes,
430 "expr": input.parsed.expr,
431 },
432 "prompt": input.prompt,
433 "workingDir": input.working_dir,
434 "chatId": input.chat_id,
435 "rootMessageId": input.root_message_id,
436 "scope": input.scope,
437 "larkAppId": input.lark_app_id,
438 "repeat": input.repeat.as_ref().map(|repeat| serde_json::json!({ "times": repeat.times })),
439 "deliver": match input.deliver.clone().unwrap_or_default() {
440 ScheduleDeliver::Origin => "origin",
441 ScheduleDeliver::Local => "local",
442 }
443 })
444}
445
446fn compute_input_hash(value: &serde_json::Value) -> Result<String, ScheduleStoreError> {
447 let canonical = canonical_json(value);
448 let mut hasher = Sha256::new();
449 hasher.update(canonical.as_bytes());
450 Ok(format!("sha256:{}", lower_hex(&hasher.finalize())))
451}
452
453fn lower_hex(bytes: &[u8]) -> String {
454 const HEX: &[u8; 16] = b"0123456789abcdef";
455 let mut out = String::with_capacity(bytes.len() * 2);
456 for byte in bytes {
457 out.push(HEX[(byte >> 4) as usize] as char);
458 out.push(HEX[(byte & 0x0f) as usize] as char);
459 }
460 out
461}
462
463fn canonical_json(value: &serde_json::Value) -> String {
464 match value {
465 serde_json::Value::Null => "null".to_string(),
466 serde_json::Value::Bool(v) => if *v { "true" } else { "false" }.to_string(),
467 serde_json::Value::Number(v) => v.to_string(),
468 serde_json::Value::String(v) => serde_json::to_string(v).expect("string serializable"),
469 serde_json::Value::Array(items) => {
470 let mut out = String::from("[");
471 let mut first = true;
472 for item in items {
473 if !first {
474 out.push(',');
475 }
476 first = false;
477 out.push_str(&canonical_json(item));
478 }
479 out.push(']');
480 out
481 }
482 serde_json::Value::Object(map) => {
483 let mut keys: Vec<_> = map.keys().collect();
484 keys.sort();
485 let mut out = String::from("{");
486 let mut first = true;
487 for key in keys {
488 if !first {
489 out.push(',');
490 }
491 first = false;
492 out.push_str(&serde_json::to_string(key).expect("key serializable"));
493 out.push(':');
494 out.push_str(&canonical_json(&map[key]));
495 }
496 out.push('}');
497 out
498 }
499 }
500}
501
502fn task_output_dir(paths: &BeamPaths, task_id: &str) -> PathBuf {
503 paths.schedules_output_dir().join(task_id)
504}
505
506struct UtcNow;
507
508impl UtcNow {
509 fn now() -> String {
510 chrono::Utc::now().to_rfc3339_opts(SecondsFormat::Millis, true)
511 }
512}
513
514fn default_true() -> bool {
515 true
516}
517
518#[cfg(test)]
519mod tests {
520 use super::*;
521 use std::time::{SystemTime, UNIX_EPOCH};
522
523 fn temp_paths(label: &str) -> BeamPaths {
524 let nanos = SystemTime::now()
525 .duration_since(UNIX_EPOCH)
526 .unwrap_or_default()
527 .as_nanos();
528 BeamPaths::from_root(std::env::temp_dir().join(format!(
529 "beam-schedule-store-{label}-{nanos}-{}",
530 std::process::id()
531 )))
532 }
533
534 fn parsed_cron() -> ParsedSchedule {
535 ParsedSchedule {
536 kind: ParsedScheduleKind::Cron,
537 run_at: None,
538 minutes: None,
539 expr: Some("0 9 * * *".to_string()),
540 display: "0 9 * * *".to_string(),
541 }
542 }
543
544 #[test]
545 fn create_task_returns_existing_when_canonical_input_matches() {
546 let paths = temp_paths("identical");
547 let input = CreateTaskInput {
548 id: Some("wf_task".to_string()),
549 name: "schedule-demo daily 9am".to_string(),
550 schedule: "0 9 * * *".to_string(),
551 parsed: parsed_cron(),
552 prompt: "Schedule demo: run workflow self-check.".to_string(),
553 working_dir: "/tmp/beam-schedule-demo".to_string(),
554 chat_id: "oc_workflow_demo".to_string(),
555 root_message_id: None,
556 scope: Some("thread".to_string()),
557 chat_type: None,
558 lark_app_id: None,
559 creator_chat_id: None,
560 creator_root_message_id: None,
561 creator_lark_app_id: None,
562 next_run_at: None,
563 repeat: None,
564 deliver: None,
565 };
566 let created = create_task(&paths, input.clone()).expect("create");
567 let returned = create_task(&paths, input).expect("create existing");
568 assert_eq!(created.id, "wf_task");
569 assert_eq!(returned.id, "wf_task");
570 assert_eq!(list_tasks(&paths).expect("list").len(), 1);
571 let _ = std::fs::remove_dir_all(paths.root());
572 }
573
574 #[test]
575 fn create_task_conflicts_when_canonical_input_differs() {
576 let paths = temp_paths("conflict");
577 let input = CreateTaskInput {
578 id: Some("wf_task".to_string()),
579 name: "schedule-demo daily 9am".to_string(),
580 schedule: "0 9 * * *".to_string(),
581 parsed: parsed_cron(),
582 prompt: "Schedule demo: run workflow self-check.".to_string(),
583 working_dir: "/tmp/beam-schedule-demo".to_string(),
584 chat_id: "oc_workflow_demo".to_string(),
585 root_message_id: None,
586 scope: Some("thread".to_string()),
587 chat_type: None,
588 lark_app_id: None,
589 creator_chat_id: None,
590 creator_root_message_id: None,
591 creator_lark_app_id: None,
592 next_run_at: None,
593 repeat: None,
594 deliver: None,
595 };
596 let _ = create_task(&paths, input).expect("create");
597 let changed = CreateTaskInput {
598 id: Some("wf_task".to_string()),
599 name: "schedule-demo daily 9am".to_string(),
600 schedule: "0 9 * * *".to_string(),
601 parsed: parsed_cron(),
602 prompt: "changed prompt".to_string(),
603 working_dir: "/tmp/beam-schedule-demo".to_string(),
604 chat_id: "oc_workflow_demo".to_string(),
605 root_message_id: None,
606 scope: Some("thread".to_string()),
607 chat_type: None,
608 lark_app_id: None,
609 creator_chat_id: None,
610 creator_root_message_id: None,
611 creator_lark_app_id: None,
612 next_run_at: None,
613 repeat: None,
614 deliver: None,
615 };
616 let err = create_task(&paths, changed).expect_err("conflict");
617 assert!(matches!(
618 err,
619 ScheduleStoreError::IdempotencyConflict { .. }
620 ));
621 let _ = std::fs::remove_dir_all(paths.root());
622 }
623
624 #[test]
625 fn mark_run_removes_finite_repeat_after_completion() {
626 let paths = temp_paths("mark");
627 let input = CreateTaskInput {
628 id: Some("wf_task".to_string()),
629 name: "schedule-demo daily 9am".to_string(),
630 schedule: "0 9 * * *".to_string(),
631 parsed: parsed_cron(),
632 prompt: "Schedule demo: run workflow self-check.".to_string(),
633 working_dir: "/tmp/beam-schedule-demo".to_string(),
634 chat_id: "oc_workflow_demo".to_string(),
635 root_message_id: None,
636 scope: Some("thread".to_string()),
637 chat_type: None,
638 lark_app_id: None,
639 creator_chat_id: None,
640 creator_root_message_id: None,
641 creator_lark_app_id: None,
642 next_run_at: None,
643 repeat: Some(ScheduleRepeat {
644 times: Some(1),
645 completed: 0,
646 }),
647 deliver: None,
648 };
649 let _ = create_task(&paths, input).expect("create");
650 mark_run(&paths, "wf_task", true, None, None).expect("mark run");
651 assert!(get_task(&paths, "wf_task").expect("get").is_none());
652 let _ = std::fs::remove_dir_all(paths.root());
653 }
654
655 #[test]
656 fn mark_run_interval_advances_next_run_at() {
657 let paths = temp_paths("interval-advance");
658 let input = CreateTaskInput {
659 id: Some("int_task".to_string()),
660 name: "every 10 min".to_string(),
661 schedule: "every 10 min".to_string(),
662 parsed: ParsedSchedule {
663 kind: ParsedScheduleKind::Interval,
664 run_at: None,
665 minutes: Some(10),
666 expr: None,
667 display: "every 10 min".to_string(),
668 },
669 prompt: "interval test".to_string(),
670 working_dir: "/tmp".to_string(),
671 chat_id: "oc_test".to_string(),
672 root_message_id: None,
673 scope: Some("thread".to_string()),
674 chat_type: None,
675 lark_app_id: None,
676 creator_chat_id: None,
677 creator_root_message_id: None,
678 creator_lark_app_id: None,
679 next_run_at: Some("2026-01-01T00:00:00.000Z".to_string()),
680 repeat: None,
681 deliver: None,
682 };
683 let _ = create_task(&paths, input).expect("create");
684 mark_run(&paths, "int_task", true, None, None).expect("mark");
685 let task = get_task(&paths, "int_task")
686 .expect("get")
687 .expect("task exists");
688 assert!(task.enabled, "interval should stay enabled");
689 assert!(task.next_run_at.is_some(), "should have next_run_at");
690 let next = task.next_run_at.unwrap();
692 let now = chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true);
693 assert!(next > now, "next_run_at={} should be > now={}", next, now);
694 let _ = std::fs::remove_dir_all(paths.root());
695 }
696
697 #[test]
698 fn mark_run_once_disables() {
699 let paths = temp_paths("once-disable");
700 let input = CreateTaskInput {
701 id: Some("once_task".to_string()),
702 name: "run once".to_string(),
703 schedule: "once".to_string(),
704 parsed: ParsedSchedule {
705 kind: ParsedScheduleKind::Once,
706 run_at: None,
707 minutes: None,
708 expr: None,
709 display: "run once".to_string(),
710 },
711 prompt: "once test".to_string(),
712 working_dir: "/tmp".to_string(),
713 chat_id: "oc_test".to_string(),
714 root_message_id: None,
715 scope: Some("thread".to_string()),
716 chat_type: None,
717 lark_app_id: None,
718 creator_chat_id: None,
719 creator_root_message_id: None,
720 creator_lark_app_id: None,
721 next_run_at: Some("2026-01-01T00:00:00.000Z".to_string()),
722 repeat: None,
723 deliver: None,
724 };
725 let _ = create_task(&paths, input).expect("create");
726 mark_run(&paths, "once_task", true, None, None).expect("mark");
727 let task = get_task(&paths, "once_task")
728 .expect("get")
729 .expect("task exists");
730 assert!(!task.enabled, "once should be disabled after run");
731 assert!(task.next_run_at.is_none(), "once should clear next_run_at");
732 let _ = std::fs::remove_dir_all(paths.root());
733 }
734}