supercode_interchange/orchestration/codec/

openclaw.rs

//! The OpenClaw codec (`openclaw.mjs`): an OpenClaw state directory ⇄ the
//! orchestration. `agents.*` become profiles — the default agent IS the `default`
//! profile, every other agent keeps its id — each rooted at its own
//! `agents/<id>/`. Channels, bindings and hooks are install-wide and land on
//! `default`; jobs, fires and obligations land on the profile their agent id
//! or session key names. `openclaw.json` is JSON5 (comments and trailing
//! commas survive the read; a re-emit is JSON, which OpenClaw's parser
//! loads); `state/openclaw.sqlite` is the pinned v2026.7.1-2 schema, its
//! rows written back column for column when their record is unchanged.

use std::collections::BTreeMap;
use std::fs;
use std::path::{Path, PathBuf};

use serde_json::{Map, Value};

use super::canonical::canonical_json;
use super::decode::{load_error, placeholder_ref, surface_key_string};
use super::folder::{empty_profile, iso_epoch, ordered_object, persona_ref, pretty_ordered};
use super::sqlite::{read_rows, table_exists, write_table, Param};
use crate::ontology::{
    ArtifactFidelity, Binding, Fidelity, HarnessId, Recurrence, Residue, SecretRef, SurfaceKey,
    Trigger, Worker,
};
use crate::orchestration::{
    ChannelConfig, Fire, FireStatus, Job, JobOrigin, Obligation, ObligationSource, ObligationState,
    Orchestration, OutboundContent, Profile, Route, RouteMatch, Schedule, Target,
    WebhookSubscription,
};
use crate::Result;

/// The config file.
pub const OPENCLAW_CONFIG: &str = "openclaw.json";
/// The shared state store.
pub const OPENCLAW_STATE_DB: &str = "state/openclaw.sqlite";
/// Keys an inline account id may sit under.
pub const OPENCLAW_ACCOUNT_KEYS: &[&str] = &[
    "accountId",
    "account_id",
    "account",
    "teamId",
    "appId",
    "userId",
];

/// The pinned v2026.7.1-2 schema, transcribed column for column.
pub const OPENCLAW_DDL: &str = r#"CREATE TABLE IF NOT EXISTS schema_meta (
  meta_key TEXT NOT NULL PRIMARY KEY,
  role TEXT NOT NULL,
  schema_version INTEGER NOT NULL,
  agent_id TEXT,
  app_version TEXT,
  created_at INTEGER NOT NULL,
  updated_at INTEGER NOT NULL
);
CREATE TABLE IF NOT EXISTS cron_jobs (
  store_key TEXT NOT NULL,
  job_id TEXT NOT NULL,
  declaration_key TEXT,
  display_name TEXT,
  owner_agent_id TEXT,
  owner_session_key TEXT,
  name TEXT NOT NULL,
  description TEXT,
  enabled INTEGER NOT NULL,
  delete_after_run INTEGER,
  created_at_ms INTEGER NOT NULL,
  agent_id TEXT,
  session_key TEXT,
  schedule_kind TEXT NOT NULL,
  schedule_expr TEXT,
  schedule_tz TEXT,
  every_ms INTEGER,
  anchor_ms INTEGER,
  at TEXT,
  stagger_ms INTEGER,
  session_target TEXT NOT NULL,
  wake_mode TEXT NOT NULL,
  trigger_script TEXT,
  trigger_once INTEGER,
  payload_kind TEXT NOT NULL,
  payload_message TEXT,
  payload_model TEXT,
  payload_fallbacks_json TEXT,
  payload_thinking TEXT,
  payload_timeout_seconds INTEGER,
  payload_allow_unsafe_external_content INTEGER,
  payload_external_content_source_json TEXT,
  payload_light_context INTEGER,
  payload_tools_allow_json TEXT,
  payload_tools_allow_is_default INTEGER,
  delivery_mode TEXT,
  delivery_channel TEXT,
  delivery_to TEXT,
  delivery_thread_id TEXT,
  delivery_thread_id_type TEXT,
  delivery_account_id TEXT,
  delivery_best_effort INTEGER,
  delivery_completion_mode TEXT,
  delivery_completion_to TEXT,
  failure_delivery_mode TEXT,
  failure_delivery_channel TEXT,
  failure_delivery_to TEXT,
  failure_delivery_account_id TEXT,
  failure_alert_disabled INTEGER,
  failure_alert_after INTEGER,
  failure_alert_channel TEXT,
  failure_alert_to TEXT,
  failure_alert_cooldown_ms INTEGER,
  failure_alert_include_skipped INTEGER,
  failure_alert_mode TEXT,
  failure_alert_account_id TEXT,
  next_run_at_ms INTEGER,
  running_at_ms INTEGER,
  last_run_at_ms INTEGER,
  last_run_status TEXT,
  last_error TEXT,
  last_duration_ms INTEGER,
  consecutive_errors INTEGER,
  consecutive_skipped INTEGER,
  schedule_error_count INTEGER,
  last_delivery_status TEXT,
  last_delivery_error TEXT,
  last_delivered INTEGER,
  last_failure_alert_at_ms INTEGER,
  job_json TEXT NOT NULL,
  state_json TEXT NOT NULL DEFAULT '{}',
  runtime_updated_at_ms INTEGER,
  schedule_identity TEXT,
  sort_order INTEGER NOT NULL DEFAULT 0,
  updated_at INTEGER NOT NULL,
  PRIMARY KEY (store_key, job_id)
);
CREATE INDEX IF NOT EXISTS idx_cron_jobs_store_updated
  ON cron_jobs(store_key, sort_order ASC, updated_at DESC, job_id);
CREATE INDEX IF NOT EXISTS idx_cron_jobs_store_order
  ON cron_jobs(store_key, sort_order ASC, updated_at ASC, job_id);
CREATE INDEX IF NOT EXISTS idx_cron_jobs_enabled_next_run
  ON cron_jobs(store_key, enabled, next_run_at_ms, job_id)
  WHERE next_run_at_ms IS NOT NULL;
CREATE INDEX IF NOT EXISTS idx_cron_jobs_agent_session
  ON cron_jobs(agent_id, session_key, updated_at DESC, job_id)
  WHERE agent_id IS NOT NULL OR session_key IS NOT NULL;
CREATE TABLE IF NOT EXISTS cron_run_logs (
  store_key TEXT NOT NULL,
  job_id TEXT NOT NULL,
  seq INTEGER NOT NULL,
  ts INTEGER NOT NULL,
  status TEXT,
  error TEXT,
  summary TEXT,
  diagnostics_summary TEXT,
  delivery_status TEXT,
  delivery_error TEXT,
  delivered INTEGER,
  session_id TEXT,
  session_key TEXT,
  run_id TEXT,
  run_at_ms INTEGER,
  duration_ms INTEGER,
  next_run_at_ms INTEGER,
  model TEXT,
  provider TEXT,
  total_tokens INTEGER,
  entry_json TEXT NOT NULL,
  created_at INTEGER NOT NULL,
  PRIMARY KEY (store_key, job_id, seq)
);
CREATE INDEX IF NOT EXISTS idx_cron_run_logs_store_ts
  ON cron_run_logs(store_key, ts DESC, seq DESC);
CREATE INDEX IF NOT EXISTS idx_cron_run_logs_job_status
  ON cron_run_logs(store_key, job_id, status, ts DESC, seq DESC);
CREATE INDEX IF NOT EXISTS idx_cron_run_logs_delivery
  ON cron_run_logs(store_key, delivery_status, ts DESC, seq DESC)
  WHERE delivery_status IS NOT NULL;
CREATE TABLE IF NOT EXISTS delivery_queue_entries (
  queue_name TEXT NOT NULL,
  id TEXT NOT NULL,
  status TEXT NOT NULL,
  entry_kind TEXT,
  session_key TEXT,
  channel TEXT,
  target TEXT,
  account_id TEXT,
  retry_count INTEGER NOT NULL DEFAULT 0,
  last_attempt_at INTEGER,
  last_error TEXT,
  recovery_state TEXT,
  platform_send_started_at INTEGER,
  entry_json TEXT NOT NULL,
  enqueued_at INTEGER NOT NULL,
  updated_at INTEGER NOT NULL,
  failed_at INTEGER,
  PRIMARY KEY (queue_name, id)
);
CREATE INDEX IF NOT EXISTS idx_delivery_queue_pending
  ON delivery_queue_entries(queue_name, status, enqueued_at, id);
CREATE INDEX IF NOT EXISTS idx_delivery_queue_failed
  ON delivery_queue_entries(queue_name, status, failed_at, id);
CREATE INDEX IF NOT EXISTS idx_delivery_queue_session
  ON delivery_queue_entries(queue_name, status, session_key, enqueued_at, id)
  WHERE session_key IS NOT NULL;
CREATE INDEX IF NOT EXISTS idx_delivery_queue_target
  ON delivery_queue_entries(queue_name, status, channel, target, enqueued_at, id)
  WHERE channel IS NOT NULL AND target IS NOT NULL;
"#;
/// `cron_jobs` columns in order.
pub const CRON_JOB_COLUMNS: &[&str] = &[
    "store_key",
    "job_id",
    "declaration_key",
    "display_name",
    "owner_agent_id",
    "owner_session_key",
    "name",
    "description",
    "enabled",
    "delete_after_run",
    "created_at_ms",
    "agent_id",
    "session_key",
    "schedule_kind",
    "schedule_expr",
    "schedule_tz",
    "every_ms",
    "anchor_ms",
    "at",
    "stagger_ms",
    "session_target",
    "wake_mode",
    "trigger_script",
    "trigger_once",
    "payload_kind",
    "payload_message",
    "payload_model",
    "payload_fallbacks_json",
    "payload_thinking",
    "payload_timeout_seconds",
    "payload_allow_unsafe_external_content",
    "payload_external_content_source_json",
    "payload_light_context",
    "payload_tools_allow_json",
    "payload_tools_allow_is_default",
    "delivery_mode",
    "delivery_channel",
    "delivery_to",
    "delivery_thread_id",
    "delivery_thread_id_type",
    "delivery_account_id",
    "delivery_best_effort",
    "delivery_completion_mode",
    "delivery_completion_to",
    "failure_delivery_mode",
    "failure_delivery_channel",
    "failure_delivery_to",
    "failure_delivery_account_id",
    "failure_alert_disabled",
    "failure_alert_after",
    "failure_alert_channel",
    "failure_alert_to",
    "failure_alert_cooldown_ms",
    "failure_alert_include_skipped",
    "failure_alert_mode",
    "failure_alert_account_id",
    "next_run_at_ms",
    "running_at_ms",
    "last_run_at_ms",
    "last_run_status",
    "last_error",
    "last_duration_ms",
    "consecutive_errors",
    "consecutive_skipped",
    "schedule_error_count",
    "last_delivery_status",
    "last_delivery_error",
    "last_delivered",
    "last_failure_alert_at_ms",
    "job_json",
    "state_json",
    "runtime_updated_at_ms",
    "schedule_identity",
    "sort_order",
    "updated_at",
];
/// `cron_run_logs` columns in order.
pub const CRON_RUN_LOG_COLUMNS: &[&str] = &[
    "store_key",
    "job_id",
    "seq",
    "ts",
    "status",
    "error",
    "summary",
    "diagnostics_summary",
    "delivery_status",
    "delivery_error",
    "delivered",
    "session_id",
    "session_key",
    "run_id",
    "run_at_ms",
    "duration_ms",
    "next_run_at_ms",
    "model",
    "provider",
    "total_tokens",
    "entry_json",
    "created_at",
];
/// `delivery_queue_entries` columns in order.
pub const DELIVERY_QUEUE_COLUMNS: &[&str] = &[
    "queue_name",
    "id",
    "status",
    "entry_kind",
    "session_key",
    "channel",
    "target",
    "account_id",
    "retry_count",
    "last_attempt_at",
    "last_error",
    "recovery_state",
    "platform_send_started_at",
    "entry_json",
    "enqueued_at",
    "updated_at",
    "failed_at",
];
/// `schema_meta` columns in order.
pub const SCHEMA_META_COLUMNS: &[&str] = &[
    "meta_key",
    "role",
    "schema_version",
    "agent_id",
    "app_version",
    "created_at",
    "updated_at",
];

// ---------------------------------------------------------------- JSON5

/// Reduce JSON5 to JSON: `//` and `/* */` comments outside strings, trailing
/// commas before `}` / `]` (`profiles.rs::strip_json5`, which the Node codec
/// transcribed; this is the same rule in the crate that owns the codec).
pub fn strip_json5(text: &str) -> String {
    let mut out = String::with_capacity(text.len());
    let mut chars = text.chars().peekable();
    let mut in_string = false;
    let mut escaped = false;
    while let Some(ch) = chars.next() {
        if in_string {
            out.push(ch);
            if escaped {
                escaped = false;
            } else if ch == '\\' {
                escaped = true;
            } else if ch == '"' {
                in_string = false;
            }
            continue;
        }
        match ch {
            '"' => {
                in_string = true;
                out.push(ch);
            }
            '/' if chars.peek() == Some(&'/') => {
                for next in chars.by_ref() {
                    if next == '\n' {
                        out.push('\n');
                        break;
                    }
                }
            }
            '/' if chars.peek() == Some(&'*') => {
                chars.next();
                let mut previous = '\0';
                for next in chars.by_ref() {
                    if previous == '*' && next == '/' {
                        break;
                    }
                    previous = next;
                }
                out.push(' ');
            }
            _ => out.push(ch),
        }
    }
    let bytes: Vec<char> = out.chars().collect();
    let mut cleaned = String::with_capacity(out.len());
    let mut index = 0usize;
    let mut in_string = false;
    let mut escaped = false;
    while index < bytes.len() {
        let ch = bytes[index];
        if in_string {
            cleaned.push(ch);
            if escaped {
                escaped = false;
            } else if ch == '\\' {
                escaped = true;
            } else if ch == '"' {
                in_string = false;
            }
            index += 1;
            continue;
        }
        if ch == '"' {
            in_string = true;
            cleaned.push(ch);
            index += 1;
            continue;
        }
        if ch == ',' {
            let mut lookahead = index + 1;
            while lookahead < bytes.len() && bytes[lookahead].is_whitespace() {
                lookahead += 1;
            }
            if lookahead < bytes.len() && (bytes[lookahead] == '}' || bytes[lookahead] == ']') {
                index += 1;
                continue;
            }
        }
        cleaned.push(ch);
        index += 1;
    }
    cleaned
}

/// Parse a JSON5 file's text.
pub fn parse_json5(file: &str, text: &str) -> Result<Value> {
    serde_json::from_str(&strip_json5(text))
        .map_err(|e| load_error(file, "", format!("JSON5: {e}")))
}

// ---------------------------------------------------------------- secrets

/// Credential-SHAPED names OpenClaw uses for things that are not secrets.
pub const NON_SECRET_KEYS: &[&str] = &[
    "sessionkey",
    "session_key",
    "storekey",
    "store_key",
    "metakey",
    "meta_key",
    "bindingkey",
    "binding_key",
    "declarationkey",
    "declaration_key",
    "idempotencykey",
    "idempotency_key",
];

