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