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