/// OpenClaw's own credential rule: the lower-cased key ENDS WITH a marker; only the name decides.
pub fn is_credential_key(key: &str) -> bool {
    let lower = key.to_ascii_lowercase();
    if NON_SECRET_KEYS.contains(&lower.as_str()) {
        return false;
    }
    ["token", "key", "secret", "password", "credential"]
        .iter()
        .any(|m| lower.ends_with(m))
}

/// The `.env` name for a credential: `OPENCLAW_<scope>_<key>`.
pub fn credential_ref_name(scope: &str, key: &str) -> String {
    format!("OPENCLAW_{scope}_{key}")
        .chars()
        .map(|c| {
            if c.is_ascii_alphanumeric() {
                c.to_ascii_uppercase()
            } else {
                '_'
            }
        })
        .collect()
}

/// Deep copy with every string under a credential-shaped key replaced by a `{dotenv}` ref, the value handed to the vault.
pub fn redact_secrets(value: &Value, scope: &str, vault: &mut BTreeMap<String, String>) -> Value {
    match value {
        Value::Array(items) => Value::Array(
            items
                .iter()
                .enumerate()
                .map(|(i, v)| redact_secrets(v, &format!("{scope}_{i}"), vault))
                .collect(),
        ),
        Value::Object(map) => {
            let mut out = Map::new();
            for (k, v) in map {
                match v {
                    Value::String(s) if is_credential_key(k) => {
                        let r = credential_ref_name(scope, k);
                        vault.insert(r.clone(), s.clone());
                        out.insert(k.clone(), Value::String(format!("${{{r}}}")));
                    }
                    Value::Object(_) | Value::Array(_) => {
                        out.insert(k.clone(), redact_secrets(v, &format!("{scope}_{k}"), vault));
                    }
                    other => {
                        out.insert(k.clone(), other.clone());
                    }
                }
            }
            Value::Object(out)
        }
        other => other.clone(),
    }
}

fn resolve_ref(name: &str, vault: &BTreeMap<String, String>, depth: usize) -> Option<String> {
    let value = vault.get(name)?;
    if depth > 4 {
        return Some(value.clone());
    }
    match placeholder_ref(value) {
        Some(next) => resolve_ref(next, vault, depth + 1).or_else(|| Some(value.clone())),
        None => Some(value.clone()),
    }
}

/// Inverse of [`redact_secrets`]: `${NAME}` references become their values.
pub fn inline_secrets(value: &Value, vault: &BTreeMap<String, String>) -> Value {
    match value {
        Value::String(s) => match placeholder_ref(s).and_then(|n| resolve_ref(n, vault, 0)) {
            Some(v) => Value::String(v),
            None => value.clone(),
        },
        Value::Array(items) => {
            Value::Array(items.iter().map(|v| inline_secrets(v, vault)).collect())
        }
        Value::Object(map) => Value::Object(
            map.iter()
                .map(|(k, v)| (k.clone(), inline_secrets(v, vault)))
                .collect(),
        ),
        other => other.clone(),
    }
}

/// Every `${NAME}` reference a write left unresolved, with its path.
fn unresolved_refs(value: &Value, path: &str) -> Vec<(String, String)> {
    match value {
        Value::String(s) => placeholder_ref(s)
            .map(|n| vec![(path.to_string(), n.to_string())])
            .unwrap_or_default(),
        Value::Array(items) => items
            .iter()
            .enumerate()
            .flat_map(|(i, v)| unresolved_refs(v, &format!("{path}[{i}]")))
            .collect(),
        Value::Object(map) => map
            .iter()
            .flat_map(|(k, v)| {
                unresolved_refs(
                    v,
                    &if path.is_empty() {
                        k.clone()
                    } else {
                        format!("{path}.{k}")
                    },
                )
            })
            .collect(),
        _ => Vec::new(),
    }
}

// ---------------------------------------------------------------- time

fn civil(ms: i64) -> (i64, u32, u32, u32, u32, u32, u32) {
    let secs = ms.div_euclid(1000);
    let sub = ms.rem_euclid(1000) as u32;
    let days = secs.div_euclid(86_400);
    let sod = secs.rem_euclid(86_400);
    let z = days + 719_468;
    let era = z.div_euclid(146_097);
    let doe = z - era * 146_097;
    let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365;
    let y = yoe + era * 400;
    let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
    let mp = (5 * doy + 2) / 153;
    let d = (doy - (153 * mp + 2) / 5 + 1) as u32;
    let m = if mp < 10 { mp + 3 } else { mp - 9 } as u32;
    let y = if m <= 2 { y + 1 } else { y };
    (
        y,
        m,
        d,
        (sod / 3600) as u32,
        ((sod % 3600) / 60) as u32,
        (sod % 60) as u32,
        sub,
    )
}

/// Unix ms → RFC 3339 at SECOND precision (`jobs.rs::iso_from_ms`).
pub fn iso_seconds(ms: Option<i64>) -> Option<String> {
    ms.map(|ms| {
        let (y, mo, d, h, mi, s, _) = civil(ms);
        format!("{y:04}-{mo:02}-{d:02}T{h:02}:{mi:02}:{s:02}Z")
    })
}

/// Unix ms → RFC 3339 with millis (`sidecar::ms_to_rfc3339`).
pub fn iso_millis(ms: Option<i64>) -> Option<String> {
    ms.map(|ms| {
        let (y, mo, d, h, mi, s, sub) = civil(ms);
        format!("{y:04}-{mo:02}-{d:02}T{h:02}:{mi:02}:{s:02}.{sub:03}Z")
    })
}

/// RFC 3339 (`YYYY-MM-DDTHH:MM:SS[.fff][Z|±HH:MM]`) → Unix ms; `None` for anything else.
pub fn ms_from_iso(iso: Option<&str>) -> Option<i64> {
    let s = iso?.trim();
    let (date, rest) = s.split_once('T')?;
    let mut dp = date.split('-');
    let (y, mo, d): (i64, i64, i64) = (
        dp.next()?.parse().ok()?,
        dp.next()?.parse().ok()?,
        dp.next()?.parse().ok()?,
    );
    let (time, offset) = if let Some(t) = rest.strip_suffix('Z') {
        (t, 0i64)
    } else if let Some(idx) = rest.rfind(['+', '-']) {
        let (t, off) = rest.split_at(idx);
        let sign = if off.starts_with('-') { -1 } else { 1 };
        let mut op = off[1..].split(':');
        let (oh, om): (i64, i64) = (
            op.next()?.parse().ok()?,
            op.next().unwrap_or("0").parse().ok()?,
        );
        (t, sign * (oh * 3600 + om * 60))
    } else {
        (rest, 0)
    };
    let (hms, frac) = match time.split_once('.') {
        Some((a, b)) => (a, b),
        None => (time, ""),
    };
    let mut tp = hms.split(':');
    let (h, mi, sec): (i64, i64, i64) = (
        tp.next()?.parse().ok()?,
        tp.next()?.parse().ok()?,
        tp.next().unwrap_or("0").parse().ok()?,
    );
    let millis: i64 = if frac.is_empty() {
        0
    } else {
        format!("{:0<3}", &frac[..frac.len().min(3)]).parse().ok()?
    };
    // days from civil (Howard Hinnant)
    let (yy, mm) = if mo <= 2 {
        (y - 1, mo + 9)
    } else {
        (y, mo - 3)
    };
    let era = yy.div_euclid(400);
    let yoe = yy - era * 400;
    let doy = (153 * mm + 2) / 5 + d - 1;
    let doe = yoe * 365 + yoe / 4 - yoe / 100 + doy;
    let days = era * 146_097 + doe - 719_468;
    Some(((days * 86_400 + h * 3600 + mi * 60 + sec) - offset) * 1000 + millis)
}

fn ms_of(v: Option<&Value>) -> Option<i64> {
    match v? {
        Value::Number(n) => n.as_i64().or_else(|| n.as_f64().map(|f| f as i64)),
        Value::String(s) => s.parse::<f64>().ok().map(|f| f as i64),
        _ => None,
    }
}

// ---------------------------------------------------------------- agents

/// The declared agents as `[id, entry]` in declaration order.
pub fn agent_entries(config: &Value) -> Vec<(String, Value)> {
    let agents = config.get("agents");
    if let Some(list) = agents.and_then(|a| a.get("list")).and_then(Value::as_array) {
        return list
            .iter()
            .filter_map(|e| {
                e.get("id")
                    .or_else(|| e.get("agentId"))
                    .and_then(Value::as_str)
                    .filter(|s| !s.is_empty())
                    .map(|id| (id.to_string(), e.clone()))
            })
            .collect();
    }
    if let Some(entries) = agents
        .and_then(|a| a.get("entries"))
        .and_then(Value::as_object)
    {
        return entries
            .iter()
            .map(|(k, v)| (k.clone(), v.clone()))
            .collect();
    }
    Vec::new()
}

/// `list` | `entries` | none.
pub fn agents_form(config: &Value) -> Option<&'static str> {
    let agents = config.get("agents")?;
    if agents.get("list").is_some_and(Value::is_array) {
        return Some("list");
    }
    if agents.get("entries").is_some_and(Value::is_object) {
        return Some("entries");
    }
    None
}

/// The default agent: the entry flagged `default: true`, else the first declared, else `main`.
pub fn default_agent_id(config: &Value) -> String {
    let entries = agent_entries(config);
    entries
        .iter()
        .find(|(_, e)| e.get("default") == Some(&Value::Bool(true)))
        .or_else(|| entries.first())
        .map(|(id, _)| id.clone())
        .unwrap_or_else(|| "main".into())
}

// ---------------------------------------------------------------- session keys

/// What an OpenClaw session key mapped to, in the orchestration's binding form.
#[derive(Debug, Clone, PartialEq)]
pub struct ParsedKey {
    /// The agent the key names, if any.
    pub agent: Option<String>,
    /// The surface (a `main` key is the DM COLLAPSE: `{main, dm, main}`).
    pub key: SurfaceKey,
    /// What the mapping had to translate (`session_key`, `dm_collapse`, `chat_kind`, `thread_word`).
    pub residue: Map<String, Value>,
    /// The job a `cron:` key names.
    pub recurrence: Option<Recurrence>,
}

const OPENCLAW_CHAT_KINDS: &[&str] = &["dm", "group", "channel", "thread"];

fn key_of(platform: &str, kind: &str, chat_id: &str, thread_id: Option<String>) -> SurfaceKey {
    SurfaceKey {
        key: None,
        platform: Some(platform.into()),
        kind: Some(kind.into()),
        chat_id: Some(chat_id.into()),
        thread_id,
        participant_id: None,
    }
}

/// An OpenClaw gateway session key → the orchestration's surface key plus what the
/// mapping had to translate (`openclaw.mjs::parseOpenclawSessionKey`). This is
/// the WORLD's form; the spine's discovery nouns keep the key's own shape
/// through `Binding::from_openclaw_key`.
pub fn parse_openclaw_session_key(key: &str) -> Option<ParsedKey> {
    let parts: Vec<&str> = key.split(':').collect();
    let mut residue = Map::new();
    residue.insert("session_key".into(), Value::String(key.into()));
    match parts.first().copied() {
        Some("agent") if parts.len() >= 3 => {
            let agent = Some(parts[1].to_string());
            if parts[2] == "main" {
                residue.insert("dm_collapse".into(), Value::Bool(true));
                return Some(ParsedKey {
                    agent,
                    key: key_of("main", "dm", "main", None),
                    residue,
                    recurrence: None,
                });
            }
            if parts.len() < 5 {
                return None;
            }
            let kind = if OPENCLAW_CHAT_KINDS.contains(&parts[3]) {
                parts[3]
            } else {
                residue.insert("chat_kind".into(), Value::String(parts[3].into()));
                "dm"
            };
            let mut thread_id = None;
            if (parts.get(5) == Some(&"thread") || parts.get(5) == Some(&"topic"))
                && parts.get(6).is_some_and(|t| !t.is_empty())
            {
                thread_id = Some(parts[6].to_string());
                if parts[5] == "topic" {
                    residue.insert("thread_word".into(), Value::String("topic".into()));
                }
            }
            let k = key_of(parts[2], kind, parts[4], thread_id);
            let recurrence = if parts[2] == "cron" {
                Some(Recurrence {
                    job_id: parts[4].into(),
                    kind: "cron".into(),
                })
            } else {
                None
            };
            Some(ParsedKey {
                agent,
                key: k,
                residue,
                recurrence,
            })
        }
        Some("cron") if parts.len() >= 2 => {
            let job_id = parts[1..].join(":");
            Some(ParsedKey {
                agent: None,
                key: key_of("cron", "dm", &job_id, None),
                residue,
                recurrence: Some(Recurrence {
                    job_id,
                    kind: "cron".into(),
                }),
            })
        }
        Some("hook") if parts.len() >= 2 => Some(ParsedKey {
            agent: None,
            key: key_of("webhook", "dm", parts[1], None),
            residue,
            recurrence: None,
        }),
        Some("acp-bridge") if parts.len() >= 2 => Some(ParsedKey {
            agent: None,
            key: key_of("acp", "dm", &parts[1..].join(":"), None),
            residue,
            recurrence: None,
        }),
        _ => None,
    }
}

// ---------------------------------------------------------------- io

/// Root bookkeeping for an OpenClaw state dir.
#[derive(Debug, Clone, Default)]
pub struct OpenclawRootIo {
    /// The state dir.
    pub state_dir: PathBuf,
    /// `openclaw.json`'s bytes as read (empty when absent).
    pub config_raw: String,
    /// Whether `openclaw.json` existed.
    pub config_present: bool,
    /// Canonical JSON of the config record as loaded.
    pub config_snapshot: String,
    /// Store rows as read, by table and id.
    pub cron_jobs: BTreeMap<String, Map<String, Value>>,
    /// Run-log rows by fire id.
    pub cron_run_logs: BTreeMap<String, Map<String, Value>>,
    /// Queue rows by id.
    pub delivery_queue_entries: BTreeMap<String, Map<String, Value>>,
    /// `schema_meta` rows.
    pub schema_meta: Vec<Map<String, Value>>,
    /// The `store_key` the jobs carry.
    pub store_key: String,
    /// Whether the store existed.
    pub db_present: bool,
    /// The default agent id.
    pub default_agent: String,
    /// The legacy `cron/jobs.json` beside the store, byte for byte, when
    /// one exists (an install the pin has not migrated yet).
    pub legacy_jobs_raw: Option<String>,
    /// Its records by id, as read.
    pub legacy_jobs: BTreeMap<String, Value>,
    /// Whether the file was `{"jobs": [...]}` rather than a bare array.
    pub legacy_jobs_object_form: bool,
}

/// Per-profile bookkeeping.
#[derive(Debug, Clone, Default)]
pub struct OpenclawProfileIo {
    /// The agent id.
    pub agent_id: String,
    /// `agents/<id>/`.
    pub source_dir: PathBuf,
    /// Canonical JSON of the store record (jobs, fires, obligations) as loaded.
    pub store_snapshot: String,
    /// Canonical JSON of the bindings as loaded.
    pub bindings_snapshot: String,
}

