Skip to main content

supercode_interchange/orchestration/codec/
openclaw.rs

1//! The OpenClaw codec (`openclaw.mjs`): an OpenClaw state directory ⇄ the
2//! orchestration. `agents.*` become profiles — the default agent IS the `default`
3//! profile, every other agent keeps its id — each rooted at its own
4//! `agents/<id>/`. Channels, bindings and hooks are install-wide and land on
5//! `default`; jobs, fires and obligations land on the profile their agent id
6//! or session key names. `openclaw.json` is JSON5 (comments and trailing
7//! commas survive the read; a re-emit is JSON, which OpenClaw's parser
8//! loads); `state/openclaw.sqlite` is the pinned v2026.7.1-2 schema, its
9//! rows written back column for column when their record is unchanged.
10
11use 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
32/// The config file.
33pub const OPENCLAW_CONFIG: &str = "openclaw.json";
34/// The shared state store.
35pub const OPENCLAW_STATE_DB: &str = "state/openclaw.sqlite";
36/// Keys an inline account id may sit under.
37pub const OPENCLAW_ACCOUNT_KEYS: &[&str] = &[
38    "accountId",
39    "account_id",
40    "account",
41    "teamId",
42    "appId",
43    "userId",
44];
45
46/// The pinned v2026.7.1-2 schema, transcribed column for column.
47pub 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"#;
207/// `cron_jobs` columns in order.
208pub 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];
285/// `cron_run_logs` columns in order.
286pub 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];
310/// `delivery_queue_entries` columns in order.
311pub 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];
330/// `schema_meta` columns in order.
331pub 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
341// ---------------------------------------------------------------- JSON5
342
343/// Reduce JSON5 to JSON: `//` and `/* */` comments outside strings, trailing
344/// commas before `}` / `]` (`profiles.rs::strip_json5`, which the Node codec
345/// transcribed; this is the same rule in the crate that owns the codec).
346pub 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
431/// Parse a JSON5 file's text.
432pub 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
437// ---------------------------------------------------------------- secrets
438
439/// Credential-SHAPED names OpenClaw uses for things that are not secrets.
440pub 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
455/// OpenClaw's own credential rule: the lower-cased key ENDS WITH a marker; only the name decides.
456pub 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
466/// The `.env` name for a credential: `OPENCLAW_<scope>_<key>`.
467pub 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
480/// Deep copy with every string under a credential-shaped key replaced by a `{dotenv}` ref, the value handed to the vault.
481pub 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
524/// Inverse of [`redact_secrets`]: `${NAME}` references become their values.
525pub 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
543/// Every `${NAME}` reference a write left unresolved, with its path.
544fn 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
571// ---------------------------------------------------------------- time
572
573fn 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
599/// Unix ms → RFC 3339 at SECOND precision (`jobs.rs::iso_from_ms`).
600pub 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
607/// Unix ms → RFC 3339 with millis (`sidecar::ms_to_rfc3339`).
608pub 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
615/// RFC 3339 (`YYYY-MM-DDTHH:MM:SS[.fff][Z|±HH:MM]`) → Unix ms; `None` for anything else.
616pub 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    // days from civil (Howard Hinnant)
655    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
676// ---------------------------------------------------------------- agents
677
678/// The declared agents as `[id, entry]` in declaration order.
679pub 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
705/// `list` | `entries` | none.
706pub 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
717/// The default agent: the entry flagged `default: true`, else the first declared, else `main`.
718pub 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// ---------------------------------------------------------------- session keys
729
730/// What an OpenClaw session key mapped to, in the orchestration's binding form.
731#[derive(Debug, Clone, PartialEq)]
732pub struct ParsedKey {
733    /// The agent the key names, if any.
734    pub agent: Option<String>,
735    /// The surface (a `main` key is the DM COLLAPSE: `{main, dm, main}`).
736    pub key: SurfaceKey,
737    /// What the mapping had to translate (`session_key`, `dm_collapse`, `chat_kind`, `thread_word`).
738    pub residue: Map<String, Value>,
739    /// The job a `cron:` key names.
740    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
756/// An OpenClaw gateway session key → the orchestration's surface key plus what the
757/// mapping had to translate (`openclaw.mjs::parseOpenclawSessionKey`). This is
758/// the WORLD's form; the spine's discovery nouns keep the key's own shape
759/// through `Binding::from_openclaw_key`.
760pub 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// ---------------------------------------------------------------- io
839
840/// Root bookkeeping for an OpenClaw state dir.
841#[derive(Debug, Clone, Default)]
842pub struct OpenclawRootIo {
843    /// The state dir.
844    pub state_dir: PathBuf,
845    /// `openclaw.json`'s bytes as read (empty when absent).
846    pub config_raw: String,
847    /// Whether `openclaw.json` existed.
848    pub config_present: bool,
849    /// Canonical JSON of the config record as loaded.
850    pub config_snapshot: String,
851    /// Store rows as read, by table and id.
852    pub cron_jobs: BTreeMap<String, Map<String, Value>>,
853    /// Run-log rows by fire id.
854    pub cron_run_logs: BTreeMap<String, Map<String, Value>>,
855    /// Queue rows by id.
856    pub delivery_queue_entries: BTreeMap<String, Map<String, Value>>,
857    /// `schema_meta` rows.
858    pub schema_meta: Vec<Map<String, Value>>,
859    /// The `store_key` the jobs carry.
860    pub store_key: String,
861    /// Whether the store existed.
862    pub db_present: bool,
863    /// The default agent id.
864    pub default_agent: String,
865    /// The legacy `cron/jobs.json` beside the store, byte for byte, when
866    /// one exists (an install the pin has not migrated yet).
867    pub legacy_jobs_raw: Option<String>,
868    /// Its records by id, as read.
869    pub legacy_jobs: BTreeMap<String, Value>,
870    /// Whether the file was `{"jobs": [...]}` rather than a bare array.
871    pub legacy_jobs_object_form: bool,
872}
873
874/// Per-profile bookkeeping.
875#[derive(Debug, Clone, Default)]
876pub struct OpenclawProfileIo {
877    /// The agent id.
878    pub agent_id: String,
879    /// `agents/<id>/`.
880    pub source_dir: PathBuf,
881    /// Canonical JSON of the store record (jobs, fires, obligations) as loaded.
882    pub store_snapshot: String,
883    /// Canonical JSON of the bindings as loaded.
884    pub bindings_snapshot: String,
885}
886
887/// A loaded OpenClaw state directory.
888#[derive(Debug, Clone)]
889pub struct OpenclawLoaded {
890    /// The orchestration.
891    pub orchestration: Orchestration,
892    /// Secret values by `.env` name; never in the orchestration.
893    pub vault: BTreeMap<String, String>,
894    /// Root bookkeeping.
895    pub root: OpenclawRootIo,
896    /// Per-profile bookkeeping by profile name.
897    pub profiles: BTreeMap<String, OpenclawProfileIo>,
898}
899
900impl OpenclawLoaded {
901    /// An orchestration that did not come from an OpenClaw store (our folder on its way
902    /// out): nothing to reuse, every artifact re-emitted, every binding refused.
903    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
945// ---------------------------------------------------------------- channels
946
947/// `channels.<name>.accounts` as `[accountId, entry]`, sorted by id.
948pub 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
1287// ---------------------------------------------------------------- routes
1288
1289const 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// ---------------------------------------------------------------- hooks
1410
1411/// What the `hooks` block carried beside its mappings.
1412#[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
1565// ---------------------------------------------------------------- jobs
1566
1567fn 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
1667/// `cron_jobs` row → Job (`openclaw.mjs::decodeJob`).
1668pub 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    // the store's writer keeps delivery in `job_json` AND in columns; a row
1719    // that has it only in the columns still has to re-emit as a delivery, so
1720    // the columns' object is what the residue keeps when `job_json` has none
1721    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
1859/// A job as a `cron_jobs` row, over the original row when one exists.
1860pub 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
2099// ---------------------------------------------------------------- fires
2100
2101fn 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
2127/// `cron_run_logs` row → Fire.
2128pub 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
2162/// A fire as a `cron_run_logs` row, over the original when one exists.
2163pub 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
2261// ---------------------------------------------------------------- obligations
2262
2263fn 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
2285/// `delivery_queue_entries` row → Obligation.
2286pub 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
2377/// An obligation as a `delivery_queue_entries` row, over the original when one exists.
2378pub 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
2453// ---------------------------------------------------------------- bindings
2454
2455/// One `current_conversation_bindings` row as a [`Binding`]: the conversation
2456/// ref is the surface (a thread's parent is the chat, the thread is the
2457/// thread), the target session is the worker, `bound_at` is when it started,
2458/// and a binding that is not `active` ended when the row last changed. The
2459/// row's own facts (binding id and key, account, target kind, status, expiry,
2460/// metadata, the full record) ride as residue.
2461fn 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
2651// ---------------------------------------------------------------- from
2652
2653fn 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
2661/// The rows the sqlite store holds for a profile; a legacy file's jobs are
2662/// the file's rows, not the store's.
2663fn 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
2672/// Compile an OpenClaw state directory into the orchestration.
2673pub 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    // install-wide config: channels, bindings, and everything else verbatim
2741    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    // the shared state DB
2779    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    // the legacy file beside the store: jobs the pin's migration has not
2822    // moved yet, read by the vocabulary that file was written in; a store
2823    // row with the same id wins
2824    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    // the store's own binding record (the pin's `current_conversation_bindings`,
2896    // ORC-12 finding 7): which agent and session a conversation is bound to,
2897    // observed rather than inferred from transcript headers. A row replaces
2898    // the inferred binding on the same surface; a store without the table
2899    // keeps the inferred ones. Read, never written back (UNI-22).
2900    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    // a fire's delivery OUTCOME rides on its own run-log row; the queue entry on
2934    // the same conversation key is the obligation that outcome is about
2935    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(); // (profile, fire id, obligation id)
2952    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// ---------------------------------------------------------------- to
3002
3003/// What an OpenClaw decompile did.
3004#[derive(Debug, Clone, Default)]
3005pub struct OpenclawReport {
3006    /// Every artifact written, with its tier.
3007    pub written: Vec<ArtifactFidelity>,
3008    /// Every write refused, with the gate named.
3009    pub refused: Vec<super::hermes::Refusal>,
3010    /// What a semantic write gave up.
3011    pub notes: Vec<String>,
3012    /// Store rows written back column for column vs re-encoded.
3013    pub rows_byte: usize,
3014    /// Store rows re-encoded.
3015    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
3147/// Flatten an ordered value back into plain JSON (for ref scans).
3148fn 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
3210// ---------------------------------------------------------------- legacy cron/jobs.json
3211
3212/// The records of a legacy `cron/jobs.json` (a bare array, or `{"jobs": [...]}`).
3213fn 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
3239/// A legacy file record as a `cron_jobs` row, in the vocabulary the pin's
3240/// migration reads: a bare cron string / `cron` / `everyMinutes` / `everyMs`
3241/// / `runAt` schedule becomes the typed `schedule` object, `delivery.kind`
3242/// is the mode, and the ISO state stamps become the row's millisecond
3243/// columns. The record is then one [`decode_job`], marked `__legacy_file`.
3244fn 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    // the column form: the store keeps the record as JSON text
3338    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
3347/// The legacy `cron/jobs.json` on the way out: the source bytes when every
3348/// file job is as it was read, else the file re-emitted from the jobs
3349/// (`job_json` records, in the file's own form).
3350fn 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
3568/// Write an OpenClaw state directory from the orchestration.
3569pub 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    // openclaw.json
3582    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    // state/openclaw.sqlite
3603    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    // bindings: read, never written back (UNI-22)
3627    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            // the source transcripts hold what the orchestration does not model; the
3645            // change has to be written INTO them, which is UNI-18's
3646            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        // our own bindings: an OpenClaw conversation's surface lives in the
3650        // session KEY inside the transcript header, so each binding becomes a
3651        // FRESH transcript whose header carries that key (the discovery door
3652        // reads `sessionKey`); the turns live in the worker's store and are
3653        // not carried. An existing transcript is never overwritten.
3654        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    // everything under the state dir the IR does not model
3677    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}