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