/// A loaded OpenClaw state directory.
#[derive(Debug, Clone)]
pub struct OpenclawLoaded {
    /// The orchestration.
    pub orchestration: Orchestration,
    /// Secret values by `.env` name; never in the orchestration.
    pub vault: BTreeMap<String, String>,
    /// Root bookkeeping.
    pub root: OpenclawRootIo,
    /// Per-profile bookkeeping by profile name.
    pub profiles: BTreeMap<String, OpenclawProfileIo>,
}

impl OpenclawLoaded {
    /// An orchestration that did not come from an OpenClaw store (our folder on its way
    /// out): nothing to reuse, every artifact re-emitted, every binding refused.
    pub fn from_orchestration(
        orchestration: Orchestration,
        vault: BTreeMap<String, String>,
    ) -> Self {
        let default_agent = orchestration.profiles["default"]
            .residue
            .config
            .get("openclaw")
            .and_then(|o| o.get("default_agent"))
            .and_then(Value::as_str)
            .unwrap_or("main")
            .to_string();
        let profiles = orchestration
            .profiles
            .keys()
            .map(|n| {
                (
                    n.clone(),
                    OpenclawProfileIo {
                        agent_id: if n == "default" {
                            default_agent.clone()
                        } else {
                            n.clone()
                        },
                        ..Default::default()
                    },
                )
            })
            .collect();
        Self {
            root: OpenclawRootIo {
                state_dir: orchestration.root.clone(),
                default_agent,
                ..Default::default()
            },
            orchestration,
            vault,
            profiles,
        }
    }
}

// ---------------------------------------------------------------- channels

/// `channels.<name>.accounts` as `[accountId, entry]`, sorted by id.
pub fn account_entries(entry: &Value) -> Vec<(String, Value)> {
    let mut out: Vec<(String, Value)> = match entry.get("accounts") {
        Some(Value::Object(m)) => m.iter().map(|(k, v)| (k.clone(), v.clone())).collect(),
        Some(Value::Array(a)) => a
            .iter()
            .filter_map(|x| {
                x.get("id")
                    .or_else(|| x.get("accountId"))
                    .and_then(Value::as_str)
                    .filter(|s| !s.is_empty())
                    .map(|id| (id.to_string(), x.clone()))
            })
            .collect(),
        _ => Vec::new(),
    };
    out.sort_by(|a, b| a.0.cmp(&b.0));
    out
}

fn credentials_of(
    block: &Value,
    scope: &str,
    vault: &mut BTreeMap<String, String>,
) -> BTreeMap<String, SecretRef> {
    let mut creds = BTreeMap::new();
    let Some(map) = block.as_object() else {
        return creds;
    };
    for (k, v) in map {
        match v {
            Value::String(s) if placeholder_ref(s).is_some() => {
                creds.insert(
                    k.clone(),
                    SecretRef::Dotenv(placeholder_ref(s).unwrap().to_string()),
                );
            }
            Value::String(s) if is_credential_key(k) => {
                let r = credential_ref_name(scope, k);
                vault.insert(r.clone(), s.clone());
                creds.insert(k.clone(), SecretRef::Dotenv(r));
            }
            _ => {}
        }
    }
    creds
}

fn decode_channels(
    config: &Value,
    vault: &mut BTreeMap<String, String>,
) -> BTreeMap<String, ChannelConfig> {
    let mut channels = BTreeMap::new();
    let Some(map) = config.get("channels").and_then(Value::as_object) else {
        return channels;
    };
    for (kind, entry) in map {
        let Some(entry_map) = entry.as_object() else {
            continue;
        };
        let mut shared = entry_map.clone();
        shared.remove("accounts");
        let mut shared_block = redact_secrets(&Value::Object(shared), kind, vault);
        let channel_enabled = entry_map.get("enabled").and_then(Value::as_bool);
        if let Some(m) = shared_block.as_object_mut() {
            m.remove("enabled");
        }
        let accounts = account_entries(entry);
        let inline_account = OPENCLAW_ACCOUNT_KEYS
            .iter()
            .find_map(|k| entry_map.get(*k).and_then(Value::as_str))
            .map(str::to_string);
        if accounts.is_empty() {
            let mut extra = BTreeMap::new();
            extra.insert("kind".into(), Value::String(kind.clone()));
            extra.insert(
                "accountId".into(),
                inline_account
                    .clone()
                    .map(Value::String)
                    .unwrap_or(Value::Null),
            );
            extra.insert(
                "account_source".into(),
                if inline_account.is_some() {
                    Value::String("entry".into())
                } else {
                    Value::Null
                },
            );
            extra.insert("accounts_form".into(), Value::Null);
            extra.insert(
                "enabled_on".into(),
                if channel_enabled.is_some() {
                    Value::String("channel".into())
                } else {
                    Value::Null
                },
            );
            extra.insert("channel_enabled".into(), Value::Null);
            extra.insert("channel_block".into(), shared_block.clone());
            channels.insert(
                kind.clone(),
                ChannelConfig {
                    platform: kind.clone(),
                    enabled: channel_enabled.unwrap_or(true),
                    credentials: credentials_of(entry, kind, vault),
                    extra,
                },
            );
            continue;
        }
        let form = if entry_map.get("accounts").is_some_and(Value::is_object) {
            "object"
        } else {
            "array"
        };
        for (id, account) in accounts {
            let name = format!("{kind}/{id}");
            let mut block = redact_secrets(&account, &name, vault);
            let account_enabled = account.get("enabled").and_then(Value::as_bool);
            if let Some(m) = block.as_object_mut() {
                m.remove("enabled");
            }
            let enabled = account_enabled.or(channel_enabled).unwrap_or(true);
            let mut credentials = credentials_of(&account, &name, vault);
            for (k, v) in credentials_of(entry, kind, vault) {
                credentials.insert(k, v);
            }
            let mut extra = BTreeMap::new();
            extra.insert("kind".into(), Value::String(kind.clone()));
            extra.insert("accountId".into(), Value::String(id.clone()));
            extra.insert("account_source".into(), Value::String("accounts".into()));
            extra.insert("accounts_form".into(), Value::String(form.into()));
            extra.insert(
                "enabled_on".into(),
                if account_enabled.is_some() {
                    Value::String("account".into())
                } else if channel_enabled.is_some() {
                    Value::String("channel".into())
                } else {
                    Value::Null
                },
            );
            extra.insert(
                "channel_enabled".into(),
                channel_enabled.map(Value::Bool).unwrap_or(Value::Null),
            );
            extra.insert("channel_block".into(), shared_block.clone());
            extra.insert("account_block".into(), block);
            channels.insert(
                name.clone(),
                ChannelConfig {
                    platform: name,
                    enabled,
                    credentials,
                    extra,
                },
            );
        }
    }
    channels
}

const CHANNEL_BOOKKEEPING: &[&str] = &[
    "kind",
    "accountId",
    "account_source",
    "accounts_form",
    "enabled_on",
    "channel_enabled",
    "channel_block",
    "account_block",
];

fn block_from_model(ch: &ChannelConfig) -> Value {
    let mut out = Map::new();
    for (k, v) in &ch.extra {
        if CHANNEL_BOOKKEEPING.contains(&k.as_str()) {
            continue;
        }
        out.insert(k.strip_prefix("extra.").unwrap_or(k).to_string(), v.clone());
    }
    for (k, r) in &ch.credentials {
        out.insert(k.clone(), serde_json::to_value(r).unwrap());
    }
    Value::Object(out)
}

fn ordered_channel_block(block: Value) -> Vec<(String, Value)> {
    let Some(map) = block.as_object() else {
        return Vec::new();
    };
    let mut out: Vec<(String, Value)> = Vec::new();
    if let Some(e) = map.get("enabled") {
        out.push(("enabled".into(), e.clone()));
    }
    for (k, v) in map {
        if k != "enabled" {
            out.push((k.clone(), v.clone()));
        }
    }
    out
}

fn encode_channels(profile: &Profile, vault: &BTreeMap<String, String>) -> Vec<(String, Value)> {
    let mut by_kind: Vec<(String, Vec<&ChannelConfig>)> = Vec::new();
    for ch in profile.channels.values() {
        let kind = ch
            .extra
            .get("kind")
            .and_then(Value::as_str)
            .unwrap_or(&ch.platform)
            .to_string();
        match by_kind.iter_mut().find(|(k, _)| *k == kind) {
            Some((_, rows)) => rows.push(ch),
            None => by_kind.push((kind, vec![ch])),
        }
    }
    let mut out = Vec::new();
    for (kind, rows) in by_kind {
        let first = rows[0];
        let channel_block_src = first
            .extra
            .get("channel_block")
            .filter(|v| v.is_object())
            .cloned()
            .unwrap_or_else(|| block_from_model(first));
        let channel_block = inline_secrets(&channel_block_src, vault);
        let channel_enabled = first.extra.get("channel_enabled").and_then(Value::as_bool);
        if first.extra.get("account_source").and_then(Value::as_str) != Some("accounts") {
            let mut single = channel_block.as_object().cloned().unwrap_or_default();
            if first.extra.get("enabled_on").and_then(Value::as_str) == Some("channel")
                || !first.enabled
            {
                single.insert("enabled".into(), Value::Bool(first.enabled));
            }
            out.push((
                kind,
                ordered_object(ordered_channel_block(Value::Object(single))),
            ));
            continue;
        }
        let form = first
            .extra
            .get("accounts_form")
            .and_then(Value::as_str)
            .unwrap_or("object");
        let mut head = channel_block.as_object().cloned().unwrap_or_default();
        if let Some(e) = channel_enabled {
            head.insert("enabled".into(), Value::Bool(e));
        }
        let blocks: Vec<(String, Value)> = rows
            .iter()
            .map(|ch| {
                let block_src = ch
                    .extra
                    .get("account_block")
                    .filter(|v| v.is_object())
                    .cloned()
                    .unwrap_or_else(|| block_from_model(ch));
                let block = inline_secrets(&block_src, vault);
                let inherited = channel_enabled.unwrap_or(true);
                let id = ch
                    .extra
                    .get("accountId")
                    .and_then(Value::as_str)
                    .unwrap_or("")
                    .to_string();
                if ch.extra.get("enabled_on").and_then(Value::as_str) == Some("account")
                    || ch.enabled != inherited
                {
                    let mut b = vec![("enabled".to_string(), Value::Bool(ch.enabled))];
                    b.extend(
                        block
                            .as_object()
                            .map(|m| {
                                m.iter()
                                    .map(|(k, v)| (k.clone(), v.clone()))
                                    .collect::<Vec<_>>()
                            })
                            .unwrap_or_default(),
                    );
                    (id, ordered_object(b))
                } else {
                    (
                        id,
                        ordered_object(
                            block
                                .as_object()
                                .map(|m| m.iter().map(|(k, v)| (k.clone(), v.clone())).collect())
                                .unwrap_or_default(),
                        ),
                    )
                }
            })
            .collect();
        let mut pairs: Vec<(String, Value)> = head.into_iter().collect();
        if form == "array" {
            pairs.push((
                "accounts".into(),
                Value::Array(
                    blocks
                        .into_iter()
                        .map(|(id, b)| {
                            let mut p = vec![("id".to_string(), Value::String(id))];
                            if let Some(items) = is_ordered_pairs(&b) {
                                p.extend(items);
                            }
                            ordered_object(p)
                        })
                        .collect(),
                ),
            ));
        } else {
            pairs.push(("accounts".into(), ordered_object(blocks)));
        }
        out.push((kind, ordered_object(pairs)));
    }
    out
}

fn is_ordered_pairs(value: &Value) -> Option<Vec<(String, Value)>> {
    let arr = value.as_array()?;
    if arr.len() == 2 && arr[0].as_str() == Some("__ordered__") {
        return arr[1].as_array().map(|items| {
            items
                .iter()
                .map(|p| {
                    (
                        p["__k"].as_str().unwrap_or("").to_string(),
                        p["__v"].clone(),
                    )
                })
                .collect()
        });
    }
    None
}

// ---------------------------------------------------------------- routes

const BINDING_MATCH_MAPPED: &[&str] = &["channel", "guildId", "peer"];

fn decode_routes(config: &Value, name_for: &dyn Fn(&str) -> String) -> Vec<Route> {
    let mut routes = Vec::new();
    let Some(list) = config.get("bindings").and_then(Value::as_array) else {
        return routes;
    };
    for (index, binding) in list.iter().enumerate() {
        let (Some(bmap), Some(agent_id)) = (
            binding.as_object(),
            binding.get("agentId").and_then(Value::as_str),
        ) else {
            continue;
        };
        let m = binding
            .get("match")
            .and_then(Value::as_object)
            .cloned()
            .unwrap_or_default();
        let text = |v: &Value| match v {
            Value::String(s) => Some(s.clone()),
            Value::Number(n) => Some(n.to_string()),
            _ => None,
        };
        let matches = RouteMatch {
            platform: m
                .get("channel")
                .and_then(Value::as_str)
                .unwrap_or("")
                .to_string(),
            guild_id: m.get("guildId").filter(|v| !v.is_null()).and_then(text),
            chat_id: m
                .get("peer")
                .and_then(|p| p.get("id"))
                .filter(|v| !v.is_null())
                .and_then(text),
            thread_id: None,
        };
        let mut match_residue = Map::new();
        for (k, v) in &m {
            if !BINDING_MATCH_MAPPED.contains(&k.as_str()) {
                match_residue.insert(k.clone(), v.clone());
            }
        }
        if let Some(peer) = m.get("peer").and_then(Value::as_object) {
            let mut pr = peer.clone();
            pr.remove("id");
            if !pr.is_empty() {
                match_residue.insert("peer".into(), Value::Object(pr));
            }
        }
        let mut residue = Residue::default();
        residue.keep("agent_id", Value::String(agent_id.into()));
        residue.keep("index", Value::from(index));
        for (k, v) in bmap {
            if k != "agentId" && k != "match" {
                residue.keep(k.clone(), v.clone());
            }
        }
        if !match_residue.is_empty() {
            residue.keep("match", Value::Object(match_residue));
        }
        routes.push(Route {
            name: None,
            matches,
            profile: name_for(agent_id),
            residue,
        });
    }
    routes
}

