1use std::path::{Path, PathBuf};
36
37use serde::{Deserialize, Serialize};
38use serde_json::Value;
39
40use crate::{
41 ClaudeCronJob, ClaudeRuntimeManifest, ClaudeWakeup, DiscoveryQuery, HarnessCatalog,
42 HarnessHomes, HarnessId, Result, Session, SessionLocator,
43};
44
45pub const JOB_HARNESSES: &[&str] = &[
49 HarnessId::CLAUDE_CODE,
50 HarnessId::HERMES,
51 HarnessId::OPENCLAW,
52 HarnessId::ORCHESTRATOR,
53];
54
55pub const CLAUDE_SESSION_SCAN_LIMIT: usize = 200;
60
61const CLAUDE_JOB_MARKERS: &[&str] = &["CronCreate", "ScheduleWakeup"];
65
66const UNKNOWN_SCHEDULE: &str = "unknown";
68
69#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
71pub struct ScheduledJob {
72 pub id: String,
75 pub harness: String,
77 pub scope: JobScope,
79 pub profile: Option<String>,
81 pub session_id: Option<String>,
84 pub schedule: JobSchedule,
86 pub payload: JobPayload,
88 pub session_target: Option<String>,
92 pub deliver: JobDeliver,
94 pub enabled: bool,
96 pub state: String,
98 pub next_run_at: Option<String>,
100 pub last_run_at: Option<String>,
102 pub last_status: Option<String>,
104 #[serde(default)]
108 pub last_delivery_error: Option<String>,
109 pub created_at: Option<String>,
111 pub recurring: bool,
113}
114
115#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
117#[serde(rename_all = "snake_case")]
118pub enum JobScope {
119 Session,
121 Install,
123}
124
125#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
127pub struct JobSchedule {
128 pub kind: String,
130 pub expr: Option<String>,
132 pub minutes: Option<f64>,
134 pub run_at: Option<String>,
136 pub display: String,
138}
139
140#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
142pub struct JobPayload {
143 pub kind: String,
145 pub text: Option<String>,
147}
148
149#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
159pub struct JobDeliver {
160 pub target: Option<String>,
165 pub chat_id: Option<String>,
168 pub thread_id: Option<String>,
171 pub account: Option<String>,
175 pub mode: Option<String>,
178}
179
180impl JobDeliver {
181 fn to(target: &str) -> Self {
184 Self {
185 target: Some(target.to_string()),
186 chat_id: None,
187 thread_id: None,
188 account: None,
189 mode: None,
190 }
191 }
192}
193
194#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
196pub struct JobSource {
197 pub harness: String,
199 pub path: PathBuf,
201 pub state: String,
203 pub profile: Option<String>,
205 pub sessions_scanned: Option<usize>,
207 pub scan_limit: Option<usize>,
209 pub detail: Option<String>,
211}
212
213impl JobSource {
214 fn store(harness: &str, path: PathBuf, state: &str, profile: Option<String>) -> Self {
215 Self {
216 harness: harness.to_string(),
217 path,
218 state: state.to_string(),
219 profile,
220 sessions_scanned: None,
221 scan_limit: None,
222 detail: None,
223 }
224 }
225}
226
227#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
229pub struct JobsListing {
230 pub jobs: Vec<ScheduledJob>,
232 pub sources: Vec<JobSource>,
234}
235
236#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
238#[serde(default)]
239pub struct JobsQuery {
240 pub harness: Option<String>,
243 pub session: Option<String>,
245 pub profile: Option<String>,
247 pub homes: HarnessHomes,
249}
250
251pub fn supports_jobs(harness: &str) -> bool {
253 JOB_HARNESSES.contains(&harness)
254}
255
256pub fn list_jobs(query: &JobsQuery) -> Result<JobsListing> {
261 let mut jobs = Vec::new();
262 let mut sources = Vec::new();
263 let wanted = query.harness.as_deref();
264 if wanted.is_none_or(|harness| harness == HarnessId::CLAUDE_CODE) {
265 collect_claude_jobs(query, &mut jobs, &mut sources)?;
266 }
267 if wanted.is_none_or(|harness| harness == HarnessId::HERMES) {
268 collect_hermes_jobs(query, &mut jobs, &mut sources);
269 }
270 if wanted.is_none_or(|harness| harness == HarnessId::OPENCLAW) {
271 collect_openclaw_jobs(query, &mut jobs, &mut sources);
272 }
273 if wanted.is_none_or(|harness| harness == HarnessId::ORCHESTRATOR) {
274 collect_hermes_shaped_jobs(HarnessId::ORCHESTRATOR, query, &mut jobs, &mut sources);
275 }
276 jobs.retain(|job| {
277 query
278 .session
279 .as_deref()
280 .is_none_or(|session| job.session_id.as_deref() == Some(session))
281 && query
282 .profile
283 .as_deref()
284 .is_none_or(|profile| job.profile.as_deref() == Some(profile))
285 });
286 Ok(JobsListing { jobs, sources })
287}
288
289pub fn get_job(
292 harness: &str,
293 id: &str,
294 homes: &HarnessHomes,
295) -> Result<Option<(ScheduledJob, Value)>> {
296 let listing = list_jobs(&JobsQuery {
297 harness: Some(harness.to_string()),
298 homes: homes.clone(),
299 ..JobsQuery::default()
300 })?;
301 let Some(job) = listing.jobs.into_iter().find(|job| job.id == id) else {
302 return Ok(None);
303 };
304 let source = native_record(&job, homes)?;
305 Ok(Some((job, source)))
306}
307
308fn native_record(job: &ScheduledJob, homes: &HarnessHomes) -> Result<Value> {
311 match job.harness.as_str() {
312 HarnessId::CLAUDE_CODE => claude_native_record(job, homes),
313 HarnessId::HERMES | HarnessId::OPENCLAW | HarnessId::ORCHESTRATOR => {
314 for store in job_store_paths(&job.harness, homes) {
315 for record in read_job_array(&store.path) {
316 if record_id(&record).as_deref() == Some(job.id.as_str()) {
317 return Ok(record);
318 }
319 }
320 }
321 Ok(Value::Null)
322 }
323 _ => Ok(Value::Null),
324 }
325}
326
327fn claude_native_record(job: &ScheduledJob, homes: &HarnessHomes) -> Result<Value> {
328 let Some(session_id) = job.session_id.as_deref() else {
329 return Ok(Value::Null);
330 };
331 for locator in claude_locators(homes, Some(session_id), usize::MAX)? {
332 let Ok(session) = Session::load(locator.storage.path()) else {
333 continue;
334 };
335 let Ok(manifest) = ClaudeRuntimeManifest::from_session(&session) else {
336 continue;
337 };
338 if let Some(cron) = manifest
339 .active_crons
340 .iter()
341 .find(|cron| cron.id == job.id)
342 .cloned()
343 {
344 return Ok(serde_json::to_value(cron)?);
345 }
346 if let Some(wakeup) = manifest
347 .pending_wakeups
348 .iter()
349 .find(|wakeup| wakeup.tool_use_id == job.id)
350 .cloned()
351 {
352 return Ok(serde_json::to_value(wakeup)?);
353 }
354 }
355 Ok(Value::Null)
356}
357
358fn claude_locators(
365 homes: &HarnessHomes,
366 session: Option<&str>,
367 limit: usize,
368) -> Result<Vec<SessionLocator>> {
369 let query = DiscoveryQuery {
370 harnesses: vec![HarnessId::new(HarnessId::CLAUDE_CODE)],
371 homes: homes.clone(),
372 limit: (limit != usize::MAX).then_some(limit),
373 ..DiscoveryQuery::default()
374 };
375 let mut found = HarnessCatalog::new().discover(&query)?;
376 if let Some(session) = session {
377 found.retain(|descriptor| descriptor.locator.session_id == session);
378 }
379 Ok(found
380 .into_iter()
381 .map(|descriptor| descriptor.locator)
382 .collect())
383}
384
385fn mentions_a_scheduling_tool(path: &Path) -> bool {
389 use std::io::BufRead;
390 let Ok(file) = std::fs::File::open(path) else {
391 return false;
392 };
393 for line in std::io::BufReader::new(file)
394 .lines()
395 .map_while(std::result::Result::ok)
396 {
397 if CLAUDE_JOB_MARKERS
398 .iter()
399 .any(|marker| line.contains(marker))
400 {
401 return true;
402 }
403 }
404 false
405}
406
407fn collect_claude_jobs(
408 query: &JobsQuery,
409 jobs: &mut Vec<ScheduledJob>,
410 sources: &mut Vec<JobSource>,
411) -> Result<()> {
412 let session = query.session.as_deref();
413 let limit = if session.is_some() {
414 usize::MAX
415 } else {
416 CLAUDE_SESSION_SCAN_LIMIT
417 };
418 let locators = claude_locators(&query.homes, session, limit)?;
419 let mut scanned = 0usize;
420 for locator in locators {
421 scanned += 1;
422 if !mentions_a_scheduling_tool(locator.storage.path()) {
423 continue;
424 }
425 let Ok(loaded) = Session::load(locator.storage.path()) else {
426 sources.push(JobSource {
427 detail: Some("session could not be loaded".into()),
428 ..JobSource::store(
429 HarnessId::CLAUDE_CODE,
430 locator.storage.path().to_path_buf(),
431 "unreadable",
432 None,
433 )
434 });
435 continue;
436 };
437 let manifest = ClaudeRuntimeManifest::from_session(&loaded)?;
438 for cron in &manifest.active_crons {
439 jobs.push(claude_cron_row(&locator.session_id, cron));
440 }
441 for wakeup in &manifest.pending_wakeups {
442 jobs.push(claude_wakeup_row(&locator.session_id, wakeup));
443 }
444 }
445 sources.push(JobSource {
446 sessions_scanned: Some(scanned),
447 scan_limit: (session.is_none()).then_some(CLAUDE_SESSION_SCAN_LIMIT),
448 ..JobSource::store(
449 HarnessId::CLAUDE_CODE,
450 query.homes.claude_code.clone(),
451 "scanned",
452 None,
453 )
454 });
455 Ok(())
456}
457
458fn claude_cron_row(session_id: &str, cron: &ClaudeCronJob) -> ScheduledJob {
459 ScheduledJob {
460 id: cron.id.clone(),
461 harness: HarnessId::CLAUDE_CODE.into(),
462 scope: JobScope::Session,
463 profile: None,
464 session_id: Some(session_id.to_string()),
465 schedule: JobSchedule {
466 kind: "cron".into(),
467 expr: Some(cron.schedule.clone()),
468 minutes: None,
469 run_at: None,
470 display: cron.schedule.clone(),
471 },
472 payload: JobPayload {
473 kind: "prompt".into(),
474 text: Some(cron.prompt.clone()),
475 },
476 session_target: None,
477 deliver: JobDeliver::to("session"),
480 enabled: true,
481 state: "active".into(),
482 next_run_at: None,
486 last_run_at: None,
487 last_status: None,
488 last_delivery_error: None,
489 created_at: cron.created_at.clone(),
490 recurring: cron.recurring,
491 }
492}
493
494fn claude_wakeup_row(session_id: &str, wakeup: &ClaudeWakeup) -> ScheduledJob {
495 ScheduledJob {
496 id: wakeup.tool_use_id.clone(),
497 harness: HarnessId::CLAUDE_CODE.into(),
498 scope: JobScope::Session,
499 profile: None,
500 session_id: Some(session_id.to_string()),
501 schedule: JobSchedule {
502 kind: "once".into(),
503 expr: None,
504 minutes: None,
505 run_at: wakeup.scheduled_for.clone(),
506 display: format!("once, +{}s", wakeup.delay_seconds),
507 },
508 payload: JobPayload {
509 kind: "wakeup".into(),
510 text: wakeup.prompt.clone().or_else(|| wakeup.reason.clone()),
511 },
512 session_target: None,
513 deliver: JobDeliver::to("session"),
514 enabled: true,
515 state: "pending".into(),
516 next_run_at: wakeup.scheduled_for.clone(),
517 last_run_at: None,
518 last_status: None,
519 last_delivery_error: None,
520 created_at: wakeup.created_at.clone(),
521 recurring: false,
522 }
523}
524
525struct JobStore {
531 path: PathBuf,
532 profile: Option<String>,
533}
534
535fn job_store_paths(harness: &str, homes: &HarnessHomes) -> Vec<JobStore> {
542 match harness {
543 HarnessId::HERMES => {
544 let home = homes
547 .hermes
548 .parent()
549 .map_or_else(|| PathBuf::from("."), Path::to_path_buf);
550 let mut stores = vec![JobStore {
551 path: home.join("cron/jobs.json"),
552 profile: None,
553 }];
554 let profiles = home.join("profiles");
555 if let Ok(entries) = std::fs::read_dir(&profiles) {
556 let mut found: Vec<JobStore> = entries
557 .flatten()
558 .filter(|entry| entry.path().is_dir())
559 .map(|entry| JobStore {
560 path: entry.path().join("cron/jobs.json"),
561 profile: entry.file_name().to_string_lossy().into_owned().into(),
562 })
563 .collect();
564 found.sort_by(|left, right| left.profile.cmp(&right.profile));
565 stores.extend(found);
566 }
567 stores
568 }
569 HarnessId::ORCHESTRATOR => crate::orchestrator_profile_dirs(&homes.orchestrator)
574 .into_iter()
575 .map(|(name, dir)| JobStore {
576 path: dir.join("cron/jobs.json"),
577 profile: (name != "default").then_some(name),
578 })
579 .collect(),
580 HarnessId::OPENCLAW => vec![
581 JobStore {
586 path: homes.openclaw.join("state/openclaw.sqlite"),
587 profile: None,
588 },
589 JobStore {
592 path: homes.openclaw.join("cron/jobs.json"),
593 profile: None,
594 },
595 ],
596 _ => Vec::new(),
597 }
598}
599
600pub(crate) fn read_job_array(path: &Path) -> Vec<Value> {
605 let Ok(text) = std::fs::read_to_string(path) else {
606 return Vec::new();
607 };
608 let Ok(value) = serde_json::from_str::<Value>(&text) else {
609 return Vec::new();
610 };
611 match value {
612 Value::Array(items) => items,
613 Value::Object(map) => map
614 .get("jobs")
615 .and_then(Value::as_array)
616 .cloned()
617 .unwrap_or_default(),
618 _ => Vec::new(),
619 }
620}
621
622pub(crate) fn record_id(record: &Value) -> Option<String> {
623 ["id", "job_id", "jobId"]
624 .iter()
625 .find_map(|key| record.get(*key).and_then(Value::as_str))
626 .map(str::to_string)
627}
628
629fn collect_hermes_jobs(
630 query: &JobsQuery,
631 jobs: &mut Vec<ScheduledJob>,
632 sources: &mut Vec<JobSource>,
633) {
634 collect_hermes_shaped_jobs(HarnessId::HERMES, query, jobs, sources);
635}
636
637fn collect_hermes_shaped_jobs(
650 harness: &str,
651 query: &JobsQuery,
652 jobs: &mut Vec<ScheduledJob>,
653 sources: &mut Vec<JobSource>,
654) {
655 use supercode_interchange::orchestration::codec::{from_hermes, load_home, Flavor};
656 let loaded = match harness {
657 HarnessId::HERMES => {
658 let home = query
659 .homes
660 .hermes
661 .parent()
662 .map_or_else(|| PathBuf::from("."), Path::to_path_buf);
663 from_hermes(&home)
664 }
665 _ => load_home(&query.homes.orchestrator, Flavor::Orchestrator),
666 };
667 let loaded = match loaded {
668 Ok(loaded) => loaded,
669 Err(error) => {
670 for store in job_store_paths(harness, &query.homes) {
671 let mut source = JobSource::store(
672 harness,
673 store.path.clone(),
674 if store.path.exists() {
675 "unreadable"
676 } else {
677 "absent_store"
678 },
679 store.profile.clone(),
680 );
681 source.detail = Some(error.to_string());
682 sources.push(source);
683 }
684 return;
685 }
686 };
687 let mut names: Vec<&String> = loaded.orchestration.profiles.keys().collect();
688 names.sort_by_key(|name| (name.as_str() != "default", name.as_str()));
689 for name in names {
690 let profile = &loaded.orchestration.profiles[name];
691 let profile_name = (name != "default").then(|| name.clone());
692 let path = profile.dir.join("cron/jobs.json");
693 sources.push(JobSource::store(
694 harness,
695 path.clone(),
696 if path.exists() {
697 "read"
698 } else {
699 "absent_store"
700 },
701 profile_name.clone(),
702 ));
703 for job in profile.jobs.values() {
704 jobs.push(ScheduledJob::from_job(harness, profile_name.clone(), job));
705 }
706 }
707}
708
709fn collect_openclaw_jobs(
713 query: &JobsQuery,
714 jobs: &mut Vec<ScheduledJob>,
715 sources: &mut Vec<JobSource>,
716) {
717 use supercode_interchange::orchestration::codec::from_openclaw;
718 let store = query.homes.openclaw.join("state/openclaw.sqlite");
719 if !store.exists() {
720 sources.push(JobSource::store(
721 HarnessId::OPENCLAW,
722 store.clone(),
723 "absent_store",
724 None,
725 ));
726 } else {
727 match from_openclaw(&query.homes.openclaw) {
728 Ok(loaded) => {
729 sources.push(JobSource::store(
730 HarnessId::OPENCLAW,
731 store.clone(),
732 "read",
733 None,
734 ));
735 let mut names: Vec<&String> = loaded.orchestration.profiles.keys().collect();
736 names.sort_by_key(|name| (name.as_str() != "default", name.as_str()));
737 for name in names {
738 let profile_name = (name != "default").then(|| name.clone());
739 for job in loaded.orchestration.profiles[name].jobs.values() {
740 jobs.push(ScheduledJob::from_job(
741 HarnessId::OPENCLAW,
742 profile_name.clone(),
743 job,
744 ));
745 }
746 }
747 }
748 Err(error) => {
749 let mut source =
750 JobSource::store(HarnessId::OPENCLAW, store.clone(), "unreadable", None);
751 source.detail = Some(error.to_string());
752 sources.push(source);
753 }
754 }
755 }
756 let legacy = query.homes.openclaw.join("cron/jobs.json");
759 sources.push(JobSource::store(
760 HarnessId::OPENCLAW,
761 legacy.clone(),
762 if legacy.exists() {
763 "read"
764 } else {
765 "absent_store"
766 },
767 None,
768 ));
769}
770
771fn text_field(record: &Value, keys: &[&str]) -> Option<String> {
772 keys.iter()
773 .find_map(|key| record.get(*key).and_then(Value::as_str))
774 .map(str::to_string)
775}
776
777fn schedule_display(
778 kind: &str,
779 expr: &Option<String>,
780 minutes: Option<f64>,
781 run_at: &Option<String>,
782) -> String {
783 match kind {
784 "cron" => expr.clone().unwrap_or_else(|| UNKNOWN_SCHEDULE.into()),
785 "interval" => minutes.map_or_else(
786 || UNKNOWN_SCHEDULE.to_string(),
787 |minutes| format!("every {} min", trim_float(minutes)),
788 ),
789 "once" => run_at
790 .clone()
791 .map_or_else(|| "once".to_string(), |run_at| format!("once @{run_at}")),
792 _ => UNKNOWN_SCHEDULE.into(),
793 }
794}
795
796fn trim_float(value: f64) -> String {
797 if (value.fract()).abs() < f64::EPSILON {
798 format!("{}", value as i64)
799 } else {
800 format!("{value}")
801 }
802}
803
804impl ScheduledJob {
805 pub fn from_job(
811 harness: &str,
812 profile: Option<String>,
813 job: &supercode_interchange::orchestration::Job,
814 ) -> Self {
815 use supercode_interchange::orchestration::{Schedule, Target};
816 let (kind, expr, minutes, run_at) = match &job.schedule {
817 Schedule::Once { run_at } => ("once", None, None, Some(run_at.clone())),
818 Schedule::Interval { minutes } => ("interval", None, Some(*minutes), None),
819 Schedule::Cron { expr, .. } => ("cron", Some(expr.clone()), None, None),
820 };
821 let script = job.residue.0.get("script").and_then(Value::as_str);
822 let payload = match script {
823 Some(script) => JobPayload {
824 kind: "script".into(),
825 text: Some(script.to_string()),
826 },
827 None => JobPayload {
828 kind: "prompt".into(),
829 text: job.prompt.clone(),
830 },
831 };
832 let payload = match job.residue.0.get("__payload") {
835 Some(native) => openclaw_payload(&serde_json::json!({ "payload": native })),
836 None => payload,
837 };
838 let deliver = Some(job.deliver.render());
839 let (explicit_chat, explicit_thread) = match &job.deliver {
840 Target::Explicit {
841 chat_id, thread_id, ..
842 } => (chat_id.clone(), thread_id.clone()),
843 _ => (None, None),
844 };
845 let chat_id = job
846 .origin
847 .as_ref()
848 .and_then(|o| o.chat_id.clone())
849 .or(explicit_chat);
850 let thread_id = job
851 .origin
852 .as_ref()
853 .and_then(|o| o.thread_id.clone())
854 .or(explicit_thread);
855 let enabled = job.enabled;
856 let oc_delivery = job.residue.0.get("__delivery").and_then(Value::as_object);
860 let oc_text = |key: &str| {
861 oc_delivery
862 .and_then(|d| d.get(key))
863 .and_then(Value::as_str)
864 .map(str::to_string)
865 };
866 let deliver = if harness == HarnessId::OPENCLAW {
868 oc_text("channel")
869 } else {
870 deliver
871 };
872 let chat_id = oc_text("to").or(chat_id);
873 let thread_id = oc_text("threadId").or(thread_id);
874 let session_target = job
875 .residue
876 .0
877 .get("__session_target")
878 .and_then(Value::as_str)
879 .map(str::to_string);
880 let profile = if harness == HarnessId::OPENCLAW {
883 job.residue
884 .0
885 .get("agentId")
886 .or_else(|| job.residue.0.get("agent_id"))
887 .and_then(Value::as_str)
888 .map(str::to_string)
889 .or(profile)
890 } else {
891 profile
892 };
893 Self {
894 id: job.id.clone(),
895 harness: harness.into(),
896 scope: JobScope::Install,
897 profile,
898 session_id: None,
899 schedule: JobSchedule {
900 display: schedule_display(kind, &expr, minutes, &run_at),
901 kind: kind.into(),
902 expr,
903 minutes,
904 run_at,
905 },
906 payload,
907 session_target,
908 deliver: JobDeliver {
909 target: deliver,
910 chat_id,
911 thread_id,
912 account: oc_text("accountId"),
913 mode: oc_text("mode"),
914 },
915 enabled,
916 state: if enabled { "active" } else { "paused" }.into(),
917 next_run_at: job.next_run_at.clone(),
918 last_run_at: job.last_run_at.clone(),
919 last_status: job.last_status.clone(),
920 last_delivery_error: job
922 .residue
923 .0
924 .get("last_delivery_error")
925 .and_then(Value::as_str)
926 .map(str::to_string),
927 created_at: job.created_at.clone(),
928 recurring: kind != "once",
929 }
930 }
931}
932
933fn openclaw_payload(record: &Value) -> JobPayload {
934 let payload = record.get("payload").cloned().unwrap_or(Value::Null);
935 let native = text_field(&payload, &["kind", "type"])
936 .or_else(|| text_field(record, &["payloadKind", "payload_kind"]));
937 let text = text_field(&payload, &["text", "message", "command", "script"])
938 .or_else(|| text_field(record, &["message", "command", "script"]));
939 let kind = match native.as_deref() {
940 Some("systemEvent" | "system_event") => "system_event",
941 Some("message" | "prompt" | "agentTurn" | "agent_turn") => "prompt",
942 Some("command") => "command",
943 Some("script") => "script",
944 Some(_) | None => {
945 if payload_has(record, &payload, "systemEvent") {
946 "system_event"
947 } else if payload_has(record, &payload, "command") {
948 "command"
949 } else if payload_has(record, &payload, "script") {
950 "script"
951 } else {
952 "prompt"
953 }
954 }
955 };
956 JobPayload {
957 kind: kind.into(),
958 text,
959 }
960}
961
962fn payload_has(record: &Value, payload: &Value, key: &str) -> bool {
963 payload.get(key).is_some() || record.get(key).is_some()
964}