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(
639 harness: &str,
640 query: &JobsQuery,
641 jobs: &mut Vec<ScheduledJob>,
642 sources: &mut Vec<JobSource>,
643) {
644 for store in job_store_paths(harness, &query.homes) {
645 if !store.path.exists() {
646 sources.push(JobSource::store(
647 harness,
648 store.path.clone(),
649 "absent_store",
650 store.profile.clone(),
651 ));
652 continue;
653 }
654 let records = read_job_array(&store.path);
655 sources.push(JobSource::store(
656 harness,
657 store.path.clone(),
658 "read",
659 store.profile.clone(),
660 ));
661 for record in records {
662 if let Some(job) = hermes_row(harness, &record, store.profile.clone()) {
663 jobs.push(job);
664 }
665 }
666 }
667}
668
669fn collect_openclaw_jobs(
670 query: &JobsQuery,
671 jobs: &mut Vec<ScheduledJob>,
672 sources: &mut Vec<JobSource>,
673) {
674 let mut seen: std::collections::BTreeSet<String> = std::collections::BTreeSet::new();
675 for store in job_store_paths(HarnessId::OPENCLAW, &query.homes) {
676 if !store.path.exists() {
677 sources.push(JobSource::store(
678 HarnessId::OPENCLAW,
679 store.path.clone(),
680 "absent_store",
681 None,
682 ));
683 continue;
684 }
685 let is_sqlite = store.path.extension().is_some_and(|ext| ext == "sqlite");
686 let records = if is_sqlite {
687 match openclaw_sqlite_records(&store.path) {
688 Ok(records) => records,
689 Err(error) => {
690 let mut source = JobSource::store(
691 HarnessId::OPENCLAW,
692 store.path.clone(),
693 "unreadable",
694 None,
695 );
696 source.detail = Some(error);
697 sources.push(source);
698 continue;
699 }
700 }
701 } else {
702 read_job_array(&store.path)
703 };
704 sources.push(JobSource::store(
705 HarnessId::OPENCLAW,
706 store.path.clone(),
707 "read",
708 None,
709 ));
710 for record in records {
711 if let Some(job) = openclaw_row(&record) {
712 if seen.insert(job.id.clone()) {
713 jobs.push(job);
714 }
715 }
716 }
717 }
718}
719
720fn openclaw_sqlite_records(path: &Path) -> std::result::Result<Vec<Value>, String> {
727 use rusqlite::{Connection, OpenFlags};
728 let plain = Connection::open_with_flags(
729 path,
730 OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_NO_MUTEX,
731 );
732 let conn = match plain {
733 Ok(conn) => conn,
734 Err(_) => Connection::open_with_flags(
735 format!("file:{}?immutable=1", path.display()),
736 OpenFlags::SQLITE_OPEN_READ_ONLY
737 | OpenFlags::SQLITE_OPEN_NO_MUTEX
738 | OpenFlags::SQLITE_OPEN_URI,
739 )
740 .map_err(|error| error.to_string())?,
741 };
742 let mut statement = conn
743 .prepare(
744 "SELECT job_id, job_json, state_json, next_run_at_ms, last_run_at_ms, \
745 last_run_status, created_at_ms, delivery_mode, delivery_channel, delivery_to, \
746 delivery_thread_id, delivery_account_id \
747 FROM cron_jobs ORDER BY sort_order, created_at_ms",
748 )
749 .map_err(|error| error.to_string())?;
750 let rows = statement
751 .query_map([], |row| {
752 Ok((
753 row.get::<_, String>(0)?,
754 row.get::<_, Option<String>>(1)?,
755 row.get::<_, Option<String>>(2)?,
756 row.get::<_, Option<i64>>(3)?,
757 row.get::<_, Option<i64>>(4)?,
758 row.get::<_, Option<String>>(5)?,
759 row.get::<_, Option<i64>>(6)?,
760 [
761 ("mode", row.get::<_, Option<String>>(7)?),
762 ("channel", row.get::<_, Option<String>>(8)?),
763 ("to", row.get::<_, Option<String>>(9)?),
764 ("threadId", row.get::<_, Option<String>>(10)?),
765 ("accountId", row.get::<_, Option<String>>(11)?),
766 ],
767 ))
768 })
769 .map_err(|error| error.to_string())?;
770 let mut records = Vec::new();
771 for row in rows.flatten() {
772 let (job_id, job_json, state_json, next_ms, last_ms, last_status, created_ms, delivery) =
773 row;
774 let mut record: Value = job_json
775 .as_deref()
776 .and_then(|text| serde_json::from_str(text).ok())
777 .unwrap_or_else(|| serde_json::json!({}));
778 if !record.is_object() {
779 record = serde_json::json!({});
780 }
781 let object = record.as_object_mut().expect("object");
782 object.entry("id").or_insert(Value::String(job_id));
783 if let Some(state) = state_json
784 .as_deref()
785 .and_then(|text| serde_json::from_str::<Value>(text).ok())
786 {
787 object.entry("state").or_insert(state);
788 }
789 if let Some(ms) = next_ms {
790 object
791 .entry("nextRunAt")
792 .or_insert(Value::String(iso_from_ms(ms)));
793 }
794 if let Some(ms) = last_ms {
795 object
796 .entry("lastRunAt")
797 .or_insert(Value::String(iso_from_ms(ms)));
798 }
799 if let Some(status) = last_status {
800 object.entry("lastStatus").or_insert(Value::String(status));
801 }
802 if let Some(ms) = created_ms {
803 object
804 .entry("createdAt")
805 .or_insert(Value::String(iso_from_ms(ms)));
806 }
807 if delivery.iter().any(|(_, value)| value.is_some()) {
812 let mut merged = object
813 .get("delivery")
814 .and_then(Value::as_object)
815 .cloned()
816 .unwrap_or_default();
817 for (key, value) in delivery {
818 if let Some(value) = value.filter(|value| !value.is_empty()) {
819 merged.entry(key).or_insert(Value::String(value));
820 }
821 }
822 object.insert("delivery".into(), Value::Object(merged));
823 }
824 records.push(record);
825 }
826 Ok(records)
827}
828
829fn iso_from_ms(ms: i64) -> String {
832 let secs = ms.div_euclid(1000);
833 let days = secs.div_euclid(86_400);
834 let sod = secs.rem_euclid(86_400);
835 let z = days + 719_468;
836 let era = z.div_euclid(146_097);
837 let doe = z - era * 146_097;
838 let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365;
839 let y = yoe + era * 400;
840 let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
841 let mp = (5 * doy + 2) / 153;
842 let d = doy - (153 * mp + 2) / 5 + 1;
843 let m = if mp < 10 { mp + 3 } else { mp - 9 };
844 let y = if m <= 2 { y + 1 } else { y };
845 format!(
846 "{y:04}-{m:02}-{d:02}T{:02}:{:02}:{:02}Z",
847 sod / 3600,
848 (sod % 3600) / 60,
849 sod % 60
850 )
851}
852
853fn text_field(record: &Value, keys: &[&str]) -> Option<String> {
854 keys.iter()
855 .find_map(|key| record.get(*key).and_then(Value::as_str))
856 .map(str::to_string)
857}
858
859fn enabled_flag(record: &Value) -> bool {
862 if let Some(enabled) = record.get("enabled").and_then(Value::as_bool) {
863 return enabled;
864 }
865 if let Some(paused) = record
866 .get("paused")
867 .or_else(|| record.get("is_paused"))
868 .and_then(Value::as_bool)
869 {
870 return !paused;
871 }
872 true
873}
874
875fn schedule_display(
876 kind: &str,
877 expr: &Option<String>,
878 minutes: Option<f64>,
879 run_at: &Option<String>,
880) -> String {
881 match kind {
882 "cron" => expr.clone().unwrap_or_else(|| UNKNOWN_SCHEDULE.into()),
883 "interval" => minutes.map_or_else(
884 || UNKNOWN_SCHEDULE.to_string(),
885 |minutes| format!("every {} min", trim_float(minutes)),
886 ),
887 "once" => run_at
888 .clone()
889 .map_or_else(|| "once".to_string(), |run_at| format!("once @{run_at}")),
890 _ => UNKNOWN_SCHEDULE.into(),
891 }
892}
893
894fn trim_float(value: f64) -> String {
895 if (value.fract()).abs() < f64::EPSILON {
896 format!("{}", value as i64)
897 } else {
898 format!("{value}")
899 }
900}
901
902fn hermes_row(harness: &str, record: &Value, profile: Option<String>) -> Option<ScheduledJob> {
909 let id = record_id(record)?;
910 let schedule = record.get("schedule").cloned().unwrap_or(Value::Null);
911 let kind = schedule
912 .get("kind")
913 .and_then(Value::as_str)
914 .unwrap_or(UNKNOWN_SCHEDULE)
915 .to_string();
916 let expr = schedule
917 .get("expr")
918 .and_then(Value::as_str)
919 .map(str::to_string);
920 let minutes = schedule.get("minutes").and_then(Value::as_f64);
921 let run_at = schedule
922 .get("run_at")
923 .and_then(Value::as_str)
924 .map(str::to_string);
925 let script = record.get("script").and_then(Value::as_str);
926 let payload = match script {
927 Some(script) => JobPayload {
928 kind: "script".into(),
929 text: Some(script.to_string()),
930 },
931 None => JobPayload {
932 kind: "prompt".into(),
933 text: text_field(record, &["prompt"]),
934 },
935 };
936 let deliver = text_field(record, &["deliver"]);
937 let explicit: Vec<&str> = deliver
941 .as_deref()
942 .map(|deliver| deliver.split(':').collect())
943 .unwrap_or_default();
944 let chat_id = record
945 .pointer("/origin/chat_id")
946 .and_then(Value::as_str)
947 .map(str::to_string)
948 .or_else(|| explicit.get(1).map(|chat| (*chat).to_string()));
949 let thread_id = record
950 .pointer("/origin/thread_id")
951 .and_then(Value::as_str)
952 .map(str::to_string)
953 .or_else(|| explicit.get(2).map(|thread| (*thread).to_string()));
954 let enabled = enabled_flag(record);
955 let recurring = record
956 .get("repeat")
957 .and_then(Value::as_bool)
958 .unwrap_or(kind != "once");
959 Some(ScheduledJob {
960 id,
961 harness: harness.into(),
962 scope: JobScope::Install,
963 profile,
964 session_id: None,
965 schedule: JobSchedule {
966 display: schedule_display(&kind, &expr, minutes, &run_at),
967 kind,
968 expr,
969 minutes,
970 run_at,
971 },
972 payload,
973 session_target: None,
976 deliver: JobDeliver {
977 target: deliver,
978 chat_id,
979 thread_id,
980 account: None,
983 mode: None,
984 },
985 enabled,
986 state: if enabled { "active" } else { "paused" }.into(),
987 next_run_at: text_field(record, &["next_run_at"]),
988 last_run_at: text_field(record, &["last_run_at"]),
989 last_status: text_field(record, &["last_status"]),
990 created_at: text_field(record, &["created_at"]),
991 recurring,
992 })
993}
994
995impl ScheduledJob {
996 pub fn from_job(
1001 harness: &str,
1002 profile: Option<String>,
1003 job: &supercode_interchange::world::Job,
1004 ) -> Self {
1005 use supercode_interchange::world::{Schedule, Target};
1006 let (kind, expr, minutes, run_at) = match &job.schedule {
1007 Schedule::Once { run_at } => ("once", None, None, Some(run_at.clone())),
1008 Schedule::Interval { minutes } => ("interval", None, Some(*minutes), None),
1009 Schedule::Cron { expr, .. } => ("cron", Some(expr.clone()), None, None),
1010 };
1011 let script = job.residue.0.get("script").and_then(Value::as_str);
1012 let payload = match script {
1013 Some(script) => JobPayload {
1014 kind: "script".into(),
1015 text: Some(script.to_string()),
1016 },
1017 None => JobPayload {
1018 kind: "prompt".into(),
1019 text: job.prompt.clone(),
1020 },
1021 };
1022 let deliver = Some(job.deliver.render());
1023 let (explicit_chat, explicit_thread) = match &job.deliver {
1024 Target::Explicit {
1025 chat_id, thread_id, ..
1026 } => (chat_id.clone(), thread_id.clone()),
1027 _ => (None, None),
1028 };
1029 let chat_id = job
1030 .origin
1031 .as_ref()
1032 .and_then(|o| o.chat_id.clone())
1033 .or(explicit_chat);
1034 let thread_id = job
1035 .origin
1036 .as_ref()
1037 .and_then(|o| o.thread_id.clone())
1038 .or(explicit_thread);
1039 let enabled = job.enabled;
1040 Self {
1041 id: job.id.clone(),
1042 harness: harness.into(),
1043 scope: JobScope::Install,
1044 profile,
1045 session_id: None,
1046 schedule: JobSchedule {
1047 display: schedule_display(kind, &expr, minutes, &run_at),
1048 kind: kind.into(),
1049 expr,
1050 minutes,
1051 run_at,
1052 },
1053 payload,
1054 session_target: None,
1055 deliver: JobDeliver {
1056 target: deliver,
1057 chat_id,
1058 thread_id,
1059 account: None,
1060 mode: None,
1061 },
1062 enabled,
1063 state: if enabled { "active" } else { "paused" }.into(),
1064 next_run_at: job.next_run_at.clone(),
1065 last_run_at: job.last_run_at.clone(),
1066 last_status: job.last_status.clone(),
1067 created_at: job.created_at.clone(),
1068 recurring: kind != "once",
1069 }
1070 }
1071}
1072
1073fn openclaw_row(record: &Value) -> Option<ScheduledJob> {
1080 let id = record_id(record)?;
1081 let schedule_obj = record.get("schedule").filter(|v| v.is_object());
1085 let expr = text_field(record, &["schedule", "cron"])
1086 .or_else(|| schedule_obj.and_then(|o| text_field(o, &["expr", "cron"])));
1087 let minutes = record
1088 .get("everyMinutes")
1089 .or_else(|| record.get("every_minutes"))
1090 .and_then(Value::as_f64)
1091 .or_else(|| {
1092 schedule_obj
1093 .and_then(|o| o.get("everyMs"))
1094 .and_then(Value::as_f64)
1095 .map(|ms| ms / 60_000.0)
1096 });
1097 let run_at = text_field(record, &["runAt", "run_at"])
1098 .or_else(|| schedule_obj.and_then(|o| text_field(o, &["at", "runAt"])));
1099 let state_obj = record.get("state").filter(|v| v.is_object());
1100 let state_ms = |key: &str| {
1101 state_obj
1102 .and_then(|o| o.get(key))
1103 .and_then(Value::as_i64)
1104 .map(iso_from_ms)
1105 };
1106 let kind = if minutes.is_some() {
1107 "interval"
1108 } else if expr.is_some() {
1109 "cron"
1110 } else if run_at.is_some() {
1111 "once"
1112 } else {
1113 UNKNOWN_SCHEDULE
1114 }
1115 .to_string();
1116 let payload = openclaw_payload(record);
1117 let delivery = record.get("delivery").cloned().unwrap_or(Value::Null);
1122 let mode = delivery
1123 .as_str()
1124 .map(str::to_string)
1125 .or_else(|| text_field(&delivery, &["mode", "kind", "type"]));
1126 let target = text_field(&delivery, &["channel"]).or_else(|| text_field(record, &["channel"]));
1127 let chat_id = text_field(&delivery, &["to"]).or_else(|| text_field(record, &["to"]));
1128 let thread_id = text_field(&delivery, &["threadId", "thread_id"]);
1129 let account = text_field(&delivery, &["accountId", "account_id"]);
1130 let enabled = enabled_flag(record);
1131 let recurring = kind != "once";
1132 Some(ScheduledJob {
1133 id,
1134 harness: HarnessId::OPENCLAW.into(),
1135 scope: JobScope::Install,
1136 profile: text_field(record, &["agentId", "agent_id"]),
1139 session_id: None,
1140 schedule: JobSchedule {
1141 display: schedule_display(&kind, &expr, minutes, &run_at),
1142 kind,
1143 expr,
1144 minutes,
1145 run_at,
1146 },
1147 payload,
1148 session_target: text_field(record, &["sessionTarget", "session_target"]),
1149 deliver: JobDeliver {
1150 target,
1151 chat_id,
1152 thread_id,
1153 account,
1154 mode,
1155 },
1156 enabled,
1157 state: if enabled { "active" } else { "paused" }.into(),
1158 next_run_at: text_field(record, &["nextRunAt", "next_run_at"])
1159 .or_else(|| state_ms("nextRunAtMs")),
1160 last_run_at: text_field(record, &["lastRunAt", "last_run_at"])
1161 .or_else(|| state_ms("lastRunAtMs")),
1162 last_status: text_field(record, &["lastStatus", "last_status"])
1163 .or_else(|| state_obj.and_then(|o| text_field(o, &["lastStatus", "lastRunStatus"]))),
1164 created_at: text_field(record, &["createdAt", "created_at"]).or_else(|| {
1165 record
1166 .get("createdAtMs")
1167 .and_then(Value::as_i64)
1168 .map(iso_from_ms)
1169 }),
1170 recurring,
1171 })
1172}
1173
1174fn openclaw_payload(record: &Value) -> JobPayload {
1175 let payload = record.get("payload").cloned().unwrap_or(Value::Null);
1176 let native = text_field(&payload, &["kind", "type"])
1177 .or_else(|| text_field(record, &["payloadKind", "payload_kind"]));
1178 let text = text_field(&payload, &["text", "message", "command", "script"])
1179 .or_else(|| text_field(record, &["message", "command", "script"]));
1180 let kind = match native.as_deref() {
1181 Some("systemEvent" | "system_event") => "system_event",
1182 Some("message" | "prompt" | "agentTurn" | "agent_turn") => "prompt",
1183 Some("command") => "command",
1184 Some("script") => "script",
1185 Some(_) | None => {
1186 if payload_has(record, &payload, "systemEvent") {
1187 "system_event"
1188 } else if payload_has(record, &payload, "command") {
1189 "command"
1190 } else if payload_has(record, &payload, "script") {
1191 "script"
1192 } else {
1193 "prompt"
1194 }
1195 }
1196 };
1197 JobPayload {
1198 kind: kind.into(),
1199 text,
1200 }
1201}
1202
1203fn payload_has(record: &Value, payload: &Value, key: &str) -> bool {
1204 payload.get(key).is_some() || record.get(key).is_some()
1205}
1206
1207#[cfg(test)]
1208mod world_projection_tests {
1209 use super::*;
1210 use supercode_interchange::world::{Job, JobOrigin, Schedule, Target};
1211
1212 #[test]
1214 fn from_job_matches_hermes_row() {
1215 let raw = serde_json::json!({
1216 "id": "coder-standup",
1217 "schedule": {"kind": "cron", "expr": "0 9 * * 1-5", "tz": "UTC"},
1218 "prompt": "Post the standup.",
1219 "deliver": "origin",
1220 "origin": {"platform": "telegram", "chat_id": "-100777", "thread_id": "55"},
1221 "enabled": true,
1222 "next_run_at": "2026-09-03T09:00:00Z",
1223 "last_run_at": "2026-09-02T10:00:00Z",
1224 "last_status": "ok",
1225 "created_at": "2026-08-28T10:00:00Z",
1226 });
1227 let job = Job {
1228 id: "coder-standup".into(),
1229 schedule: Schedule::Cron {
1230 expr: "0 9 * * 1-5".into(),
1231 tz: "UTC".into(),
1232 },
1233 prompt: Some("Post the standup.".into()),
1234 workdir: None,
1235 model: None,
1236 skills: Vec::new(),
1237 context_from: None,
1238 deliver: Target::Origin,
1239 failure_deliver: None,
1240 origin: Some(JobOrigin {
1241 platform: "telegram".into(),
1242 chat_type: None,
1243 chat_id: Some("-100777".into()),
1244 thread_id: Some("55".into()),
1245 }),
1246 attach_to_session: None,
1247 repeat: None,
1248 enabled: true,
1249 next_run_at: Some("2026-09-03T09:00:00Z".into()),
1250 last_run_at: Some("2026-09-02T10:00:00Z".into()),
1251 last_status: Some("ok".into()),
1252 created_at: Some("2026-08-28T10:00:00Z".into()),
1253 residue: Default::default(),
1254 };
1255 let from_raw = hermes_row("hermes", &raw, Some("coder".into())).unwrap();
1256 let from_typed = ScheduledJob::from_job("hermes", Some("coder".into()), &job);
1257 assert_eq!(from_typed, from_raw);
1258
1259 let explicit_raw = serde_json::json!({
1260 "id": "digest", "schedule": {"kind": "interval", "minutes": 120}, "prompt": "Digest.",
1261 "deliver": "slack:C0FIXTURE:t1", "enabled": false, "repeat": {"times": null, "completed": 0},
1262 });
1263 let explicit = Job {
1264 id: "digest".into(),
1265 schedule: Schedule::Interval { minutes: 120.0 },
1266 prompt: Some("Digest.".into()),
1267 deliver: Target::Explicit {
1268 platform: "slack".into(),
1269 chat_id: Some("C0FIXTURE".into()),
1270 thread_id: Some("t1".into()),
1271 },
1272 enabled: false,
1273 ..job.clone()
1274 };
1275 let explicit = Job {
1276 origin: None,
1277 next_run_at: None,
1278 last_run_at: None,
1279 last_status: None,
1280 created_at: None,
1281 ..explicit
1282 };
1283 assert_eq!(
1284 ScheduledJob::from_job("hermes", None, &explicit),
1285 hermes_row("hermes", &explicit_raw, None).unwrap()
1286 );
1287 }
1288}