fn encode_routes(profile: &Profile, id_for_name: &dyn Fn(&str) -> String) -> Vec<Value> {
    profile
        .routes
        .iter()
        .map(|r| {
            let mut residue = r.residue.0.clone();
            let agent_id = residue
                .remove("agent_id")
                .and_then(|v| v.as_str().map(str::to_string))
                .unwrap_or_else(|| id_for_name(&r.profile));
            residue.remove("index");
            let match_residue = residue
                .remove("match")
                .and_then(|v| v.as_object().cloned())
                .unwrap_or_default();
            let mut m: Vec<(String, Value)> = Vec::new();
            if !r.matches.platform.is_empty() {
                m.push(("channel".into(), Value::String(r.matches.platform.clone())));
            }
            for (k, v) in &match_residue {
                if k != "peer" {
                    m.push((k.clone(), v.clone()));
                }
            }
            if let Some(g) = &r.matches.guild_id {
                m.push(("guildId".into(), Value::String(g.clone())));
            }
            if r.matches.chat_id.is_some()
                || match_residue.get("peer").is_some_and(Value::is_object)
            {
                let mut peer: Vec<(String, Value)> = match_residue
                    .get("peer")
                    .and_then(Value::as_object)
                    .map(|p| p.iter().map(|(k, v)| (k.clone(), v.clone())).collect())
                    .unwrap_or_default();
                if let Some(c) = &r.matches.chat_id {
                    peer.push(("id".into(), Value::String(c.clone())));
                }
                m.push(("peer".into(), ordered_object(peer)));
            }
            let mut pairs: Vec<(String, Value)> = residue.into_iter().collect();
            pairs.push(("agentId".into(), Value::String(agent_id)));
            pairs.push(("match".into(), ordered_object(m)));
            ordered_object(pairs)
        })
        .collect()
}

// ---------------------------------------------------------------- hooks

/// What the `hooks` block carried beside its mappings.
#[derive(Debug, Clone, Default, PartialEq)]
struct HooksMeta {
    block: Map<String, Value>,
    has_token: bool,
}

fn decode_hooks(
    config: &Value,
    vault: &mut BTreeMap<String, String>,
    name_for: &dyn Fn(&str) -> String,
    profiles: &mut BTreeMap<String, Profile>,
) -> Option<HooksMeta> {
    let hooks = config.get("hooks").and_then(Value::as_object)?;
    let mut secret = None;
    if let Some(Value::String(token)) = hooks.get("token") {
        let r = credential_ref_name("hooks", "token");
        vault.insert(r.clone(), token.clone());
        secret = Some(SecretRef::Dotenv(r));
    }
    if let Some(mappings) = hooks.get("mappings").and_then(Value::as_array) {
        for (index, mapping) in mappings.iter().enumerate() {
            let Some(mm) = mapping.as_object() else {
                continue;
            };
            let name = mm
                .get("id")
                .and_then(Value::as_str)
                .filter(|s| !s.is_empty())
                .map(str::to_string)
                .unwrap_or_else(|| format!("hook-{index}"));
            let mut residue_v = redact_secrets(mapping, &format!("hook_{name}"), vault)
                .as_object()
                .cloned()
                .unwrap_or_default();
            residue_v.remove("deliver");
            residue_v.remove("to");
            residue_v.insert("__index".into(), Value::from(index));
            let owner = mm
                .get("agentId")
                .and_then(Value::as_str)
                .map(name_for)
                .filter(|n| profiles.contains_key(n))
                .unwrap_or_else(|| "default".into());
            let deliver =
                mm.get("deliver")
                    .and_then(Value::as_str)
                    .map(|platform| Target::Explicit {
                        platform: platform.into(),
                        chat_id: mm.get("to").and_then(|t| match t {
                            Value::String(s) => Some(s.clone()),
                            Value::Number(n) => Some(n.to_string()),
                            _ => None,
                        }),
                        thread_id: None,
                    });
            profiles.get_mut(&owner).unwrap().subscriptions.insert(
                name.clone(),
                WebhookSubscription {
                    name,
                    secret: secret.clone(),
                    events: None,
                    prompt_template: String::new(),
                    deliver,
                    skills: Vec::new(),
                    description: None,
                    created_at: None,
                    residue: Residue(residue_v.into_iter().collect()),
                },
            );
        }
    }
    let mut block = hooks.clone();
    block.remove("token");
    block.remove("mappings");
    Some(HooksMeta {
        block,
        has_token: secret.is_some(),
    })
}

fn encode_hooks(
    orchestration: &Orchestration,
    meta: Option<&HooksMeta>,
    vault: &BTreeMap<String, String>,
) -> Option<Value> {
    let mut subs: Vec<&WebhookSubscription> = orchestration
        .profiles
        .values()
        .flat_map(|p| p.subscriptions.values())
        .collect();
    if meta.is_none() && subs.is_empty() {
        return None;
    }
    subs.sort_by_key(|s| {
        s.residue
            .0
            .get("__index")
            .and_then(Value::as_i64)
            .unwrap_or(0)
    });
    let mut pairs: Vec<(String, Value)> = meta
        .map(|m| {
            inline_secrets(&Value::Object(m.block.clone()), vault)
                .as_object()
                .map(|o| o.iter().map(|(k, v)| (k.clone(), v.clone())).collect())
                .unwrap_or_default()
        })
        .unwrap_or_default();
    if meta.is_some_and(|m| m.has_token) {
        let r = subs
            .iter()
            .find_map(|s| s.secret.clone())
            .unwrap_or_else(|| SecretRef::Dotenv(credential_ref_name("hooks", "token")));
        pairs.push((
            "token".into(),
            inline_secrets(&serde_json::to_value(&r).unwrap(), vault),
        ));
    }
    if !subs.is_empty() {
        pairs.push((
            "mappings".into(),
            Value::Array(
                subs.iter()
                    .map(|sub| {
                        let mut residue = inline_secrets(
                            &Value::Object(sub.residue.0.clone().into_iter().collect()),
                            vault,
                        )
                        .as_object()
                        .cloned()
                        .unwrap_or_default();
                        residue.remove("__index");
                        let mut mp: Vec<(String, Value)> =
                            vec![("id".into(), Value::String(sub.name.clone()))];
                        mp.extend(residue);
                        if let Some(Target::Explicit {
                            platform, chat_id, ..
                        }) = &sub.deliver
                        {
                            mp.push(("deliver".into(), Value::String(platform.clone())));
                            if let Some(c) = chat_id {
                                mp.push(("to".into(), Value::String(c.clone())));
                            }
                        }
                        ordered_object(mp)
                    })
                    .collect(),
            ),
        ));
    }
    Some(ordered_object(pairs))
}

// ---------------------------------------------------------------- jobs

fn json_object_of(text: Option<&Value>) -> Map<String, Value> {
    text.and_then(Value::as_str)
        .and_then(|s| serde_json::from_str::<Value>(s).ok())
        .and_then(|v| v.as_object().cloned())
        .unwrap_or_default()
}

fn opt_text(v: Option<&Value>) -> Option<String> {
    match v {
        Some(Value::String(s)) if !s.is_empty() => Some(s.clone()),
        Some(Value::Number(n)) => Some(n.to_string()),
        _ => None,
    }
}

fn delivery_target(
    delivery: Option<&Map<String, Value>>,
    row: Option<&Map<String, Value>>,
) -> Option<Target> {
    let d = |k: &str| delivery.and_then(|d| d.get(k)).filter(|v| !v.is_null());
    let r = |k: &str| row.and_then(|r| r.get(k)).filter(|v| !v.is_null());
    let mode = d("mode")
        .or_else(|| d("kind"))
        .or_else(|| d("type"))
        .or_else(|| r("delivery_mode"))
        .and_then(Value::as_str)
        .map(str::to_string);
    let channel = d("channel")
        .or_else(|| r("delivery_channel"))
        .and_then(Value::as_str)
        .map(str::to_string);
    let to = opt_text(d("to").or_else(|| r("delivery_to")));
    let thread = opt_text(
        d("threadId")
            .or_else(|| d("thread_id"))
            .or_else(|| r("delivery_thread_id")),
    );
    if mode.as_deref() == Some("none") {
        return Some(Target::Local);
    }
    if matches!(channel.as_deref(), Some("last") | Some("origin")) {
        return Some(Target::Origin);
    }
    let Some(channel) = channel else {
        return if mode.is_some() {
            Some(Target::Local)
        } else {
            None
        };
    };
    Some(Target::Explicit {
        platform: channel,
        chat_id: to,
        thread_id: thread,
    })
}

fn failure_target(record: &Map<String, Value>, row: &Map<String, Value>) -> Option<Target> {
    let f = record.get("failureDelivery").and_then(Value::as_object);
    let channel = f
        .and_then(|f| f.get("channel"))
        .or_else(|| row.get("failure_delivery_channel"))
        .filter(|v| !v.is_null())
        .cloned();
    let mode = f
        .and_then(|f| f.get("mode"))
        .or_else(|| row.get("failure_delivery_mode"))
        .filter(|v| !v.is_null())
        .cloned();
    if channel.is_none() && mode.is_none() {
        return None;
    }
    let mut synth = Map::new();
    if let Some(m) = mode {
        synth.insert("mode".into(), m);
    }
    if let Some(c) = channel {
        synth.insert("channel".into(), c);
    }
    if let Some(t) = f
        .and_then(|f| f.get("to"))
        .or_else(|| row.get("failure_delivery_to"))
        .filter(|v| !v.is_null())
    {
        synth.insert("to".into(), t.clone());
    }
    delivery_target(Some(&synth), None)
}

fn origin_of(row: &Map<String, Value>) -> Option<JobOrigin> {
    let key = row.get("owner_session_key").and_then(Value::as_str)?;
    let parsed = parse_openclaw_session_key(key)?;
    Some(JobOrigin {
        platform: parsed.key.platform.unwrap_or_default(),
        chat_type: parsed.key.kind,
        chat_id: parsed.key.chat_id,
        thread_id: parsed.key.thread_id,
    })
}

/// `cron_jobs` row → Job (`openclaw.mjs::decodeJob`).
pub fn decode_job(row: &Map<String, Value>) -> Job {
    let record = json_object_of(row.get("job_json"));
    let state_json = json_object_of(row.get("state_json"));
    let raw_schedule = record
        .get("schedule")
        .and_then(Value::as_object)
        .cloned()
        .unwrap_or_default();
    let mut schedule_residue = raw_schedule.clone();
    let kind = raw_schedule.get("kind").and_then(Value::as_str);
    let schedule =
        if kind == Some("every") && raw_schedule.get("everyMs").is_some_and(Value::is_number) {
            schedule_residue.remove("kind");
            schedule_residue.remove("everyMs");
            Schedule::Interval {
                minutes: raw_schedule["everyMs"].as_f64().unwrap() / 60000.0,
            }
        } else if kind == Some("cron") && raw_schedule.get("expr").is_some_and(Value::is_string) {
            schedule_residue.remove("kind");
            schedule_residue.remove("expr");
            schedule_residue.remove("tz");
            Schedule::Cron {
                expr: raw_schedule["expr"].as_str().unwrap().into(),
                tz: raw_schedule
                    .get("tz")
                    .and_then(Value::as_str)
                    .map(str::to_string),
            }
        } else if kind == Some("at") && raw_schedule.get("at").is_some_and(Value::is_string) {
            schedule_residue.remove("kind");
            schedule_residue.remove("at");
            Schedule::Once {
                run_at: raw_schedule["at"].as_str().unwrap().into(),
            }
        } else {
            schedule_residue.insert("__unmapped".into(), Value::Bool(true));
            Schedule::Cron {
                expr: String::new(),
                tz: None,
            }
        };
    let payload = record
        .get("payload")
        .and_then(Value::as_object)
        .cloned()
        .unwrap_or_default();
    let prompt = ["message", "text", "command", "script"]
        .iter()
        .find_map(|k| payload.get(*k).and_then(Value::as_str))
        .map(str::to_string);
    // the store's writer keeps delivery in `job_json` AND in columns; a row
    // that has it only in the columns still has to re-emit as a delivery, so
    // the columns' object is what the residue keeps when `job_json` has none
    let delivery = record
        .get("delivery")
        .and_then(Value::as_object)
        .cloned()
        .or_else(|| {
            let mut d = Map::new();
            for (column, key) in [
                ("delivery_mode", "mode"),
                ("delivery_channel", "channel"),
                ("delivery_to", "to"),
                ("delivery_thread_id", "threadId"),
                ("delivery_account_id", "accountId"),
            ] {
                if let Some(v) = row.get(column).filter(|v| !v.is_null()) {
                    if v.as_str().is_some_and(str::is_empty) {
                        continue;
                    }
                    d.insert(key.into(), v.clone());
                }
            }
            (!d.is_empty()).then_some(d)
        });
    let session_target = record
        .get("sessionTarget")
        .and_then(Value::as_str)
        .map(str::to_string)
        .or_else(|| {
            row.get("session_target")
                .and_then(Value::as_str)
                .map(str::to_string)
        });
    let job_id = opt_text(row.get("job_id")).unwrap_or_default();
    let mut residue = Residue::default();
    residue.keep(
        "__session_target",
        session_target.map(Value::String).unwrap_or(Value::Null),
    );
    residue.keep(
        "name",
        record
            .get("name")
            .and_then(Value::as_str)
            .map(|s| Value::String(s.into()))
            .unwrap_or_else(|| {
                row.get("name")
                    .filter(|v| !v.is_null())
                    .cloned()
                    .unwrap_or(Value::String(job_id.clone()))
            }),
    );
    for (k, v) in &record {
        if [
            "id",
            "name",
            "enabled",
            "schedule",
            "payload",
            "delivery",
            "sessionTarget",
            "createdAtMs",
            "state",
        ]
        .contains(&k.as_str())
        {
            continue;
        }
        residue.keep(k.clone(), v.clone());
    }
    if !schedule_residue.is_empty() {
        residue.keep("__schedule", Value::Object(schedule_residue));
    }
    if !payload.is_empty() {
        residue.keep("__payload", Value::Object(payload.clone()));
    }
    if let Some(d) = &delivery {
        residue.keep("__delivery", Value::Object(d.clone()));
    }
    if let Some(f) = record.get("failureDelivery").filter(|v| v.is_object()) {
        residue.keep("__failure_delivery", f.clone());
    }
    if !state_json.is_empty() {
        residue.keep("__state", Value::Object(state_json.clone()));
    }
    let enabled = match row.get("enabled") {
        None | Some(Value::Null) => true,
        Some(Value::Bool(b)) => *b,
        Some(Value::Number(n)) => n.as_f64() != Some(0.0),
        Some(other) => !matches!(other, Value::String(s) if s.is_empty()),
    };
    Job {
        id: job_id,
        schedule,
        prompt,
        workdir: None,
        model: payload
            .get("model")
            .and_then(Value::as_str)
            .map(str::to_string)
            .or_else(|| {
                row.get("payload_model")
                    .and_then(Value::as_str)
                    .map(str::to_string)
            }),
        skills: Vec::new(),
        context_from: None,
        deliver: delivery_target(delivery.as_ref(), Some(row)).unwrap_or(Target::Local),
        failure_deliver: failure_target(&record, row),
        origin: origin_of(row),
        attach_to_session: None,
        machine: None,
        repeat: None,
        enabled,
        next_run_at: iso_seconds(
            ms_of(row.get("next_run_at_ms").filter(|v| !v.is_null()))
                .or_else(|| ms_of(state_json.get("nextRunAtMs"))),
        ),
        last_run_at: iso_seconds(
            ms_of(row.get("last_run_at_ms").filter(|v| !v.is_null()))
                .or_else(|| ms_of(state_json.get("lastRunAtMs"))),
        ),
        last_status: row
            .get("last_run_status")
            .and_then(Value::as_str)
            .map(str::to_string),
        created_at: iso_seconds(
            ms_of(row.get("created_at_ms").filter(|v| !v.is_null()))
                .or_else(|| ms_of(record.get("createdAtMs"))),
        ),
        residue,
    }
}

