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, 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::world::{
ChannelConfig, Fire, FireStatus, Job, JobOrigin, Obligation, ObligationSource, ObligationState,
OutboundContent, Profile, Route, RouteMatch, Schedule, Target, WebhookSubscription, World,
};
use crate::Result;
pub const OPENCLAW_CONFIG: &str = "openclaw.json";
pub const OPENCLAW_STATE_DB: &str = "state/openclaw.sqlite";
pub const OPENCLAW_ACCOUNT_KEYS: &[&str] = &[
"accountId",
"account_id",
"account",
"teamId",
"appId",
"userId",
];
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;
"#;
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",
];
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",
];
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",
];
pub const SCHEMA_META_COLUMNS: &[&str] = &[
"meta_key",
"role",
"schema_version",
"agent_id",
"app_version",
"created_at",
"updated_at",
];
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
}
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}")))
}
pub const NON_SECRET_KEYS: &[&str] = &[
"sessionkey",
"session_key",
"storekey",
"store_key",
"metakey",
"meta_key",
"bindingkey",
"binding_key",
"declarationkey",
"declaration_key",
"idempotencykey",
"idempotency_key",
];
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))
}
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()
}
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(), serde_json::json!({"dotenv": 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()),
}
}
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) => {
if map.len() == 1 {
if let Some(Value::String(n)) = map.get("dotenv").or_else(|| map.get("env")) {
if let Some(v) = resolve_ref(n, vault, 0) {
return Value::String(v);
}
return value.clone();
}
}
Value::Object(
map.iter()
.map(|(k, v)| (k.clone(), inline_secrets(v, vault)))
.collect(),
)
}
other => other.clone(),
}
}
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) => {
if map.len() == 1 {
if let Some(Value::String(n)) = map.get("dotenv").or_else(|| map.get("env")) {
return vec![(path.to_string(), n.clone())];
}
}
map.iter()
.flat_map(|(k, v)| {
unresolved_refs(
v,
&if path.is_empty() {
k.clone()
} else {
format!("{path}.{k}")
},
)
})
.collect()
}
_ => Vec::new(),
}
}
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,
)
}
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")
})
}
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")
})
}
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()?
};
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,
}
}
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()
}
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
}
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())
}
#[derive(Debug, Clone, PartialEq)]
pub struct ParsedKey {
pub agent: Option<String>,
pub key: SurfaceKey,
pub residue: Map<String, Value>,
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,
}
}
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,
}
}
#[derive(Debug, Clone, Default)]
pub struct OpenclawRootIo {
pub state_dir: PathBuf,
pub config_raw: String,
pub config_present: bool,
pub config_snapshot: String,
pub cron_jobs: BTreeMap<String, Map<String, Value>>,
pub cron_run_logs: BTreeMap<String, Map<String, Value>>,
pub delivery_queue_entries: BTreeMap<String, Map<String, Value>>,
pub schema_meta: Vec<Map<String, Value>>,
pub store_key: String,
pub db_present: bool,
pub default_agent: String,
}
#[derive(Debug, Clone, Default)]
pub struct OpenclawProfileIo {
pub agent_id: String,
pub source_dir: PathBuf,
pub store_snapshot: String,
pub bindings_snapshot: String,
}
#[derive(Debug, Clone)]
pub struct OpenclawLoaded {
pub world: World,
pub vault: BTreeMap<String, String>,
pub root: OpenclawRootIo,
pub profiles: BTreeMap<String, OpenclawProfileIo>,
}
impl OpenclawLoaded {
pub fn from_world(world: World, vault: BTreeMap<String, String>) -> Self {
let default_agent = world.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 = world
.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: world.root.clone(),
default_agent,
..Default::default()
},
world,
vault,
profiles,
}
}
}
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 is_credential_key(k) => {
let r = credential_ref_name(scope, k);
vault.insert(r.clone(), s.clone());
creds.insert(k.clone(), SecretRef::Dotenv(r));
}
Value::Object(o) if o.len() == 1 => {
if let Some(Value::String(n)) = o.get("dotenv") {
creds.insert(k.clone(), SecretRef::Dotenv(n.clone()));
} else if let Some(Value::String(n)) = o.get("env") {
creds.insert(k.clone(), SecretRef::Env(n.clone()));
}
}
_ => {}
}
}
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
}
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()
}
#[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(
world: &World,
meta: Option<&HooksMeta>,
vault: &BTreeMap<String, String>,
) -> Option<Value> {
let mut subs: Vec<&WebhookSubscription> = world
.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))
}
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,
})
}
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)
.unwrap_or("UTC")
.into(),
}
} 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: "UTC".into(),
}
};
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);
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,
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()
}
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 tz != "UTC" {
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
}
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",
];
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,
}
}
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
}
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",
];
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,
}
}
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
}
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}-"))
{
continue;
}
out.push(rel);
}
Ok(())
}
walk(state_dir, state_dir, &mut out)?;
Ok(out)
}
fn config_record(world: &World) -> Value {
let root = &world.profiles["default"];
serde_json::json!({
"channels": root.channels, "routes": root.routes, "residue": root.residue.config,
"profiles": world.profiles.iter().map(|(n, p)| (n.clone(), serde_json::json!({"agent": p.residue.config.get("openclaw_agent"), "subscriptions": p.subscriptions}))).collect::<BTreeMap<_, _>>(),
})
}
fn store_record(profile: &Profile) -> Value {
serde_json::json!({ "jobs": profile.jobs, "fires": profile.fires, "obligations": profile.obligations })
}
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()
},
);
}
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()
};
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);
}
}
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);
}
}
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();
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(); 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 world = World {
root: state_dir.to_path_buf(),
profiles,
};
root_io.config_snapshot = canonical_json(&config_record(&world));
for (name, profile) in world.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();
}
}
world.profiles.get_mut("default").unwrap().residue.files = list_unmodeled(state_dir)?;
Ok(OpenclawLoaded {
world,
vault,
root: root_io,
profiles: ios,
})
}
#[derive(Debug, Clone, Default)]
pub struct OpenclawReport {
pub written: Vec<ArtifactFidelity>,
pub refused: Vec<super::hermes::Refusal>,
pub notes: Vec<String>,
pub rows_byte: usize,
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 world = &loaded.world;
let root = &world.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)>)> = world
.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(world, hooks_meta.as_ref(), &loaded.vault) {
pairs.push(("hooks".into(), hooks));
}
ordered_object(pairs)
}
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()
}
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.world.profiles.values() {
for job in profile.jobs.values() {
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(())
}
pub fn to_openclaw(loaded: &OpenclawLoaded, dest: &Path) -> Result<OpenclawReport> {
let mut report = OpenclawReport::default();
let world = &loaded.world;
if !world.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)?;
let cfg_unchanged = loaded.root.config_snapshot == canonical_json(&config_record(world));
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"));
}
}
let store_unchanged = world.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 = world
.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)?;
}
for (name, profile) in &world.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 why = if snapshot.is_none() {
"these bindings did not come from an OpenClaw store"
} else {
"bindings changed since import"
};
let agent = io
.map(|io| io.agent_id.clone())
.unwrap_or_else(|| name.clone());
report.refused.push(super::hermes::Refusal { file: format!("agents/{agent}/sessions/*.jsonl"), reason: format!("{why}; an OpenClaw conversation's surface lives in its session KEY inside the transcript header, and writing OpenClaw session stores is behind the UNI-22 stability gate") });
}
copy_unmodeled(
&world.profiles["default"].residue.files,
&loaded.root.state_dir,
dest,
&mut report,
"",
)?;
for (name, profile) in &world.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)
}