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 mut seen: std::collections::BTreeSet<String> = std::collections::BTreeSet::new();
712 let store = query.homes.openclaw.join("state/openclaw.sqlite");
713 if !store.exists() {
714 sources.push(JobSource::store(
715 HarnessId::OPENCLAW,
716 store.clone(),
717 "absent_store",
718 None,
719 ));
720 } else {
721 match from_openclaw(&query.homes.openclaw) {
722 Ok(loaded) => {
723 sources.push(JobSource::store(
724 HarnessId::OPENCLAW,
725 store.clone(),
726 "read",
727 None,
728 ));
729 let mut names: Vec<&String> = loaded.orchestration.profiles.keys().collect();
730 names.sort_by_key(|name| (name.as_str() != "default", name.as_str()));
731 for name in names {
732 let profile_name = (name != "default").then(|| name.clone());
733 for job in loaded.orchestration.profiles[name].jobs.values() {
734 if seen.insert(job.id.clone()) {
735 jobs.push(ScheduledJob::from_job(
736 HarnessId::OPENCLAW,
737 profile_name.clone(),
738 job,
739 ));
740 }
741 }
742 }
743 }
744 Err(error) => {
745 let mut source =
746 JobSource::store(HarnessId::OPENCLAW, store.clone(), "unreadable", None);
747 source.detail = Some(error.to_string());
748 sources.push(source);
749 }
750 }
751 }
752 let legacy = query.homes.openclaw.join("cron/jobs.json");
753 if !legacy.exists() {
754 sources.push(JobSource::store(
755 HarnessId::OPENCLAW,
756 legacy,
757 "absent_store",
758 None,
759 ));
760 return;
761 }
762 sources.push(JobSource::store(
763 HarnessId::OPENCLAW,
764 legacy.clone(),
765 "read",
766 None,
767 ));
768 for record in read_job_array(&legacy) {
769 if let Some(job) = openclaw_row(&record) {
770 if seen.insert(job.id.clone()) {
771 jobs.push(job);
772 }
773 }
774 }
775}
776
777fn iso_from_ms(ms: i64) -> String {
780 let secs = ms.div_euclid(1000);
781 let days = secs.div_euclid(86_400);
782 let sod = secs.rem_euclid(86_400);
783 let z = days + 719_468;
784 let era = z.div_euclid(146_097);
785 let doe = z - era * 146_097;
786 let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365;
787 let y = yoe + era * 400;
788 let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
789 let mp = (5 * doy + 2) / 153;
790 let d = doy - (153 * mp + 2) / 5 + 1;
791 let m = if mp < 10 { mp + 3 } else { mp - 9 };
792 let y = if m <= 2 { y + 1 } else { y };
793 format!(
794 "{y:04}-{m:02}-{d:02}T{:02}:{:02}:{:02}Z",
795 sod / 3600,
796 (sod % 3600) / 60,
797 sod % 60
798 )
799}
800
801fn text_field(record: &Value, keys: &[&str]) -> Option<String> {
802 keys.iter()
803 .find_map(|key| record.get(*key).and_then(Value::as_str))
804 .map(str::to_string)
805}
806
807fn enabled_flag(record: &Value) -> bool {
810 if let Some(enabled) = record.get("enabled").and_then(Value::as_bool) {
811 return enabled;
812 }
813 if let Some(paused) = record
814 .get("paused")
815 .or_else(|| record.get("is_paused"))
816 .and_then(Value::as_bool)
817 {
818 return !paused;
819 }
820 true
821}
822
823fn schedule_display(
824 kind: &str,
825 expr: &Option<String>,
826 minutes: Option<f64>,
827 run_at: &Option<String>,
828) -> String {
829 match kind {
830 "cron" => expr.clone().unwrap_or_else(|| UNKNOWN_SCHEDULE.into()),
831 "interval" => minutes.map_or_else(
832 || UNKNOWN_SCHEDULE.to_string(),
833 |minutes| format!("every {} min", trim_float(minutes)),
834 ),
835 "once" => run_at
836 .clone()
837 .map_or_else(|| "once".to_string(), |run_at| format!("once @{run_at}")),
838 _ => UNKNOWN_SCHEDULE.into(),
839 }
840}
841
842fn trim_float(value: f64) -> String {
843 if (value.fract()).abs() < f64::EPSILON {
844 format!("{}", value as i64)
845 } else {
846 format!("{value}")
847 }
848}
849
850impl ScheduledJob {
851 pub fn from_job(
857 harness: &str,
858 profile: Option<String>,
859 job: &supercode_interchange::orchestration::Job,
860 ) -> Self {
861 use supercode_interchange::orchestration::{Schedule, Target};
862 let (kind, expr, minutes, run_at) = match &job.schedule {
863 Schedule::Once { run_at } => ("once", None, None, Some(run_at.clone())),
864 Schedule::Interval { minutes } => ("interval", None, Some(*minutes), None),
865 Schedule::Cron { expr, .. } => ("cron", Some(expr.clone()), None, None),
866 };
867 let script = job.residue.0.get("script").and_then(Value::as_str);
868 let payload = match script {
869 Some(script) => JobPayload {
870 kind: "script".into(),
871 text: Some(script.to_string()),
872 },
873 None => JobPayload {
874 kind: "prompt".into(),
875 text: job.prompt.clone(),
876 },
877 };
878 let payload = match job.residue.0.get("__payload") {
881 Some(native) => openclaw_payload(&serde_json::json!({ "payload": native })),
882 None => payload,
883 };
884 let deliver = Some(job.deliver.render());
885 let (explicit_chat, explicit_thread) = match &job.deliver {
886 Target::Explicit {
887 chat_id, thread_id, ..
888 } => (chat_id.clone(), thread_id.clone()),
889 _ => (None, None),
890 };
891 let chat_id = job
892 .origin
893 .as_ref()
894 .and_then(|o| o.chat_id.clone())
895 .or(explicit_chat);
896 let thread_id = job
897 .origin
898 .as_ref()
899 .and_then(|o| o.thread_id.clone())
900 .or(explicit_thread);
901 let enabled = job.enabled;
902 let oc_delivery = job.residue.0.get("__delivery").and_then(Value::as_object);
906 let oc_text = |key: &str| {
907 oc_delivery
908 .and_then(|d| d.get(key))
909 .and_then(Value::as_str)
910 .map(str::to_string)
911 };
912 let deliver = if harness == HarnessId::OPENCLAW {
914 oc_text("channel")
915 } else {
916 deliver
917 };
918 let chat_id = oc_text("to").or(chat_id);
919 let thread_id = oc_text("threadId").or(thread_id);
920 let session_target = job
921 .residue
922 .0
923 .get("__session_target")
924 .and_then(Value::as_str)
925 .map(str::to_string);
926 Self {
927 id: job.id.clone(),
928 harness: harness.into(),
929 scope: JobScope::Install,
930 profile,
931 session_id: None,
932 schedule: JobSchedule {
933 display: schedule_display(kind, &expr, minutes, &run_at),
934 kind: kind.into(),
935 expr,
936 minutes,
937 run_at,
938 },
939 payload,
940 session_target,
941 deliver: JobDeliver {
942 target: deliver,
943 chat_id,
944 thread_id,
945 account: oc_text("accountId"),
946 mode: oc_text("mode"),
947 },
948 enabled,
949 state: if enabled { "active" } else { "paused" }.into(),
950 next_run_at: job.next_run_at.clone(),
951 last_run_at: job.last_run_at.clone(),
952 last_status: job.last_status.clone(),
953 created_at: job.created_at.clone(),
954 recurring: kind != "once",
955 }
956 }
957}
958
959fn openclaw_row(record: &Value) -> Option<ScheduledJob> {
966 let id = record_id(record)?;
967 let schedule_obj = record.get("schedule").filter(|v| v.is_object());
971 let expr = text_field(record, &["schedule", "cron"])
972 .or_else(|| schedule_obj.and_then(|o| text_field(o, &["expr", "cron"])));
973 let minutes = record
974 .get("everyMinutes")
975 .or_else(|| record.get("every_minutes"))
976 .and_then(Value::as_f64)
977 .or_else(|| {
978 schedule_obj
979 .and_then(|o| o.get("everyMs"))
980 .and_then(Value::as_f64)
981 .map(|ms| ms / 60_000.0)
982 });
983 let run_at = text_field(record, &["runAt", "run_at"])
984 .or_else(|| schedule_obj.and_then(|o| text_field(o, &["at", "runAt"])));
985 let state_obj = record.get("state").filter(|v| v.is_object());
986 let state_ms = |key: &str| {
987 state_obj
988 .and_then(|o| o.get(key))
989 .and_then(Value::as_i64)
990 .map(iso_from_ms)
991 };
992 let kind = if minutes.is_some() {
993 "interval"
994 } else if expr.is_some() {
995 "cron"
996 } else if run_at.is_some() {
997 "once"
998 } else {
999 UNKNOWN_SCHEDULE
1000 }
1001 .to_string();
1002 let payload = openclaw_payload(record);
1003 let delivery = record.get("delivery").cloned().unwrap_or(Value::Null);
1008 let mode = delivery
1009 .as_str()
1010 .map(str::to_string)
1011 .or_else(|| text_field(&delivery, &["mode", "kind", "type"]));
1012 let target = text_field(&delivery, &["channel"]).or_else(|| text_field(record, &["channel"]));
1013 let chat_id = text_field(&delivery, &["to"]).or_else(|| text_field(record, &["to"]));
1014 let thread_id = text_field(&delivery, &["threadId", "thread_id"]);
1015 let account = text_field(&delivery, &["accountId", "account_id"]);
1016 let enabled = enabled_flag(record);
1017 let recurring = kind != "once";
1018 Some(ScheduledJob {
1019 id,
1020 harness: HarnessId::OPENCLAW.into(),
1021 scope: JobScope::Install,
1022 profile: text_field(record, &["agentId", "agent_id"]),
1025 session_id: None,
1026 schedule: JobSchedule {
1027 display: schedule_display(&kind, &expr, minutes, &run_at),
1028 kind,
1029 expr,
1030 minutes,
1031 run_at,
1032 },
1033 payload,
1034 session_target: text_field(record, &["sessionTarget", "session_target"]),
1035 deliver: JobDeliver {
1036 target,
1037 chat_id,
1038 thread_id,
1039 account,
1040 mode,
1041 },
1042 enabled,
1043 state: if enabled { "active" } else { "paused" }.into(),
1044 next_run_at: text_field(record, &["nextRunAt", "next_run_at"])
1045 .or_else(|| state_ms("nextRunAtMs")),
1046 last_run_at: text_field(record, &["lastRunAt", "last_run_at"])
1047 .or_else(|| state_ms("lastRunAtMs")),
1048 last_status: text_field(record, &["lastStatus", "last_status"])
1049 .or_else(|| state_obj.and_then(|o| text_field(o, &["lastStatus", "lastRunStatus"]))),
1050 created_at: text_field(record, &["createdAt", "created_at"]).or_else(|| {
1051 record
1052 .get("createdAtMs")
1053 .and_then(Value::as_i64)
1054 .map(iso_from_ms)
1055 }),
1056 recurring,
1057 })
1058}
1059
1060fn openclaw_payload(record: &Value) -> JobPayload {
1061 let payload = record.get("payload").cloned().unwrap_or(Value::Null);
1062 let native = text_field(&payload, &["kind", "type"])
1063 .or_else(|| text_field(record, &["payloadKind", "payload_kind"]));
1064 let text = text_field(&payload, &["text", "message", "command", "script"])
1065 .or_else(|| text_field(record, &["message", "command", "script"]));
1066 let kind = match native.as_deref() {
1067 Some("systemEvent" | "system_event") => "system_event",
1068 Some("message" | "prompt" | "agentTurn" | "agent_turn") => "prompt",
1069 Some("command") => "command",
1070 Some("script") => "script",
1071 Some(_) | None => {
1072 if payload_has(record, &payload, "systemEvent") {
1073 "system_event"
1074 } else if payload_has(record, &payload, "command") {
1075 "command"
1076 } else if payload_has(record, &payload, "script") {
1077 "script"
1078 } else {
1079 "prompt"
1080 }
1081 }
1082 };
1083 JobPayload {
1084 kind: kind.into(),
1085 text,
1086 }
1087}
1088
1089fn payload_has(record: &Value, payload: &Value, key: &str) -> bool {
1090 payload.get(key).is_some() || record.get(key).is_some()
1091}