fn empty_row(columns: &[&str]) -> Map<String, Value> {
    columns
        .iter()
        .map(|c| (c.to_string(), Value::Null))
        .collect()
}

/// A job as a `cron_jobs` row, over the original row when one exists.
pub fn encode_job_row(
    job: &Job,
    raw: Option<&Map<String, Value>>,
    store_key: &str,
) -> Map<String, Value> {
    let residue = &job.residue.0;
    let mut row = raw.cloned().unwrap_or_else(|| empty_row(CRON_JOB_COLUMNS));
    let mut schedule = Map::new();
    match &job.schedule {
        Schedule::Interval { minutes } => {
            schedule.insert("kind".into(), "every".into());
            schedule.insert(
                "everyMs".into(),
                Value::from((minutes * 60000.0).round() as i64),
            );
        }
        Schedule::Cron { expr, tz } => {
            schedule.insert("kind".into(), "cron".into());
            schedule.insert("expr".into(), Value::String(expr.clone()));
            if let Some(tz) = tz {
                schedule.insert("tz".into(), Value::String(tz.clone()));
            }
        }
        Schedule::Once { run_at } => {
            schedule.insert("kind".into(), "at".into());
            schedule.insert("at".into(), Value::String(run_at.clone()));
        }
    }
    if let Some(Value::Object(extra)) = residue.get("__schedule") {
        for (k, v) in extra {
            schedule.insert(k.clone(), v.clone());
        }
    }
    schedule.remove("__unmapped");
    let mut record = Map::new();
    record.insert("id".into(), Value::String(job.id.clone()));
    record.insert(
        "name".into(),
        residue
            .get("name")
            .cloned()
            .unwrap_or(Value::String(job.id.clone())),
    );
    record.insert("enabled".into(), Value::Bool(job.enabled));
    if let Some(ms) = ms_from_iso(job.created_at.as_deref()) {
        record.insert("createdAtMs".into(), Value::from(ms));
    }
    record.insert("schedule".into(), Value::Object(schedule.clone()));
    if let Some(st) = residue.get("__session_target").filter(|v| !v.is_null()) {
        record.insert("sessionTarget".into(), st.clone());
    }
    match residue.get("__payload").and_then(Value::as_object) {
        Some(p) => {
            let mut p = p.clone();
            if let Some(prompt) = &job.prompt {
                for k in ["message", "text", "command", "script"] {
                    if p.contains_key(k) {
                        p.insert(k.into(), Value::String(prompt.clone()));
                        break;
                    }
                }
            }
            record.insert("payload".into(), Value::Object(p));
        }
        None => {
            if let Some(prompt) = &job.prompt {
                record.insert(
                    "payload".into(),
                    serde_json::json!({"kind": "agentTurn", "message": prompt}),
                );
            }
        }
    }
    if let Some(d) = residue.get("__delivery") {
        record.insert("delivery".into(), d.clone());
    }
    if let Some(f) = residue.get("__failure_delivery") {
        record.insert("failureDelivery".into(), f.clone());
    }
    record.insert(
        "state".into(),
        residue
            .get("__state")
            .cloned()
            .unwrap_or_else(|| Value::Object(Map::new())),
    );
    for (k, v) in residue {
        if !k.starts_with("__") {
            record.insert(k.clone(), v.clone());
        }
    }
    let get = |m: &Map<String, Value>, k: &str| m.get(k).filter(|v| !v.is_null()).cloned();
    let delivery = record
        .get("delivery")
        .and_then(Value::as_object)
        .cloned()
        .unwrap_or_default();
    let payload = record
        .get("payload")
        .and_then(Value::as_object)
        .cloned()
        .unwrap_or_default();
    row.insert(
        "store_key".into(),
        get(&row, "store_key").unwrap_or(Value::String(store_key.into())),
    );
    row.insert("job_id".into(), Value::String(job.id.clone()));
    row.insert("name".into(), record["name"].clone());
    row.insert("enabled".into(), Value::from(i64::from(job.enabled)));
    row.insert(
        "created_at_ms".into(),
        record
            .get("createdAtMs")
            .cloned()
            .or_else(|| get(&row, "created_at_ms"))
            .unwrap_or(Value::from(0)),
    );
    row.insert(
        "schedule_kind".into(),
        schedule
            .get("kind")
            .cloned()
            .unwrap_or(Value::String("cron".into())),
    );
    row.insert(
        "schedule_expr".into(),
        schedule.get("expr").cloned().unwrap_or(Value::Null),
    );
    row.insert(
        "schedule_tz".into(),
        schedule.get("tz").cloned().unwrap_or(Value::Null),
    );
    row.insert(
        "every_ms".into(),
        schedule.get("everyMs").cloned().unwrap_or(Value::Null),
    );
    row.insert(
        "anchor_ms".into(),
        schedule.get("anchorMs").cloned().unwrap_or(Value::Null),
    );
    row.insert(
        "at".into(),
        schedule.get("at").cloned().unwrap_or(Value::Null),
    );
    row.insert(
        "session_target".into(),
        record
            .get("sessionTarget")
            .cloned()
            .or_else(|| get(&row, "session_target"))
            .unwrap_or(Value::String("isolated".into())),
    );
    row.insert(
        "wake_mode".into(),
        get(&row, "wake_mode").unwrap_or(Value::String("now".into())),
    );
    row.insert(
        "payload_kind".into(),
        payload
            .get("kind")
            .cloned()
            .or_else(|| get(&row, "payload_kind"))
            .unwrap_or(Value::String("agentTurn".into())),
    );
    row.insert(
        "payload_message".into(),
        job.prompt.clone().map(Value::String).unwrap_or(Value::Null),
    );
    row.insert(
        "payload_model".into(),
        job.model.clone().map(Value::String).unwrap_or(Value::Null),
    );
    row.insert(
        "delivery_mode".into(),
        delivery
            .get("mode")
            .or_else(|| delivery.get("kind"))
            .cloned()
            .unwrap_or(Value::Null),
    );
    row.insert(
        "delivery_channel".into(),
        delivery.get("channel").cloned().unwrap_or(Value::Null),
    );
    row.insert(
        "delivery_to".into(),
        delivery.get("to").cloned().unwrap_or(Value::Null),
    );
    row.insert(
        "delivery_thread_id".into(),
        delivery.get("threadId").cloned().unwrap_or(Value::Null),
    );
    row.insert(
        "delivery_account_id".into(),
        delivery.get("accountId").cloned().unwrap_or(Value::Null),
    );
    row.insert(
        "next_run_at_ms".into(),
        ms_from_iso(job.next_run_at.as_deref())
            .map(Value::from)
            .unwrap_or(Value::Null),
    );
    row.insert(
        "last_run_at_ms".into(),
        ms_from_iso(job.last_run_at.as_deref())
            .map(Value::from)
            .unwrap_or(Value::Null),
    );
    row.insert(
        "last_run_status".into(),
        job.last_status
            .clone()
            .map(Value::String)
            .unwrap_or(Value::Null),
    );
    row.insert(
        "job_json".into(),
        Value::String(serde_json::to_string(&Value::Object(record.clone())).unwrap()),
    );
    row.insert(
        "state_json".into(),
        Value::String(
            serde_json::to_string(record.get("state").unwrap_or(&Value::Object(Map::new())))
                .unwrap(),
        ),
    );
    row.insert(
        "sort_order".into(),
        get(&row, "sort_order").unwrap_or(Value::from(0)),
    );
    row.insert(
        "updated_at".into(),
        get(&row, "updated_at")
            .or_else(|| get(&row, "created_at_ms"))
            .unwrap_or(Value::from(0)),
    );
    row
}

// ---------------------------------------------------------------- fires

fn run_status(word: Option<&str>) -> FireStatus {
    match word {
        Some("ok") => FireStatus::Succeeded,
        Some("error") => FireStatus::Failed,
        _ => FireStatus::Unknown,
    }
}

fn run_status_back(status: FireStatus) -> &'static str {
    match status {
        FireStatus::Succeeded | FireStatus::Claimed | FireStatus::Running => "ok",
        FireStatus::Failed | FireStatus::Timeout => "error",
        FireStatus::Unknown => "skipped",
    }
}

const FIRE_MAPPED: &[&str] = &[
    "job_id",
    "seq",
    "ts",
    "error",
    "run_id",
    "run_at_ms",
    "session_id",
];

/// `cron_run_logs` row → Fire.
pub fn decode_fire(row: &Map<String, Value>) -> Fire {
    let job_id = opt_text(row.get("job_id")).unwrap_or_default();
    let id = opt_text(row.get("run_id")).unwrap_or_else(|| {
        format!(
            "{job_id}#{}",
            row.get("seq").map(|v| v.to_string()).unwrap_or_default()
        )
    });
    let started = iso_millis(ms_of(row.get("run_at_ms").filter(|v| !v.is_null())));
    let finished = iso_millis(ms_of(row.get("ts").filter(|v| !v.is_null())));
    let mut residue = Residue::default();
    residue.keep("__no_claim", Value::Bool(true));
    for (k, v) in row {
        if !FIRE_MAPPED.contains(&k.as_str()) && !v.is_null() {
            residue.keep(k.clone(), v.clone());
        }
    }
    Fire {
        id,
        job_id,
        session_id: opt_text(row.get("session_id")),
        status: run_status(row.get("status").and_then(Value::as_str)),
        claimed_at: started
            .clone()
            .or_else(|| finished.clone())
            .unwrap_or_default(),
        started_at: started,
        finished_at: finished,
        error: opt_text(row.get("error")),
        obligation_id: None,
        residue,
    }
}

/// A fire as a `cron_run_logs` row, over the original when one exists.
pub fn encode_fire_row(
    fire: &Fire,
    raw: Option<&Map<String, Value>>,
    store_key: &str,
) -> Map<String, Value> {
    let residue = &fire.residue.0;
    let mut row = raw
        .cloned()
        .unwrap_or_else(|| empty_row(CRON_RUN_LOG_COLUMNS));
    let get = |m: &Map<String, Value>, k: &str| m.get(k).filter(|v| !v.is_null()).cloned();
    row.insert(
        "store_key".into(),
        get(&row, "store_key")
            .or_else(|| residue.get("store_key").cloned())
            .unwrap_or(Value::String(store_key.into())),
    );
    row.insert("job_id".into(), Value::String(fire.job_id.clone()));
    row.insert(
        "seq".into(),
        get(&row, "seq")
            .or_else(|| residue.get("seq").cloned())
            .unwrap_or(Value::Null),
    );
    let ts = ms_from_iso(fire.finished_at.as_deref())
        .map(Value::from)
        .or_else(|| get(&row, "ts"))
        .unwrap_or(Value::from(0));
    row.insert("ts".into(), ts.clone());
    row.insert(
        "status".into(),
        residue
            .get("status")
            .cloned()
            .unwrap_or(Value::String(run_status_back(fire.status).into())),
    );
    row.insert(
        "error".into(),
        fire.error.clone().map(Value::String).unwrap_or(Value::Null),
    );
    row.insert(
        "delivery_status".into(),
        residue
            .get("delivery_status")
            .cloned()
            .unwrap_or(Value::Null),
    );
    row.insert(
        "delivery_error".into(),
        residue
            .get("delivery_error")
            .cloned()
            .unwrap_or(Value::Null),
    );
    row.insert(
        "delivered".into(),
        residue.get("delivered").cloned().unwrap_or(Value::Null),
    );
    row.insert(
        "session_id".into(),
        fire.session_id
            .clone()
            .map(Value::String)
            .unwrap_or(Value::Null),
    );
    row.insert(
        "session_key".into(),
        residue.get("session_key").cloned().unwrap_or(Value::Null),
    );
    row.insert(
        "run_id".into(),
        if fire.id.contains('#') {
            Value::Null
        } else {
            Value::String(fire.id.clone())
        },
    );
    row.insert(
        "run_at_ms".into(),
        ms_from_iso(fire.started_at.as_deref())
            .map(Value::from)
            .unwrap_or(Value::Null),
    );
    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())));
    row.insert(
        "created_at".into(),
        residue.get("created_at").cloned().unwrap_or(ts),
    );
    for c in CRON_RUN_LOG_COLUMNS {
        if !row.contains_key(*c) {
            row.insert(
                c.to_string(),
                residue.get(*c).cloned().unwrap_or(Value::Null),
            );
        }
    }
    row
}

// ---------------------------------------------------------------- obligations

fn obl_state(native: &str) -> ObligationState {
    match native {
        "pending" | "queued" | "sending" | "in_flight" | "retrying" => ObligationState::Pending,
        "delivered" | "sent" => ObligationState::Sent,
        "failed" => ObligationState::Failed,
        "dropped" | "dead" | "cancelled" | "canceled" => ObligationState::Dropped,
        _ => ObligationState::Pending,
    }
}

const OBL_MAPPED: &[&str] = &[
    "id",
    "status",
    "session_key",
    "channel",
    "target",
    "retry_count",
    "last_error",
    "enqueued_at",
    "updated_at",
];

/// `delivery_queue_entries` row → Obligation.
pub fn decode_obligation(row: &Map<String, Value>) -> Obligation {
    let parsed = row
        .get("session_key")
        .and_then(Value::as_str)
        .and_then(parse_openclaw_session_key);
    let native = row
        .get("status")
        .and_then(Value::as_str)
        .map(|s| s.to_ascii_lowercase())
        .unwrap_or_default();
    let state = obl_state(&native);
    let mut content = OutboundContent {
        text: String::new(),
        attachments: None,
        reply_to: None,
        format: None,
    };
    let mut residue = Residue::default();
    residue.keep("status", row.get("status").cloned().unwrap_or(Value::Null));
    if let Some(entry) = row
        .get("entry_json")
        .and_then(Value::as_str)
        .and_then(|s| serde_json::from_str::<Value>(s).ok())
        .and_then(|v| v.as_object().cloned())
    {
        if let Some(t) = ["text", "message", "content", "body"]
            .iter()
            .find_map(|k| entry.get(*k).and_then(Value::as_str))
        {
            content.text = t.into();
        }
    }
    for (k, v) in row {
        if !OBL_MAPPED.contains(&k.as_str()) && !v.is_null() {
            residue.keep(k.clone(), v.clone());
        }
    }
    let text = |k: &str| {
        row.get(k).and_then(|v| match v {
            Value::String(s) => Some(s.clone()),
            Value::Number(n) => Some(n.to_string()),
            _ => None,
        })
    };
    let created_at = text("enqueued_at").unwrap_or_else(|| "0".into());
    let updated_at = text("updated_at").unwrap_or_else(|| created_at.clone());
    Obligation {
        id: opt_text(row.get("id")).unwrap_or_default(),
        target: SurfaceKey {
            key: None,
            platform: Some(
                text("channel")
                    .filter(|s| !s.is_empty())
                    .or_else(|| parsed.as_ref().and_then(|p| p.key.platform.clone()))
                    .unwrap_or_default(),
            ),
            kind: parsed.as_ref().and_then(|p| p.key.kind.clone()),
            chat_id: Some(text("target").unwrap_or_default()),
            thread_id: parsed.as_ref().and_then(|p| p.key.thread_id.clone()),
            participant_id: None,
        },
        session_key: row
            .get("session_key")
            .and_then(Value::as_str)
            .map(str::to_string),
        content,
        state,
        attempts: ms_of(row.get("retry_count")).unwrap_or(0).max(0) as u64,
        last_error: row
            .get("last_error")
            .and_then(Value::as_str)
            .map(str::to_string),
        delivered_at: if state == ObligationState::Sent {
            iso_millis(
                ms_of(row.get("updated_at").filter(|v| !v.is_null()))
                    .or_else(|| ms_of(row.get("enqueued_at"))),
            )
        } else {
            None
        },
        created_at,
        updated_at,
        posted: None,
        source: match parsed {
            Some(p) => ObligationSource::Turn { key: Some(p.key) },
            None => ObligationSource::Turn { key: None },
        },
        residue,
    }
}

