Skip to main content

supercode_interchange/world/codec/
openclaw.rs

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