1use std::collections::BTreeMap;
12use std::fs;
13use std::path::{Path, PathBuf};
14
15use serde_json::{Map, Value};
16
17use super::canonical::canonical_json;
18use super::decode::{load_error, placeholder_ref, surface_key_string};
19use super::folder::{empty_profile, ordered_object, persona_ref, pretty_ordered};
20use super::sqlite::{read_rows, table_exists, write_table, Param};
21use crate::ontology::{
22 ArtifactFidelity, Binding, Fidelity, HarnessId, Recurrence, Residue, SecretRef, SurfaceKey,
23 Trigger, Worker,
24};
25use crate::orchestration::{
26 ChannelConfig, Fire, FireStatus, Job, JobOrigin, Obligation, ObligationSource, ObligationState,
27 Orchestration, OutboundContent, Profile, Route, RouteMatch, Schedule, Target,
28 WebhookSubscription,
29};
30use crate::Result;
31
32pub const OPENCLAW_CONFIG: &str = "openclaw.json";
34pub const OPENCLAW_STATE_DB: &str = "state/openclaw.sqlite";
36pub const OPENCLAW_ACCOUNT_KEYS: &[&str] = &[
38 "accountId",
39 "account_id",
40 "account",
41 "teamId",
42 "appId",
43 "userId",
44];
45
46pub const OPENCLAW_DDL: &str = r#"CREATE TABLE IF NOT EXISTS schema_meta (
48 meta_key TEXT NOT NULL PRIMARY KEY,
49 role TEXT NOT NULL,
50 schema_version INTEGER NOT NULL,
51 agent_id TEXT,
52 app_version TEXT,
53 created_at INTEGER NOT NULL,
54 updated_at INTEGER NOT NULL
55);
56CREATE TABLE IF NOT EXISTS cron_jobs (
57 store_key TEXT NOT NULL,
58 job_id TEXT NOT NULL,
59 declaration_key TEXT,
60 display_name TEXT,
61 owner_agent_id TEXT,
62 owner_session_key TEXT,
63 name TEXT NOT NULL,
64 description TEXT,
65 enabled INTEGER NOT NULL,
66 delete_after_run INTEGER,
67 created_at_ms INTEGER NOT NULL,
68 agent_id TEXT,
69 session_key TEXT,
70 schedule_kind TEXT NOT NULL,
71 schedule_expr TEXT,
72 schedule_tz TEXT,
73 every_ms INTEGER,
74 anchor_ms INTEGER,
75 at TEXT,
76 stagger_ms INTEGER,
77 session_target TEXT NOT NULL,
78 wake_mode TEXT NOT NULL,
79 trigger_script TEXT,
80 trigger_once INTEGER,
81 payload_kind TEXT NOT NULL,
82 payload_message TEXT,
83 payload_model TEXT,
84 payload_fallbacks_json TEXT,
85 payload_thinking TEXT,
86 payload_timeout_seconds INTEGER,
87 payload_allow_unsafe_external_content INTEGER,
88 payload_external_content_source_json TEXT,
89 payload_light_context INTEGER,
90 payload_tools_allow_json TEXT,
91 payload_tools_allow_is_default INTEGER,
92 delivery_mode TEXT,
93 delivery_channel TEXT,
94 delivery_to TEXT,
95 delivery_thread_id TEXT,
96 delivery_thread_id_type TEXT,
97 delivery_account_id TEXT,
98 delivery_best_effort INTEGER,
99 delivery_completion_mode TEXT,
100 delivery_completion_to TEXT,
101 failure_delivery_mode TEXT,
102 failure_delivery_channel TEXT,
103 failure_delivery_to TEXT,
104 failure_delivery_account_id TEXT,
105 failure_alert_disabled INTEGER,
106 failure_alert_after INTEGER,
107 failure_alert_channel TEXT,
108 failure_alert_to TEXT,
109 failure_alert_cooldown_ms INTEGER,
110 failure_alert_include_skipped INTEGER,
111 failure_alert_mode TEXT,
112 failure_alert_account_id TEXT,
113 next_run_at_ms INTEGER,
114 running_at_ms INTEGER,
115 last_run_at_ms INTEGER,
116 last_run_status TEXT,
117 last_error TEXT,
118 last_duration_ms INTEGER,
119 consecutive_errors INTEGER,
120 consecutive_skipped INTEGER,
121 schedule_error_count INTEGER,
122 last_delivery_status TEXT,
123 last_delivery_error TEXT,
124 last_delivered INTEGER,
125 last_failure_alert_at_ms INTEGER,
126 job_json TEXT NOT NULL,
127 state_json TEXT NOT NULL DEFAULT '{}',
128 runtime_updated_at_ms INTEGER,
129 schedule_identity TEXT,
130 sort_order INTEGER NOT NULL DEFAULT 0,
131 updated_at INTEGER NOT NULL,
132 PRIMARY KEY (store_key, job_id)
133);
134CREATE INDEX IF NOT EXISTS idx_cron_jobs_store_updated
135 ON cron_jobs(store_key, sort_order ASC, updated_at DESC, job_id);
136CREATE INDEX IF NOT EXISTS idx_cron_jobs_store_order
137 ON cron_jobs(store_key, sort_order ASC, updated_at ASC, job_id);
138CREATE INDEX IF NOT EXISTS idx_cron_jobs_enabled_next_run
139 ON cron_jobs(store_key, enabled, next_run_at_ms, job_id)
140 WHERE next_run_at_ms IS NOT NULL;
141CREATE INDEX IF NOT EXISTS idx_cron_jobs_agent_session
142 ON cron_jobs(agent_id, session_key, updated_at DESC, job_id)
143 WHERE agent_id IS NOT NULL OR session_key IS NOT NULL;
144CREATE TABLE IF NOT EXISTS cron_run_logs (
145 store_key TEXT NOT NULL,
146 job_id TEXT NOT NULL,
147 seq INTEGER NOT NULL,
148 ts INTEGER NOT NULL,
149 status TEXT,
150 error TEXT,
151 summary TEXT,
152 diagnostics_summary TEXT,
153 delivery_status TEXT,
154 delivery_error TEXT,
155 delivered INTEGER,
156 session_id TEXT,
157 session_key TEXT,
158 run_id TEXT,
159 run_at_ms INTEGER,
160 duration_ms INTEGER,
161 next_run_at_ms INTEGER,
162 model TEXT,
163 provider TEXT,
164 total_tokens INTEGER,
165 entry_json TEXT NOT NULL,
166 created_at INTEGER NOT NULL,
167 PRIMARY KEY (store_key, job_id, seq)
168);
169CREATE INDEX IF NOT EXISTS idx_cron_run_logs_store_ts
170 ON cron_run_logs(store_key, ts DESC, seq DESC);
171CREATE INDEX IF NOT EXISTS idx_cron_run_logs_job_status
172 ON cron_run_logs(store_key, job_id, status, ts DESC, seq DESC);
173CREATE INDEX IF NOT EXISTS idx_cron_run_logs_delivery
174 ON cron_run_logs(store_key, delivery_status, ts DESC, seq DESC)
175 WHERE delivery_status IS NOT NULL;
176CREATE TABLE IF NOT EXISTS delivery_queue_entries (
177 queue_name TEXT NOT NULL,
178 id TEXT NOT NULL,
179 status TEXT NOT NULL,
180 entry_kind TEXT,
181 session_key TEXT,
182 channel TEXT,
183 target TEXT,
184 account_id TEXT,
185 retry_count INTEGER NOT NULL DEFAULT 0,
186 last_attempt_at INTEGER,
187 last_error TEXT,
188 recovery_state TEXT,
189 platform_send_started_at INTEGER,
190 entry_json TEXT NOT NULL,
191 enqueued_at INTEGER NOT NULL,
192 updated_at INTEGER NOT NULL,
193 failed_at INTEGER,
194 PRIMARY KEY (queue_name, id)
195);
196CREATE INDEX IF NOT EXISTS idx_delivery_queue_pending
197 ON delivery_queue_entries(queue_name, status, enqueued_at, id);
198CREATE INDEX IF NOT EXISTS idx_delivery_queue_failed
199 ON delivery_queue_entries(queue_name, status, failed_at, id);
200CREATE INDEX IF NOT EXISTS idx_delivery_queue_session
201 ON delivery_queue_entries(queue_name, status, session_key, enqueued_at, id)
202 WHERE session_key IS NOT NULL;
203CREATE INDEX IF NOT EXISTS idx_delivery_queue_target
204 ON delivery_queue_entries(queue_name, status, channel, target, enqueued_at, id)
205 WHERE channel IS NOT NULL AND target IS NOT NULL;
206"#;
207pub const CRON_JOB_COLUMNS: &[&str] = &[
209 "store_key",
210 "job_id",
211 "declaration_key",
212 "display_name",
213 "owner_agent_id",
214 "owner_session_key",
215 "name",
216 "description",
217 "enabled",
218 "delete_after_run",
219 "created_at_ms",
220 "agent_id",
221 "session_key",
222 "schedule_kind",
223 "schedule_expr",
224 "schedule_tz",
225 "every_ms",
226 "anchor_ms",
227 "at",
228 "stagger_ms",
229 "session_target",
230 "wake_mode",
231 "trigger_script",
232 "trigger_once",
233 "payload_kind",
234 "payload_message",
235 "payload_model",
236 "payload_fallbacks_json",
237 "payload_thinking",
238 "payload_timeout_seconds",
239 "payload_allow_unsafe_external_content",
240 "payload_external_content_source_json",
241 "payload_light_context",
242 "payload_tools_allow_json",
243 "payload_tools_allow_is_default",
244 "delivery_mode",
245 "delivery_channel",
246 "delivery_to",
247 "delivery_thread_id",
248 "delivery_thread_id_type",
249 "delivery_account_id",
250 "delivery_best_effort",
251 "delivery_completion_mode",
252 "delivery_completion_to",
253 "failure_delivery_mode",
254 "failure_delivery_channel",
255 "failure_delivery_to",
256 "failure_delivery_account_id",
257 "failure_alert_disabled",
258 "failure_alert_after",
259 "failure_alert_channel",
260 "failure_alert_to",
261 "failure_alert_cooldown_ms",
262 "failure_alert_include_skipped",
263 "failure_alert_mode",
264 "failure_alert_account_id",
265 "next_run_at_ms",
266 "running_at_ms",
267 "last_run_at_ms",
268 "last_run_status",
269 "last_error",
270 "last_duration_ms",
271 "consecutive_errors",
272 "consecutive_skipped",
273 "schedule_error_count",
274 "last_delivery_status",
275 "last_delivery_error",
276 "last_delivered",
277 "last_failure_alert_at_ms",
278 "job_json",
279 "state_json",
280 "runtime_updated_at_ms",
281 "schedule_identity",
282 "sort_order",
283 "updated_at",
284];
285pub const CRON_RUN_LOG_COLUMNS: &[&str] = &[
287 "store_key",
288 "job_id",
289 "seq",
290 "ts",
291 "status",
292 "error",
293 "summary",
294 "diagnostics_summary",
295 "delivery_status",
296 "delivery_error",
297 "delivered",
298 "session_id",
299 "session_key",
300 "run_id",
301 "run_at_ms",
302 "duration_ms",
303 "next_run_at_ms",
304 "model",
305 "provider",
306 "total_tokens",
307 "entry_json",
308 "created_at",
309];
310pub const DELIVERY_QUEUE_COLUMNS: &[&str] = &[
312 "queue_name",
313 "id",
314 "status",
315 "entry_kind",
316 "session_key",
317 "channel",
318 "target",
319 "account_id",
320 "retry_count",
321 "last_attempt_at",
322 "last_error",
323 "recovery_state",
324 "platform_send_started_at",
325 "entry_json",
326 "enqueued_at",
327 "updated_at",
328 "failed_at",
329];
330pub const SCHEMA_META_COLUMNS: &[&str] = &[
332 "meta_key",
333 "role",
334 "schema_version",
335 "agent_id",
336 "app_version",
337 "created_at",
338 "updated_at",
339];
340
341pub fn strip_json5(text: &str) -> String {
347 let mut out = String::with_capacity(text.len());
348 let mut chars = text.chars().peekable();
349 let mut in_string = false;
350 let mut escaped = false;
351 while let Some(ch) = chars.next() {
352 if in_string {
353 out.push(ch);
354 if escaped {
355 escaped = false;
356 } else if ch == '\\' {
357 escaped = true;
358 } else if ch == '"' {
359 in_string = false;
360 }
361 continue;
362 }
363 match ch {
364 '"' => {
365 in_string = true;
366 out.push(ch);
367 }
368 '/' if chars.peek() == Some(&'/') => {
369 for next in chars.by_ref() {
370 if next == '\n' {
371 out.push('\n');
372 break;
373 }
374 }
375 }
376 '/' if chars.peek() == Some(&'*') => {
377 chars.next();
378 let mut previous = '\0';
379 for next in chars.by_ref() {
380 if previous == '*' && next == '/' {
381 break;
382 }
383 previous = next;
384 }
385 out.push(' ');
386 }
387 _ => out.push(ch),
388 }
389 }
390 let bytes: Vec<char> = out.chars().collect();
391 let mut cleaned = String::with_capacity(out.len());
392 let mut index = 0usize;
393 let mut in_string = false;
394 let mut escaped = false;
395 while index < bytes.len() {
396 let ch = bytes[index];
397 if in_string {
398 cleaned.push(ch);
399 if escaped {
400 escaped = false;
401 } else if ch == '\\' {
402 escaped = true;
403 } else if ch == '"' {
404 in_string = false;
405 }
406 index += 1;
407 continue;
408 }
409 if ch == '"' {
410 in_string = true;
411 cleaned.push(ch);
412 index += 1;
413 continue;
414 }
415 if ch == ',' {
416 let mut lookahead = index + 1;
417 while lookahead < bytes.len() && bytes[lookahead].is_whitespace() {
418 lookahead += 1;
419 }
420 if lookahead < bytes.len() && (bytes[lookahead] == '}' || bytes[lookahead] == ']') {
421 index += 1;
422 continue;
423 }
424 }
425 cleaned.push(ch);
426 index += 1;
427 }
428 cleaned
429}
430
431pub fn parse_json5(file: &str, text: &str) -> Result<Value> {
433 serde_json::from_str(&strip_json5(text))
434 .map_err(|e| load_error(file, "", format!("JSON5: {e}")))
435}
436
437pub const NON_SECRET_KEYS: &[&str] = &[
441 "sessionkey",
442 "session_key",
443 "storekey",
444 "store_key",
445 "metakey",
446 "meta_key",
447 "bindingkey",
448 "binding_key",
449 "declarationkey",
450 "declaration_key",
451 "idempotencykey",
452 "idempotency_key",
453];
454
455pub fn is_credential_key(key: &str) -> bool {
457 let lower = key.to_ascii_lowercase();
458 if NON_SECRET_KEYS.contains(&lower.as_str()) {
459 return false;
460 }
461 ["token", "key", "secret", "password", "credential"]
462 .iter()
463 .any(|m| lower.ends_with(m))
464}
465
466pub fn credential_ref_name(scope: &str, key: &str) -> String {
468 format!("OPENCLAW_{scope}_{key}")
469 .chars()
470 .map(|c| {
471 if c.is_ascii_alphanumeric() {
472 c.to_ascii_uppercase()
473 } else {
474 '_'
475 }
476 })
477 .collect()
478}
479
480pub fn redact_secrets(value: &Value, scope: &str, vault: &mut BTreeMap<String, String>) -> Value {
482 match value {
483 Value::Array(items) => Value::Array(
484 items
485 .iter()
486 .enumerate()
487 .map(|(i, v)| redact_secrets(v, &format!("{scope}_{i}"), vault))
488 .collect(),
489 ),
490 Value::Object(map) => {
491 let mut out = Map::new();
492 for (k, v) in map {
493 match v {
494 Value::String(s) if is_credential_key(k) => {
495 let r = credential_ref_name(scope, k);
496 vault.insert(r.clone(), s.clone());
497 out.insert(k.clone(), serde_json::json!({"dotenv": r}));
498 }
499 Value::Object(_) | Value::Array(_) => {
500 out.insert(k.clone(), redact_secrets(v, &format!("{scope}_{k}"), vault));
501 }
502 other => {
503 out.insert(k.clone(), other.clone());
504 }
505 }
506 }
507 Value::Object(out)
508 }
509 other => other.clone(),
510 }
511}
512
513fn resolve_ref(name: &str, vault: &BTreeMap<String, String>, depth: usize) -> Option<String> {
514 let value = vault.get(name)?;
515 if depth > 4 {
516 return Some(value.clone());
517 }
518 match placeholder_ref(value) {
519 Some(next) => resolve_ref(next, vault, depth + 1).or_else(|| Some(value.clone())),
520 None => Some(value.clone()),
521 }
522}
523
524pub fn inline_secrets(value: &Value, vault: &BTreeMap<String, String>) -> Value {
526 match value {
527 Value::String(s) => match placeholder_ref(s).and_then(|n| resolve_ref(n, vault, 0)) {
528 Some(v) => Value::String(v),
529 None => value.clone(),
530 },
531 Value::Array(items) => {
532 Value::Array(items.iter().map(|v| inline_secrets(v, vault)).collect())
533 }
534 Value::Object(map) => {
535 if map.len() == 1 {
536 if let Some(Value::String(n)) = map.get("dotenv").or_else(|| map.get("env")) {
537 if let Some(v) = resolve_ref(n, vault, 0) {
538 return Value::String(v);
539 }
540 return value.clone();
541 }
542 }
543 Value::Object(
544 map.iter()
545 .map(|(k, v)| (k.clone(), inline_secrets(v, vault)))
546 .collect(),
547 )
548 }
549 other => other.clone(),
550 }
551}
552
553fn unresolved_refs(value: &Value, path: &str) -> Vec<(String, String)> {
555 match value {
556 Value::String(s) => placeholder_ref(s)
557 .map(|n| vec![(path.to_string(), n.to_string())])
558 .unwrap_or_default(),
559 Value::Array(items) => items
560 .iter()
561 .enumerate()
562 .flat_map(|(i, v)| unresolved_refs(v, &format!("{path}[{i}]")))
563 .collect(),
564 Value::Object(map) => {
565 if map.len() == 1 {
566 if let Some(Value::String(n)) = map.get("dotenv").or_else(|| map.get("env")) {
567 return vec![(path.to_string(), n.clone())];
568 }
569 }
570 map.iter()
571 .flat_map(|(k, v)| {
572 unresolved_refs(
573 v,
574 &if path.is_empty() {
575 k.clone()
576 } else {
577 format!("{path}.{k}")
578 },
579 )
580 })
581 .collect()
582 }
583 _ => Vec::new(),
584 }
585}
586
587fn civil(ms: i64) -> (i64, u32, u32, u32, u32, u32, u32) {
590 let secs = ms.div_euclid(1000);
591 let sub = ms.rem_euclid(1000) as u32;
592 let days = secs.div_euclid(86_400);
593 let sod = secs.rem_euclid(86_400);
594 let z = days + 719_468;
595 let era = z.div_euclid(146_097);
596 let doe = z - era * 146_097;
597 let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365;
598 let y = yoe + era * 400;
599 let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
600 let mp = (5 * doy + 2) / 153;
601 let d = (doy - (153 * mp + 2) / 5 + 1) as u32;
602 let m = if mp < 10 { mp + 3 } else { mp - 9 } as u32;
603 let y = if m <= 2 { y + 1 } else { y };
604 (
605 y,
606 m,
607 d,
608 (sod / 3600) as u32,
609 ((sod % 3600) / 60) as u32,
610 (sod % 60) as u32,
611 sub,
612 )
613}
614
615pub fn iso_seconds(ms: Option<i64>) -> Option<String> {
617 ms.map(|ms| {
618 let (y, mo, d, h, mi, s, _) = civil(ms);
619 format!("{y:04}-{mo:02}-{d:02}T{h:02}:{mi:02}:{s:02}Z")
620 })
621}
622
623pub fn iso_millis(ms: Option<i64>) -> Option<String> {
625 ms.map(|ms| {
626 let (y, mo, d, h, mi, s, sub) = civil(ms);
627 format!("{y:04}-{mo:02}-{d:02}T{h:02}:{mi:02}:{s:02}.{sub:03}Z")
628 })
629}
630
631pub fn ms_from_iso(iso: Option<&str>) -> Option<i64> {
633 let s = iso?.trim();
634 let (date, rest) = s.split_once('T')?;
635 let mut dp = date.split('-');
636 let (y, mo, d): (i64, i64, i64) = (
637 dp.next()?.parse().ok()?,
638 dp.next()?.parse().ok()?,
639 dp.next()?.parse().ok()?,
640 );
641 let (time, offset) = if let Some(t) = rest.strip_suffix('Z') {
642 (t, 0i64)
643 } else if let Some(idx) = rest.rfind(['+', '-']) {
644 let (t, off) = rest.split_at(idx);
645 let sign = if off.starts_with('-') { -1 } else { 1 };
646 let mut op = off[1..].split(':');
647 let (oh, om): (i64, i64) = (
648 op.next()?.parse().ok()?,
649 op.next().unwrap_or("0").parse().ok()?,
650 );
651 (t, sign * (oh * 3600 + om * 60))
652 } else {
653 (rest, 0)
654 };
655 let (hms, frac) = match time.split_once('.') {
656 Some((a, b)) => (a, b),
657 None => (time, ""),
658 };
659 let mut tp = hms.split(':');
660 let (h, mi, sec): (i64, i64, i64) = (
661 tp.next()?.parse().ok()?,
662 tp.next()?.parse().ok()?,
663 tp.next().unwrap_or("0").parse().ok()?,
664 );
665 let millis: i64 = if frac.is_empty() {
666 0
667 } else {
668 format!("{:0<3}", &frac[..frac.len().min(3)]).parse().ok()?
669 };
670 let (yy, mm) = if mo <= 2 {
672 (y - 1, mo + 9)
673 } else {
674 (y, mo - 3)
675 };
676 let era = yy.div_euclid(400);
677 let yoe = yy - era * 400;
678 let doy = (153 * mm + 2) / 5 + d - 1;
679 let doe = yoe * 365 + yoe / 4 - yoe / 100 + doy;
680 let days = era * 146_097 + doe - 719_468;
681 Some(((days * 86_400 + h * 3600 + mi * 60 + sec) - offset) * 1000 + millis)
682}
683
684fn ms_of(v: Option<&Value>) -> Option<i64> {
685 match v? {
686 Value::Number(n) => n.as_i64().or_else(|| n.as_f64().map(|f| f as i64)),
687 Value::String(s) => s.parse::<f64>().ok().map(|f| f as i64),
688 _ => None,
689 }
690}
691
692pub fn agent_entries(config: &Value) -> Vec<(String, Value)> {
696 let agents = config.get("agents");
697 if let Some(list) = agents.and_then(|a| a.get("list")).and_then(Value::as_array) {
698 return list
699 .iter()
700 .filter_map(|e| {
701 e.get("id")
702 .or_else(|| e.get("agentId"))
703 .and_then(Value::as_str)
704 .filter(|s| !s.is_empty())
705 .map(|id| (id.to_string(), e.clone()))
706 })
707 .collect();
708 }
709 if let Some(entries) = agents
710 .and_then(|a| a.get("entries"))
711 .and_then(Value::as_object)
712 {
713 return entries
714 .iter()
715 .map(|(k, v)| (k.clone(), v.clone()))
716 .collect();
717 }
718 Vec::new()
719}
720
721pub fn agents_form(config: &Value) -> Option<&'static str> {
723 let agents = config.get("agents")?;
724 if agents.get("list").is_some_and(Value::is_array) {
725 return Some("list");
726 }
727 if agents.get("entries").is_some_and(Value::is_object) {
728 return Some("entries");
729 }
730 None
731}
732
733pub fn default_agent_id(config: &Value) -> String {
735 let entries = agent_entries(config);
736 entries
737 .iter()
738 .find(|(_, e)| e.get("default") == Some(&Value::Bool(true)))
739 .or_else(|| entries.first())
740 .map(|(id, _)| id.clone())
741 .unwrap_or_else(|| "main".into())
742}
743
744#[derive(Debug, Clone, PartialEq)]
748pub struct ParsedKey {
749 pub agent: Option<String>,
751 pub key: SurfaceKey,
753 pub residue: Map<String, Value>,
755 pub recurrence: Option<Recurrence>,
757}
758
759const OPENCLAW_CHAT_KINDS: &[&str] = &["dm", "group", "channel", "thread"];
760
761fn key_of(platform: &str, kind: &str, chat_id: &str, thread_id: Option<String>) -> SurfaceKey {
762 SurfaceKey {
763 key: None,
764 platform: Some(platform.into()),
765 kind: Some(kind.into()),
766 chat_id: Some(chat_id.into()),
767 thread_id,
768 participant_id: None,
769 }
770}
771
772pub fn parse_openclaw_session_key(key: &str) -> Option<ParsedKey> {
777 let parts: Vec<&str> = key.split(':').collect();
778 let mut residue = Map::new();
779 residue.insert("session_key".into(), Value::String(key.into()));
780 match parts.first().copied() {
781 Some("agent") if parts.len() >= 3 => {
782 let agent = Some(parts[1].to_string());
783 if parts[2] == "main" {
784 residue.insert("dm_collapse".into(), Value::Bool(true));
785 return Some(ParsedKey {
786 agent,
787 key: key_of("main", "dm", "main", None),
788 residue,
789 recurrence: None,
790 });
791 }
792 if parts.len() < 5 {
793 return None;
794 }
795 let kind = if OPENCLAW_CHAT_KINDS.contains(&parts[3]) {
796 parts[3]
797 } else {
798 residue.insert("chat_kind".into(), Value::String(parts[3].into()));
799 "dm"
800 };
801 let mut thread_id = None;
802 if (parts.get(5) == Some(&"thread") || parts.get(5) == Some(&"topic"))
803 && parts.get(6).is_some_and(|t| !t.is_empty())
804 {
805 thread_id = Some(parts[6].to_string());
806 if parts[5] == "topic" {
807 residue.insert("thread_word".into(), Value::String("topic".into()));
808 }
809 }
810 let k = key_of(parts[2], kind, parts[4], thread_id);
811 let recurrence = if parts[2] == "cron" {
812 Some(Recurrence {
813 job_id: parts[4].into(),
814 kind: "cron".into(),
815 })
816 } else {
817 None
818 };
819 Some(ParsedKey {
820 agent,
821 key: k,
822 residue,
823 recurrence,
824 })
825 }
826 Some("cron") if parts.len() >= 2 => {
827 let job_id = parts[1..].join(":");
828 Some(ParsedKey {
829 agent: None,
830 key: key_of("cron", "dm", &job_id, None),
831 residue,
832 recurrence: Some(Recurrence {
833 job_id,
834 kind: "cron".into(),
835 }),
836 })
837 }
838 Some("hook") if parts.len() >= 2 => Some(ParsedKey {
839 agent: None,
840 key: key_of("webhook", "dm", parts[1], None),
841 residue,
842 recurrence: None,
843 }),
844 Some("acp-bridge") if parts.len() >= 2 => Some(ParsedKey {
845 agent: None,
846 key: key_of("acp", "dm", &parts[1..].join(":"), None),
847 residue,
848 recurrence: None,
849 }),
850 _ => None,
851 }
852}
853
854#[derive(Debug, Clone, Default)]
858pub struct OpenclawRootIo {
859 pub state_dir: PathBuf,
861 pub config_raw: String,
863 pub config_present: bool,
865 pub config_snapshot: String,
867 pub cron_jobs: BTreeMap<String, Map<String, Value>>,
869 pub cron_run_logs: BTreeMap<String, Map<String, Value>>,
871 pub delivery_queue_entries: BTreeMap<String, Map<String, Value>>,
873 pub schema_meta: Vec<Map<String, Value>>,
875 pub store_key: String,
877 pub db_present: bool,
879 pub default_agent: String,
881}
882
883#[derive(Debug, Clone, Default)]
885pub struct OpenclawProfileIo {
886 pub agent_id: String,
888 pub source_dir: PathBuf,
890 pub store_snapshot: String,
892 pub bindings_snapshot: String,
894}
895
896#[derive(Debug, Clone)]
898pub struct OpenclawLoaded {
899 pub orchestration: Orchestration,
901 pub vault: BTreeMap<String, String>,
903 pub root: OpenclawRootIo,
905 pub profiles: BTreeMap<String, OpenclawProfileIo>,
907}
908
909impl OpenclawLoaded {
910 pub fn from_orchestration(
913 orchestration: Orchestration,
914 vault: BTreeMap<String, String>,
915 ) -> Self {
916 let default_agent = orchestration.profiles["default"]
917 .residue
918 .config
919 .get("openclaw")
920 .and_then(|o| o.get("default_agent"))
921 .and_then(Value::as_str)
922 .unwrap_or("main")
923 .to_string();
924 let profiles = orchestration
925 .profiles
926 .keys()
927 .map(|n| {
928 (
929 n.clone(),
930 OpenclawProfileIo {
931 agent_id: if n == "default" {
932 default_agent.clone()
933 } else {
934 n.clone()
935 },
936 ..Default::default()
937 },
938 )
939 })
940 .collect();
941 Self {
942 root: OpenclawRootIo {
943 state_dir: orchestration.root.clone(),
944 default_agent,
945 ..Default::default()
946 },
947 orchestration,
948 vault,
949 profiles,
950 }
951 }
952}
953
954pub fn account_entries(entry: &Value) -> Vec<(String, Value)> {
958 let mut out: Vec<(String, Value)> = match entry.get("accounts") {
959 Some(Value::Object(m)) => m.iter().map(|(k, v)| (k.clone(), v.clone())).collect(),
960 Some(Value::Array(a)) => a
961 .iter()
962 .filter_map(|x| {
963 x.get("id")
964 .or_else(|| x.get("accountId"))
965 .and_then(Value::as_str)
966 .filter(|s| !s.is_empty())
967 .map(|id| (id.to_string(), x.clone()))
968 })
969 .collect(),
970 _ => Vec::new(),
971 };
972 out.sort_by(|a, b| a.0.cmp(&b.0));
973 out
974}
975
976fn credentials_of(
977 block: &Value,
978 scope: &str,
979 vault: &mut BTreeMap<String, String>,
980) -> BTreeMap<String, SecretRef> {
981 let mut creds = BTreeMap::new();
982 let Some(map) = block.as_object() else {
983 return creds;
984 };
985 for (k, v) in map {
986 match v {
987 Value::String(s) if is_credential_key(k) => {
988 let r = credential_ref_name(scope, k);
989 vault.insert(r.clone(), s.clone());
990 creds.insert(k.clone(), SecretRef::Dotenv(r));
991 }
992 Value::Object(o) if o.len() == 1 => {
993 if let Some(Value::String(n)) = o.get("dotenv") {
994 creds.insert(k.clone(), SecretRef::Dotenv(n.clone()));
995 } else if let Some(Value::String(n)) = o.get("env") {
996 creds.insert(k.clone(), SecretRef::Env(n.clone()));
997 }
998 }
999 _ => {}
1000 }
1001 }
1002 creds
1003}
1004
1005fn decode_channels(
1006 config: &Value,
1007 vault: &mut BTreeMap<String, String>,
1008) -> BTreeMap<String, ChannelConfig> {
1009 let mut channels = BTreeMap::new();
1010 let Some(map) = config.get("channels").and_then(Value::as_object) else {
1011 return channels;
1012 };
1013 for (kind, entry) in map {
1014 let Some(entry_map) = entry.as_object() else {
1015 continue;
1016 };
1017 let mut shared = entry_map.clone();
1018 shared.remove("accounts");
1019 let mut shared_block = redact_secrets(&Value::Object(shared), kind, vault);
1020 let channel_enabled = entry_map.get("enabled").and_then(Value::as_bool);
1021 if let Some(m) = shared_block.as_object_mut() {
1022 m.remove("enabled");
1023 }
1024 let accounts = account_entries(entry);
1025 let inline_account = OPENCLAW_ACCOUNT_KEYS
1026 .iter()
1027 .find_map(|k| entry_map.get(*k).and_then(Value::as_str))
1028 .map(str::to_string);
1029 if accounts.is_empty() {
1030 let mut extra = BTreeMap::new();
1031 extra.insert("kind".into(), Value::String(kind.clone()));
1032 extra.insert(
1033 "accountId".into(),
1034 inline_account
1035 .clone()
1036 .map(Value::String)
1037 .unwrap_or(Value::Null),
1038 );
1039 extra.insert(
1040 "account_source".into(),
1041 if inline_account.is_some() {
1042 Value::String("entry".into())
1043 } else {
1044 Value::Null
1045 },
1046 );
1047 extra.insert("accounts_form".into(), Value::Null);
1048 extra.insert(
1049 "enabled_on".into(),
1050 if channel_enabled.is_some() {
1051 Value::String("channel".into())
1052 } else {
1053 Value::Null
1054 },
1055 );
1056 extra.insert("channel_enabled".into(), Value::Null);
1057 extra.insert("channel_block".into(), shared_block.clone());
1058 channels.insert(
1059 kind.clone(),
1060 ChannelConfig {
1061 platform: kind.clone(),
1062 enabled: channel_enabled.unwrap_or(true),
1063 credentials: credentials_of(entry, kind, vault),
1064 extra,
1065 },
1066 );
1067 continue;
1068 }
1069 let form = if entry_map.get("accounts").is_some_and(Value::is_object) {
1070 "object"
1071 } else {
1072 "array"
1073 };
1074 for (id, account) in accounts {
1075 let name = format!("{kind}/{id}");
1076 let mut block = redact_secrets(&account, &name, vault);
1077 let account_enabled = account.get("enabled").and_then(Value::as_bool);
1078 if let Some(m) = block.as_object_mut() {
1079 m.remove("enabled");
1080 }
1081 let enabled = account_enabled.or(channel_enabled).unwrap_or(true);
1082 let mut credentials = credentials_of(&account, &name, vault);
1083 for (k, v) in credentials_of(entry, kind, vault) {
1084 credentials.insert(k, v);
1085 }
1086 let mut extra = BTreeMap::new();
1087 extra.insert("kind".into(), Value::String(kind.clone()));
1088 extra.insert("accountId".into(), Value::String(id.clone()));
1089 extra.insert("account_source".into(), Value::String("accounts".into()));
1090 extra.insert("accounts_form".into(), Value::String(form.into()));
1091 extra.insert(
1092 "enabled_on".into(),
1093 if account_enabled.is_some() {
1094 Value::String("account".into())
1095 } else if channel_enabled.is_some() {
1096 Value::String("channel".into())
1097 } else {
1098 Value::Null
1099 },
1100 );
1101 extra.insert(
1102 "channel_enabled".into(),
1103 channel_enabled.map(Value::Bool).unwrap_or(Value::Null),
1104 );
1105 extra.insert("channel_block".into(), shared_block.clone());
1106 extra.insert("account_block".into(), block);
1107 channels.insert(
1108 name.clone(),
1109 ChannelConfig {
1110 platform: name,
1111 enabled,
1112 credentials,
1113 extra,
1114 },
1115 );
1116 }
1117 }
1118 channels
1119}
1120
1121const CHANNEL_BOOKKEEPING: &[&str] = &[
1122 "kind",
1123 "accountId",
1124 "account_source",
1125 "accounts_form",
1126 "enabled_on",
1127 "channel_enabled",
1128 "channel_block",
1129 "account_block",
1130];
1131
1132fn block_from_model(ch: &ChannelConfig) -> Value {
1133 let mut out = Map::new();
1134 for (k, v) in &ch.extra {
1135 if CHANNEL_BOOKKEEPING.contains(&k.as_str()) {
1136 continue;
1137 }
1138 out.insert(k.strip_prefix("extra.").unwrap_or(k).to_string(), v.clone());
1139 }
1140 for (k, r) in &ch.credentials {
1141 out.insert(k.clone(), serde_json::to_value(r).unwrap());
1142 }
1143 Value::Object(out)
1144}
1145
1146fn ordered_channel_block(block: Value) -> Vec<(String, Value)> {
1147 let Some(map) = block.as_object() else {
1148 return Vec::new();
1149 };
1150 let mut out: Vec<(String, Value)> = Vec::new();
1151 if let Some(e) = map.get("enabled") {
1152 out.push(("enabled".into(), e.clone()));
1153 }
1154 for (k, v) in map {
1155 if k != "enabled" {
1156 out.push((k.clone(), v.clone()));
1157 }
1158 }
1159 out
1160}
1161
1162fn encode_channels(profile: &Profile, vault: &BTreeMap<String, String>) -> Vec<(String, Value)> {
1163 let mut by_kind: Vec<(String, Vec<&ChannelConfig>)> = Vec::new();
1164 for ch in profile.channels.values() {
1165 let kind = ch
1166 .extra
1167 .get("kind")
1168 .and_then(Value::as_str)
1169 .unwrap_or(&ch.platform)
1170 .to_string();
1171 match by_kind.iter_mut().find(|(k, _)| *k == kind) {
1172 Some((_, rows)) => rows.push(ch),
1173 None => by_kind.push((kind, vec![ch])),
1174 }
1175 }
1176 let mut out = Vec::new();
1177 for (kind, rows) in by_kind {
1178 let first = rows[0];
1179 let channel_block_src = first
1180 .extra
1181 .get("channel_block")
1182 .filter(|v| v.is_object())
1183 .cloned()
1184 .unwrap_or_else(|| block_from_model(first));
1185 let channel_block = inline_secrets(&channel_block_src, vault);
1186 let channel_enabled = first.extra.get("channel_enabled").and_then(Value::as_bool);
1187 if first.extra.get("account_source").and_then(Value::as_str) != Some("accounts") {
1188 let mut single = channel_block.as_object().cloned().unwrap_or_default();
1189 if first.extra.get("enabled_on").and_then(Value::as_str) == Some("channel")
1190 || !first.enabled
1191 {
1192 single.insert("enabled".into(), Value::Bool(first.enabled));
1193 }
1194 out.push((
1195 kind,
1196 ordered_object(ordered_channel_block(Value::Object(single))),
1197 ));
1198 continue;
1199 }
1200 let form = first
1201 .extra
1202 .get("accounts_form")
1203 .and_then(Value::as_str)
1204 .unwrap_or("object");
1205 let mut head = channel_block.as_object().cloned().unwrap_or_default();
1206 if let Some(e) = channel_enabled {
1207 head.insert("enabled".into(), Value::Bool(e));
1208 }
1209 let blocks: Vec<(String, Value)> = rows
1210 .iter()
1211 .map(|ch| {
1212 let block_src = ch
1213 .extra
1214 .get("account_block")
1215 .filter(|v| v.is_object())
1216 .cloned()
1217 .unwrap_or_else(|| block_from_model(ch));
1218 let block = inline_secrets(&block_src, vault);
1219 let inherited = channel_enabled.unwrap_or(true);
1220 let id = ch
1221 .extra
1222 .get("accountId")
1223 .and_then(Value::as_str)
1224 .unwrap_or("")
1225 .to_string();
1226 if ch.extra.get("enabled_on").and_then(Value::as_str) == Some("account")
1227 || ch.enabled != inherited
1228 {
1229 let mut b = vec![("enabled".to_string(), Value::Bool(ch.enabled))];
1230 b.extend(
1231 block
1232 .as_object()
1233 .map(|m| {
1234 m.iter()
1235 .map(|(k, v)| (k.clone(), v.clone()))
1236 .collect::<Vec<_>>()
1237 })
1238 .unwrap_or_default(),
1239 );
1240 (id, ordered_object(b))
1241 } else {
1242 (
1243 id,
1244 ordered_object(
1245 block
1246 .as_object()
1247 .map(|m| m.iter().map(|(k, v)| (k.clone(), v.clone())).collect())
1248 .unwrap_or_default(),
1249 ),
1250 )
1251 }
1252 })
1253 .collect();
1254 let mut pairs: Vec<(String, Value)> = head.into_iter().collect();
1255 if form == "array" {
1256 pairs.push((
1257 "accounts".into(),
1258 Value::Array(
1259 blocks
1260 .into_iter()
1261 .map(|(id, b)| {
1262 let mut p = vec![("id".to_string(), Value::String(id))];
1263 if let Some(items) = is_ordered_pairs(&b) {
1264 p.extend(items);
1265 }
1266 ordered_object(p)
1267 })
1268 .collect(),
1269 ),
1270 ));
1271 } else {
1272 pairs.push(("accounts".into(), ordered_object(blocks)));
1273 }
1274 out.push((kind, ordered_object(pairs)));
1275 }
1276 out
1277}
1278
1279fn is_ordered_pairs(value: &Value) -> Option<Vec<(String, Value)>> {
1280 let arr = value.as_array()?;
1281 if arr.len() == 2 && arr[0].as_str() == Some("__ordered__") {
1282 return arr[1].as_array().map(|items| {
1283 items
1284 .iter()
1285 .map(|p| {
1286 (
1287 p["__k"].as_str().unwrap_or("").to_string(),
1288 p["__v"].clone(),
1289 )
1290 })
1291 .collect()
1292 });
1293 }
1294 None
1295}
1296
1297const BINDING_MATCH_MAPPED: &[&str] = &["channel", "guildId", "peer"];
1300
1301fn decode_routes(config: &Value, name_for: &dyn Fn(&str) -> String) -> Vec<Route> {
1302 let mut routes = Vec::new();
1303 let Some(list) = config.get("bindings").and_then(Value::as_array) else {
1304 return routes;
1305 };
1306 for (index, binding) in list.iter().enumerate() {
1307 let (Some(bmap), Some(agent_id)) = (
1308 binding.as_object(),
1309 binding.get("agentId").and_then(Value::as_str),
1310 ) else {
1311 continue;
1312 };
1313 let m = binding
1314 .get("match")
1315 .and_then(Value::as_object)
1316 .cloned()
1317 .unwrap_or_default();
1318 let text = |v: &Value| match v {
1319 Value::String(s) => Some(s.clone()),
1320 Value::Number(n) => Some(n.to_string()),
1321 _ => None,
1322 };
1323 let matches = RouteMatch {
1324 platform: m
1325 .get("channel")
1326 .and_then(Value::as_str)
1327 .unwrap_or("")
1328 .to_string(),
1329 guild_id: m.get("guildId").filter(|v| !v.is_null()).and_then(text),
1330 chat_id: m
1331 .get("peer")
1332 .and_then(|p| p.get("id"))
1333 .filter(|v| !v.is_null())
1334 .and_then(text),
1335 thread_id: None,
1336 };
1337 let mut match_residue = Map::new();
1338 for (k, v) in &m {
1339 if !BINDING_MATCH_MAPPED.contains(&k.as_str()) {
1340 match_residue.insert(k.clone(), v.clone());
1341 }
1342 }
1343 if let Some(peer) = m.get("peer").and_then(Value::as_object) {
1344 let mut pr = peer.clone();
1345 pr.remove("id");
1346 if !pr.is_empty() {
1347 match_residue.insert("peer".into(), Value::Object(pr));
1348 }
1349 }
1350 let mut residue = Residue::default();
1351 residue.keep("agent_id", Value::String(agent_id.into()));
1352 residue.keep("index", Value::from(index));
1353 for (k, v) in bmap {
1354 if k != "agentId" && k != "match" {
1355 residue.keep(k.clone(), v.clone());
1356 }
1357 }
1358 if !match_residue.is_empty() {
1359 residue.keep("match", Value::Object(match_residue));
1360 }
1361 routes.push(Route {
1362 name: None,
1363 matches,
1364 profile: name_for(agent_id),
1365 residue,
1366 });
1367 }
1368 routes
1369}
1370
1371fn encode_routes(profile: &Profile, id_for_name: &dyn Fn(&str) -> String) -> Vec<Value> {
1372 profile
1373 .routes
1374 .iter()
1375 .map(|r| {
1376 let mut residue = r.residue.0.clone();
1377 let agent_id = residue
1378 .remove("agent_id")
1379 .and_then(|v| v.as_str().map(str::to_string))
1380 .unwrap_or_else(|| id_for_name(&r.profile));
1381 residue.remove("index");
1382 let match_residue = residue
1383 .remove("match")
1384 .and_then(|v| v.as_object().cloned())
1385 .unwrap_or_default();
1386 let mut m: Vec<(String, Value)> = Vec::new();
1387 if !r.matches.platform.is_empty() {
1388 m.push(("channel".into(), Value::String(r.matches.platform.clone())));
1389 }
1390 for (k, v) in &match_residue {
1391 if k != "peer" {
1392 m.push((k.clone(), v.clone()));
1393 }
1394 }
1395 if let Some(g) = &r.matches.guild_id {
1396 m.push(("guildId".into(), Value::String(g.clone())));
1397 }
1398 if r.matches.chat_id.is_some()
1399 || match_residue.get("peer").is_some_and(Value::is_object)
1400 {
1401 let mut peer: Vec<(String, Value)> = match_residue
1402 .get("peer")
1403 .and_then(Value::as_object)
1404 .map(|p| p.iter().map(|(k, v)| (k.clone(), v.clone())).collect())
1405 .unwrap_or_default();
1406 if let Some(c) = &r.matches.chat_id {
1407 peer.push(("id".into(), Value::String(c.clone())));
1408 }
1409 m.push(("peer".into(), ordered_object(peer)));
1410 }
1411 let mut pairs: Vec<(String, Value)> = residue.into_iter().collect();
1412 pairs.push(("agentId".into(), Value::String(agent_id)));
1413 pairs.push(("match".into(), ordered_object(m)));
1414 ordered_object(pairs)
1415 })
1416 .collect()
1417}
1418
1419#[derive(Debug, Clone, Default, PartialEq)]
1423struct HooksMeta {
1424 block: Map<String, Value>,
1425 has_token: bool,
1426}
1427
1428fn decode_hooks(
1429 config: &Value,
1430 vault: &mut BTreeMap<String, String>,
1431 name_for: &dyn Fn(&str) -> String,
1432 profiles: &mut BTreeMap<String, Profile>,
1433) -> Option<HooksMeta> {
1434 let hooks = config.get("hooks").and_then(Value::as_object)?;
1435 let mut secret = None;
1436 if let Some(Value::String(token)) = hooks.get("token") {
1437 let r = credential_ref_name("hooks", "token");
1438 vault.insert(r.clone(), token.clone());
1439 secret = Some(SecretRef::Dotenv(r));
1440 }
1441 if let Some(mappings) = hooks.get("mappings").and_then(Value::as_array) {
1442 for (index, mapping) in mappings.iter().enumerate() {
1443 let Some(mm) = mapping.as_object() else {
1444 continue;
1445 };
1446 let name = mm
1447 .get("id")
1448 .and_then(Value::as_str)
1449 .filter(|s| !s.is_empty())
1450 .map(str::to_string)
1451 .unwrap_or_else(|| format!("hook-{index}"));
1452 let mut residue_v = redact_secrets(mapping, &format!("hook_{name}"), vault)
1453 .as_object()
1454 .cloned()
1455 .unwrap_or_default();
1456 residue_v.remove("deliver");
1457 residue_v.remove("to");
1458 residue_v.insert("__index".into(), Value::from(index));
1459 let owner = mm
1460 .get("agentId")
1461 .and_then(Value::as_str)
1462 .map(name_for)
1463 .filter(|n| profiles.contains_key(n))
1464 .unwrap_or_else(|| "default".into());
1465 let deliver =
1466 mm.get("deliver")
1467 .and_then(Value::as_str)
1468 .map(|platform| Target::Explicit {
1469 platform: platform.into(),
1470 chat_id: mm.get("to").and_then(|t| match t {
1471 Value::String(s) => Some(s.clone()),
1472 Value::Number(n) => Some(n.to_string()),
1473 _ => None,
1474 }),
1475 thread_id: None,
1476 });
1477 profiles.get_mut(&owner).unwrap().subscriptions.insert(
1478 name.clone(),
1479 WebhookSubscription {
1480 name,
1481 secret: secret.clone(),
1482 events: None,
1483 prompt_template: String::new(),
1484 deliver,
1485 skills: Vec::new(),
1486 description: None,
1487 created_at: None,
1488 residue: Residue(residue_v.into_iter().collect()),
1489 },
1490 );
1491 }
1492 }
1493 let mut block = hooks.clone();
1494 block.remove("token");
1495 block.remove("mappings");
1496 Some(HooksMeta {
1497 block,
1498 has_token: secret.is_some(),
1499 })
1500}
1501
1502fn encode_hooks(
1503 orchestration: &Orchestration,
1504 meta: Option<&HooksMeta>,
1505 vault: &BTreeMap<String, String>,
1506) -> Option<Value> {
1507 let mut subs: Vec<&WebhookSubscription> = orchestration
1508 .profiles
1509 .values()
1510 .flat_map(|p| p.subscriptions.values())
1511 .collect();
1512 if meta.is_none() && subs.is_empty() {
1513 return None;
1514 }
1515 subs.sort_by_key(|s| {
1516 s.residue
1517 .0
1518 .get("__index")
1519 .and_then(Value::as_i64)
1520 .unwrap_or(0)
1521 });
1522 let mut pairs: Vec<(String, Value)> = meta
1523 .map(|m| {
1524 inline_secrets(&Value::Object(m.block.clone()), vault)
1525 .as_object()
1526 .map(|o| o.iter().map(|(k, v)| (k.clone(), v.clone())).collect())
1527 .unwrap_or_default()
1528 })
1529 .unwrap_or_default();
1530 if meta.is_some_and(|m| m.has_token) {
1531 let r = subs
1532 .iter()
1533 .find_map(|s| s.secret.clone())
1534 .unwrap_or_else(|| SecretRef::Dotenv(credential_ref_name("hooks", "token")));
1535 pairs.push((
1536 "token".into(),
1537 inline_secrets(&serde_json::to_value(&r).unwrap(), vault),
1538 ));
1539 }
1540 if !subs.is_empty() {
1541 pairs.push((
1542 "mappings".into(),
1543 Value::Array(
1544 subs.iter()
1545 .map(|sub| {
1546 let mut residue = inline_secrets(
1547 &Value::Object(sub.residue.0.clone().into_iter().collect()),
1548 vault,
1549 )
1550 .as_object()
1551 .cloned()
1552 .unwrap_or_default();
1553 residue.remove("__index");
1554 let mut mp: Vec<(String, Value)> =
1555 vec![("id".into(), Value::String(sub.name.clone()))];
1556 mp.extend(residue);
1557 if let Some(Target::Explicit {
1558 platform, chat_id, ..
1559 }) = &sub.deliver
1560 {
1561 mp.push(("deliver".into(), Value::String(platform.clone())));
1562 if let Some(c) = chat_id {
1563 mp.push(("to".into(), Value::String(c.clone())));
1564 }
1565 }
1566 ordered_object(mp)
1567 })
1568 .collect(),
1569 ),
1570 ));
1571 }
1572 Some(ordered_object(pairs))
1573}
1574
1575fn json_object_of(text: Option<&Value>) -> Map<String, Value> {
1578 text.and_then(Value::as_str)
1579 .and_then(|s| serde_json::from_str::<Value>(s).ok())
1580 .and_then(|v| v.as_object().cloned())
1581 .unwrap_or_default()
1582}
1583
1584fn opt_text(v: Option<&Value>) -> Option<String> {
1585 match v {
1586 Some(Value::String(s)) if !s.is_empty() => Some(s.clone()),
1587 Some(Value::Number(n)) => Some(n.to_string()),
1588 _ => None,
1589 }
1590}
1591
1592fn delivery_target(
1593 delivery: Option<&Map<String, Value>>,
1594 row: Option<&Map<String, Value>>,
1595) -> Option<Target> {
1596 let d = |k: &str| delivery.and_then(|d| d.get(k)).filter(|v| !v.is_null());
1597 let r = |k: &str| row.and_then(|r| r.get(k)).filter(|v| !v.is_null());
1598 let mode = d("mode")
1599 .or_else(|| d("kind"))
1600 .or_else(|| d("type"))
1601 .or_else(|| r("delivery_mode"))
1602 .and_then(Value::as_str)
1603 .map(str::to_string);
1604 let channel = d("channel")
1605 .or_else(|| r("delivery_channel"))
1606 .and_then(Value::as_str)
1607 .map(str::to_string);
1608 let to = opt_text(d("to").or_else(|| r("delivery_to")));
1609 let thread = opt_text(
1610 d("threadId")
1611 .or_else(|| d("thread_id"))
1612 .or_else(|| r("delivery_thread_id")),
1613 );
1614 if mode.as_deref() == Some("none") {
1615 return Some(Target::Local);
1616 }
1617 if matches!(channel.as_deref(), Some("last") | Some("origin")) {
1618 return Some(Target::Origin);
1619 }
1620 let Some(channel) = channel else {
1621 return if mode.is_some() {
1622 Some(Target::Local)
1623 } else {
1624 None
1625 };
1626 };
1627 Some(Target::Explicit {
1628 platform: channel,
1629 chat_id: to,
1630 thread_id: thread,
1631 })
1632}
1633
1634fn failure_target(record: &Map<String, Value>, row: &Map<String, Value>) -> Option<Target> {
1635 let f = record.get("failureDelivery").and_then(Value::as_object);
1636 let channel = f
1637 .and_then(|f| f.get("channel"))
1638 .or_else(|| row.get("failure_delivery_channel"))
1639 .filter(|v| !v.is_null())
1640 .cloned();
1641 let mode = f
1642 .and_then(|f| f.get("mode"))
1643 .or_else(|| row.get("failure_delivery_mode"))
1644 .filter(|v| !v.is_null())
1645 .cloned();
1646 if channel.is_none() && mode.is_none() {
1647 return None;
1648 }
1649 let mut synth = Map::new();
1650 if let Some(m) = mode {
1651 synth.insert("mode".into(), m);
1652 }
1653 if let Some(c) = channel {
1654 synth.insert("channel".into(), c);
1655 }
1656 if let Some(t) = f
1657 .and_then(|f| f.get("to"))
1658 .or_else(|| row.get("failure_delivery_to"))
1659 .filter(|v| !v.is_null())
1660 {
1661 synth.insert("to".into(), t.clone());
1662 }
1663 delivery_target(Some(&synth), None)
1664}
1665
1666fn origin_of(row: &Map<String, Value>) -> Option<JobOrigin> {
1667 let key = row.get("owner_session_key").and_then(Value::as_str)?;
1668 let parsed = parse_openclaw_session_key(key)?;
1669 Some(JobOrigin {
1670 platform: parsed.key.platform.unwrap_or_default(),
1671 chat_type: parsed.key.kind,
1672 chat_id: parsed.key.chat_id,
1673 thread_id: parsed.key.thread_id,
1674 })
1675}
1676
1677pub fn decode_job(row: &Map<String, Value>) -> Job {
1679 let record = json_object_of(row.get("job_json"));
1680 let state_json = json_object_of(row.get("state_json"));
1681 let raw_schedule = record
1682 .get("schedule")
1683 .and_then(Value::as_object)
1684 .cloned()
1685 .unwrap_or_default();
1686 let mut schedule_residue = raw_schedule.clone();
1687 let kind = raw_schedule.get("kind").and_then(Value::as_str);
1688 let schedule =
1689 if kind == Some("every") && raw_schedule.get("everyMs").is_some_and(Value::is_number) {
1690 schedule_residue.remove("kind");
1691 schedule_residue.remove("everyMs");
1692 Schedule::Interval {
1693 minutes: raw_schedule["everyMs"].as_f64().unwrap() / 60000.0,
1694 }
1695 } else if kind == Some("cron") && raw_schedule.get("expr").is_some_and(Value::is_string) {
1696 schedule_residue.remove("kind");
1697 schedule_residue.remove("expr");
1698 schedule_residue.remove("tz");
1699 Schedule::Cron {
1700 expr: raw_schedule["expr"].as_str().unwrap().into(),
1701 tz: raw_schedule
1702 .get("tz")
1703 .and_then(Value::as_str)
1704 .unwrap_or("UTC")
1705 .into(),
1706 }
1707 } else if kind == Some("at") && raw_schedule.get("at").is_some_and(Value::is_string) {
1708 schedule_residue.remove("kind");
1709 schedule_residue.remove("at");
1710 Schedule::Once {
1711 run_at: raw_schedule["at"].as_str().unwrap().into(),
1712 }
1713 } else {
1714 schedule_residue.insert("__unmapped".into(), Value::Bool(true));
1715 Schedule::Cron {
1716 expr: String::new(),
1717 tz: "UTC".into(),
1718 }
1719 };
1720 let payload = record
1721 .get("payload")
1722 .and_then(Value::as_object)
1723 .cloned()
1724 .unwrap_or_default();
1725 let prompt = ["message", "text", "command", "script"]
1726 .iter()
1727 .find_map(|k| payload.get(*k).and_then(Value::as_str))
1728 .map(str::to_string);
1729 let delivery = record
1733 .get("delivery")
1734 .and_then(Value::as_object)
1735 .cloned()
1736 .or_else(|| {
1737 let mut d = Map::new();
1738 for (column, key) in [
1739 ("delivery_mode", "mode"),
1740 ("delivery_channel", "channel"),
1741 ("delivery_to", "to"),
1742 ("delivery_thread_id", "threadId"),
1743 ("delivery_account_id", "accountId"),
1744 ] {
1745 if let Some(v) = row.get(column).filter(|v| !v.is_null()) {
1746 if v.as_str().is_some_and(str::is_empty) {
1747 continue;
1748 }
1749 d.insert(key.into(), v.clone());
1750 }
1751 }
1752 (!d.is_empty()).then_some(d)
1753 });
1754 let session_target = record
1755 .get("sessionTarget")
1756 .and_then(Value::as_str)
1757 .map(str::to_string)
1758 .or_else(|| {
1759 row.get("session_target")
1760 .and_then(Value::as_str)
1761 .map(str::to_string)
1762 });
1763 let job_id = opt_text(row.get("job_id")).unwrap_or_default();
1764 let mut residue = Residue::default();
1765 residue.keep(
1766 "__session_target",
1767 session_target.map(Value::String).unwrap_or(Value::Null),
1768 );
1769 residue.keep(
1770 "name",
1771 record
1772 .get("name")
1773 .and_then(Value::as_str)
1774 .map(|s| Value::String(s.into()))
1775 .unwrap_or_else(|| {
1776 row.get("name")
1777 .filter(|v| !v.is_null())
1778 .cloned()
1779 .unwrap_or(Value::String(job_id.clone()))
1780 }),
1781 );
1782 for (k, v) in &record {
1783 if [
1784 "id",
1785 "name",
1786 "enabled",
1787 "schedule",
1788 "payload",
1789 "delivery",
1790 "sessionTarget",
1791 "createdAtMs",
1792 "state",
1793 ]
1794 .contains(&k.as_str())
1795 {
1796 continue;
1797 }
1798 residue.keep(k.clone(), v.clone());
1799 }
1800 if !schedule_residue.is_empty() {
1801 residue.keep("__schedule", Value::Object(schedule_residue));
1802 }
1803 if !payload.is_empty() {
1804 residue.keep("__payload", Value::Object(payload.clone()));
1805 }
1806 if let Some(d) = &delivery {
1807 residue.keep("__delivery", Value::Object(d.clone()));
1808 }
1809 if let Some(f) = record.get("failureDelivery").filter(|v| v.is_object()) {
1810 residue.keep("__failure_delivery", f.clone());
1811 }
1812 if !state_json.is_empty() {
1813 residue.keep("__state", Value::Object(state_json.clone()));
1814 }
1815 let enabled = match row.get("enabled") {
1816 None | Some(Value::Null) => true,
1817 Some(Value::Bool(b)) => *b,
1818 Some(Value::Number(n)) => n.as_f64() != Some(0.0),
1819 Some(other) => !matches!(other, Value::String(s) if s.is_empty()),
1820 };
1821 Job {
1822 id: job_id,
1823 schedule,
1824 prompt,
1825 workdir: None,
1826 model: payload
1827 .get("model")
1828 .and_then(Value::as_str)
1829 .map(str::to_string)
1830 .or_else(|| {
1831 row.get("payload_model")
1832 .and_then(Value::as_str)
1833 .map(str::to_string)
1834 }),
1835 skills: Vec::new(),
1836 context_from: None,
1837 deliver: delivery_target(delivery.as_ref(), Some(row)).unwrap_or(Target::Local),
1838 failure_deliver: failure_target(&record, row),
1839 origin: origin_of(row),
1840 attach_to_session: None,
1841 repeat: None,
1842 enabled,
1843 next_run_at: iso_seconds(
1844 ms_of(row.get("next_run_at_ms").filter(|v| !v.is_null()))
1845 .or_else(|| ms_of(state_json.get("nextRunAtMs"))),
1846 ),
1847 last_run_at: iso_seconds(
1848 ms_of(row.get("last_run_at_ms").filter(|v| !v.is_null()))
1849 .or_else(|| ms_of(state_json.get("lastRunAtMs"))),
1850 ),
1851 last_status: row
1852 .get("last_run_status")
1853 .and_then(Value::as_str)
1854 .map(str::to_string),
1855 created_at: iso_seconds(
1856 ms_of(row.get("created_at_ms").filter(|v| !v.is_null()))
1857 .or_else(|| ms_of(record.get("createdAtMs"))),
1858 ),
1859 residue,
1860 }
1861}
1862
1863fn empty_row(columns: &[&str]) -> Map<String, Value> {
1864 columns
1865 .iter()
1866 .map(|c| (c.to_string(), Value::Null))
1867 .collect()
1868}
1869
1870pub fn encode_job_row(
1872 job: &Job,
1873 raw: Option<&Map<String, Value>>,
1874 store_key: &str,
1875) -> Map<String, Value> {
1876 let residue = &job.residue.0;
1877 let mut row = raw.cloned().unwrap_or_else(|| empty_row(CRON_JOB_COLUMNS));
1878 let mut schedule = Map::new();
1879 match &job.schedule {
1880 Schedule::Interval { minutes } => {
1881 schedule.insert("kind".into(), "every".into());
1882 schedule.insert(
1883 "everyMs".into(),
1884 Value::from((minutes * 60000.0).round() as i64),
1885 );
1886 }
1887 Schedule::Cron { expr, tz } => {
1888 schedule.insert("kind".into(), "cron".into());
1889 schedule.insert("expr".into(), Value::String(expr.clone()));
1890 if tz != "UTC" {
1891 schedule.insert("tz".into(), Value::String(tz.clone()));
1892 }
1893 }
1894 Schedule::Once { run_at } => {
1895 schedule.insert("kind".into(), "at".into());
1896 schedule.insert("at".into(), Value::String(run_at.clone()));
1897 }
1898 }
1899 if let Some(Value::Object(extra)) = residue.get("__schedule") {
1900 for (k, v) in extra {
1901 schedule.insert(k.clone(), v.clone());
1902 }
1903 }
1904 schedule.remove("__unmapped");
1905 let mut record = Map::new();
1906 record.insert("id".into(), Value::String(job.id.clone()));
1907 record.insert(
1908 "name".into(),
1909 residue
1910 .get("name")
1911 .cloned()
1912 .unwrap_or(Value::String(job.id.clone())),
1913 );
1914 record.insert("enabled".into(), Value::Bool(job.enabled));
1915 if let Some(ms) = ms_from_iso(job.created_at.as_deref()) {
1916 record.insert("createdAtMs".into(), Value::from(ms));
1917 }
1918 record.insert("schedule".into(), Value::Object(schedule.clone()));
1919 if let Some(st) = residue.get("__session_target").filter(|v| !v.is_null()) {
1920 record.insert("sessionTarget".into(), st.clone());
1921 }
1922 match residue.get("__payload").and_then(Value::as_object) {
1923 Some(p) => {
1924 let mut p = p.clone();
1925 if let Some(prompt) = &job.prompt {
1926 for k in ["message", "text", "command", "script"] {
1927 if p.contains_key(k) {
1928 p.insert(k.into(), Value::String(prompt.clone()));
1929 break;
1930 }
1931 }
1932 }
1933 record.insert("payload".into(), Value::Object(p));
1934 }
1935 None => {
1936 if let Some(prompt) = &job.prompt {
1937 record.insert(
1938 "payload".into(),
1939 serde_json::json!({"kind": "agentTurn", "message": prompt}),
1940 );
1941 }
1942 }
1943 }
1944 if let Some(d) = residue.get("__delivery") {
1945 record.insert("delivery".into(), d.clone());
1946 }
1947 if let Some(f) = residue.get("__failure_delivery") {
1948 record.insert("failureDelivery".into(), f.clone());
1949 }
1950 record.insert(
1951 "state".into(),
1952 residue
1953 .get("__state")
1954 .cloned()
1955 .unwrap_or_else(|| Value::Object(Map::new())),
1956 );
1957 for (k, v) in residue {
1958 if !k.starts_with("__") {
1959 record.insert(k.clone(), v.clone());
1960 }
1961 }
1962 let get = |m: &Map<String, Value>, k: &str| m.get(k).filter(|v| !v.is_null()).cloned();
1963 let delivery = record
1964 .get("delivery")
1965 .and_then(Value::as_object)
1966 .cloned()
1967 .unwrap_or_default();
1968 let payload = record
1969 .get("payload")
1970 .and_then(Value::as_object)
1971 .cloned()
1972 .unwrap_or_default();
1973 row.insert(
1974 "store_key".into(),
1975 get(&row, "store_key").unwrap_or(Value::String(store_key.into())),
1976 );
1977 row.insert("job_id".into(), Value::String(job.id.clone()));
1978 row.insert("name".into(), record["name"].clone());
1979 row.insert("enabled".into(), Value::from(i64::from(job.enabled)));
1980 row.insert(
1981 "created_at_ms".into(),
1982 record
1983 .get("createdAtMs")
1984 .cloned()
1985 .or_else(|| get(&row, "created_at_ms"))
1986 .unwrap_or(Value::from(0)),
1987 );
1988 row.insert(
1989 "schedule_kind".into(),
1990 schedule
1991 .get("kind")
1992 .cloned()
1993 .unwrap_or(Value::String("cron".into())),
1994 );
1995 row.insert(
1996 "schedule_expr".into(),
1997 schedule.get("expr").cloned().unwrap_or(Value::Null),
1998 );
1999 row.insert(
2000 "schedule_tz".into(),
2001 schedule.get("tz").cloned().unwrap_or(Value::Null),
2002 );
2003 row.insert(
2004 "every_ms".into(),
2005 schedule.get("everyMs").cloned().unwrap_or(Value::Null),
2006 );
2007 row.insert(
2008 "anchor_ms".into(),
2009 schedule.get("anchorMs").cloned().unwrap_or(Value::Null),
2010 );
2011 row.insert(
2012 "at".into(),
2013 schedule.get("at").cloned().unwrap_or(Value::Null),
2014 );
2015 row.insert(
2016 "session_target".into(),
2017 record
2018 .get("sessionTarget")
2019 .cloned()
2020 .or_else(|| get(&row, "session_target"))
2021 .unwrap_or(Value::String("isolated".into())),
2022 );
2023 row.insert(
2024 "wake_mode".into(),
2025 get(&row, "wake_mode").unwrap_or(Value::String("now".into())),
2026 );
2027 row.insert(
2028 "payload_kind".into(),
2029 payload
2030 .get("kind")
2031 .cloned()
2032 .or_else(|| get(&row, "payload_kind"))
2033 .unwrap_or(Value::String("agentTurn".into())),
2034 );
2035 row.insert(
2036 "payload_message".into(),
2037 job.prompt.clone().map(Value::String).unwrap_or(Value::Null),
2038 );
2039 row.insert(
2040 "payload_model".into(),
2041 job.model.clone().map(Value::String).unwrap_or(Value::Null),
2042 );
2043 row.insert(
2044 "delivery_mode".into(),
2045 delivery
2046 .get("mode")
2047 .or_else(|| delivery.get("kind"))
2048 .cloned()
2049 .unwrap_or(Value::Null),
2050 );
2051 row.insert(
2052 "delivery_channel".into(),
2053 delivery.get("channel").cloned().unwrap_or(Value::Null),
2054 );
2055 row.insert(
2056 "delivery_to".into(),
2057 delivery.get("to").cloned().unwrap_or(Value::Null),
2058 );
2059 row.insert(
2060 "delivery_thread_id".into(),
2061 delivery.get("threadId").cloned().unwrap_or(Value::Null),
2062 );
2063 row.insert(
2064 "delivery_account_id".into(),
2065 delivery.get("accountId").cloned().unwrap_or(Value::Null),
2066 );
2067 row.insert(
2068 "next_run_at_ms".into(),
2069 ms_from_iso(job.next_run_at.as_deref())
2070 .map(Value::from)
2071 .unwrap_or(Value::Null),
2072 );
2073 row.insert(
2074 "last_run_at_ms".into(),
2075 ms_from_iso(job.last_run_at.as_deref())
2076 .map(Value::from)
2077 .unwrap_or(Value::Null),
2078 );
2079 row.insert(
2080 "last_run_status".into(),
2081 job.last_status
2082 .clone()
2083 .map(Value::String)
2084 .unwrap_or(Value::Null),
2085 );
2086 row.insert(
2087 "job_json".into(),
2088 Value::String(serde_json::to_string(&Value::Object(record.clone())).unwrap()),
2089 );
2090 row.insert(
2091 "state_json".into(),
2092 Value::String(
2093 serde_json::to_string(record.get("state").unwrap_or(&Value::Object(Map::new())))
2094 .unwrap(),
2095 ),
2096 );
2097 row.insert(
2098 "sort_order".into(),
2099 get(&row, "sort_order").unwrap_or(Value::from(0)),
2100 );
2101 row.insert(
2102 "updated_at".into(),
2103 get(&row, "updated_at")
2104 .or_else(|| get(&row, "created_at_ms"))
2105 .unwrap_or(Value::from(0)),
2106 );
2107 row
2108}
2109
2110fn run_status(word: Option<&str>) -> FireStatus {
2113 match word {
2114 Some("ok") => FireStatus::Succeeded,
2115 Some("error") => FireStatus::Failed,
2116 _ => FireStatus::Unknown,
2117 }
2118}
2119
2120fn run_status_back(status: FireStatus) -> &'static str {
2121 match status {
2122 FireStatus::Succeeded | FireStatus::Claimed | FireStatus::Running => "ok",
2123 FireStatus::Failed | FireStatus::Timeout => "error",
2124 FireStatus::Unknown => "skipped",
2125 }
2126}
2127
2128const FIRE_MAPPED: &[&str] = &[
2129 "job_id",
2130 "seq",
2131 "ts",
2132 "error",
2133 "run_id",
2134 "run_at_ms",
2135 "session_id",
2136];
2137
2138pub fn decode_fire(row: &Map<String, Value>) -> Fire {
2140 let job_id = opt_text(row.get("job_id")).unwrap_or_default();
2141 let id = opt_text(row.get("run_id")).unwrap_or_else(|| {
2142 format!(
2143 "{job_id}#{}",
2144 row.get("seq").map(|v| v.to_string()).unwrap_or_default()
2145 )
2146 });
2147 let started = iso_millis(ms_of(row.get("run_at_ms").filter(|v| !v.is_null())));
2148 let finished = iso_millis(ms_of(row.get("ts").filter(|v| !v.is_null())));
2149 let mut residue = Residue::default();
2150 residue.keep("__no_claim", Value::Bool(true));
2151 for (k, v) in row {
2152 if !FIRE_MAPPED.contains(&k.as_str()) && !v.is_null() {
2153 residue.keep(k.clone(), v.clone());
2154 }
2155 }
2156 Fire {
2157 id,
2158 job_id,
2159 session_id: opt_text(row.get("session_id")),
2160 status: run_status(row.get("status").and_then(Value::as_str)),
2161 claimed_at: started
2162 .clone()
2163 .or_else(|| finished.clone())
2164 .unwrap_or_default(),
2165 started_at: started,
2166 finished_at: finished,
2167 error: opt_text(row.get("error")),
2168 obligation_id: None,
2169 residue,
2170 }
2171}
2172
2173pub fn encode_fire_row(
2175 fire: &Fire,
2176 raw: Option<&Map<String, Value>>,
2177 store_key: &str,
2178) -> Map<String, Value> {
2179 let residue = &fire.residue.0;
2180 let mut row = raw
2181 .cloned()
2182 .unwrap_or_else(|| empty_row(CRON_RUN_LOG_COLUMNS));
2183 let get = |m: &Map<String, Value>, k: &str| m.get(k).filter(|v| !v.is_null()).cloned();
2184 row.insert(
2185 "store_key".into(),
2186 get(&row, "store_key")
2187 .or_else(|| residue.get("store_key").cloned())
2188 .unwrap_or(Value::String(store_key.into())),
2189 );
2190 row.insert("job_id".into(), Value::String(fire.job_id.clone()));
2191 row.insert(
2192 "seq".into(),
2193 get(&row, "seq")
2194 .or_else(|| residue.get("seq").cloned())
2195 .unwrap_or(Value::Null),
2196 );
2197 let ts = ms_from_iso(fire.finished_at.as_deref())
2198 .map(Value::from)
2199 .or_else(|| get(&row, "ts"))
2200 .unwrap_or(Value::from(0));
2201 row.insert("ts".into(), ts.clone());
2202 row.insert(
2203 "status".into(),
2204 residue
2205 .get("status")
2206 .cloned()
2207 .unwrap_or(Value::String(run_status_back(fire.status).into())),
2208 );
2209 row.insert(
2210 "error".into(),
2211 fire.error.clone().map(Value::String).unwrap_or(Value::Null),
2212 );
2213 row.insert(
2214 "delivery_status".into(),
2215 residue
2216 .get("delivery_status")
2217 .cloned()
2218 .unwrap_or(Value::Null),
2219 );
2220 row.insert(
2221 "delivery_error".into(),
2222 residue
2223 .get("delivery_error")
2224 .cloned()
2225 .unwrap_or(Value::Null),
2226 );
2227 row.insert(
2228 "delivered".into(),
2229 residue.get("delivered").cloned().unwrap_or(Value::Null),
2230 );
2231 row.insert(
2232 "session_id".into(),
2233 fire.session_id
2234 .clone()
2235 .map(Value::String)
2236 .unwrap_or(Value::Null),
2237 );
2238 row.insert(
2239 "session_key".into(),
2240 residue.get("session_key").cloned().unwrap_or(Value::Null),
2241 );
2242 row.insert(
2243 "run_id".into(),
2244 if fire.id.contains('#') {
2245 Value::Null
2246 } else {
2247 Value::String(fire.id.clone())
2248 },
2249 );
2250 row.insert(
2251 "run_at_ms".into(),
2252 ms_from_iso(fire.started_at.as_deref())
2253 .map(Value::from)
2254 .unwrap_or(Value::Null),
2255 );
2256 row.insert("entry_json".into(), residue.get("entry_json").cloned().unwrap_or_else(|| Value::String(serde_json::json!({"action": "finished", "ts": ts, "jobId": fire.job_id, "status": row["status"]}).to_string())));
2257 row.insert(
2258 "created_at".into(),
2259 residue.get("created_at").cloned().unwrap_or(ts),
2260 );
2261 for c in CRON_RUN_LOG_COLUMNS {
2262 if !row.contains_key(*c) {
2263 row.insert(
2264 c.to_string(),
2265 residue.get(*c).cloned().unwrap_or(Value::Null),
2266 );
2267 }
2268 }
2269 row
2270}
2271
2272fn obl_state(native: &str) -> ObligationState {
2275 match native {
2276 "pending" | "queued" | "sending" | "in_flight" | "retrying" => ObligationState::Pending,
2277 "delivered" | "sent" => ObligationState::Sent,
2278 "failed" => ObligationState::Failed,
2279 "dropped" | "dead" | "cancelled" | "canceled" => ObligationState::Dropped,
2280 _ => ObligationState::Pending,
2281 }
2282}
2283
2284const OBL_MAPPED: &[&str] = &[
2285 "id",
2286 "status",
2287 "session_key",
2288 "channel",
2289 "target",
2290 "retry_count",
2291 "last_error",
2292 "enqueued_at",
2293 "updated_at",
2294];
2295
2296pub fn decode_obligation(row: &Map<String, Value>) -> Obligation {
2298 let parsed = row
2299 .get("session_key")
2300 .and_then(Value::as_str)
2301 .and_then(parse_openclaw_session_key);
2302 let native = row
2303 .get("status")
2304 .and_then(Value::as_str)
2305 .map(|s| s.to_ascii_lowercase())
2306 .unwrap_or_default();
2307 let state = obl_state(&native);
2308 let mut content = OutboundContent {
2309 text: String::new(),
2310 attachments: None,
2311 reply_to: None,
2312 format: None,
2313 };
2314 let mut residue = Residue::default();
2315 residue.keep("status", row.get("status").cloned().unwrap_or(Value::Null));
2316 if let Some(entry) = row
2317 .get("entry_json")
2318 .and_then(Value::as_str)
2319 .and_then(|s| serde_json::from_str::<Value>(s).ok())
2320 .and_then(|v| v.as_object().cloned())
2321 {
2322 if let Some(t) = ["text", "message", "content", "body"]
2323 .iter()
2324 .find_map(|k| entry.get(*k).and_then(Value::as_str))
2325 {
2326 content.text = t.into();
2327 }
2328 }
2329 for (k, v) in row {
2330 if !OBL_MAPPED.contains(&k.as_str()) && !v.is_null() {
2331 residue.keep(k.clone(), v.clone());
2332 }
2333 }
2334 let text = |k: &str| {
2335 row.get(k).and_then(|v| match v {
2336 Value::String(s) => Some(s.clone()),
2337 Value::Number(n) => Some(n.to_string()),
2338 _ => None,
2339 })
2340 };
2341 let created_at = text("enqueued_at").unwrap_or_else(|| "0".into());
2342 let updated_at = text("updated_at").unwrap_or_else(|| created_at.clone());
2343 Obligation {
2344 id: opt_text(row.get("id")).unwrap_or_default(),
2345 target: SurfaceKey {
2346 key: None,
2347 platform: Some(
2348 text("channel")
2349 .filter(|s| !s.is_empty())
2350 .or_else(|| parsed.as_ref().and_then(|p| p.key.platform.clone()))
2351 .unwrap_or_default(),
2352 ),
2353 kind: parsed.as_ref().and_then(|p| p.key.kind.clone()),
2354 chat_id: Some(text("target").unwrap_or_default()),
2355 thread_id: parsed.as_ref().and_then(|p| p.key.thread_id.clone()),
2356 participant_id: None,
2357 },
2358 session_key: row
2359 .get("session_key")
2360 .and_then(Value::as_str)
2361 .map(str::to_string),
2362 content,
2363 state,
2364 attempts: ms_of(row.get("retry_count")).unwrap_or(0).max(0) as u64,
2365 last_error: row
2366 .get("last_error")
2367 .and_then(Value::as_str)
2368 .map(str::to_string),
2369 delivered_at: if state == ObligationState::Sent {
2370 iso_millis(
2371 ms_of(row.get("updated_at").filter(|v| !v.is_null()))
2372 .or_else(|| ms_of(row.get("enqueued_at"))),
2373 )
2374 } else {
2375 None
2376 },
2377 created_at,
2378 updated_at,
2379 posted: None,
2380 source: match parsed {
2381 Some(p) => ObligationSource::Turn { key: Some(p.key) },
2382 None => ObligationSource::Turn { key: None },
2383 },
2384 residue,
2385 }
2386}
2387
2388pub fn encode_obligation_row(
2390 o: &Obligation,
2391 raw: Option<&Map<String, Value>>,
2392) -> Map<String, Value> {
2393 let residue = &o.residue.0;
2394 let mut row = raw
2395 .cloned()
2396 .unwrap_or_else(|| empty_row(DELIVERY_QUEUE_COLUMNS));
2397 let get = |m: &Map<String, Value>, k: &str| m.get(k).filter(|v| !v.is_null()).cloned();
2398 row.insert(
2399 "queue_name".into(),
2400 get(&row, "queue_name")
2401 .or_else(|| residue.get("queue_name").cloned())
2402 .unwrap_or(Value::String("default".into())),
2403 );
2404 row.insert("id".into(), Value::String(o.id.clone()));
2405 row.insert(
2406 "status".into(),
2407 residue
2408 .get("status")
2409 .filter(|v| !v.is_null())
2410 .cloned()
2411 .unwrap_or(Value::String(o.state.hermes_word().into())),
2412 );
2413 row.insert(
2414 "session_key".into(),
2415 o.session_key
2416 .clone()
2417 .map(Value::String)
2418 .unwrap_or(Value::Null),
2419 );
2420 row.insert(
2421 "channel".into(),
2422 o.target
2423 .platform
2424 .clone()
2425 .filter(|p| !p.is_empty())
2426 .map(Value::String)
2427 .unwrap_or(Value::Null),
2428 );
2429 row.insert(
2430 "target".into(),
2431 o.target
2432 .chat_id
2433 .clone()
2434 .filter(|c| !c.is_empty())
2435 .map(Value::String)
2436 .unwrap_or(Value::Null),
2437 );
2438 row.insert(
2439 "last_error".into(),
2440 o.last_error
2441 .clone()
2442 .map(Value::String)
2443 .unwrap_or(Value::Null),
2444 );
2445 let enq = o.created_at.parse::<f64>().map(|f| f as i64).unwrap_or(0);
2446 row.insert("enqueued_at".into(), Value::from(enq));
2447 row.insert(
2448 "updated_at".into(),
2449 Value::from(o.updated_at.parse::<f64>().map(|f| f as i64).unwrap_or(enq)),
2450 );
2451 row.insert("entry_json".into(), residue.get("entry_json").cloned().unwrap_or_else(|| Value::String(serde_json::json!({"kind": residue.get("entry_kind").cloned().unwrap_or(Value::String("message".into())), "text": o.content.text}).to_string())));
2452 for c in DELIVERY_QUEUE_COLUMNS {
2453 if row.get(*c).is_none_or(Value::is_null) {
2454 row.insert(
2455 c.to_string(),
2456 residue.get(*c).cloned().unwrap_or(Value::Null),
2457 );
2458 }
2459 }
2460 row.insert("retry_count".into(), Value::from(o.attempts));
2461 row
2462}
2463
2464fn bindings_for_agent(agent_dir: &Path, agent_id: &str) -> Result<Vec<Binding>> {
2467 let mut out = Vec::new();
2468 let sessions = agent_dir.join("sessions");
2469 if !sessions.is_dir() {
2470 return Ok(out);
2471 }
2472 let mut entries: Vec<_> = fs::read_dir(&sessions)?.flatten().collect();
2473 entries.sort_by_key(|e| e.file_name());
2474 for entry in entries {
2475 let name = entry.file_name().to_string_lossy().into_owned();
2476 if !name.ends_with(".jsonl") || name.ends_with(".trajectory.jsonl") {
2477 continue;
2478 }
2479 let path = entry.path();
2480 let Ok(text) = fs::read_to_string(&path) else {
2481 continue;
2482 };
2483 let lines: Vec<&str> = text.lines().filter(|l| !l.trim().is_empty()).collect();
2484 let Some(first) = lines.first() else { continue };
2485 let Ok(header) = serde_json::from_str::<Value>(first) else {
2486 continue;
2487 };
2488 if header.get("type").and_then(Value::as_str) != Some("session") {
2489 continue;
2490 }
2491 let Some(key) = header
2492 .get("sessionKey")
2493 .or_else(|| header.get("__openclaw").and_then(|o| o.get("sessionKey")))
2494 .and_then(Value::as_str)
2495 else {
2496 continue;
2497 };
2498 let Some(parsed) = parse_openclaw_session_key(key) else {
2499 continue;
2500 };
2501 let header_ts = header
2502 .get("timestamp")
2503 .and_then(Value::as_str)
2504 .map(str::to_string);
2505 let mut last = header_ts.clone();
2506 for line in lines.iter().rev() {
2507 if let Ok(rec) = serde_json::from_str::<Value>(line) {
2508 if let Some(ts) = rec.get("timestamp").and_then(Value::as_str) {
2509 last = Some(ts.into());
2510 break;
2511 }
2512 }
2513 }
2514 let mut residue = Residue(parsed.residue.into_iter().collect());
2515 residue.keep(
2516 "agent_id",
2517 Value::String(parsed.agent.clone().unwrap_or_else(|| agent_id.into())),
2518 );
2519 out.push(Binding {
2520 trigger: if parsed.recurrence.is_some() {
2521 Trigger::Cron
2522 } else {
2523 Trigger::Channel
2524 },
2525 key: parsed.key,
2526 profile: None,
2527 worker: Worker {
2528 harness: HarnessId::new(HarnessId::OPENCLAW),
2529 session_id: Some(
2530 header
2531 .get("id")
2532 .and_then(Value::as_str)
2533 .map(str::to_string)
2534 .unwrap_or_else(|| name.trim_end_matches(".jsonl").into()),
2535 ),
2536 locator: Some(path.display().to_string()),
2537 },
2538 recurrence: parsed.recurrence,
2539 handoff: None,
2540 started_at: header_ts.clone().or_else(|| last.clone()),
2541 last_activity_at: last.or(header_ts),
2542 ended_at: None,
2543 end_reason: None,
2544 residue,
2545 });
2546 }
2547 Ok(out)
2548}
2549
2550fn list_unmodeled(state_dir: &Path) -> Result<Vec<String>> {
2551 let mut out = Vec::new();
2552 fn walk(base: &Path, dir: &Path, out: &mut Vec<String>) -> Result<()> {
2553 let mut entries: Vec<_> = fs::read_dir(dir)?.flatten().collect();
2554 entries.sort_by_key(|e| e.file_name());
2555 for entry in entries {
2556 let name = entry.file_name().to_string_lossy().into_owned();
2557 if name == "node_modules" || name == ".git" {
2558 continue;
2559 }
2560 let p = entry.path();
2561 let rel = p
2562 .strip_prefix(base)
2563 .unwrap_or(&p)
2564 .to_string_lossy()
2565 .replace('\\', "/");
2566 let Ok(st) = fs::symlink_metadata(&p) else {
2567 continue;
2568 };
2569 if st.is_dir() {
2570 walk(base, &p, out)?;
2571 continue;
2572 }
2573 if rel == OPENCLAW_CONFIG
2574 || rel == OPENCLAW_STATE_DB
2575 || rel.starts_with(&format!("{OPENCLAW_STATE_DB}-"))
2576 {
2577 continue;
2578 }
2579 out.push(rel);
2580 }
2581 Ok(())
2582 }
2583 walk(state_dir, state_dir, &mut out)?;
2584 Ok(out)
2585}
2586
2587fn config_record(orchestration: &Orchestration) -> Value {
2590 let root = &orchestration.profiles["default"];
2591 serde_json::json!({
2592 "channels": root.channels, "routes": root.routes, "residue": root.residue.config,
2593 "profiles": orchestration.profiles.iter().map(|(n, p)| (n.clone(), serde_json::json!({"agent": p.residue.config.get("openclaw_agent"), "subscriptions": p.subscriptions}))).collect::<BTreeMap<_, _>>(),
2594 })
2595}
2596
2597fn store_record(profile: &Profile) -> Value {
2598 serde_json::json!({ "jobs": profile.jobs, "fires": profile.fires, "obligations": profile.obligations })
2599}
2600
2601pub fn from_openclaw(state_dir: &Path) -> Result<OpenclawLoaded> {
2603 if !state_dir.is_dir() {
2604 return Err(load_error(
2605 &state_dir.display().to_string(),
2606 "",
2607 "not a directory",
2608 ));
2609 }
2610 let mut vault = BTreeMap::new();
2611 let config_path = state_dir.join(OPENCLAW_CONFIG);
2612 let config_text = if config_path.is_file() {
2613 Some(fs::read_to_string(&config_path)?)
2614 } else {
2615 None
2616 };
2617 let config = match &config_text {
2618 Some(t) => parse_json5(&config_path.display().to_string(), t)?,
2619 None => Value::Object(Map::new()),
2620 };
2621 if !config.is_object() {
2622 return Err(load_error(
2623 &config_path.display().to_string(),
2624 "",
2625 "expected a JSON5 object",
2626 ));
2627 }
2628 let entries = agent_entries(&config);
2629 let default_id = default_agent_id(&config);
2630 let declared: Vec<(String, Value)> = if entries.is_empty() {
2631 vec![(default_id.clone(), Value::Object(Map::new()))]
2632 } else {
2633 entries
2634 };
2635 let name_for = |agent_id: &str| -> String {
2636 if agent_id == default_id {
2637 "default".into()
2638 } else {
2639 agent_id.into()
2640 }
2641 };
2642
2643 let mut profiles: BTreeMap<String, Profile> = BTreeMap::new();
2644 let mut ios: BTreeMap<String, OpenclawProfileIo> = BTreeMap::new();
2645 for (id, entry) in &declared {
2646 let name = name_for(id);
2647 let dir = state_dir.join("agents").join(id);
2648 let mut profile = empty_profile(&name, &dir);
2649 profile.residue.config.insert(
2650 "openclaw_agent".into(),
2651 redact_secrets(entry, &format!("agent_{id}"), &mut vault),
2652 );
2653 if let Ok(text) = fs::read_to_string(dir.join("AGENTS.md")) {
2654 profile.persona = Some(persona_ref(&text));
2655 }
2656 for b in bindings_for_agent(&dir, id)? {
2657 profile.bindings.insert(surface_key_string(&b.key), b);
2658 }
2659 profiles.insert(name.clone(), profile);
2660 ios.insert(
2661 name,
2662 OpenclawProfileIo {
2663 agent_id: id.clone(),
2664 source_dir: dir,
2665 ..Default::default()
2666 },
2667 );
2668 }
2669 let channels = decode_channels(&config, &mut vault);
2671 let routes = decode_routes(&config, &name_for);
2672 let mut rest = Map::new();
2673 for (k, v) in config.as_object().unwrap() {
2674 if !["channels", "bindings", "agents", "hooks"].contains(&k.as_str()) {
2675 rest.insert(k.clone(), v.clone());
2676 }
2677 }
2678 let mut agents_rest = Map::new();
2679 if let Some(a) = config.get("agents").and_then(Value::as_object) {
2680 for (k, v) in a {
2681 if k != "list" && k != "entries" {
2682 agents_rest.insert(k.clone(), v.clone());
2683 }
2684 }
2685 }
2686 let hooks = decode_hooks(&config, &mut vault, &name_for, &mut profiles);
2687 {
2688 let root = profiles.get_mut("default").unwrap();
2689 root.channels = channels;
2690 root.routes = routes;
2691 root.residue.config.insert("openclaw".into(), serde_json::json!({
2692 "default_agent": default_id,
2693 "agents_form": agents_form(&config),
2694 "agents_rest": redact_secrets(&Value::Object(agents_rest), "agents", &mut vault),
2695 "hooks": hooks.as_ref().map(|h| serde_json::json!({"block": h.block, "has_token": h.has_token})),
2696 "rest": redact_secrets(&Value::Object(rest), "config", &mut vault),
2697 }));
2698 }
2699 let mut root_io = OpenclawRootIo {
2700 state_dir: state_dir.to_path_buf(),
2701 config_raw: config_text.clone().unwrap_or_default(),
2702 config_present: config_text.is_some(),
2703 default_agent: default_id.clone(),
2704 ..Default::default()
2705 };
2706
2707 let db_path = state_dir.join(OPENCLAW_STATE_DB);
2709 let mut store_key = state_dir.join("cron/jobs.json").display().to_string();
2710 let mut job_owner: BTreeMap<String, String> = BTreeMap::new();
2711 if table_exists(&db_path, "cron_jobs") {
2712 for row in read_rows(
2713 &db_path,
2714 "select * from cron_jobs order by sort_order, created_at_ms, job_id",
2715 &[],
2716 )?
2717 .unwrap_or_default()
2718 {
2719 let job_id = opt_text(row.get("job_id")).unwrap_or_default();
2720 if let Some(k) = row
2721 .get("store_key")
2722 .and_then(Value::as_str)
2723 .filter(|s| !s.is_empty())
2724 {
2725 store_key = k.into();
2726 }
2727 let record = json_object_of(row.get("job_json"));
2728 let agent_id = record
2729 .get("agentId")
2730 .or_else(|| row.get("agent_id"))
2731 .or_else(|| row.get("owner_agent_id"))
2732 .and_then(Value::as_str)
2733 .unwrap_or(&default_id)
2734 .to_string();
2735 let owner = if profiles.contains_key(&name_for(&agent_id)) {
2736 name_for(&agent_id)
2737 } else {
2738 "default".into()
2739 };
2740 let job = decode_job(&row);
2741 root_io.cron_jobs.insert(job_id.clone(), row);
2742 profiles
2743 .get_mut(&owner)
2744 .unwrap()
2745 .jobs
2746 .insert(job.id.clone(), job);
2747 job_owner.insert(job_id, owner);
2748 }
2749 }
2750 if table_exists(&db_path, "cron_run_logs") {
2751 for row in read_rows(
2752 &db_path,
2753 "select * from cron_run_logs order by ts, seq",
2754 &[],
2755 )?
2756 .unwrap_or_default()
2757 {
2758 let fire = decode_fire(&row);
2759 root_io.cron_run_logs.insert(fire.id.clone(), row);
2760 let owner = job_owner
2761 .get(&fire.job_id)
2762 .cloned()
2763 .unwrap_or_else(|| "default".into());
2764 profiles.get_mut(&owner).unwrap().fires.push(fire);
2765 }
2766 }
2767 if table_exists(&db_path, "delivery_queue_entries") {
2768 for row in read_rows(
2769 &db_path,
2770 "select * from delivery_queue_entries order by enqueued_at, id",
2771 &[],
2772 )?
2773 .unwrap_or_default()
2774 {
2775 let o = decode_obligation(&row);
2776 root_io.delivery_queue_entries.insert(o.id.clone(), row);
2777 let owner = o
2778 .session_key
2779 .as_deref()
2780 .and_then(parse_openclaw_session_key)
2781 .and_then(|p| p.agent)
2782 .map(|a| name_for(&a))
2783 .filter(|n| profiles.contains_key(n))
2784 .unwrap_or_else(|| "default".into());
2785 profiles.get_mut(&owner).unwrap().obligations.push(o);
2786 }
2787 }
2788 if table_exists(&db_path, "schema_meta") {
2789 root_io.schema_meta =
2790 read_rows(&db_path, "select * from schema_meta order by meta_key", &[])?
2791 .unwrap_or_default();
2792 }
2793 root_io.store_key = store_key;
2794 root_io.db_present = db_path.exists();
2795
2796 let mut fire_by_key: BTreeMap<String, (String, String, Option<String>)> = BTreeMap::new();
2799 for (name, profile) in &profiles {
2800 for fire in &profile.fires {
2801 let Some(key) = fire.residue.0.get("session_key").and_then(Value::as_str) else {
2802 continue;
2803 };
2804 let finished = fire.finished_at.clone();
2805 match fire_by_key.get(key) {
2806 Some((_, _, held))
2807 if held.clone().unwrap_or_default() > finished.clone().unwrap_or_default() => {}
2808 _ => {
2809 fire_by_key.insert(key.into(), (name.clone(), fire.id.clone(), finished));
2810 }
2811 }
2812 }
2813 }
2814 let mut links: Vec<(String, String, String)> = Vec::new(); for profile in profiles.values_mut() {
2816 for o in &mut profile.obligations {
2817 let Some((owner, fire_id, _)) =
2818 o.session_key.as_deref().and_then(|k| fire_by_key.get(k))
2819 else {
2820 continue;
2821 };
2822 o.source = ObligationSource::Fire {
2823 fire_id: fire_id.clone(),
2824 };
2825 links.push((owner.clone(), fire_id.clone(), o.id.clone()));
2826 }
2827 }
2828 for (owner, fire_id, obligation_id) in links {
2829 if let Some(fire) = profiles
2830 .get_mut(&owner)
2831 .and_then(|p| p.fires.iter_mut().find(|f| f.id == fire_id))
2832 {
2833 fire.obligation_id = Some(obligation_id);
2834 }
2835 }
2836
2837 let mut orchestration = Orchestration {
2838 root: state_dir.to_path_buf(),
2839 profiles,
2840 };
2841 root_io.config_snapshot = canonical_json(&config_record(&orchestration));
2842 for (name, profile) in orchestration.profiles.iter_mut() {
2843 let io = ios.get_mut(name).unwrap();
2844 io.store_snapshot = canonical_json(&store_record(profile));
2845 io.bindings_snapshot = canonical_json(&serde_json::to_value(&profile.bindings).unwrap());
2846 if name != "default" {
2847 profile.residue.files = Vec::new();
2848 }
2849 }
2850 orchestration
2851 .profiles
2852 .get_mut("default")
2853 .unwrap()
2854 .residue
2855 .files = list_unmodeled(state_dir)?;
2856 Ok(OpenclawLoaded {
2857 orchestration,
2858 vault,
2859 root: root_io,
2860 profiles: ios,
2861 })
2862}
2863
2864#[derive(Debug, Clone, Default)]
2868pub struct OpenclawReport {
2869 pub written: Vec<ArtifactFidelity>,
2871 pub refused: Vec<super::hermes::Refusal>,
2873 pub notes: Vec<String>,
2875 pub rows_byte: usize,
2877 pub rows_emitted: usize,
2879}
2880
2881fn write_atomic(path: &Path, text: &str) -> Result<()> {
2882 if let Some(parent) = path.parent() {
2883 fs::create_dir_all(parent)?;
2884 }
2885 let tmp = path.with_file_name(format!(
2886 "{}.tmp-{}",
2887 path.file_name().unwrap().to_string_lossy(),
2888 std::process::id()
2889 ));
2890 fs::write(&tmp, text)?;
2891 fs::rename(&tmp, path)?;
2892 Ok(())
2893}
2894
2895fn encode_config(loaded: &OpenclawLoaded) -> Value {
2896 let orchestration = &loaded.orchestration;
2897 let root = &orchestration.profiles["default"];
2898 let own = root
2899 .residue
2900 .config
2901 .get("openclaw")
2902 .and_then(Value::as_object)
2903 .cloned()
2904 .unwrap_or_default();
2905 let default_id = own
2906 .get("default_agent")
2907 .and_then(Value::as_str)
2908 .map(str::to_string)
2909 .unwrap_or_else(|| loaded.root.default_agent.clone());
2910 let id_for_name = |name: &str| -> String {
2911 if name == "default" {
2912 default_id.clone()
2913 } else {
2914 loaded
2915 .profiles
2916 .get(name)
2917 .map(|io| io.agent_id.clone())
2918 .unwrap_or_else(|| name.into())
2919 }
2920 };
2921 let mut pairs: Vec<(String, Value)> = inline_secrets(
2922 own.get("rest").unwrap_or(&Value::Object(Map::new())),
2923 &loaded.vault,
2924 )
2925 .as_object()
2926 .map(|m| m.iter().map(|(k, v)| (k.clone(), v.clone())).collect())
2927 .unwrap_or_default();
2928 pairs.push((
2929 "channels".into(),
2930 ordered_object(encode_channels(root, &loaded.vault)),
2931 ));
2932 let agent_blocks: Vec<(String, Vec<(String, Value)>)> = orchestration
2933 .profiles
2934 .iter()
2935 .map(|(name, p)| {
2936 let id = id_for_name(name);
2937 let entry = inline_secrets(
2938 p.residue
2939 .config
2940 .get("openclaw_agent")
2941 .unwrap_or(&Value::Object(Map::new())),
2942 &loaded.vault,
2943 );
2944 let mut block: Vec<(String, Value)> = vec![("id".into(), Value::String(id.clone()))];
2945 if let Some(m) = entry.as_object() {
2946 for (k, v) in m {
2947 if k != "id" {
2948 block.push((k.clone(), v.clone()));
2949 }
2950 }
2951 }
2952 (id, block)
2953 })
2954 .collect();
2955 let mut agents: Vec<(String, Value)> = inline_secrets(
2956 own.get("agents_rest").unwrap_or(&Value::Object(Map::new())),
2957 &loaded.vault,
2958 )
2959 .as_object()
2960 .map(|m| m.iter().map(|(k, v)| (k.clone(), v.clone())).collect())
2961 .unwrap_or_default();
2962 if own.get("agents_form").and_then(Value::as_str) == Some("entries") {
2963 agents.push((
2964 "entries".into(),
2965 ordered_object(
2966 agent_blocks
2967 .into_iter()
2968 .map(|(id, block)| {
2969 (
2970 id,
2971 ordered_object(block.into_iter().filter(|(k, _)| k != "id").collect()),
2972 )
2973 })
2974 .collect(),
2975 ),
2976 ));
2977 } else {
2978 agents.push((
2979 "list".into(),
2980 Value::Array(
2981 agent_blocks
2982 .into_iter()
2983 .map(|(_, block)| ordered_object(block))
2984 .collect(),
2985 ),
2986 ));
2987 }
2988 pairs.push(("agents".into(), ordered_object(agents)));
2989 pairs.push((
2990 "bindings".into(),
2991 Value::Array(encode_routes(root, &id_for_name)),
2992 ));
2993 let hooks_meta = own
2994 .get("hooks")
2995 .and_then(Value::as_object)
2996 .map(|h| HooksMeta {
2997 block: h
2998 .get("block")
2999 .and_then(Value::as_object)
3000 .cloned()
3001 .unwrap_or_default(),
3002 has_token: h.get("has_token").and_then(Value::as_bool).unwrap_or(false),
3003 });
3004 if let Some(hooks) = encode_hooks(orchestration, hooks_meta.as_ref(), &loaded.vault) {
3005 pairs.push(("hooks".into(), hooks));
3006 }
3007 ordered_object(pairs)
3008}
3009
3010fn plain(value: &Value) -> Value {
3012 if let Some(pairs) = is_ordered_pairs(value) {
3013 return Value::Object(pairs.into_iter().map(|(k, v)| (k, plain(&v))).collect());
3014 }
3015 match value {
3016 Value::Array(items) => Value::Array(items.iter().map(plain).collect()),
3017 Value::Object(m) => Value::Object(m.iter().map(|(k, v)| (k.clone(), plain(v))).collect()),
3018 other => other.clone(),
3019 }
3020}
3021
3022fn resequence(rows: &mut [Map<String, Value>]) {
3023 let key = |r: &Map<String, Value>| {
3024 format!(
3025 "{}\u{0}{}",
3026 r.get("store_key")
3027 .map(|v| v.to_string())
3028 .unwrap_or_default(),
3029 r.get("job_id").map(|v| v.to_string()).unwrap_or_default()
3030 )
3031 };
3032 let mut taken: BTreeMap<String, Vec<i64>> = BTreeMap::new();
3033 for row in rows.iter_mut() {
3034 let Some(seq) = row.get("seq").and_then(Value::as_i64) else {
3035 continue;
3036 };
3037 let seen = taken.entry(key(row)).or_default();
3038 if seen.contains(&seq) {
3039 row.insert("seq".into(), Value::Null);
3040 continue;
3041 }
3042 seen.push(seq);
3043 }
3044 for row in rows.iter_mut() {
3045 if row.get("seq").is_some_and(|v| !v.is_null()) {
3046 continue;
3047 }
3048 let seen = taken.entry(key(row)).or_default();
3049 let mut next = 1;
3050 while seen.contains(&next) {
3051 next += 1;
3052 }
3053 row.insert("seq".into(), Value::from(next));
3054 seen.push(next);
3055 }
3056}
3057
3058fn insert_for(table: &str, columns: &[&str]) -> String {
3059 format!(
3060 "insert into {table} ({}) values ({})",
3061 columns.join(", "),
3062 columns.iter().map(|_| "?").collect::<Vec<_>>().join(",")
3063 )
3064}
3065
3066fn params_of(row: &Map<String, Value>, columns: &[&str]) -> Vec<Param> {
3067 columns
3068 .iter()
3069 .map(|c| Param::from(row.get(*c).unwrap_or(&Value::Null)))
3070 .collect()
3071}
3072
3073fn write_store(loaded: &OpenclawLoaded, dest: &Path, report: &mut OpenclawReport) -> Result<()> {
3074 let store_key = if loaded.root.store_key.is_empty() {
3075 dest.join("cron/jobs.json").display().to_string()
3076 } else {
3077 loaded.root.store_key.clone()
3078 };
3079 let mut job_rows = Vec::new();
3080 let mut fire_rows = Vec::new();
3081 let mut obligation_rows = Vec::new();
3082 let (mut byte_rows, mut emitted) = (0usize, 0usize);
3083 for profile in loaded.orchestration.profiles.values() {
3084 for job in profile.jobs.values() {
3085 let original = loaded.root.cron_jobs.get(&job.id);
3086 let unchanged = original.is_some_and(|o| {
3087 canonical_json(&serde_json::to_value(decode_job(o)).unwrap())
3088 == canonical_json(&serde_json::to_value(job).unwrap())
3089 });
3090 if unchanged {
3091 job_rows.push(original.unwrap().clone());
3092 byte_rows += 1;
3093 } else {
3094 job_rows.push(encode_job_row(job, original, &store_key));
3095 emitted += 1;
3096 }
3097 }
3098 for fire in &profile.fires {
3099 let original = loaded.root.cron_run_logs.get(&fire.id);
3100 let unchanged = original.is_some_and(|o| {
3101 let mut d = decode_fire(o);
3102 d.obligation_id = fire.obligation_id.clone();
3103 canonical_json(&serde_json::to_value(d).unwrap())
3104 == canonical_json(&serde_json::to_value(fire).unwrap())
3105 });
3106 if unchanged {
3107 fire_rows.push(original.unwrap().clone());
3108 byte_rows += 1;
3109 } else {
3110 fire_rows.push(encode_fire_row(fire, original, &store_key));
3111 emitted += 1;
3112 }
3113 }
3114 resequence(&mut fire_rows);
3115 for o in &profile.obligations {
3116 let original = loaded.root.delivery_queue_entries.get(&o.id);
3117 let unchanged = original.is_some_and(|r| {
3118 canonical_json(&serde_json::to_value(decode_obligation(r)).unwrap())
3119 == canonical_json(&serde_json::to_value(o).unwrap())
3120 });
3121 if unchanged {
3122 obligation_rows.push(original.unwrap().clone());
3123 byte_rows += 1;
3124 } else {
3125 obligation_rows.push(encode_obligation_row(o, original));
3126 emitted += 1;
3127 }
3128 }
3129 }
3130 let target = dest.join(OPENCLAW_STATE_DB);
3131 fs::create_dir_all(dest.join("state"))?;
3132 let tmp = target.with_file_name(format!("openclaw.sqlite.tmp-{}", std::process::id()));
3133 let _ = fs::remove_file(&tmp);
3134 let meta: Vec<Map<String, Value>> = if loaded.root.schema_meta.is_empty() {
3135 vec![serde_json::from_value(serde_json::json!({"meta_key": "global", "role": "global", "schema_version": 1, "agent_id": null, "app_version": null, "created_at": 0, "updated_at": 0})).unwrap()]
3136 } else {
3137 loaded.root.schema_meta.clone()
3138 };
3139 write_table(
3140 &tmp,
3141 OPENCLAW_DDL,
3142 &insert_for("schema_meta", SCHEMA_META_COLUMNS),
3143 &meta
3144 .iter()
3145 .map(|r| params_of(r, SCHEMA_META_COLUMNS))
3146 .collect::<Vec<_>>(),
3147 )?;
3148 write_table(
3149 &tmp,
3150 "",
3151 &insert_for("cron_jobs", CRON_JOB_COLUMNS),
3152 &job_rows
3153 .iter()
3154 .map(|r| params_of(r, CRON_JOB_COLUMNS))
3155 .collect::<Vec<_>>(),
3156 )?;
3157 write_table(
3158 &tmp,
3159 "",
3160 &insert_for("cron_run_logs", CRON_RUN_LOG_COLUMNS),
3161 &fire_rows
3162 .iter()
3163 .map(|r| params_of(r, CRON_RUN_LOG_COLUMNS))
3164 .collect::<Vec<_>>(),
3165 )?;
3166 write_table(
3167 &tmp,
3168 "",
3169 &insert_for("delivery_queue_entries", DELIVERY_QUEUE_COLUMNS),
3170 &obligation_rows
3171 .iter()
3172 .map(|r| params_of(r, DELIVERY_QUEUE_COLUMNS))
3173 .collect::<Vec<_>>(),
3174 )?;
3175 fs::rename(&tmp, &target)?;
3176 report.written.push(ArtifactFidelity {
3177 path: OPENCLAW_STATE_DB.into(),
3178 fidelity: if emitted == 0 {
3179 Fidelity::ByteLossless
3180 } else {
3181 Fidelity::Semantic
3182 },
3183 loss: Vec::new(),
3184 });
3185 report.rows_byte = byte_rows;
3186 report.rows_emitted = emitted;
3187 Ok(())
3188}
3189
3190fn copy_unmodeled(
3191 files: &[String],
3192 src: &Path,
3193 into: &Path,
3194 report: &mut OpenclawReport,
3195 prefix: &str,
3196) -> Result<()> {
3197 for rel in files {
3198 let from = src.join(rel);
3199 if !from.exists() {
3200 continue;
3201 }
3202 let to = into.join(rel);
3203 if let Some(parent) = to.parent() {
3204 fs::create_dir_all(parent)?;
3205 }
3206 fs::copy(&from, &to)?;
3207 report
3208 .written
3209 .push(ArtifactFidelity::byte(format!("{prefix}{rel}")));
3210 }
3211 Ok(())
3212}
3213
3214pub fn to_openclaw(loaded: &OpenclawLoaded, dest: &Path) -> Result<OpenclawReport> {
3216 let mut report = OpenclawReport::default();
3217 let orchestration = &loaded.orchestration;
3218 if !orchestration.profiles.contains_key("default") {
3219 return Err(load_error(
3220 &dest.display().to_string(),
3221 "",
3222 "no `default` profile: an OpenClaw install always has a default agent",
3223 ));
3224 }
3225 fs::create_dir_all(dest)?;
3226
3227 let cfg_unchanged =
3229 loaded.root.config_snapshot == canonical_json(&config_record(orchestration));
3230 if cfg_unchanged && loaded.root.config_present {
3231 write_atomic(&dest.join(OPENCLAW_CONFIG), &loaded.root.config_raw)?;
3232 report.written.push(ArtifactFidelity::byte(OPENCLAW_CONFIG));
3233 } else {
3234 let encoded = encode_config(loaded);
3235 let missing = unresolved_refs(&plain(&encoded), "");
3236 if !missing.is_empty() {
3237 report.refused.push(super::hermes::Refusal { file: OPENCLAW_CONFIG.into(), reason: format!("the vault has no value for {}; openclaw would read the reference itself as the credential", missing.iter().map(|(p, r)| format!("{r} ({p})")).collect::<Vec<_>>().join(", ")) });
3238 } else {
3239 write_atomic(
3240 &dest.join(OPENCLAW_CONFIG),
3241 &format!("{}\n", pretty_ordered(&encoded, 0)),
3242 )?;
3243 report.written.push(ArtifactFidelity::semantic(OPENCLAW_CONFIG, vec!["re-emitted as JSON: openclaw reads it with a JSON5 parser, so it loads, but the source's comments, trailing commas and key order are gone".into()]));
3244 report.notes.push(format!("{OPENCLAW_CONFIG}: re-emitted as JSON; comments, trailing commas and key order are gone"));
3245 }
3246 }
3247
3248 let store_unchanged = orchestration.profiles.iter().all(|(n, p)| {
3250 loaded
3251 .profiles
3252 .get(n)
3253 .map(|io| io.store_snapshot == canonical_json(&store_record(p)))
3254 .unwrap_or(false)
3255 });
3256 let src_db = loaded.root.state_dir.join(OPENCLAW_STATE_DB);
3257 let has_rows = orchestration
3258 .profiles
3259 .values()
3260 .any(|p| !p.jobs.is_empty() || !p.fires.is_empty() || !p.obligations.is_empty());
3261 if store_unchanged && loaded.root.db_present && src_db.exists() {
3262 fs::create_dir_all(dest.join("state"))?;
3263 fs::copy(&src_db, dest.join(OPENCLAW_STATE_DB))?;
3264 report
3265 .written
3266 .push(ArtifactFidelity::byte(OPENCLAW_STATE_DB));
3267 } else if has_rows || loaded.root.db_present {
3268 write_store(loaded, dest, &mut report)?;
3269 }
3270
3271 for (name, profile) in &orchestration.profiles {
3273 if profile.bindings.is_empty() {
3274 continue;
3275 }
3276 let io = loaded.profiles.get(name);
3277 let snapshot = io
3278 .map(|io| io.bindings_snapshot.clone())
3279 .filter(|s| !s.is_empty());
3280 if snapshot.as_deref()
3281 == Some(canonical_json(&serde_json::to_value(&profile.bindings).unwrap()).as_str())
3282 {
3283 continue;
3284 }
3285 let agent = io
3286 .map(|io| io.agent_id.clone())
3287 .unwrap_or_else(|| name.clone());
3288 if snapshot.is_some() {
3289 report.refused.push(super::hermes::Refusal { file: format!("agents/{agent}/sessions/*.jsonl"), reason: "bindings changed since import; writing the change into an OpenClaw transcript is UNI-18 (the first write into another harness's live session store), not this codec's".into() });
3292 continue;
3293 }
3294 let sessions_dir = dest.join("agents").join(&agent).join("sessions");
3300 for (slot, b) in &profile.bindings {
3301 let id = b.worker.session_id.clone().unwrap_or_else(|| slot.clone());
3302 let file = sessions_dir.join(format!("{id}.jsonl"));
3303 let rel = format!("agents/{agent}/sessions/{id}.jsonl");
3304 if file.exists() {
3305 report.refused.push(super::hermes::Refusal { file: rel, reason: "the destination already holds this transcript; writing into a live OpenClaw session store is UNI-18, not this codec's".into() });
3306 continue;
3307 }
3308 fs::create_dir_all(&sessions_dir)?;
3309 let header = serde_json::json!({
3310 "type": "session",
3311 "version": 3,
3312 "id": id,
3313 "timestamp": b.started_at.clone().or_else(|| b.last_activity_at.clone()).unwrap_or_default(),
3314 "sessionKey": crate::ontology::render_openclaw_session_key(&agent, b),
3315 });
3316 fs::write(&file, format!("{header}\n"))?;
3317 report.written.push(ArtifactFidelity::semantic(rel, vec!["a fresh transcript header carrying the session key; the turns live in the worker's store and are not carried".into()]));
3318 }
3319 }
3320
3321 copy_unmodeled(
3323 &orchestration.profiles["default"].residue.files,
3324 &loaded.root.state_dir,
3325 dest,
3326 &mut report,
3327 "",
3328 )?;
3329 for (name, profile) in &orchestration.profiles {
3330 if name == "default" {
3331 continue;
3332 }
3333 let Some(io) = loaded.profiles.get(name) else {
3334 continue;
3335 };
3336 copy_unmodeled(
3337 &profile.residue.files,
3338 &io.source_dir,
3339 &dest.join("agents").join(&io.agent_id),
3340 &mut report,
3341 &format!("agents/{}/", io.agent_id),
3342 )?;
3343 }
3344 Ok(report)
3345}