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