/// An obligation as a `delivery_queue_entries` row, over the original when one exists.
pub fn encode_obligation_row(
    o: &Obligation,
    raw: Option<&Map<String, Value>>,
) -> Map<String, Value> {
    let residue = &o.residue.0;
    let mut row = raw
        .cloned()
        .unwrap_or_else(|| empty_row(DELIVERY_QUEUE_COLUMNS));
    let get = |m: &Map<String, Value>, k: &str| m.get(k).filter(|v| !v.is_null()).cloned();
    row.insert(
        "queue_name".into(),
        get(&row, "queue_name")
            .or_else(|| residue.get("queue_name").cloned())
            .unwrap_or(Value::String("default".into())),
    );
    row.insert("id".into(), Value::String(o.id.clone()));
    row.insert(
        "status".into(),
        residue
            .get("status")
            .filter(|v| !v.is_null())
            .cloned()
            .unwrap_or(Value::String(o.state.hermes_word().into())),
    );
    row.insert(
        "session_key".into(),
        o.session_key
            .clone()
            .map(Value::String)
            .unwrap_or(Value::Null),
    );
    row.insert(
        "channel".into(),
        o.target
            .platform
            .clone()
            .filter(|p| !p.is_empty())
            .map(Value::String)
            .unwrap_or(Value::Null),
    );
    row.insert(
        "target".into(),
        o.target
            .chat_id
            .clone()
            .filter(|c| !c.is_empty())
            .map(Value::String)
            .unwrap_or(Value::Null),
    );
    row.insert(
        "last_error".into(),
        o.last_error
            .clone()
            .map(Value::String)
            .unwrap_or(Value::Null),
    );
    let enq = o.created_at.parse::<f64>().map(|f| f as i64).unwrap_or(0);
    row.insert("enqueued_at".into(), Value::from(enq));
    row.insert(
        "updated_at".into(),
        Value::from(o.updated_at.parse::<f64>().map(|f| f as i64).unwrap_or(enq)),
    );
    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())));
    for c in DELIVERY_QUEUE_COLUMNS {
        if row.get(*c).is_none_or(Value::is_null) {
            row.insert(
                c.to_string(),
                residue.get(*c).cloned().unwrap_or(Value::Null),
            );
        }
    }
    row.insert("retry_count".into(), Value::from(o.attempts));
    row
}

// ---------------------------------------------------------------- bindings

/// One `current_conversation_bindings` row as a [`Binding`]: the conversation
/// ref is the surface (a thread's parent is the chat, the thread is the
/// thread), the target session is the worker, `bound_at` is when it started,
/// and a binding that is not `active` ended when the row last changed. The
/// row's own facts (binding id and key, account, target kind, status, expiry,
/// metadata, the full record) ride as residue.
fn decode_conversation_binding(row: &Map<String, Value>) -> Option<Binding> {
    let text = |k: &str| opt_text(row.get(k));
    let channel = text("channel")?;
    let conversation_id = text("conversation_id")?;
    let kind = text("conversation_kind");
    let parent = text("parent_conversation_id");
    let (chat_id, thread_id) = if kind.as_deref() == Some("thread") && parent.is_some() {
        (parent.clone(), Some(conversation_id.clone()))
    } else {
        (Some(conversation_id.clone()), None)
    };
    let status = text("status");
    let started_at = iso_millis(ms_of(row.get("bound_at").filter(|v| !v.is_null())));
    let updated_at = iso_millis(ms_of(row.get("updated_at").filter(|v| !v.is_null())));
    let mut residue = Residue::default();
    residue.keep("__observed", Value::Bool(true));
    for k in [
        "binding_id",
        "binding_key",
        "account_id",
        "target_kind",
        "target_session_key",
        "status",
        "expires_at",
    ] {
        if let Some(v) = row.get(k).filter(|v| !v.is_null()) {
            residue.keep(k.to_string(), v.clone());
        }
    }
    for k in ["metadata_json", "record_json"] {
        if let Some(t) = text(k) {
            let parsed = serde_json::from_str::<Value>(&t).unwrap_or(Value::String(t));
            residue.keep(k.trim_end_matches("_json").to_string(), parsed);
        }
    }
    if let Some(agent) = text("target_agent_id") {
        residue.keep("agent_id", Value::String(agent));
    }
    Some(Binding {
        key: SurfaceKey {
            key: text("target_session_key"),
            platform: Some(channel),
            kind,
            chat_id,
            thread_id,
            participant_id: None,
        },
        profile: None,
        worker: Worker {
            harness: HarnessId::new(HarnessId::OPENCLAW),
            session_id: text("target_session_id"),
            locator: None,
        },
        trigger: Trigger::Channel,
        recurrence: None,
        handoff: None,
        started_at: started_at.clone(),
        last_activity_at: updated_at.clone().or(started_at),
        ended_at: if status.as_deref().is_some_and(|s| s != "active") {
            updated_at
        } else {
            None
        },
        end_reason: None,
        residue,
    })
}

fn bindings_for_agent(agent_dir: &Path, agent_id: &str) -> Result<Vec<Binding>> {
    let mut out = Vec::new();
    let sessions = agent_dir.join("sessions");
    if !sessions.is_dir() {
        return Ok(out);
    }
    let mut entries: Vec<_> = fs::read_dir(&sessions)?.flatten().collect();
    entries.sort_by_key(|e| e.file_name());
    for entry in entries {
        let name = entry.file_name().to_string_lossy().into_owned();
        if !name.ends_with(".jsonl") || name.ends_with(".trajectory.jsonl") {
            continue;
        }
        let path = entry.path();
        let Ok(text) = fs::read_to_string(&path) else {
            continue;
        };
        let lines: Vec<&str> = text.lines().filter(|l| !l.trim().is_empty()).collect();
        let Some(first) = lines.first() else { continue };
        let Ok(header) = serde_json::from_str::<Value>(first) else {
            continue;
        };
        if header.get("type").and_then(Value::as_str) != Some("session") {
            continue;
        }
        let Some(key) = header
            .get("sessionKey")
            .or_else(|| header.get("__openclaw").and_then(|o| o.get("sessionKey")))
            .and_then(Value::as_str)
        else {
            continue;
        };
        let Some(parsed) = parse_openclaw_session_key(key) else {
            continue;
        };
        let header_ts = header
            .get("timestamp")
            .and_then(Value::as_str)
            .map(str::to_string);
        let mut last = header_ts.clone();
        for line in lines.iter().rev() {
            if let Ok(rec) = serde_json::from_str::<Value>(line) {
                if let Some(ts) = rec.get("timestamp").and_then(Value::as_str) {
                    last = Some(ts.into());
                    break;
                }
            }
        }
        let mut residue = Residue(parsed.residue.into_iter().collect());
        residue.keep(
            "agent_id",
            Value::String(parsed.agent.clone().unwrap_or_else(|| agent_id.into())),
        );
        out.push(Binding {
            trigger: if parsed.recurrence.is_some() {
                Trigger::Cron
            } else {
                Trigger::Channel
            },
            key: parsed.key,
            profile: None,
            worker: Worker {
                harness: HarnessId::new(HarnessId::OPENCLAW),
                session_id: Some(
                    header
                        .get("id")
                        .and_then(Value::as_str)
                        .map(str::to_string)
                        .unwrap_or_else(|| name.trim_end_matches(".jsonl").into()),
                ),
                locator: Some(path.display().to_string()),
            },
            recurrence: parsed.recurrence,
            handoff: None,
            started_at: header_ts.clone().or_else(|| last.clone()),
            last_activity_at: last.or(header_ts),
            ended_at: None,
            end_reason: None,
            residue,
        });
    }
    Ok(out)
}

fn list_unmodeled(state_dir: &Path) -> Result<Vec<String>> {
    let mut out = Vec::new();
    fn walk(base: &Path, dir: &Path, out: &mut Vec<String>) -> Result<()> {
        let mut entries: Vec<_> = fs::read_dir(dir)?.flatten().collect();
        entries.sort_by_key(|e| e.file_name());
        for entry in entries {
            let name = entry.file_name().to_string_lossy().into_owned();
            if name == "node_modules" || name == ".git" {
                continue;
            }
            let p = entry.path();
            let rel = p
                .strip_prefix(base)
                .unwrap_or(&p)
                .to_string_lossy()
                .replace('\\', "/");
            let Ok(st) = fs::symlink_metadata(&p) else {
                continue;
            };
            if st.is_dir() {
                walk(base, &p, out)?;
                continue;
            }
            if rel == OPENCLAW_CONFIG
                || rel == OPENCLAW_STATE_DB
                || rel.starts_with(&format!("{OPENCLAW_STATE_DB}-"))
                || rel == "cron/jobs.json"
            {
                continue;
            }
            out.push(rel);
        }
        Ok(())
    }
    walk(state_dir, state_dir, &mut out)?;
    Ok(out)
}

// ---------------------------------------------------------------- from

fn config_record(orchestration: &Orchestration) -> Value {
    let root = &orchestration.profiles["default"];
    serde_json::json!({
        "channels": root.channels, "routes": root.routes, "residue": root.residue.config,
        "profiles": orchestration.profiles.iter().map(|(n, p)| (n.clone(), serde_json::json!({"agent": p.residue.config.get("openclaw_agent"), "subscriptions": p.subscriptions}))).collect::<BTreeMap<_, _>>(),
    })
}

/// The rows the sqlite store holds for a profile; a legacy file's jobs are
/// the file's rows, not the store's.
fn store_record(profile: &Profile) -> Value {
    let jobs: BTreeMap<&String, &Job> = profile
        .jobs
        .iter()
        .filter(|(_, job)| !is_legacy_file_job(job))
        .collect();
    serde_json::json!({ "jobs": jobs, "fires": profile.fires, "obligations": profile.obligations })
}

