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