use std::fs;
use std::path::Path;
use serde_json::Value;
use super::canonical::canonical_json;
use super::decode::{encode_obligation_row, OBLIGATION_COLUMNS};
use super::folder::{
config_record, copy_unmodeled, encode_config, encode_jobs_file, encode_subscriptions_file,
write_executions, Flavor, LoadedHome, ProfileIo,
};
use super::openclaw::ms_from_iso;
use super::sqlite::{write_table, Param};
use crate::ontology::{
hermes_source_for_binding, render_hermes_session_key, ArtifactFidelity, Binding, Fidelity,
};
use crate::orchestration::Profile;
const HERMES_STATE_V22: &str = include_str!("hermes_state_v22.sql");
const HERMES_STATE_V22_VERSION: i64 = 22;
const SESSION_COLUMNS: &[&str] = &[
"id",
"source",
"user_id",
"session_key",
"chat_id",
"chat_type",
"thread_id",
"expiry_finalized",
"started_at",
"ended_at",
"end_reason",
"handoff_state",
"handoff_platform",
"handoff_error",
"profile_name",
];
fn epoch_seconds(iso: Option<&str>) -> Option<f64> {
ms_from_iso(iso).map(|ms| ms as f64 / 1000.0)
}
fn hermes_session_row(profile: &str, slot: &str, b: &Binding) -> Vec<Param> {
let text = |v: &Option<String>| v.clone().map(Param::Text).unwrap_or(Param::Null);
let hermes_profile = if profile == "default" {
"main"
} else {
profile
};
let started = epoch_seconds(b.started_at.as_deref())
.or_else(|| epoch_seconds(b.last_activity_at.as_deref()))
.unwrap_or(0.0);
vec![
Param::Text(
b.worker
.session_id
.clone()
.unwrap_or_else(|| slot.to_string()),
),
Param::Text(hermes_source_for_binding(b)),
text(&b.key.participant_id),
Param::Text(
b.key
.key
.clone()
.unwrap_or_else(|| render_hermes_session_key(hermes_profile, &b.key)),
),
text(&b.key.chat_id),
text(&b.key.kind),
text(&b.key.thread_id),
Param::Int(i64::from(b.ended_at.is_some())),
Param::Real(started),
epoch_seconds(b.ended_at.as_deref())
.map(Param::Real)
.unwrap_or(Param::Null),
b.end_reason
.map(|r| Param::Text(r.as_str().into()))
.unwrap_or(Param::Null),
text(&b.handoff.as_ref().map(|h| h.state.clone())),
text(&b.handoff.as_ref().and_then(|h| h.to.clone())),
text(&b.handoff.as_ref().and_then(|h| h.error.clone())),
if profile == "default" {
Param::Null
} else {
Param::Text(profile.into())
},
]
}
fn write_fresh_store(profiles: &[(&str, &Profile)], path: &Path) -> Result<()> {
let placeholders = |n: usize| std::iter::repeat_n("?", n).collect::<Vec<_>>().join(",");
write_table(
path,
HERMES_STATE_V22,
"insert into schema_version (version) values (?)",
&[vec![Param::Int(HERMES_STATE_V22_VERSION)]],
)?;
let sessions: Vec<Vec<Param>> = profiles
.iter()
.flat_map(|(name, profile)| {
profile
.bindings
.iter()
.map(move |(slot, b)| hermes_session_row(name, slot, b))
})
.collect();
write_table(
path,
"",
&format!(
"insert into sessions ({}) values ({})",
SESSION_COLUMNS.join(", "),
placeholders(SESSION_COLUMNS.len())
),
&sessions,
)?;
let obligations: Vec<Vec<Param>> = profiles
.iter()
.flat_map(|(_, profile)| profile.obligations.iter())
.map(|o| encode_obligation_row(o).iter().map(Param::from).collect())
.collect();
write_table(
path,
"",
&format!(
"insert into delivery_obligations ({}) values ({})",
OBLIGATION_COLUMNS.join(", "),
placeholders(OBLIGATION_COLUMNS.len())
),
&obligations,
)
}
use crate::Result;
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Refusal {
pub file: String,
pub reason: String,
}
#[derive(Debug, Clone, Default)]
pub struct HermesReport {
pub written: Vec<ArtifactFidelity>,
pub refused: Vec<Refusal>,
}
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(())
}
pub fn from_hermes(home: &Path) -> Result<LoadedHome> {
super::folder::load_home(home, Flavor::Hermes)
}
pub fn to_hermes(loaded: &LoadedHome, dest: &Path, only: Option<&str>) -> Result<HermesReport> {
let mut report = HermesReport::default();
let empty = ProfileIo {
raw: Default::default(),
snapshot: Default::default(),
source_dir: None,
flavor: Flavor::Orchestrator,
jobs_form: None,
routes_at_top: false,
borrowed_from: None,
lenders: Vec::new(),
};
let mut fresh_rows: Vec<(&str, &Profile)> = Vec::new();
for (name, profile) in &loaded.orchestration.profiles {
if only.is_some_and(|o| o != name) {
continue;
}
let dir = if name == "default" {
dest.to_path_buf()
} else {
dest.join("profiles").join(name)
};
fs::create_dir_all(dir.join("cron"))?;
let meta = loaded.io.get(name).unwrap_or(&empty);
let rel = |p: &str| {
if name == "default" {
p.to_string()
} else {
format!("profiles/{name}/{p}")
}
};
let unchanged =
|file: &str, record: &Value| meta.snapshot.get(file) == Some(&canonical_json(record));
fn emit_file(
written: &mut Vec<ArtifactFidelity>,
dir: &Path,
path: String,
file: &str,
text: &str,
tier: Fidelity,
) -> Result<()> {
write_atomic(&dir.join(file), text)?;
written.push(ArtifactFidelity {
path,
fidelity: tier,
loss: Vec::new(),
});
Ok(())
}
macro_rules! emit {
($file:expr, $text:expr, $tier:expr) => {
emit_file(&mut report.written, &dir, rel($file), $file, $text, $tier)?
};
}
let cfg_record = config_record(profile);
let cfg_empty = profile.routes.is_empty()
&& profile.channels.is_empty()
&& profile.residue.config.is_empty()
&& profile.worker.is_none()
&& profile.home.is_none()
&& profile.expiry == Default::default();
let hermes_bytes = meta.flavor == Flavor::Hermes;
if hermes_bytes
&& unchanged("config.yaml", &cfg_record)
&& meta.raw.contains_key("config.yaml")
{
emit!(
"config.yaml",
&meta.raw["config.yaml"],
Fidelity::ByteLossless
);
} else if !cfg_empty || meta.raw.contains_key("config.yaml") {
emit!(
"config.yaml",
&encode_config(profile, Some(meta), Some(&loaded.vault), Flavor::Hermes),
Fidelity::Semantic
);
}
if let Some(persona) = &profile.persona {
let src_name = if meta.raw.contains_key("SOUL.md") {
"SOUL.md"
} else {
"AGENTS.md"
};
let record = serde_json::to_value(&profile.persona).unwrap();
if unchanged(src_name, &record) && meta.raw.contains_key(src_name) {
emit!(
"SOUL.md",
&meta.raw[src_name],
if src_name == "SOUL.md" {
Fidelity::ByteLossless
} else {
Fidelity::Semantic
}
);
} else {
emit!(
"SOUL.md",
persona.text.as_deref().unwrap_or(""),
Fidelity::Semantic
);
}
}
let jobs_record: Vec<Value> = profile
.jobs
.values()
.map(|j| serde_json::to_value(j).unwrap())
.collect();
if !profile.jobs.is_empty() || meta.raw.contains_key("cron/jobs.json") {
if unchanged("cron/jobs.json", &Value::Array(jobs_record))
&& meta.raw.contains_key("cron/jobs.json")
{
emit!(
"cron/jobs.json",
&meta.raw["cron/jobs.json"],
Fidelity::ByteLossless
);
} else {
let mut view: Profile = profile.clone();
for job in view.jobs.values_mut() {
if let Some(w) = &job.workdir {
if !w.starts_with('/') {
job.workdir = Some(dir.join(w).display().to_string());
}
}
}
emit!(
"cron/jobs.json",
&encode_jobs_file(&view, Some(meta)),
Fidelity::ByteLossless
);
}
}
let src_exec = meta
.source_dir
.as_ref()
.map(|d| d.join("cron/executions.db"));
if !profile.fires.is_empty() || src_exec.as_ref().is_some_and(|p| p.exists()) {
let target = dir.join("cron/executions.db");
let fires_record = serde_json::to_value(&profile.fires).unwrap();
if unchanged("cron/executions.db", &fires_record)
&& src_exec.as_ref().is_some_and(|p| p.exists())
{
fs::copy(src_exec.as_ref().unwrap(), &target)?;
} else {
let tmp =
target.with_file_name(format!("executions.db.tmp-{}", std::process::id()));
let _ = fs::remove_file(&tmp);
write_executions(&tmp, &profile.fires)?;
fs::rename(&tmp, &target)?;
}
report
.written
.push(ArtifactFidelity::byte(rel("cron/executions.db")));
}
let subs_record: Vec<Value> = profile
.subscriptions
.values()
.map(|s| serde_json::to_value(s).unwrap())
.collect();
if !profile.subscriptions.is_empty() || meta.raw.contains_key("webhook_subscriptions.json")
{
if hermes_bytes
&& unchanged("webhook_subscriptions.json", &Value::Array(subs_record))
&& meta.raw.contains_key("webhook_subscriptions.json")
{
emit!(
"webhook_subscriptions.json",
&meta.raw["webhook_subscriptions.json"],
Fidelity::ByteLossless
);
} else {
emit!(
"webhook_subscriptions.json",
&encode_subscriptions_file(profile, Some(&loaded.vault)),
Fidelity::ByteLossless
);
}
}
if meta.borrowed_from.is_none() {
let state_record = serde_json::json!({ "bindings": profile.bindings, "obligations": profile.obligations });
let src_state = meta.source_dir.as_ref().map(|d| d.join("state.db"));
let lenders_unchanged = meta.lenders.iter().all(|n| {
let p = &loaded.orchestration.profiles[n];
loaded.io.get(n).and_then(|m| m.snapshot.get("state.db")) == Some(&canonical_json(&serde_json::json!({ "bindings": p.bindings, "obligations": p.obligations })))
});
let dest_store = dir.join("state.db");
if src_state.as_ref().is_some_and(|p| p.exists()) && meta.flavor == Flavor::Hermes {
if unchanged("state.db", &state_record) && lenders_unchanged {
fs::copy(src_state.as_ref().unwrap(), &dest_store)?;
report.written.push(ArtifactFidelity::byte(rel("state.db")));
} else {
report.refused.push(Refusal { file: rel("state.db"), reason: "bindings/obligations changed since import; writing the change into a Hermes session store is UNI-18 (the first shared-WAL write), not this codec's".into() });
}
} else if !profile.bindings.is_empty() || !profile.obligations.is_empty() {
let _ = dest_store;
fresh_rows.push((name.as_str(), profile));
}
}
copy_unmodeled(profile, meta, &dir)?;
for f in &profile.residue.files {
report.written.push(ArtifactFidelity::byte(rel(f)));
}
}
if !fresh_rows.is_empty() {
let root_store = dest.join("state.db");
if root_store.exists() {
report.refused.push(Refusal { file: "state.db".into(), reason: "the destination already holds a Hermes session store; writing into a live store is UNI-18 (the first shared-WAL write), not this codec's".into() });
} else {
write_fresh_store(&fresh_rows, &root_store)?;
let bindings: usize = fresh_rows.iter().map(|(_, p)| p.bindings.len()).sum();
let obligations: usize = fresh_rows.iter().map(|(_, p)| p.obligations.len()).sum();
report.written.push(ArtifactFidelity::semantic(
"state.db",
vec![format!(
"a fresh store at schema {HERMES_STATE_V22_VERSION} (Hermes migrates it on open): {bindings} binding(s) as sessions rows across {} profile(s) partitioned by profile_name, {obligations} obligation(s) as delivery rows; the transcripts live in the worker's store and are not carried",
fresh_rows.len()
)],
));
}
}
Ok(report)
}