/// Compile an OpenClaw state directory into the orchestration.
pub fn from_openclaw(state_dir: &Path) -> Result<OpenclawLoaded> {
    if !state_dir.is_dir() {
        return Err(load_error(
            &state_dir.display().to_string(),
            "",
            "not a directory",
        ));
    }
    let mut vault = BTreeMap::new();
    let config_path = state_dir.join(OPENCLAW_CONFIG);
    let config_text = if config_path.is_file() {
        Some(fs::read_to_string(&config_path)?)
    } else {
        None
    };
    let config = match &config_text {
        Some(t) => parse_json5(&config_path.display().to_string(), t)?,
        None => Value::Object(Map::new()),
    };
    if !config.is_object() {
        return Err(load_error(
            &config_path.display().to_string(),
            "",
            "expected a JSON5 object",
        ));
    }
    let entries = agent_entries(&config);
    let default_id = default_agent_id(&config);
    let declared: Vec<(String, Value)> = if entries.is_empty() {
        vec![(default_id.clone(), Value::Object(Map::new()))]
    } else {
        entries
    };
    let name_for = |agent_id: &str| -> String {
        if agent_id == default_id {
            "default".into()
        } else {
            agent_id.into()
        }
    };

    let mut profiles: BTreeMap<String, Profile> = BTreeMap::new();
    let mut ios: BTreeMap<String, OpenclawProfileIo> = BTreeMap::new();
    for (id, entry) in &declared {
        let name = name_for(id);
        let dir = state_dir.join("agents").join(id);
        let mut profile = empty_profile(&name, &dir);
        profile.residue.config.insert(
            "openclaw_agent".into(),
            redact_secrets(entry, &format!("agent_{id}"), &mut vault),
        );
        if let Ok(text) = fs::read_to_string(dir.join("AGENTS.md")) {
            profile.persona = Some(persona_ref(&text));
        }
        for b in bindings_for_agent(&dir, id)? {
            profile.bindings.insert(surface_key_string(&b.key), b);
        }
        profiles.insert(name.clone(), profile);
        ios.insert(
            name,
            OpenclawProfileIo {
                agent_id: id.clone(),
                source_dir: dir,
                ..Default::default()
            },
        );
    }
    // install-wide config: channels, bindings, and everything else verbatim
    let channels = decode_channels(&config, &mut vault);
    let routes = decode_routes(&config, &name_for);
    let mut rest = Map::new();
    for (k, v) in config.as_object().unwrap() {
        if !["channels", "bindings", "agents", "hooks"].contains(&k.as_str()) {
            rest.insert(k.clone(), v.clone());
        }
    }
    let mut agents_rest = Map::new();
    if let Some(a) = config.get("agents").and_then(Value::as_object) {
        for (k, v) in a {
            if k != "list" && k != "entries" {
                agents_rest.insert(k.clone(), v.clone());
            }
        }
    }
    let hooks = decode_hooks(&config, &mut vault, &name_for, &mut profiles);
    {
        let root = profiles.get_mut("default").unwrap();
        root.channels = channels;
        root.routes = routes;
        root.residue.config.insert("openclaw".into(), serde_json::json!({
            "default_agent": default_id,
            "agents_form": agents_form(&config),
            "agents_rest": redact_secrets(&Value::Object(agents_rest), "agents", &mut vault),
            "hooks": hooks.as_ref().map(|h| serde_json::json!({"block": h.block, "has_token": h.has_token})),
            "rest": redact_secrets(&Value::Object(rest), "config", &mut vault),
        }));
    }
    let mut root_io = OpenclawRootIo {
        state_dir: state_dir.to_path_buf(),
        config_raw: config_text.clone().unwrap_or_default(),
        config_present: config_text.is_some(),
        default_agent: default_id.clone(),
        ..Default::default()
    };

    // the shared state DB
    let db_path = state_dir.join(OPENCLAW_STATE_DB);
    let mut store_key = state_dir.join("cron/jobs.json").display().to_string();
    let mut job_owner: BTreeMap<String, String> = BTreeMap::new();
    if table_exists(&db_path, "cron_jobs") {
        for row in read_rows(
            &db_path,
            "select * from cron_jobs order by sort_order, created_at_ms, job_id",
            &[],
        )?
        .unwrap_or_default()
        {
            let job_id = opt_text(row.get("job_id")).unwrap_or_default();
            if let Some(k) = row
                .get("store_key")
                .and_then(Value::as_str)
                .filter(|s| !s.is_empty())
            {
                store_key = k.into();
            }
            let record = json_object_of(row.get("job_json"));
            let agent_id = record
                .get("agentId")
                .or_else(|| row.get("agent_id"))
                .or_else(|| row.get("owner_agent_id"))
                .and_then(Value::as_str)
                .unwrap_or(&default_id)
                .to_string();
            let owner = if profiles.contains_key(&name_for(&agent_id)) {
                name_for(&agent_id)
            } else {
                "default".into()
            };
            let job = decode_job(&row);
            root_io.cron_jobs.insert(job_id.clone(), row);
            profiles
                .get_mut(&owner)
                .unwrap()
                .jobs
                .insert(job.id.clone(), job);
            job_owner.insert(job_id, owner);
        }
    }
    // the legacy file beside the store: jobs the pin's migration has not
    // moved yet, read by the vocabulary that file was written in; a store
    // row with the same id wins
    let legacy_path = state_dir.join("cron/jobs.json");
    if let Ok(text) = fs::read_to_string(&legacy_path) {
        let (records, object_form) = legacy_job_records(&text);
        root_io.legacy_jobs_raw = Some(text);
        root_io.legacy_jobs_object_form = object_form;
        for record in records {
            let Some(id) = legacy_job_id(&record) else {
                continue;
            };
            if profiles.values().any(|p| p.jobs.contains_key(&id)) {
                continue;
            }
            let agent_id = record
                .get("agentId")
                .or_else(|| record.get("agent_id"))
                .and_then(Value::as_str)
                .unwrap_or(&default_id)
                .to_string();
            let owner = if profiles.contains_key(&name_for(&agent_id)) {
                name_for(&agent_id)
            } else {
                "default".into()
            };
            let job = decode_legacy_job(&record);
            root_io.legacy_jobs.insert(id.clone(), record);
            profiles
                .get_mut(&owner)
                .unwrap()
                .jobs
                .insert(id.clone(), job);
            job_owner.insert(id, owner);
        }
    }
    if table_exists(&db_path, "cron_run_logs") {
        for row in read_rows(
            &db_path,
            "select * from cron_run_logs order by ts, seq",
            &[],
        )?
        .unwrap_or_default()
        {
            let fire = decode_fire(&row);
            root_io.cron_run_logs.insert(fire.id.clone(), row);
            let owner = job_owner
                .get(&fire.job_id)
                .cloned()
                .unwrap_or_else(|| "default".into());
            profiles.get_mut(&owner).unwrap().fires.push(fire);
        }
    }
    if table_exists(&db_path, "delivery_queue_entries") {
        for row in read_rows(
            &db_path,
            "select * from delivery_queue_entries order by enqueued_at, id",
            &[],
        )?
        .unwrap_or_default()
        {
            let o = decode_obligation(&row);
            root_io.delivery_queue_entries.insert(o.id.clone(), row);
            let owner = o
                .session_key
                .as_deref()
                .and_then(parse_openclaw_session_key)
                .and_then(|p| p.agent)
                .map(|a| name_for(&a))
                .filter(|n| profiles.contains_key(n))
                .unwrap_or_else(|| "default".into());
            profiles.get_mut(&owner).unwrap().obligations.push(o);
        }
    }
    // the store's own binding record (the pin's `current_conversation_bindings`,
    // ORC-12 finding 7): which agent and session a conversation is bound to,
    // observed rather than inferred from transcript headers. A row replaces
    // the inferred binding on the same surface; a store without the table
    // keeps the inferred ones. Read, never written back (UNI-22).
    if table_exists(&db_path, "current_conversation_bindings") {
        for row in read_rows(
            &db_path,
            "select * from current_conversation_bindings order by bound_at, binding_key",
            &[],
        )?
        .unwrap_or_default()
        {
            let Some(b) = decode_conversation_binding(&row) else {
                continue;
            };
            let agent_id =
                opt_text(row.get("target_agent_id")).unwrap_or_else(|| default_id.clone());
            let owner = if profiles.contains_key(&name_for(&agent_id)) {
                name_for(&agent_id)
            } else {
                "default".into()
            };
            profiles
                .get_mut(&owner)
                .unwrap()
                .bindings
                .insert(surface_key_string(&b.key), b);
        }
    }
    if table_exists(&db_path, "schema_meta") {
        root_io.schema_meta =
            read_rows(&db_path, "select * from schema_meta order by meta_key", &[])?
                .unwrap_or_default();
    }
    root_io.store_key = store_key;
    root_io.db_present = db_path.exists();

    // a fire's delivery OUTCOME rides on its own run-log row; the queue entry on
    // the same conversation key is the obligation that outcome is about
    let mut fire_by_key: BTreeMap<String, (String, String, Option<String>)> = BTreeMap::new();
    for (name, profile) in &profiles {
        for fire in &profile.fires {
            let Some(key) = fire.residue.0.get("session_key").and_then(Value::as_str) else {
                continue;
            };
            let finished = fire.finished_at.clone();
            match fire_by_key.get(key) {
                Some((_, _, held))
                    if held.clone().unwrap_or_default() > finished.clone().unwrap_or_default() => {}
                _ => {
                    fire_by_key.insert(key.into(), (name.clone(), fire.id.clone(), finished));
                }
            }
        }
    }
    let mut links: Vec<(String, String, String)> = Vec::new(); // (profile, fire id, obligation id)
    for profile in profiles.values_mut() {
        for o in &mut profile.obligations {
            let Some((owner, fire_id, _)) =
                o.session_key.as_deref().and_then(|k| fire_by_key.get(k))
            else {
                continue;
            };
            o.source = ObligationSource::Fire {
                fire_id: fire_id.clone(),
            };
            links.push((owner.clone(), fire_id.clone(), o.id.clone()));
        }
    }
    for (owner, fire_id, obligation_id) in links {
        if let Some(fire) = profiles
            .get_mut(&owner)
            .and_then(|p| p.fires.iter_mut().find(|f| f.id == fire_id))
        {
            fire.obligation_id = Some(obligation_id);
        }
    }

    let mut orchestration = Orchestration {
        root: state_dir.to_path_buf(),
        profiles,
        workflow: crate::orchestration::hermes_instance(),
    };
    root_io.config_snapshot = canonical_json(&config_record(&orchestration));
    for (name, profile) in orchestration.profiles.iter_mut() {
        let io = ios.get_mut(name).unwrap();
        io.store_snapshot = canonical_json(&store_record(profile));
        io.bindings_snapshot = canonical_json(&serde_json::to_value(&profile.bindings).unwrap());
        if name != "default" {
            profile.residue.files = Vec::new();
        }
    }
    orchestration
        .profiles
        .get_mut("default")
        .unwrap()
        .residue
        .files = list_unmodeled(state_dir)?;
    Ok(OpenclawLoaded {
        orchestration,
        vault,
        root: root_io,
        profiles: ios,
    })
}

// ---------------------------------------------------------------- to

/// What an OpenClaw decompile did.
#[derive(Debug, Clone, Default)]
pub struct OpenclawReport {
    /// Every artifact written, with its tier.
    pub written: Vec<ArtifactFidelity>,
    /// Every write refused, with the gate named.
    pub refused: Vec<super::hermes::Refusal>,
    /// What a semantic write gave up.
    pub notes: Vec<String>,
    /// Store rows written back column for column vs re-encoded.
    pub rows_byte: usize,
    /// Store rows re-encoded.
    pub rows_emitted: usize,
}

fn write_atomic(path: &Path, text: &str) -> Result<()> {
    if let Some(parent) = path.parent() {
        fs::create_dir_all(parent)?;
    }
    let tmp = path.with_file_name(format!(
        "{}.tmp-{}",
        path.file_name().unwrap().to_string_lossy(),
        std::process::id()
    ));
    fs::write(&tmp, text)?;
    fs::rename(&tmp, path)?;
    Ok(())
}

fn encode_config(loaded: &OpenclawLoaded) -> Value {
    let orchestration = &loaded.orchestration;
    let root = &orchestration.profiles["default"];
    let own = root
        .residue
        .config
        .get("openclaw")
        .and_then(Value::as_object)
        .cloned()
        .unwrap_or_default();
    let default_id = own
        .get("default_agent")
        .and_then(Value::as_str)
        .map(str::to_string)
        .unwrap_or_else(|| loaded.root.default_agent.clone());
    let id_for_name = |name: &str| -> String {
        if name == "default" {
            default_id.clone()
        } else {
            loaded
                .profiles
                .get(name)
                .map(|io| io.agent_id.clone())
                .unwrap_or_else(|| name.into())
        }
    };
    let mut pairs: Vec<(String, Value)> = inline_secrets(
        own.get("rest").unwrap_or(&Value::Object(Map::new())),
        &loaded.vault,
    )
    .as_object()
    .map(|m| m.iter().map(|(k, v)| (k.clone(), v.clone())).collect())
    .unwrap_or_default();
    pairs.push((
        "channels".into(),
        ordered_object(encode_channels(root, &loaded.vault)),
    ));
    let agent_blocks: Vec<(String, Vec<(String, Value)>)> = orchestration
        .profiles
        .iter()
        .map(|(name, p)| {
            let id = id_for_name(name);
            let entry = inline_secrets(
                p.residue
                    .config
                    .get("openclaw_agent")
                    .unwrap_or(&Value::Object(Map::new())),
                &loaded.vault,
            );
            let mut block: Vec<(String, Value)> = vec![("id".into(), Value::String(id.clone()))];
            if let Some(m) = entry.as_object() {
                for (k, v) in m {
                    if k != "id" {
                        block.push((k.clone(), v.clone()));
                    }
                }
            }
            (id, block)
        })
        .collect();
    let mut agents: Vec<(String, Value)> = inline_secrets(
        own.get("agents_rest").unwrap_or(&Value::Object(Map::new())),
        &loaded.vault,
    )
    .as_object()
    .map(|m| m.iter().map(|(k, v)| (k.clone(), v.clone())).collect())
    .unwrap_or_default();
    if own.get("agents_form").and_then(Value::as_str) == Some("entries") {
        agents.push((
            "entries".into(),
            ordered_object(
                agent_blocks
                    .into_iter()
                    .map(|(id, block)| {
                        (
                            id,
                            ordered_object(block.into_iter().filter(|(k, _)| k != "id").collect()),
                        )
                    })
                    .collect(),
            ),
        ));
    } else {
        agents.push((
            "list".into(),
            Value::Array(
                agent_blocks
                    .into_iter()
                    .map(|(_, block)| ordered_object(block))
                    .collect(),
            ),
        ));
    }
    pairs.push(("agents".into(), ordered_object(agents)));
    pairs.push((
        "bindings".into(),
        Value::Array(encode_routes(root, &id_for_name)),
    ));
    let hooks_meta = own
        .get("hooks")
        .and_then(Value::as_object)
        .map(|h| HooksMeta {
            block: h
                .get("block")
                .and_then(Value::as_object)
                .cloned()
                .unwrap_or_default(),
            has_token: h.get("has_token").and_then(Value::as_bool).unwrap_or(false),
        });
    if let Some(hooks) = encode_hooks(orchestration, hooks_meta.as_ref(), &loaded.vault) {
        pairs.push(("hooks".into(), hooks));
    }
    ordered_object(pairs)
}

/// Flatten an ordered value back into plain JSON (for ref scans).
fn plain(value: &Value) -> Value {
    if let Some(pairs) = is_ordered_pairs(value) {
        return Value::Object(pairs.into_iter().map(|(k, v)| (k, plain(&v))).collect());
    }
    match value {
        Value::Array(items) => Value::Array(items.iter().map(plain).collect()),
        Value::Object(m) => Value::Object(m.iter().map(|(k, v)| (k.clone(), plain(v))).collect()),
        other => other.clone(),
    }
}

fn resequence(rows: &mut [Map<String, Value>]) {
    let key = |r: &Map<String, Value>| {
        format!(
            "{}\u{0}{}",
            r.get("store_key")
                .map(|v| v.to_string())
                .unwrap_or_default(),
            r.get("job_id").map(|v| v.to_string()).unwrap_or_default()
        )
    };
    let mut taken: BTreeMap<String, Vec<i64>> = BTreeMap::new();
    for row in rows.iter_mut() {
        let Some(seq) = row.get("seq").and_then(Value::as_i64) else {
            continue;
        };
        let seen = taken.entry(key(row)).or_default();
        if seen.contains(&seq) {
            row.insert("seq".into(), Value::Null);
            continue;
        }
        seen.push(seq);
    }
    for row in rows.iter_mut() {
        if row.get("seq").is_some_and(|v| !v.is_null()) {
            continue;
        }
        let seen = taken.entry(key(row)).or_default();
        let mut next = 1;
        while seen.contains(&next) {
            next += 1;
        }
        row.insert("seq".into(), Value::from(next));
        seen.push(next);
    }
}

fn insert_for(table: &str, columns: &[&str]) -> String {
    format!(
        "insert into {table} ({}) values ({})",
        columns.join(", "),
        columns.iter().map(|_| "?").collect::<Vec<_>>().join(",")
    )
}

fn params_of(row: &Map<String, Value>, columns: &[&str]) -> Vec<Param> {
    columns
        .iter()
        .map(|c| Param::from(row.get(*c).unwrap_or(&Value::Null)))
        .collect()
}

// ---------------------------------------------------------------- legacy cron/jobs.json

/// The records of a legacy `cron/jobs.json` (a bare array, or `{"jobs": [...]}`).
fn legacy_job_records(text: &str) -> (Vec<Value>, bool) {
    match serde_json::from_str::<Value>(text) {
        Ok(Value::Array(items)) => (items, false),
        Ok(Value::Object(map)) => (
            map.get("jobs")
                .and_then(Value::as_array)
                .cloned()
                .unwrap_or_default(),
            true,
        ),
        _ => (Vec::new(), false),
    }
}

fn legacy_job_id(record: &Value) -> Option<String> {
    ["id", "job_id", "jobId"]
        .iter()
        .find_map(|k| record.get(*k).and_then(Value::as_str))
        .filter(|s| !s.is_empty())
        .map(str::to_string)
}

fn is_legacy_file_job(job: &Job) -> bool {
    job.residue.0.get("__legacy_file") == Some(&Value::Bool(true))
}

