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:{:x}", hasher.finalize()))
451}
452
453fn canonical_json(value: &serde_json::Value) -> String {
454 match value {
455 serde_json::Value::Null => "null".to_string(),
456 serde_json::Value::Bool(v) => if *v { "true" } else { "false" }.to_string(),
457 serde_json::Value::Number(v) => v.to_string(),
458 serde_json::Value::String(v) => serde_json::to_string(v).expect("string serializable"),
459 serde_json::Value::Array(items) => {
460 let mut out = String::from("[");
461 let mut first = true;
462 for item in items {
463 if !first {
464 out.push(',');
465 }
466 first = false;
467 out.push_str(&canonical_json(item));
468 }
469 out.push(']');
470 out
471 }
472 serde_json::Value::Object(map) => {
473 let mut keys: Vec<_> = map.keys().collect();
474 keys.sort();
475 let mut out = String::from("{");
476 let mut first = true;
477 for key in keys {
478 if !first {
479 out.push(',');
480 }
481 first = false;
482 out.push_str(&serde_json::to_string(key).expect("key serializable"));
483 out.push(':');
484 out.push_str(&canonical_json(&map[key]));
485 }
486 out.push('}');
487 out
488 }
489 }
490}
491
492fn task_output_dir(paths: &BeamPaths, task_id: &str) -> PathBuf {
493 paths.schedules_output_dir().join(task_id)
494}
495
496struct UtcNow;
497
498impl UtcNow {
499 fn now() -> String {
500 chrono::Utc::now().to_rfc3339_opts(SecondsFormat::Millis, true)
501 }
502}
503
504fn default_true() -> bool {
505 true
506}
507
508#[cfg(test)]
509mod tests {
510 use super::*;
511 use std::time::{SystemTime, UNIX_EPOCH};
512
513 fn temp_paths(label: &str) -> BeamPaths {
514 let nanos = SystemTime::now()
515 .duration_since(UNIX_EPOCH)
516 .unwrap_or_default()
517 .as_nanos();
518 BeamPaths::from_root(std::env::temp_dir().join(format!(
519 "beam-schedule-store-{label}-{nanos}-{}",
520 std::process::id()
521 )))
522 }
523
524 fn parsed_cron() -> ParsedSchedule {
525 ParsedSchedule {
526 kind: ParsedScheduleKind::Cron,
527 run_at: None,
528 minutes: None,
529 expr: Some("0 9 * * *".to_string()),
530 display: "0 9 * * *".to_string(),
531 }
532 }
533
534 #[test]
535 fn create_task_returns_existing_when_canonical_input_matches() {
536 let paths = temp_paths("identical");
537 let input = CreateTaskInput {
538 id: Some("wf_task".to_string()),
539 name: "schedule-demo daily 9am".to_string(),
540 schedule: "0 9 * * *".to_string(),
541 parsed: parsed_cron(),
542 prompt: "Schedule demo: run workflow self-check.".to_string(),
543 working_dir: "/tmp/beam-schedule-demo".to_string(),
544 chat_id: "oc_workflow_demo".to_string(),
545 root_message_id: None,
546 scope: Some("thread".to_string()),
547 chat_type: None,
548 lark_app_id: None,
549 creator_chat_id: None,
550 creator_root_message_id: None,
551 creator_lark_app_id: None,
552 next_run_at: None,
553 repeat: None,
554 deliver: None,
555 };
556 let created = create_task(&paths, input.clone()).expect("create");
557 let returned = create_task(&paths, input).expect("create existing");
558 assert_eq!(created.id, "wf_task");
559 assert_eq!(returned.id, "wf_task");
560 assert_eq!(list_tasks(&paths).expect("list").len(), 1);
561 let _ = std::fs::remove_dir_all(paths.root());
562 }
563
564 #[test]
565 fn create_task_conflicts_when_canonical_input_differs() {
566 let paths = temp_paths("conflict");
567 let input = CreateTaskInput {
568 id: Some("wf_task".to_string()),
569 name: "schedule-demo daily 9am".to_string(),
570 schedule: "0 9 * * *".to_string(),
571 parsed: parsed_cron(),
572 prompt: "Schedule demo: run workflow self-check.".to_string(),
573 working_dir: "/tmp/beam-schedule-demo".to_string(),
574 chat_id: "oc_workflow_demo".to_string(),
575 root_message_id: None,
576 scope: Some("thread".to_string()),
577 chat_type: None,
578 lark_app_id: None,
579 creator_chat_id: None,
580 creator_root_message_id: None,
581 creator_lark_app_id: None,
582 next_run_at: None,
583 repeat: None,
584 deliver: None,
585 };
586 let _ = create_task(&paths, input).expect("create");
587 let changed = CreateTaskInput {
588 id: Some("wf_task".to_string()),
589 name: "schedule-demo daily 9am".to_string(),
590 schedule: "0 9 * * *".to_string(),
591 parsed: parsed_cron(),
592 prompt: "changed prompt".to_string(),
593 working_dir: "/tmp/beam-schedule-demo".to_string(),
594 chat_id: "oc_workflow_demo".to_string(),
595 root_message_id: None,
596 scope: Some("thread".to_string()),
597 chat_type: None,
598 lark_app_id: None,
599 creator_chat_id: None,
600 creator_root_message_id: None,
601 creator_lark_app_id: None,
602 next_run_at: None,
603 repeat: None,
604 deliver: None,
605 };
606 let err = create_task(&paths, changed).expect_err("conflict");
607 assert!(matches!(
608 err,
609 ScheduleStoreError::IdempotencyConflict { .. }
610 ));
611 let _ = std::fs::remove_dir_all(paths.root());
612 }
613
614 #[test]
615 fn mark_run_removes_finite_repeat_after_completion() {
616 let paths = temp_paths("mark");
617 let input = CreateTaskInput {
618 id: Some("wf_task".to_string()),
619 name: "schedule-demo daily 9am".to_string(),
620 schedule: "0 9 * * *".to_string(),
621 parsed: parsed_cron(),
622 prompt: "Schedule demo: run workflow self-check.".to_string(),
623 working_dir: "/tmp/beam-schedule-demo".to_string(),
624 chat_id: "oc_workflow_demo".to_string(),
625 root_message_id: None,
626 scope: Some("thread".to_string()),
627 chat_type: None,
628 lark_app_id: None,
629 creator_chat_id: None,
630 creator_root_message_id: None,
631 creator_lark_app_id: None,
632 next_run_at: None,
633 repeat: Some(ScheduleRepeat {
634 times: Some(1),
635 completed: 0,
636 }),
637 deliver: None,
638 };
639 let _ = create_task(&paths, input).expect("create");
640 mark_run(&paths, "wf_task", true, None, None).expect("mark run");
641 assert!(get_task(&paths, "wf_task").expect("get").is_none());
642 let _ = std::fs::remove_dir_all(paths.root());
643 }
644
645 #[test]
646 fn mark_run_interval_advances_next_run_at() {
647 let paths = temp_paths("interval-advance");
648 let input = CreateTaskInput {
649 id: Some("int_task".to_string()),
650 name: "every 10 min".to_string(),
651 schedule: "every 10 min".to_string(),
652 parsed: ParsedSchedule {
653 kind: ParsedScheduleKind::Interval,
654 run_at: None,
655 minutes: Some(10),
656 expr: None,
657 display: "every 10 min".to_string(),
658 },
659 prompt: "interval test".to_string(),
660 working_dir: "/tmp".to_string(),
661 chat_id: "oc_test".to_string(),
662 root_message_id: None,
663 scope: Some("thread".to_string()),
664 chat_type: None,
665 lark_app_id: None,
666 creator_chat_id: None,
667 creator_root_message_id: None,
668 creator_lark_app_id: None,
669 next_run_at: Some("2026-01-01T00:00:00.000Z".to_string()),
670 repeat: None,
671 deliver: None,
672 };
673 let _ = create_task(&paths, input).expect("create");
674 mark_run(&paths, "int_task", true, None, None).expect("mark");
675 let task = get_task(&paths, "int_task")
676 .expect("get")
677 .expect("task exists");
678 assert!(task.enabled, "interval should stay enabled");
679 assert!(task.next_run_at.is_some(), "should have next_run_at");
680 let next = task.next_run_at.unwrap();
682 let now = chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true);
683 assert!(next > now, "next_run_at={} should be > now={}", next, now);
684 let _ = std::fs::remove_dir_all(paths.root());
685 }
686
687 #[test]
688 fn mark_run_once_disables() {
689 let paths = temp_paths("once-disable");
690 let input = CreateTaskInput {
691 id: Some("once_task".to_string()),
692 name: "run once".to_string(),
693 schedule: "once".to_string(),
694 parsed: ParsedSchedule {
695 kind: ParsedScheduleKind::Once,
696 run_at: None,
697 minutes: None,
698 expr: None,
699 display: "run once".to_string(),
700 },
701 prompt: "once test".to_string(),
702 working_dir: "/tmp".to_string(),
703 chat_id: "oc_test".to_string(),
704 root_message_id: None,
705 scope: Some("thread".to_string()),
706 chat_type: None,
707 lark_app_id: None,
708 creator_chat_id: None,
709 creator_root_message_id: None,
710 creator_lark_app_id: None,
711 next_run_at: Some("2026-01-01T00:00:00.000Z".to_string()),
712 repeat: None,
713 deliver: None,
714 };
715 let _ = create_task(&paths, input).expect("create");
716 mark_run(&paths, "once_task", true, None, None).expect("mark");
717 let task = get_task(&paths, "once_task")
718 .expect("get")
719 .expect("task exists");
720 assert!(!task.enabled, "once should be disabled after run");
721 assert!(task.next_run_at.is_none(), "once should clear next_run_at");
722 let _ = std::fs::remove_dir_all(paths.root());
723 }
724}