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 pub created_at: Option<String>,
106 pub recurring: bool,
108}
109
110#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
112#[serde(rename_all = "snake_case")]
113pub enum JobScope {
114 Session,
116 Install,
118}
119
120#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
122pub struct JobSchedule {
123 pub kind: String,
125 pub expr: Option<String>,
127 pub minutes: Option<f64>,
129 pub run_at: Option<String>,
131 pub display: String,
133}
134
135#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
137pub struct JobPayload {
138 pub kind: String,
140 pub text: Option<String>,
142}
143
144#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
154pub struct JobDeliver {
155 pub target: Option<String>,
160 pub chat_id: Option<String>,
163 pub thread_id: Option<String>,
166 pub account: Option<String>,
170 pub mode: Option<String>,
173}
174
175impl JobDeliver {
176 fn to(target: &str) -> Self {
179 Self {
180 target: Some(target.to_string()),
181 chat_id: None,
182 thread_id: None,
183 account: None,
184 mode: None,
185 }
186 }
187}
188
189#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
191pub struct JobSource {
192 pub harness: String,
194 pub path: PathBuf,
196 pub state: String,
198 pub profile: Option<String>,
200 pub sessions_scanned: Option<usize>,
202 pub scan_limit: Option<usize>,
204 pub detail: Option<String>,
206}
207
208impl JobSource {
209 fn store(harness: &str, path: PathBuf, state: &str, profile: Option<String>) -> Self {
210 Self {
211 harness: harness.to_string(),
212 path,
213 state: state.to_string(),
214 profile,
215 sessions_scanned: None,
216 scan_limit: None,
217 detail: None,
218 }
219 }
220}
221
222#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
224pub struct JobsListing {
225 pub jobs: Vec<ScheduledJob>,
227 pub sources: Vec<JobSource>,
229}
230
231#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
233#[serde(default)]
234pub struct JobsQuery {
235 pub harness: Option<String>,
238 pub session: Option<String>,
240 pub profile: Option<String>,
242 pub homes: HarnessHomes,
244}
245
246pub fn supports_jobs(harness: &str) -> bool {
248 JOB_HARNESSES.contains(&harness)
249}
250
251pub fn list_jobs(query: &JobsQuery) -> Result<JobsListing> {
256 let mut jobs = Vec::new();
257 let mut sources = Vec::new();
258 let wanted = query.harness.as_deref();
259 if wanted.is_none_or(|harness| harness == HarnessId::CLAUDE_CODE) {
260 collect_claude_jobs(query, &mut jobs, &mut sources)?;
261 }
262 if wanted.is_none_or(|harness| harness == HarnessId::HERMES) {
263 collect_hermes_jobs(query, &mut jobs, &mut sources);
264 }
265 if wanted.is_none_or(|harness| harness == HarnessId::OPENCLAW) {
266 collect_openclaw_jobs(query, &mut jobs, &mut sources);
267 }
268 if wanted.is_none_or(|harness| harness == HarnessId::ORCHESTRATOR) {
269 collect_hermes_shaped_jobs(HarnessId::ORCHESTRATOR, query, &mut jobs, &mut sources);
270 }
271 jobs.retain(|job| {
272 query
273 .session
274 .as_deref()
275 .is_none_or(|session| job.session_id.as_deref() == Some(session))
276 && query
277 .profile
278 .as_deref()
279 .is_none_or(|profile| job.profile.as_deref() == Some(profile))
280 });
281 Ok(JobsListing { jobs, sources })
282}
283
284pub fn get_job(
287 harness: &str,
288 id: &str,
289 homes: &HarnessHomes,
290) -> Result<Option<(ScheduledJob, Value)>> {
291 let listing = list_jobs(&JobsQuery {
292 harness: Some(harness.to_string()),
293 homes: homes.clone(),
294 ..JobsQuery::default()
295 })?;
296 let Some(job) = listing.jobs.into_iter().find(|job| job.id == id) else {
297 return Ok(None);
298 };
299 let source = native_record(&job, homes)?;
300 Ok(Some((job, source)))
301}
302
303fn native_record(job: &ScheduledJob, homes: &HarnessHomes) -> Result<Value> {
306 match job.harness.as_str() {
307 HarnessId::CLAUDE_CODE => claude_native_record(job, homes),
308 HarnessId::HERMES | HarnessId::OPENCLAW | HarnessId::ORCHESTRATOR => {
309 for store in job_store_paths(&job.harness, homes) {
310 for record in read_job_array(&store.path) {
311 if record_id(&record).as_deref() == Some(job.id.as_str()) {
312 return Ok(record);
313 }
314 }
315 }
316 Ok(Value::Null)
317 }
318 _ => Ok(Value::Null),
319 }
320}
321
322fn claude_native_record(job: &ScheduledJob, homes: &HarnessHomes) -> Result<Value> {
323 let Some(session_id) = job.session_id.as_deref() else {
324 return Ok(Value::Null);
325 };
326 for locator in claude_locators(homes, Some(session_id), usize::MAX)? {
327 let Ok(session) = Session::load(locator.storage.path()) else {
328 continue;
329 };
330 let Ok(manifest) = ClaudeRuntimeManifest::from_session(&session) else {
331 continue;
332 };
333 if let Some(cron) = manifest
334 .active_crons
335 .iter()
336 .find(|cron| cron.id == job.id)
337 .cloned()
338 {
339 return Ok(serde_json::to_value(cron)?);
340 }
341 if let Some(wakeup) = manifest
342 .pending_wakeups
343 .iter()
344 .find(|wakeup| wakeup.tool_use_id == job.id)
345 .cloned()
346 {
347 return Ok(serde_json::to_value(wakeup)?);
348 }
349 }
350 Ok(Value::Null)
351}
352
353fn claude_locators(
360 homes: &HarnessHomes,
361 session: Option<&str>,
362 limit: usize,
363) -> Result<Vec<SessionLocator>> {
364 let query = DiscoveryQuery {
365 harnesses: vec![HarnessId::new(HarnessId::CLAUDE_CODE)],
366 homes: homes.clone(),
367 limit: (limit != usize::MAX).then_some(limit),
368 ..DiscoveryQuery::default()
369 };
370 let mut found = HarnessCatalog::new().discover(&query)?;
371 if let Some(session) = session {
372 found.retain(|descriptor| descriptor.locator.session_id == session);
373 }
374 Ok(found
375 .into_iter()
376 .map(|descriptor| descriptor.locator)
377 .collect())
378}
379
380fn mentions_a_scheduling_tool(path: &Path) -> bool {
384 use std::io::BufRead;
385 let Ok(file) = std::fs::File::open(path) else {
386 return false;
387 };
388 for line in std::io::BufReader::new(file)
389 .lines()
390 .map_while(std::result::Result::ok)
391 {
392 if CLAUDE_JOB_MARKERS
393 .iter()
394 .any(|marker| line.contains(marker))
395 {
396 return true;
397 }
398 }
399 false
400}
401
402fn collect_claude_jobs(
403 query: &JobsQuery,
404 jobs: &mut Vec<ScheduledJob>,
405 sources: &mut Vec<JobSource>,
406) -> Result<()> {
407 let session = query.session.as_deref();
408 let limit = if session.is_some() {
409 usize::MAX
410 } else {
411 CLAUDE_SESSION_SCAN_LIMIT
412 };
413 let locators = claude_locators(&query.homes, session, limit)?;
414 let mut scanned = 0usize;
415 for locator in locators {
416 scanned += 1;
417 if !mentions_a_scheduling_tool(locator.storage.path()) {
418 continue;
419 }
420 let Ok(loaded) = Session::load(locator.storage.path()) else {
421 sources.push(JobSource {
422 detail: Some("session could not be loaded".into()),
423 ..JobSource::store(
424 HarnessId::CLAUDE_CODE,
425 locator.storage.path().to_path_buf(),
426 "unreadable",
427 None,
428 )
429 });
430 continue;
431 };
432 let manifest = ClaudeRuntimeManifest::from_session(&loaded)?;
433 for cron in &manifest.active_crons {
434 jobs.push(claude_cron_row(&locator.session_id, cron));
435 }
436 for wakeup in &manifest.pending_wakeups {
437 jobs.push(claude_wakeup_row(&locator.session_id, wakeup));
438 }
439 }
440 sources.push(JobSource {
441 sessions_scanned: Some(scanned),
442 scan_limit: (session.is_none()).then_some(CLAUDE_SESSION_SCAN_LIMIT),
443 ..JobSource::store(
444 HarnessId::CLAUDE_CODE,
445 query.homes.claude_code.clone(),
446 "scanned",
447 None,
448 )
449 });
450 Ok(())
451}
452
453fn claude_cron_row(session_id: &str, cron: &ClaudeCronJob) -> ScheduledJob {
454 ScheduledJob {
455 id: cron.id.clone(),
456 harness: HarnessId::CLAUDE_CODE.into(),
457 scope: JobScope::Session,
458 profile: None,
459 session_id: Some(session_id.to_string()),
460 schedule: JobSchedule {
461 kind: "cron".into(),
462 expr: Some(cron.schedule.clone()),
463 minutes: None,
464 run_at: None,
465 display: cron.schedule.clone(),
466 },
467 payload: JobPayload {
468 kind: "prompt".into(),
469 text: Some(cron.prompt.clone()),
470 },
471 session_target: None,
472 deliver: JobDeliver::to("session"),
475 enabled: true,
476 state: "active".into(),
477 next_run_at: None,
481 last_run_at: None,
482 last_status: None,
483 created_at: cron.created_at.clone(),
484 recurring: cron.recurring,
485 }
486}
487
488fn claude_wakeup_row(session_id: &str, wakeup: &ClaudeWakeup) -> ScheduledJob {
489 ScheduledJob {
490 id: wakeup.tool_use_id.clone(),
491 harness: HarnessId::CLAUDE_CODE.into(),
492 scope: JobScope::Session,
493 profile: None,
494 session_id: Some(session_id.to_string()),
495 schedule: JobSchedule {
496 kind: "once".into(),
497 expr: None,
498 minutes: None,
499 run_at: wakeup.scheduled_for.clone(),
500 display: format!("once, +{}s", wakeup.delay_seconds),
501 },
502 payload: JobPayload {
503 kind: "wakeup".into(),
504 text: wakeup.prompt.clone().or_else(|| wakeup.reason.clone()),
505 },
506 session_target: None,
507 deliver: JobDeliver::to("session"),
508 enabled: true,
509 state: "pending".into(),
510 next_run_at: wakeup.scheduled_for.clone(),
511 last_run_at: None,
512 last_status: None,
513 created_at: wakeup.created_at.clone(),
514 recurring: false,
515 }
516}
517
518struct JobStore {
524 path: PathBuf,
525 profile: Option<String>,
526}
527
528fn job_store_paths(harness: &str, homes: &HarnessHomes) -> Vec<JobStore> {
535 match harness {
536 HarnessId::HERMES => {
537 let home = homes
540 .hermes
541 .parent()
542 .map_or_else(|| PathBuf::from("."), Path::to_path_buf);
543 let mut stores = vec![JobStore {
544 path: home.join("cron/jobs.json"),
545 profile: None,
546 }];
547 let profiles = home.join("profiles");
548 if let Ok(entries) = std::fs::read_dir(&profiles) {
549 let mut found: Vec<JobStore> = entries
550 .flatten()
551 .filter(|entry| entry.path().is_dir())
552 .map(|entry| JobStore {
553 path: entry.path().join("cron/jobs.json"),
554 profile: entry.file_name().to_string_lossy().into_owned().into(),
555 })
556 .collect();
557 found.sort_by(|left, right| left.profile.cmp(&right.profile));
558 stores.extend(found);
559 }
560 stores
561 }
562 HarnessId::ORCHESTRATOR => crate::orchestrator_profile_dirs(&homes.orchestrator)
567 .into_iter()
568 .map(|(name, dir)| JobStore {
569 path: dir.join("cron/jobs.json"),
570 profile: (name != "default").then_some(name),
571 })
572 .collect(),
573 HarnessId::OPENCLAW => vec![
574 JobStore {
579 path: homes.openclaw.join("state/openclaw.sqlite"),
580 profile: None,
581 },
582 JobStore {
585 path: homes.openclaw.join("cron/jobs.json"),
586 profile: None,
587 },
588 ],
589 _ => Vec::new(),
590 }
591}
592
593pub(crate) fn read_job_array(path: &Path) -> Vec<Value> {
598 let Ok(text) = std::fs::read_to_string(path) else {
599 return Vec::new();
600 };
601 let Ok(value) = serde_json::from_str::<Value>(&text) else {
602 return Vec::new();
603 };
604 match value {
605 Value::Array(items) => items,
606 Value::Object(map) => map
607 .get("jobs")
608 .and_then(Value::as_array)
609 .cloned()
610 .unwrap_or_default(),
611 _ => Vec::new(),
612 }
613}
614
615pub(crate) fn record_id(record: &Value) -> Option<String> {
616 ["id", "job_id", "jobId"]
617 .iter()
618 .find_map(|key| record.get(*key).and_then(Value::as_str))
619 .map(str::to_string)
620}
621
622fn collect_hermes_jobs(
623 query: &JobsQuery,
624 jobs: &mut Vec<ScheduledJob>,
625 sources: &mut Vec<JobSource>,
626) {
627 collect_hermes_shaped_jobs(HarnessId::HERMES, query, jobs, sources);
628}
629
630fn collect_hermes_shaped_jobs(
643 harness: &str,
644 query: &JobsQuery,
645 jobs: &mut Vec<ScheduledJob>,
646 sources: &mut Vec<JobSource>,
647) {
648 use supercode_interchange::orchestration::codec::{from_hermes, load_home, Flavor};
649 let loaded = match harness {
650 HarnessId::HERMES => {
651 let home = query
652 .homes
653 .hermes
654 .parent()
655 .map_or_else(|| PathBuf::from("."), Path::to_path_buf);
656 from_hermes(&home)
657 }
658 _ => load_home(&query.homes.orchestrator, Flavor::Orchestrator),
659 };
660 let loaded = match loaded {
661 Ok(loaded) => loaded,
662 Err(error) => {
663 for store in job_store_paths(harness, &query.homes) {
664 let mut source = JobSource::store(
665 harness,
666 store.path.clone(),
667 if store.path.exists() {
668 "unreadable"
669 } else {
670 "absent_store"
671 },
672 store.profile.clone(),
673 );
674 source.detail = Some(error.to_string());
675 sources.push(source);
676 }
677 return;
678 }
679 };
680 let mut names: Vec<&String> = loaded.orchestration.profiles.keys().collect();
681 names.sort_by_key(|name| (name.as_str() != "default", name.as_str()));
682 for name in names {
683 let profile = &loaded.orchestration.profiles[name];
684 let profile_name = (name != "default").then(|| name.clone());
685 let path = profile.dir.join("cron/jobs.json");
686 sources.push(JobSource::store(
687 harness,
688 path.clone(),
689 if path.exists() {
690 "read"
691 } else {
692 "absent_store"
693 },
694 profile_name.clone(),
695 ));
696 for job in profile.jobs.values() {
697 jobs.push(ScheduledJob::from_job(harness, profile_name.clone(), job));
698 }
699 }
700}
701
702fn collect_openclaw_jobs(
706 query: &JobsQuery,
707 jobs: &mut Vec<ScheduledJob>,
708 sources: &mut Vec<JobSource>,
709) {
710 use supercode_interchange::orchestration::codec::from_openclaw;
711 let store = query.homes.openclaw.join("state/openclaw.sqlite");
712 if !store.exists() {
713 sources.push(JobSource::store(
714 HarnessId::OPENCLAW,
715 store.clone(),
716 "absent_store",
717 None,
718 ));
719 } else {
720 match from_openclaw(&query.homes.openclaw) {
721 Ok(loaded) => {
722 sources.push(JobSource::store(
723 HarnessId::OPENCLAW,
724 store.clone(),
725 "read",
726 None,
727 ));
728 let mut names: Vec<&String> = loaded.orchestration.profiles.keys().collect();
729 names.sort_by_key(|name| (name.as_str() != "default", name.as_str()));
730 for name in names {
731 let profile_name = (name != "default").then(|| name.clone());
732 for job in loaded.orchestration.profiles[name].jobs.values() {
733 jobs.push(ScheduledJob::from_job(
734 HarnessId::OPENCLAW,
735 profile_name.clone(),
736 job,
737 ));
738 }
739 }
740 }
741 Err(error) => {
742 let mut source =
743 JobSource::store(HarnessId::OPENCLAW, store.clone(), "unreadable", None);
744 source.detail = Some(error.to_string());
745 sources.push(source);
746 }
747 }
748 }
749 let legacy = query.homes.openclaw.join("cron/jobs.json");
752 sources.push(JobSource::store(
753 HarnessId::OPENCLAW,
754 legacy.clone(),
755 if legacy.exists() {
756 "read"
757 } else {
758 "absent_store"
759 },
760 None,
761 ));
762}
763
764fn text_field(record: &Value, keys: &[&str]) -> Option<String> {
765 keys.iter()
766 .find_map(|key| record.get(*key).and_then(Value::as_str))
767 .map(str::to_string)
768}
769
770fn schedule_display(
771 kind: &str,
772 expr: &Option<String>,
773 minutes: Option<f64>,
774 run_at: &Option<String>,
775) -> String {
776 match kind {
777 "cron" => expr.clone().unwrap_or_else(|| UNKNOWN_SCHEDULE.into()),
778 "interval" => minutes.map_or_else(
779 || UNKNOWN_SCHEDULE.to_string(),
780 |minutes| format!("every {} min", trim_float(minutes)),
781 ),
782 "once" => run_at
783 .clone()
784 .map_or_else(|| "once".to_string(), |run_at| format!("once @{run_at}")),
785 _ => UNKNOWN_SCHEDULE.into(),
786 }
787}
788
789fn trim_float(value: f64) -> String {
790 if (value.fract()).abs() < f64::EPSILON {
791 format!("{}", value as i64)
792 } else {
793 format!("{value}")
794 }
795}
796
797impl ScheduledJob {
798 pub fn from_job(
804 harness: &str,
805 profile: Option<String>,
806 job: &supercode_interchange::orchestration::Job,
807 ) -> Self {
808 use supercode_interchange::orchestration::{Schedule, Target};
809 let (kind, expr, minutes, run_at) = match &job.schedule {
810 Schedule::Once { run_at } => ("once", None, None, Some(run_at.clone())),
811 Schedule::Interval { minutes } => ("interval", None, Some(*minutes), None),
812 Schedule::Cron { expr, .. } => ("cron", Some(expr.clone()), None, None),
813 };
814 let script = job.residue.0.get("script").and_then(Value::as_str);
815 let payload = match script {
816 Some(script) => JobPayload {
817 kind: "script".into(),
818 text: Some(script.to_string()),
819 },
820 None => JobPayload {
821 kind: "prompt".into(),
822 text: job.prompt.clone(),
823 },
824 };
825 let payload = match job.residue.0.get("__payload") {
828 Some(native) => openclaw_payload(&serde_json::json!({ "payload": native })),
829 None => payload,
830 };
831 let deliver = Some(job.deliver.render());
832 let (explicit_chat, explicit_thread) = match &job.deliver {
833 Target::Explicit {
834 chat_id, thread_id, ..
835 } => (chat_id.clone(), thread_id.clone()),
836 _ => (None, None),
837 };
838 let chat_id = job
839 .origin
840 .as_ref()
841 .and_then(|o| o.chat_id.clone())
842 .or(explicit_chat);
843 let thread_id = job
844 .origin
845 .as_ref()
846 .and_then(|o| o.thread_id.clone())
847 .or(explicit_thread);
848 let enabled = job.enabled;
849 let oc_delivery = job.residue.0.get("__delivery").and_then(Value::as_object);
853 let oc_text = |key: &str| {
854 oc_delivery
855 .and_then(|d| d.get(key))
856 .and_then(Value::as_str)
857 .map(str::to_string)
858 };
859 let deliver = if harness == HarnessId::OPENCLAW {
861 oc_text("channel")
862 } else {
863 deliver
864 };
865 let chat_id = oc_text("to").or(chat_id);
866 let thread_id = oc_text("threadId").or(thread_id);
867 let session_target = job
868 .residue
869 .0
870 .get("__session_target")
871 .and_then(Value::as_str)
872 .map(str::to_string);
873 Self {
874 id: job.id.clone(),
875 harness: harness.into(),
876 scope: JobScope::Install,
877 profile,
878 session_id: None,
879 schedule: JobSchedule {
880 display: schedule_display(kind, &expr, minutes, &run_at),
881 kind: kind.into(),
882 expr,
883 minutes,
884 run_at,
885 },
886 payload,
887 session_target,
888 deliver: JobDeliver {
889 target: deliver,
890 chat_id,
891 thread_id,
892 account: oc_text("accountId"),
893 mode: oc_text("mode"),
894 },
895 enabled,
896 state: if enabled { "active" } else { "paused" }.into(),
897 next_run_at: job.next_run_at.clone(),
898 last_run_at: job.last_run_at.clone(),
899 last_status: job.last_status.clone(),
900 created_at: job.created_at.clone(),
901 recurring: kind != "once",
902 }
903 }
904}
905
906fn openclaw_payload(record: &Value) -> JobPayload {
907 let payload = record.get("payload").cloned().unwrap_or(Value::Null);
908 let native = text_field(&payload, &["kind", "type"])
909 .or_else(|| text_field(record, &["payloadKind", "payload_kind"]));
910 let text = text_field(&payload, &["text", "message", "command", "script"])
911 .or_else(|| text_field(record, &["message", "command", "script"]));
912 let kind = match native.as_deref() {
913 Some("systemEvent" | "system_event") => "system_event",
914 Some("message" | "prompt" | "agentTurn" | "agent_turn") => "prompt",
915 Some("command") => "command",
916 Some("script") => "script",
917 Some(_) | None => {
918 if payload_has(record, &payload, "systemEvent") {
919 "system_event"
920 } else if payload_has(record, &payload, "command") {
921 "command"
922 } else if payload_has(record, &payload, "script") {
923 "script"
924 } else {
925 "prompt"
926 }
927 }
928 };
929 JobPayload {
930 kind: kind.into(),
931 text,
932 }
933}
934
935fn payload_has(record: &Value, payload: &Value, key: &str) -> bool {
936 payload.get(key).is_some() || record.get(key).is_some()
937}