Skip to main content

supercode_interchange/orchestration/codec/
openclaw.rs

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