/// A legacy file record as a `cron_jobs` row, in the vocabulary the pin's
/// migration reads: a bare cron string / `cron` / `everyMinutes` / `everyMs`
/// / `runAt` schedule becomes the typed `schedule` object, `delivery.kind`
/// is the mode, and the ISO state stamps become the row's millisecond
/// columns. The record is then one [`decode_job`], marked `__legacy_file`.
fn decode_legacy_job(record: &Value) -> Job {
    let mut modern = record.as_object().cloned().unwrap_or_default();
    let text = |v: Option<&Value>| match v {
        Some(Value::String(s)) if !s.is_empty() => Some(s.clone()),
        Some(Value::Number(n)) => Some(n.to_string()),
        _ => None,
    };
    let ms = |v: &Value| match v {
        Value::Number(n) => n.as_i64(),
        Value::String(s) => iso_epoch(s).map(|secs| (secs * 1000.0) as i64),
        _ => None,
    };
    let schedule = match modern.get("schedule") {
        Some(Value::Object(o)) if o.get("kind").is_some() => Some(Value::Object(o.clone())),
        Some(Value::String(expr)) => Some(serde_json::json!({ "kind": "cron", "expr": expr })),
        _ => None,
    }
    .or_else(|| {
        text(modern.get("cron")).map(|expr| serde_json::json!({ "kind": "cron", "expr": expr }))
    })
    .or_else(|| {
        modern
            .get("everyMinutes")
            .and_then(Value::as_f64)
            .map(|m| serde_json::json!({ "kind": "every", "everyMs": (m * 60_000.0) as i64 }))
    })
    .or_else(|| {
        modern
            .get("everyMs")
            .and_then(Value::as_f64)
            .map(|ms| serde_json::json!({ "kind": "every", "everyMs": ms as i64 }))
    })
    .or_else(|| {
        text(modern.get("runAt").or_else(|| modern.get("run_at")))
            .map(|at| serde_json::json!({ "kind": "at", "at": at }))
    });
    for k in ["cron", "everyMinutes", "everyMs", "runAt", "run_at"] {
        modern.remove(k);
    }
    if let Some(schedule) = schedule {
        modern.insert("schedule".into(), schedule);
    }
    match modern.get("delivery").cloned() {
        Some(Value::String(mode)) => {
            modern.insert("delivery".into(), serde_json::json!({ "mode": mode }));
        }
        Some(Value::Object(mut d)) => {
            if !d.contains_key("mode") {
                if let Some(kind) = d.remove("kind").or_else(|| d.remove("type")) {
                    d.insert("mode".into(), kind);
                }
            }
            modern.insert("delivery".into(), Value::Object(d));
        }
        _ => {}
    }
    let mut row = Map::new();
    row.insert(
        "job_id".into(),
        Value::String(legacy_job_id(record).unwrap_or_default()),
    );
    if let Some(name) = modern.get("name").cloned() {
        row.insert("name".into(), name);
    }
    if let Some(enabled) = modern.get("enabled").cloned() {
        row.insert("enabled".into(), enabled);
    }
    for (legacy, column) in [
        ("nextRunAt", "next_run_at_ms"),
        ("next_run_at", "next_run_at_ms"),
        ("lastRunAt", "last_run_at_ms"),
        ("last_run_at", "last_run_at_ms"),
    ] {
        if let Some(v) = modern.remove(legacy) {
            if let Some(at) = ms(&v) {
                row.insert(column.into(), Value::from(at));
            }
        }
    }
    if let Some(status) = modern
        .remove("lastStatus")
        .or_else(|| modern.remove("last_status"))
    {
        row.insert("last_run_status".into(), status);
    }
    if let Some(v) = modern
        .remove("createdAt")
        .or_else(|| modern.remove("created_at"))
    {
        if let Some(at) = ms(&v) {
            modern.insert("createdAtMs".into(), Value::from(at));
        }
    }
    // the column form: the store keeps the record as JSON text
    row.insert(
        "job_json".into(),
        Value::String(serde_json::to_string(&Value::Object(modern)).unwrap()),
    );
    let mut job = decode_job(&row);
    job.residue.keep("__legacy_file", Value::Bool(true));
    job
}

/// The legacy `cron/jobs.json` on the way out: the source bytes when every
/// file job is as it was read, else the file re-emitted from the jobs
/// (`job_json` records, in the file's own form).
fn write_legacy_jobs(
    loaded: &OpenclawLoaded,
    dest: &Path,
    report: &mut OpenclawReport,
) -> Result<()> {
    let jobs: Vec<&Job> = loaded
        .orchestration
        .profiles
        .values()
        .flat_map(|p| p.jobs.values())
        .filter(|j| is_legacy_file_job(j))
        .collect();
    if jobs.is_empty() && loaded.root.legacy_jobs_raw.is_none() {
        return Ok(());
    }
    let target = dest.join("cron/jobs.json");
    if let Some(parent) = target.parent() {
        fs::create_dir_all(parent)?;
    }
    let unchanged = loaded.root.legacy_jobs_raw.is_some()
        && jobs.len() == loaded.root.legacy_jobs.len()
        && jobs.iter().all(|job| {
            loaded.root.legacy_jobs.get(&job.id).is_some_and(|record| {
                canonical_json(&serde_json::to_value(decode_legacy_job(record)).unwrap())
                    == canonical_json(&serde_json::to_value(job).unwrap())
            })
        });
    if unchanged {
        fs::write(
            &target,
            loaded.root.legacy_jobs_raw.as_deref().unwrap_or(""),
        )?;
        report
            .written
            .push(ArtifactFidelity::byte("cron/jobs.json"));
        return Ok(());
    }
    if jobs.is_empty() {
        return Ok(());
    }
    let store_key = target.display().to_string();
    let records: Vec<Value> = jobs
        .iter()
        .map(|job| {
            let raw = loaded
                .root
                .legacy_jobs
                .get(&job.id)
                .and_then(Value::as_object);
            match encode_job_row(job, raw, &store_key).remove("job_json") {
                Some(Value::String(text)) => {
                    serde_json::from_str(&text).unwrap_or(Value::String(text))
                }
                Some(other) => other,
                None => Value::Null,
            }
        })
        .collect();
    let body = if loaded.root.legacy_jobs_object_form {
        serde_json::json!({ "jobs": records })
    } else {
        Value::Array(records)
    };
    fs::write(
        &target,
        format!("{}\n", serde_json::to_string_pretty(&body).unwrap()),
    )?;
    report.written.push(ArtifactFidelity::semantic(
        "cron/jobs.json",
        vec!["re-emitted: a file job changed".into()],
    ));
    Ok(())
}

fn write_store(loaded: &OpenclawLoaded, dest: &Path, report: &mut OpenclawReport) -> Result<()> {
    let store_key = if loaded.root.store_key.is_empty() {
        dest.join("cron/jobs.json").display().to_string()
    } else {
        loaded.root.store_key.clone()
    };
    let mut job_rows = Vec::new();
    let mut fire_rows = Vec::new();
    let mut obligation_rows = Vec::new();
    let (mut byte_rows, mut emitted) = (0usize, 0usize);
    for profile in loaded.orchestration.profiles.values() {
        for job in profile.jobs.values() {
            if is_legacy_file_job(job) {
                continue;
            }
            let original = loaded.root.cron_jobs.get(&job.id);
            let unchanged = original.is_some_and(|o| {
                canonical_json(&serde_json::to_value(decode_job(o)).unwrap())
                    == canonical_json(&serde_json::to_value(job).unwrap())
            });
            if unchanged {
                job_rows.push(original.unwrap().clone());
                byte_rows += 1;
            } else {
                job_rows.push(encode_job_row(job, original, &store_key));
                emitted += 1;
            }
        }
        for fire in &profile.fires {
            let original = loaded.root.cron_run_logs.get(&fire.id);
            let unchanged = original.is_some_and(|o| {
                let mut d = decode_fire(o);
                d.obligation_id = fire.obligation_id.clone();
                canonical_json(&serde_json::to_value(d).unwrap())
                    == canonical_json(&serde_json::to_value(fire).unwrap())
            });
            if unchanged {
                fire_rows.push(original.unwrap().clone());
                byte_rows += 1;
            } else {
                fire_rows.push(encode_fire_row(fire, original, &store_key));
                emitted += 1;
            }
        }
        resequence(&mut fire_rows);
        for o in &profile.obligations {
            let original = loaded.root.delivery_queue_entries.get(&o.id);
            let unchanged = original.is_some_and(|r| {
                canonical_json(&serde_json::to_value(decode_obligation(r)).unwrap())
                    == canonical_json(&serde_json::to_value(o).unwrap())
            });
            if unchanged {
                obligation_rows.push(original.unwrap().clone());
                byte_rows += 1;
            } else {
                obligation_rows.push(encode_obligation_row(o, original));
                emitted += 1;
            }
        }
    }
    let target = dest.join(OPENCLAW_STATE_DB);
    fs::create_dir_all(dest.join("state"))?;
    let tmp = target.with_file_name(format!("openclaw.sqlite.tmp-{}", std::process::id()));
    let _ = fs::remove_file(&tmp);
    let meta: Vec<Map<String, Value>> = if loaded.root.schema_meta.is_empty() {
        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()]
    } else {
        loaded.root.schema_meta.clone()
    };
    write_table(
        &tmp,
        OPENCLAW_DDL,
        &insert_for("schema_meta", SCHEMA_META_COLUMNS),
        &meta
            .iter()
            .map(|r| params_of(r, SCHEMA_META_COLUMNS))
            .collect::<Vec<_>>(),
    )?;
    write_table(
        &tmp,
        "",
        &insert_for("cron_jobs", CRON_JOB_COLUMNS),
        &job_rows
            .iter()
            .map(|r| params_of(r, CRON_JOB_COLUMNS))
            .collect::<Vec<_>>(),
    )?;
    write_table(
        &tmp,
        "",
        &insert_for("cron_run_logs", CRON_RUN_LOG_COLUMNS),
        &fire_rows
            .iter()
            .map(|r| params_of(r, CRON_RUN_LOG_COLUMNS))
            .collect::<Vec<_>>(),
    )?;
    write_table(
        &tmp,
        "",
        &insert_for("delivery_queue_entries", DELIVERY_QUEUE_COLUMNS),
        &obligation_rows
            .iter()
            .map(|r| params_of(r, DELIVERY_QUEUE_COLUMNS))
            .collect::<Vec<_>>(),
    )?;
    fs::rename(&tmp, &target)?;
    report.written.push(ArtifactFidelity {
        path: OPENCLAW_STATE_DB.into(),
        fidelity: if emitted == 0 {
            Fidelity::ByteLossless
        } else {
            Fidelity::Semantic
        },
        loss: Vec::new(),
    });
    report.rows_byte = byte_rows;
    report.rows_emitted = emitted;
    Ok(())
}

fn copy_unmodeled(
    files: &[String],
    src: &Path,
    into: &Path,
    report: &mut OpenclawReport,
    prefix: &str,
) -> Result<()> {
    for rel in files {
        let from = src.join(rel);
        if !from.exists() {
            continue;
        }
        let to = into.join(rel);
        if let Some(parent) = to.parent() {
            fs::create_dir_all(parent)?;
        }
        fs::copy(&from, &to)?;
        report
            .written
            .push(ArtifactFidelity::byte(format!("{prefix}{rel}")));
    }
    Ok(())
}

/// Write an OpenClaw state directory from the orchestration.
pub fn to_openclaw(loaded: &OpenclawLoaded, dest: &Path) -> Result<OpenclawReport> {
    let mut report = OpenclawReport::default();
    let orchestration = &loaded.orchestration;
    if !orchestration.profiles.contains_key("default") {
        return Err(load_error(
            &dest.display().to_string(),
            "",
            "no `default` profile: an OpenClaw install always has a default agent",
        ));
    }
    fs::create_dir_all(dest)?;

    // openclaw.json
    let cfg_unchanged =
        loaded.root.config_snapshot == canonical_json(&config_record(orchestration));
    if cfg_unchanged && loaded.root.config_present {
        write_atomic(&dest.join(OPENCLAW_CONFIG), &loaded.root.config_raw)?;
        report.written.push(ArtifactFidelity::byte(OPENCLAW_CONFIG));
    } else {
        let encoded = encode_config(loaded);
        let missing = unresolved_refs(&plain(&encoded), "");
        if !missing.is_empty() {
            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(", ")) });
        } else {
            write_atomic(
                &dest.join(OPENCLAW_CONFIG),
                &format!("{}\n", pretty_ordered(&encoded, 0)),
            )?;
            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()]));
            report.notes.push(format!("{OPENCLAW_CONFIG}: re-emitted as JSON; comments, trailing commas and key order are gone"));
        }
    }

    // state/openclaw.sqlite
    let store_unchanged = orchestration.profiles.iter().all(|(n, p)| {
        loaded
            .profiles
            .get(n)
            .map(|io| io.store_snapshot == canonical_json(&store_record(p)))
            .unwrap_or(false)
    });
    let src_db = loaded.root.state_dir.join(OPENCLAW_STATE_DB);
    let has_rows = orchestration
        .profiles
        .values()
        .any(|p| !p.jobs.is_empty() || !p.fires.is_empty() || !p.obligations.is_empty());
    if store_unchanged && loaded.root.db_present && src_db.exists() {
        fs::create_dir_all(dest.join("state"))?;
        fs::copy(&src_db, dest.join(OPENCLAW_STATE_DB))?;
        report
            .written
            .push(ArtifactFidelity::byte(OPENCLAW_STATE_DB));
    } else if has_rows || loaded.root.db_present {
        write_store(loaded, dest, &mut report)?;
    }
    write_legacy_jobs(loaded, dest, &mut report)?;

    // bindings: read, never written back (UNI-22)
    for (name, profile) in &orchestration.profiles {
        if profile.bindings.is_empty() {
            continue;
        }
        let io = loaded.profiles.get(name);
        let snapshot = io
            .map(|io| io.bindings_snapshot.clone())
            .filter(|s| !s.is_empty());
        if snapshot.as_deref()
            == Some(canonical_json(&serde_json::to_value(&profile.bindings).unwrap()).as_str())
        {
            continue;
        }
        let agent = io
            .map(|io| io.agent_id.clone())
            .unwrap_or_else(|| name.clone());
        if snapshot.is_some() {
            // the source transcripts hold what the orchestration does not model; the
            // change has to be written INTO them, which is UNI-18's
            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() });
            continue;
        }
        // our own bindings: an OpenClaw conversation's surface lives in the
        // session KEY inside the transcript header, so each binding becomes a
        // FRESH transcript whose header carries that key (the discovery door
        // reads `sessionKey`); the turns live in the worker's store and are
        // not carried. An existing transcript is never overwritten.
        let sessions_dir = dest.join("agents").join(&agent).join("sessions");
        for (slot, b) in &profile.bindings {
            let id = b.worker.session_id.clone().unwrap_or_else(|| slot.clone());
            let file = sessions_dir.join(format!("{id}.jsonl"));
            let rel = format!("agents/{agent}/sessions/{id}.jsonl");
            if file.exists() {
                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() });
                continue;
            }
            fs::create_dir_all(&sessions_dir)?;
            let header = serde_json::json!({
                "type": "session",
                "version": 3,
                "id": id,
                "timestamp": b.started_at.clone().or_else(|| b.last_activity_at.clone()).unwrap_or_default(),
                "sessionKey": crate::ontology::render_openclaw_session_key(&agent, b),
            });
            fs::write(&file, format!("{header}\n"))?;
            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()]));
        }
    }

    // everything under the state dir the IR does not model
    copy_unmodeled(
        &orchestration.profiles["default"].residue.files,
        &loaded.root.state_dir,
        dest,
        &mut report,
        "",
    )?;
    for (name, profile) in &orchestration.profiles {
        if name == "default" {
            continue;
        }
        let Some(io) = loaded.profiles.get(name) else {
            continue;
        };
        copy_unmodeled(
            &profile.residue.files,
            &io.source_dir,
            &dest.join("agents").join(&io.agent_id),
            &mut report,
            &format!("agents/{}/", io.agent_id),
        )?;
    }
    Ok(report)
}