use std::collections::{BTreeMap, BTreeSet};
use std::fs;
use std::path::{Path, PathBuf};
use serde_json::{Map, Value};
use sha2::{Digest, Sha256};
use super::canonical::canonical_json;
use super::decode::{
decode_access, decode_binding_row, decode_channel, decode_expiry, decode_fire_row, decode_home,
decode_job, decode_obligation_row, decode_route, decode_subscription, decode_worker,
encode_access, encode_channel, encode_fire_row, encode_job, encode_obligation_row,
encode_route, encode_subscription, encode_surface_key, load_error, surface_key_string,
EXECUTION_COLUMNS, OBLIGATION_COLUMNS,
};
use super::dotenv::{parse_dotenv, render_dotenv};
use super::sqlite::{read_rows, table_exists, write_table, Param};
use crate::ontology::{
hermes_cron_job_id, parse_hermes_session_key, Binding, EndReason, Handoff, HarnessId,
HermesSessionRow, Recurrence, Residue, SurfaceKey, Trigger, Worker,
};
use crate::orchestration::{ExpiryPolicy, Orchestration, PersonaRef, Profile, ProfileResidue};
use crate::Result;
pub const OWNED_FILES: &[&str] = &[
"config.yaml",
"AGENTS.md",
"CLAUDE.md",
"access.yaml",
".env",
"cron/jobs.json",
"cron/executions.db",
"webhook_subscriptions.json",
"state.db",
];
const CONFIG_O_KEYS: &[&str] = &["worker", "expiry", "home"];
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Flavor {
Orchestrator,
Hermes,
}
#[derive(Debug, Clone, PartialEq)]
pub struct JobsForm {
pub object: bool,
pub extras: Map<String, Value>,
}
#[derive(Debug, Clone)]
pub struct ProfileIo {
pub raw: BTreeMap<String, String>,
pub snapshot: BTreeMap<String, String>,
pub source_dir: Option<PathBuf>,
pub flavor: Flavor,
pub jobs_form: Option<JobsForm>,
pub routes_at_top: bool,
pub borrowed_from: Option<PathBuf>,
pub lenders: Vec<String>,
}
impl ProfileIo {
fn new(flavor: Flavor) -> Self {
Self {
raw: BTreeMap::new(),
snapshot: BTreeMap::new(),
source_dir: None,
flavor,
jobs_form: None,
routes_at_top: false,
borrowed_from: None,
lenders: Vec::new(),
}
}
}
#[derive(Debug, Clone)]
pub struct LoadedHome {
pub orchestration: Orchestration,
pub vault: BTreeMap<String, String>,
pub io: BTreeMap<String, ProfileIo>,
}
fn sha256_hex(text: &str) -> String {
let mut h = Sha256::new();
h.update(text.as_bytes());
h.finalize().iter().map(|b| format!("{b:02x}")).collect()
}
pub fn persona_ref(text: &str) -> PersonaRef {
PersonaRef {
path: "AGENTS.md".into(),
text: Some(text.to_string()),
sha256: sha256_hex(text),
}
}
fn read_text(dir: &Path, rel: &str) -> Result<Option<String>> {
let p = dir.join(rel);
if !p.is_file() {
return Ok(None);
}
Ok(Some(fs::read_to_string(&p)?))
}
fn yaml_to_json(file: &str, text: &str) -> Result<Value> {
let value: serde_yaml::Value =
serde_yaml::from_str(text).map_err(|e| load_error(file, "", format!("YAML: {e}")))?;
let json: Value =
serde_json::to_value(value).map_err(|e| load_error(file, "", format!("YAML: {e}")))?;
Ok(if json.is_null() {
Value::Object(Map::new())
} else {
json
})
}
fn json_to_yaml(value: &Value) -> String {
let y: serde_yaml::Value =
serde_json::from_value(value.clone()).unwrap_or(serde_yaml::Value::Null);
match value.as_object() {
Some(m) if m.is_empty() => String::new(),
_ => serde_yaml::to_string(&y).unwrap_or_default(),
}
}
pub fn empty_profile(name: &str, dir: &Path) -> Profile {
Profile {
name: name.to_string(),
dir: dir.to_path_buf(),
worker: None,
persona: None,
channels: BTreeMap::new(),
routes: Vec::new(),
expiry: ExpiryPolicy::default(),
home: None,
jobs: BTreeMap::new(),
subscriptions: BTreeMap::new(),
access: Default::default(),
bindings: BTreeMap::new(),
fires: Vec::new(),
obligations: Vec::new(),
residue: ProfileResidue::default(),
}
}
pub fn config_record(profile: &Profile) -> Value {
serde_json::json!({
"worker": profile.worker, "expiry": profile.expiry, "home": profile.home.as_ref().map(encode_surface_key),
"routes": profile.routes, "channels": profile.channels, "residue": profile.residue.config,
})
}
fn state_record(profile: &Profile) -> Value {
serde_json::json!({ "bindings": profile.bindings, "obligations": profile.obligations })
}
pub fn load_home(root: &Path, flavor: Flavor) -> Result<LoadedHome> {
if !root.is_dir() {
return Err(load_error(
&root.display().to_string(),
"",
"not a directory",
));
}
let mut vault = BTreeMap::new();
let mut io = BTreeMap::new();
let mut profiles = BTreeMap::new();
let (default, default_io) = load_profile_dir("default", root, flavor, &mut vault)?;
profiles.insert("default".to_string(), default);
io.insert("default".to_string(), default_io);
let profiles_dir = root.join("profiles");
if profiles_dir.is_dir() {
let mut names: Vec<String> = fs::read_dir(&profiles_dir)?
.flatten()
.filter(|e| e.path().is_dir())
.filter_map(|e| e.file_name().into_string().ok())
.filter(|n| n != "node_modules" && !n.starts_with('.'))
.collect();
names.sort();
for name in names {
let dir = profiles_dir.join(&name);
if name == "default" {
return Err(load_error(
&dir.display().to_string(),
"",
"\"default\" is the root folder, not a named profile",
));
}
let (profile, meta) = load_profile_dir(&name, &dir, flavor, &mut vault)?;
profiles.insert(name.clone(), profile);
io.insert(name, meta);
}
}
let mut loaded = LoadedHome {
orchestration: Orchestration {
root: root.to_path_buf(),
profiles,
},
vault,
io,
};
if flavor == Flavor::Hermes {
partition_shared_store(&mut loaded, root)?;
}
link_fires(&mut loaded, root)?;
Ok(loaded)
}
fn link_fires(loaded: &mut LoadedHome, root: &Path) -> Result<()> {
let names: Vec<String> = loaded.orchestration.profiles.keys().cloned().collect();
let root_store = root.join("state.db");
for name in &names {
let own = loaded.orchestration.profiles[name].dir.join("state.db");
let store = if own.is_file() {
own
} else {
root_store.clone()
};
let sessions: Vec<Map<String, Value>> = if table_exists(&store, "sessions") {
read_rows(
&store,
"select id, started_at, end_reason, parent_session_id, session_key from sessions",
&[],
)?
.unwrap_or_default()
} else {
Vec::new()
};
let holders: Vec<String> = {
let io = &loaded.io[name];
if io.borrowed_from.is_some() || !io.lenders.is_empty() {
let mut v = vec!["default".to_string()];
v.extend(loaded.io["default"].lenders.iter().cloned());
v
} else {
vec![name.clone()]
}
};
let mut links: Vec<(usize, Option<String>, Option<(String, String)>)> = Vec::new();
let profile = &loaded.orchestration.profiles[name];
for (index, fire) in profile.fires.iter().enumerate() {
let session_id = fire_session(&sessions, fire);
let session_key = session_id.as_deref().and_then(|id| {
sessions
.iter()
.find(|row| row.get("id").and_then(Value::as_str) == Some(id))
.and_then(|row| row.get("session_key"))
.and_then(Value::as_str)
.filter(|k| !k.is_empty())
.map(str::to_string)
});
let surface = profile.jobs.get(&fire.job_id).and_then(job_surface);
let (Some(from), to) = (
iso_epoch(&fire.claimed_at),
fire.finished_at
.as_deref()
.and_then(iso_epoch)
.unwrap_or(f64::MAX),
) else {
links.push((index, session_id, None));
continue;
};
let in_window = |o: &&crate::orchestration::Obligation| {
o.created_at
.parse::<f64>()
.is_ok_and(|at| at >= from && at <= to)
};
let latest = |mut found: Vec<(&String, &crate::orchestration::Obligation)>| {
found.sort_by(|a, b| {
let at = |o: &crate::orchestration::Obligation| {
o.created_at.parse::<f64>().unwrap_or(0.0)
};
at(b.1)
.partial_cmp(&at(a.1))
.unwrap_or(std::cmp::Ordering::Equal)
});
found
.first()
.map(|(holder, o)| ((*holder).clone(), o.id.clone()))
};
let candidates = |pick: &dyn Fn(&crate::orchestration::Obligation) -> bool| {
holders
.iter()
.flat_map(|h| {
loaded.orchestration.profiles[h]
.obligations
.iter()
.filter(in_window)
.filter(|o| pick(o))
.map(move |o| (h, o))
})
.collect::<Vec<_>>()
};
let obligation = match &session_key {
Some(key) => latest(candidates(&|o| o.session_key.as_deref() == Some(key))),
None => None,
}
.or_else(|| {
let (platform, chat_id) = surface.as_ref()?;
latest(candidates(&|o| {
o.target.platform.as_deref() == Some(platform)
&& o.target.chat_id.as_deref() == Some(chat_id)
}))
});
links.push((index, session_id, obligation));
}
for (index, session_id, obligation) in links {
let fire_id = {
let fire = &mut loaded.orchestration.profiles.get_mut(name).unwrap().fires[index];
fire.session_id = session_id;
fire.obligation_id = obligation.as_ref().map(|(_, id)| id.clone());
fire.id.clone()
};
if let Some((holder, obligation_id)) = obligation {
if let Some(o) = loaded
.orchestration
.profiles
.get_mut(&holder)
.and_then(|p| p.obligations.iter_mut().find(|o| o.id == obligation_id))
{
o.source = crate::orchestration::ObligationSource::Fire { fire_id };
}
}
}
}
for name in &names {
let profile = &loaded.orchestration.profiles[name];
let io = loaded.io.get_mut(name).unwrap();
io.snapshot.insert(
"cron/executions.db".into(),
canonical_json(&serde_json::to_value(&profile.fires).unwrap()),
);
io.snapshot
.insert("state.db".into(), canonical_json(&state_record(profile)));
}
Ok(())
}
fn job_surface(job: &crate::orchestration::Job) -> Option<(String, String)> {
use crate::orchestration::Target;
match &job.deliver {
Target::Origin => {
let origin = job.origin.as_ref()?;
Some((origin.platform.clone(), origin.chat_id.clone()?))
}
Target::Explicit {
platform, chat_id, ..
} => Some((platform.clone(), chat_id.clone()?)),
Target::Home | Target::Local => None,
}
}
fn fire_session(
sessions: &[Map<String, Value>],
fire: &crate::orchestration::Fire,
) -> Option<String> {
let claimed = instant_key(&fire.claimed_at)?;
let finished = fire.finished_at.as_deref().and_then(instant_key);
let candidates = sessions.iter().filter_map(|row| {
let id = row.get("id").and_then(Value::as_str)?;
let key = cron_session_instant(id, &fire.job_id)?;
(key >= claimed && finished.is_none_or(|f| key <= f)).then(|| (key, id.to_string()))
});
let chosen = match finished {
Some(_) => candidates.max_by_key(|(key, _)| *key),
None => candidates.min_by_key(|(key, _)| *key),
}?;
let mut current = chosen.1;
for _ in 0..32 {
let row = sessions
.iter()
.find(|row| row.get("id").and_then(Value::as_str) == Some(current.as_str()));
let compressed = row
.and_then(|row| row.get("end_reason"))
.and_then(Value::as_str)
== Some("compression");
if !compressed {
return Some(current);
}
let next = sessions
.iter()
.filter(|row| {
row.get("parent_session_id").and_then(Value::as_str) == Some(current.as_str())
})
.max_by(|a, b| {
let at = |r: &Map<String, Value>| {
r.get("started_at").and_then(Value::as_f64).unwrap_or(0.0)
};
at(a)
.partial_cmp(&at(b))
.unwrap_or(std::cmp::Ordering::Equal)
.then_with(|| {
let id = |r: &Map<String, Value>| {
r.get("id")
.and_then(Value::as_str)
.unwrap_or("")
.to_string()
};
id(a).cmp(&id(b))
})
})
.and_then(|row| row.get("id").and_then(Value::as_str).map(str::to_string));
match next {
None => return Some(current),
Some(next) => current = next,
}
}
Some(current)
}
fn instant_key(iso: &str) -> Option<u64> {
let digits: String = iso
.chars()
.take_while(|c| *c != '+' && *c != 'Z')
.filter(char::is_ascii_digit)
.collect();
(digits.len() >= 14).then(|| digits[..14].parse().ok())?
}
fn cron_session_instant(session_id: &str, job_id: &str) -> Option<u64> {
if hermes_cron_job_id(session_id).as_deref() != Some(job_id) {
return None;
}
let stamp = session_id.rsplit_once('_')?;
let date = stamp.0.rsplit_once('_')?.1;
format!("{date}{}", stamp.1).parse().ok()
}
fn iso_epoch(iso: &str) -> Option<f64> {
let (instant, offset) = if let Some(instant) = iso.strip_suffix('Z') {
(instant, 0.0)
} else {
let time_at = iso.find('T')?;
let sign_at = iso[time_at..].find(['+', '-']).map(|i| i + time_at)?;
let (instant, offset) = iso.split_at(sign_at);
let (hours, minutes) = offset[1..].split_once(':')?;
let seconds = hours.parse::<f64>().ok()? * 3_600.0 + minutes.parse::<f64>().ok()? * 60.0;
(
instant,
if offset.starts_with('-') {
-seconds
} else {
seconds
},
)
};
let (date, time) = instant.split_once('T')?;
let mut date = date.splitn(3, '-');
let year: i64 = date.next()?.parse().ok()?;
let month: i64 = date.next()?.parse().ok()?;
let day: i64 = date.next()?.parse().ok()?;
let mut clock = time.splitn(3, ':');
let hour: i64 = clock.next()?.parse().ok()?;
let minute: i64 = clock.next()?.parse().ok()?;
let seconds: f64 = clock.next()?.parse().ok()?;
let year = year - i64::from(month <= 2);
let era = year.div_euclid(400);
let yoe = year - era * 400;
let doy = (153 * (if month > 2 { month - 3 } else { month + 9 }) + 2) / 5 + day - 1;
let doe = yoe * 365 + yoe / 4 - yoe / 100 + doy;
let days = era * 146_097 + doe - 719_468;
Some((days * 86_400 + hour * 3_600 + minute * 60) as f64 + seconds - offset)
}
fn partition_shared_store(loaded: &mut LoadedHome, root: &Path) -> Result<()> {
let root_path = root.join("state.db");
if !root_path.is_file() {
return Ok(());
}
let names: Vec<String> = loaded
.orchestration
.profiles
.keys()
.filter(|n| *n != "default")
.cloned()
.collect();
for name in names {
let has_own = loaded.orchestration.profiles[&name]
.dir
.join("state.db")
.is_file();
if has_own {
continue;
}
loaded.io.get_mut(&name).unwrap().borrowed_from = Some(root_path.clone());
loaded
.io
.get_mut("default")
.unwrap()
.lenders
.push(name.clone());
if table_exists(&root_path, "sessions") {
let rows = read_rows(
&root_path,
"select * from sessions where profile_name = ?1 order by started_at, id",
&[&name],
)?
.unwrap_or_default();
for row in rows {
if let Some(b) = binding_from_hermes_session(&root_path, &row, &name) {
loaded
.orchestration
.profiles
.get_mut(&name)
.unwrap()
.bindings
.insert(surface_key_string(&b.key), b);
}
}
}
let root_profile = loaded.orchestration.profiles.get_mut("default").unwrap();
let (mine, rest): (Vec<_>, Vec<_>) = root_profile.obligations.drain(..).partition(|o| {
o.session_key
.as_deref()
.and_then(parse_hermes_session_key)
.and_then(|(_, p)| p)
.as_deref()
== Some(name.as_str())
});
root_profile.obligations = rest;
loaded
.orchestration
.profiles
.get_mut(&name)
.unwrap()
.obligations = mine;
let snap = canonical_json(&state_record(&loaded.orchestration.profiles[&name]));
loaded
.io
.get_mut(&name)
.unwrap()
.snapshot
.insert("state.db".into(), snap);
}
let snap = canonical_json(&state_record(&loaded.orchestration.profiles["default"]));
loaded
.io
.get_mut("default")
.unwrap()
.snapshot
.insert("state.db".into(), snap);
Ok(())
}
const HERMES_SESSION_MAPPED: &[&str] = &[
"id",
"source",
"session_key",
"chat_id",
"chat_type",
"thread_id",
"user_id",
"profile_name",
"handoff_state",
"handoff_platform",
"handoff_error",
"started_at",
"ended_at",
"end_reason",
];
pub fn binding_from_hermes_session(
file: &Path,
row: &Map<String, Value>,
profile_name: &str,
) -> Option<Binding> {
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,
})
.filter(|s| !s.is_empty())
};
let num = |k: &str| row.get(k).and_then(Value::as_f64);
let id = text("id")?;
let source = text("source");
let parsed = text("session_key").and_then(|k| parse_hermes_session_key(&k));
let chat_type = text("chat_type").or_else(|| parsed.as_ref().and_then(|(k, _)| k.kind.clone()));
let key = if text("session_key").is_some()
&& chat_type
.as_deref()
.is_some_and(|c| super::decode::CHAT_TYPES.contains(&c))
{
SurfaceKey {
key: None,
platform: source
.clone()
.or_else(|| parsed.as_ref().and_then(|(k, _)| k.platform.clone())),
kind: chat_type,
chat_id: text("chat_id")
.or_else(|| parsed.as_ref().and_then(|(k, _)| k.chat_id.clone())),
thread_id: text("thread_id")
.or_else(|| parsed.as_ref().and_then(|(k, _)| k.thread_id.clone())),
participant_id: parsed.as_ref().and_then(|(k, _)| k.participant_id.clone()),
}
} else if source.as_deref() == Some("cron") {
let job = crate::ontology::hermes_cron_job_id(&id).unwrap_or_else(|| id.clone());
SurfaceKey {
key: None,
platform: Some("cron".into()),
kind: Some("dm".into()),
chat_id: Some(job),
thread_id: None,
participant_id: None,
}
} else {
return None;
};
if let Some(p) = text("profile_name") {
if p != profile_name && !(profile_name == "default" && p == "main") {
return None; }
}
let iso = |v: Option<f64>| v.map(epoch_iso);
let end_word = text("end_reason");
let end_reason = end_word.as_deref().and_then(EndReason::parse);
let mut residue = Residue::default();
for (k, v) in row {
if !HERMES_SESSION_MAPPED.contains(&k.as_str()) && !v.is_null() {
residue.keep(k.clone(), v.clone());
}
}
if let (Some(word), None) = (&end_word, end_reason) {
residue.keep("end_reason", Value::String(word.clone()));
}
if let Some(u) = text("user_id") {
residue.keep("user_id", Value::String(u));
}
let recurrence = if key.platform.as_deref() == Some("cron") {
key.chat_id.clone().map(|job_id| Recurrence {
job_id,
kind: "cron".into(),
})
} else {
None
};
Some(Binding {
trigger: match (recurrence.is_some(), source.as_deref()) {
(true, _) => Trigger::Cron,
(_, Some(s)) => crate::ontology::hermes_trigger_for_source(s),
_ => Trigger::Unknown,
},
key,
profile: None,
worker: Worker {
harness: HarnessId::new(HarnessId::HERMES),
session_id: Some(id),
locator: Some(file.display().to_string()),
},
recurrence,
handoff: text("handoff_state").map(|state| Handoff {
to: text("handoff_platform"),
state,
error: text("handoff_error"),
}),
started_at: iso(num("started_at")),
last_activity_at: iso(num("ended_at").or_else(|| num("started_at"))),
ended_at: iso(num("ended_at")),
end_reason,
residue,
})
}
fn epoch_iso(seconds: f64) -> String {
let row = HermesSessionRow {
started_at: Some(seconds),
..Default::default()
};
Binding::from_hermes_row(&row, None)
.started_at
.unwrap_or_default()
}
fn load_profile_dir(
name: &str,
dir: &Path,
flavor: Flavor,
vault: &mut BTreeMap<String, String>,
) -> Result<(Profile, ProfileIo)> {
let mut profile = empty_profile(name, dir);
let mut config_normalized = false;
let mut subs_normalized = false;
let mut meta = ProfileIo::new(flavor);
meta.source_dir = Some(dir.to_path_buf());
let remember = |meta: &mut ProfileIo, rel: &str, raw: Option<String>, record: &Value| {
if let Some(raw) = raw {
meta.raw.insert(rel.to_string(), raw);
}
meta.snapshot
.insert(rel.to_string(), canonical_json(record));
};
if let Some(env) = read_text(dir, ".env")? {
for (k, v) in parse_dotenv(&env) {
vault.insert(k, v);
}
meta.raw.insert(".env".into(), env);
}
let cfg_file = dir.join("config.yaml").display().to_string();
let cfg_text = read_text(dir, "config.yaml")?;
let cfg = match &cfg_text {
Some(text) => yaml_to_json(&cfg_file, text)?,
None => Value::Object(Map::new()),
};
let cfg_map = cfg
.as_object()
.ok_or_else(|| load_error(&cfg_file, "", "expected a mapping"))?;
profile.worker = decode_worker(&cfg_file, cfg_map.get("worker"))?;
profile.expiry = decode_expiry(&cfg_file, cfg_map.get("expiry"))?;
profile.home = decode_home(&cfg_file, cfg_map.get("home"))?;
let gateway = cfg_map.get("gateway").and_then(Value::as_object);
let routes_raw: Vec<Value> = match cfg_map.get("profile_routes").and_then(Value::as_array) {
Some(a) => {
meta.routes_at_top = true;
a.clone()
}
None => gateway
.and_then(|g| g.get("profile_routes"))
.and_then(Value::as_array)
.cloned()
.unwrap_or_default(),
};
for (i, r) in routes_raw.iter().enumerate() {
profile.routes.push(decode_route(&cfg_file, i, r)?);
}
if let Some(platforms) = cfg_map.get("platforms") {
let map = platforms
.as_object()
.ok_or_else(|| load_error(&cfg_file, "platforms", "expected a map"))?;
let known: BTreeSet<String> = vault.keys().cloned().collect();
for (platform, raw) in map {
profile.channels.insert(
platform.clone(),
decode_channel(&cfg_file, platform, raw, vault)?,
);
}
config_normalized =
flavor == Flavor::Orchestrator && vault.keys().any(|k| !known.contains(k));
}
for (k, v) in cfg_map {
if CONFIG_O_KEYS.contains(&k.as_str()) || k == "platforms" || k == "profile_routes" {
continue;
}
if k == "gateway" {
let mut g = v.as_object().cloned().unwrap_or_default();
g.remove("profile_routes");
if !g.is_empty() {
profile
.residue
.config
.insert("gateway".into(), Value::Object(g));
}
continue;
}
profile.residue.config.insert(k.clone(), v.clone());
}
remember(&mut meta, "config.yaml", cfg_text, &config_record(&profile));
if config_normalized {
meta.raw.remove("config.yaml");
meta.snapshot.remove("config.yaml");
}
let persona_file = if flavor == Flavor::Hermes {
"SOUL.md"
} else {
"AGENTS.md"
};
let persona_text = read_text(dir, persona_file)?;
profile.persona = persona_text.as_ref().map(|t| PersonaRef {
path: "AGENTS.md".into(),
text: Some(t.clone()),
sha256: sha256_hex(t),
});
remember(
&mut meta,
persona_file,
persona_text,
&serde_json::to_value(&profile.persona).unwrap(),
);
let jobs_file = dir.join("cron/jobs.json").display().to_string();
let jobs_text = read_text(dir, "cron/jobs.json")?;
if let Some(text) = &jobs_text {
let parsed: Value = serde_json::from_str(text)
.map_err(|e| load_error(&jobs_file, "", format!("JSON: {e}")))?;
let arr: Vec<Value> = match &parsed {
Value::Array(a) => {
meta.jobs_form = Some(JobsForm {
object: false,
extras: Map::new(),
});
a.clone()
}
Value::Object(o) => match o.get("jobs") {
Some(Value::Array(a)) => {
let mut extras = o.clone();
extras.remove("jobs");
meta.jobs_form = Some(JobsForm {
object: true,
extras,
});
a.clone()
}
Some(Value::Object(m)) => {
let mut extras = o.clone();
extras.remove("jobs");
meta.jobs_form = Some(JobsForm {
object: true,
extras,
});
m.iter()
.map(|(id, j)| {
let mut j = j.as_object().cloned().unwrap_or_default();
j.insert("id".into(), Value::String(id.clone()));
Value::Object(j)
})
.collect()
}
_ => {
return Err(load_error(
&jobs_file,
"",
"expected an array of jobs or {\"jobs\": [...]}",
))
}
},
_ => {
return Err(load_error(
&jobs_file,
"",
"expected an array of jobs or {\"jobs\": [...]}",
))
}
};
for raw in &arr {
let job = decode_job(&jobs_file, raw)?;
if profile.jobs.contains_key(&job.id) {
return Err(load_error(&jobs_file, &job.id, "duplicate job id"));
}
profile.jobs.insert(job.id.clone(), job);
}
}
let jobs_record: Vec<Value> = profile
.jobs
.values()
.map(|j| serde_json::to_value(j).unwrap())
.collect();
remember(
&mut meta,
"cron/jobs.json",
jobs_text,
&Value::Array(jobs_record),
);
let exec_path = dir.join("cron/executions.db");
if table_exists(&exec_path, "executions") {
for row in read_rows(
&exec_path,
"select * from executions order by claimed_at, id",
&[],
)?
.unwrap_or_default()
{
profile
.fires
.push(decode_fire_row(&exec_path.display().to_string(), &row)?);
}
}
remember(
&mut meta,
"cron/executions.db",
None,
&serde_json::to_value(&profile.fires).unwrap(),
);
let subs_file = dir.join("webhook_subscriptions.json").display().to_string();
let subs_text = read_text(dir, "webhook_subscriptions.json")?;
if let Some(text) = &subs_text {
let parsed: Value = serde_json::from_str(text)
.map_err(|e| load_error(&subs_file, "", format!("JSON: {e}")))?;
let map = parsed
.as_object()
.ok_or_else(|| load_error(&subs_file, "", "expected a map"))?;
let known: BTreeSet<String> = vault.keys().cloned().collect();
for (n, raw) in map {
profile
.subscriptions
.insert(n.clone(), decode_subscription(&subs_file, n, raw, vault)?);
}
subs_normalized =
flavor == Flavor::Orchestrator && vault.keys().any(|k| !known.contains(k));
}
let subs_record: Vec<Value> = profile
.subscriptions
.values()
.map(|s| serde_json::to_value(s).unwrap())
.collect();
remember(
&mut meta,
"webhook_subscriptions.json",
subs_text,
&Value::Array(subs_record),
);
if subs_normalized {
meta.raw.remove("webhook_subscriptions.json");
meta.snapshot.remove("webhook_subscriptions.json");
}
let access_file = dir.join("access.yaml").display().to_string();
let access_text = read_text(dir, "access.yaml")?;
let access_raw = match &access_text {
Some(t) => Some(yaml_to_json(&access_file, t)?),
None => None,
};
profile.access = decode_access(&access_file, access_raw.as_ref())?;
remember(
&mut meta,
"access.yaml",
access_text,
&serde_json::to_value(&profile.access).unwrap(),
);
let state_path = dir.join("state.db");
if table_exists(&state_path, "delivery_obligations") {
for row in read_rows(
&state_path,
"select * from delivery_obligations order by created_at, obligation_id",
&[],
)?
.unwrap_or_default()
{
profile.obligations.push(decode_obligation_row(
&state_path.display().to_string(),
&row,
)?);
}
}
if flavor == Flavor::Orchestrator {
if table_exists(&state_path, "bindings") {
for row in read_rows(
&state_path,
"select * from bindings order by started_at, slot",
&[],
)?
.unwrap_or_default()
{
let b = decode_binding_row(&state_path.display().to_string(), &row)?;
let slot = row
.get("slot")
.and_then(Value::as_str)
.map(str::to_string)
.unwrap_or_else(|| surface_key_string(&b.key));
profile.bindings.insert(slot, b);
}
}
} else if table_exists(&state_path, "sessions") {
for row in read_rows(
&state_path,
"select * from sessions order by started_at, id",
&[],
)?
.unwrap_or_default()
{
if let Some(b) = binding_from_hermes_session(&state_path, &row, name) {
profile.bindings.insert(surface_key_string(&b.key), b);
}
}
}
remember(&mut meta, "state.db", None, &state_record(&profile));
profile.residue.files = list_unmodeled(dir, flavor)?;
Ok((profile, meta))
}
fn list_unmodeled(dir: &Path, flavor: Flavor) -> Result<Vec<String>> {
let mut owned: Vec<&str> = OWNED_FILES.to_vec();
if flavor == Flavor::Hermes {
owned.push("SOUL.md");
owned.retain(|f| !["AGENTS.md", "CLAUDE.md", "access.yaml"].contains(f));
}
let runtime_artifacts = ["orchestrator.lock", "orchestrator.sock", "service"];
let mut out = Vec::new();
fn walk(
base: &Path,
d: &Path,
owned: &[&str],
runtime: &[&str],
out: &mut Vec<String>,
) -> Result<()> {
let mut entries: Vec<_> = fs::read_dir(d)?.flatten().collect();
entries.sort_by_key(|e| e.file_name());
for entry in entries {
let p = entry.path();
let rel = p
.strip_prefix(base)
.unwrap_or(&p)
.to_string_lossy()
.replace('\\', "/");
let name = entry.file_name().to_string_lossy().into_owned();
if rel == "profiles"
|| name == "node_modules"
|| name == ".git"
|| rel.starts_with("state.db")
|| rel.starts_with("cron/executions.db")
{
continue;
}
if runtime.contains(&rel.as_str()) || regex_tmp(&name) {
continue;
}
let st = fs::symlink_metadata(&p)?;
if st.is_dir() {
walk(base, &p, owned, runtime, out)?;
continue;
}
if !st.is_file() {
continue;
}
if owned.contains(&rel.as_str()) {
continue;
}
out.push(rel);
}
Ok(())
}
walk(dir, dir, &owned, &runtime_artifacts, &mut out)?;
Ok(out)
}
fn regex_tmp(name: &str) -> bool {
name.rsplit_once(".tmp-")
.is_some_and(|(_, pid)| !pid.is_empty() && pid.chars().all(|c| c.is_ascii_digit()))
}
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 encode_config(
profile: &Profile,
meta: Option<&ProfileIo>,
vault: Option<&BTreeMap<String, String>>,
flavor: Flavor,
) -> String {
let mut out = Map::new();
for (k, v) in &profile.residue.config {
if k != "gateway" {
out.insert(k.clone(), v.clone());
}
}
if let Some(w) = &profile.worker {
let mut wm = Map::new();
wm.insert("harness".into(), Value::String(w.harness.as_str().into()));
if let Some(m) = &w.model {
wm.insert("model".into(), Value::String(m.clone()));
}
if let Some(p) = &w.preset {
wm.insert("preset".into(), Value::String(p.clone()));
}
if w.cwd != "." {
wm.insert("cwd".into(), Value::String(w.cwd.clone()));
}
if !w.env.is_empty() {
wm.insert("env".into(), serde_json::to_value(&w.env).unwrap());
}
if w.permission.timeout_seconds != 300
|| w.permission.default != crate::orchestration::PermissionDefault::Deny
{
wm.insert(
"permission".into(),
serde_json::to_value(&w.permission).unwrap(),
);
}
out.insert("worker".into(), Value::Object(wm));
}
if flavor == Flavor::Orchestrator || profile.expiry != ExpiryPolicy::default() {
out.insert(
"expiry".into(),
serde_json::to_value(&profile.expiry).unwrap(),
);
}
if let Some(h) = &profile.home {
out.insert("home".into(), encode_surface_key(h));
}
let mut gateway = profile
.residue
.config
.get("gateway")
.and_then(Value::as_object)
.cloned()
.unwrap_or_default();
let routes: Vec<Value> = profile.routes.iter().map(encode_route).collect();
if meta.is_some_and(|m| m.routes_at_top) {
if !routes.is_empty() {
out.insert("profile_routes".into(), Value::Array(routes));
}
} else if !routes.is_empty() {
gateway.insert("profile_routes".into(), Value::Array(routes));
}
if !gateway.is_empty() {
out.insert("gateway".into(), Value::Object(gateway));
}
let mut platforms = Map::new();
for (p, ch) in &profile.channels {
platforms.insert(
p.clone(),
encode_channel(
ch,
if flavor == Flavor::Hermes {
vault
} else {
None
},
),
);
}
if !platforms.is_empty() {
out.insert("platforms".into(), Value::Object(platforms));
}
json_to_yaml(&Value::Object(out))
}
pub fn encode_jobs_file(profile: &Profile, meta: Option<&ProfileIo>) -> String {
let jobs: Vec<Value> = profile
.jobs
.values()
.map(|j| ordered_object(encode_job(j)))
.collect();
let form = meta.and_then(|m| m.jobs_form.clone()).unwrap_or(JobsForm {
object: true,
extras: Map::new(),
});
let body = if form.object {
let mut pairs = vec![("jobs".to_string(), Value::Array(jobs))];
pairs.extend(form.extras.iter().map(|(k, v)| (k.clone(), v.clone())));
ordered_object(pairs)
} else {
Value::Array(jobs)
};
format!("{}\n", pretty_ordered(&body, 0))
}
pub(crate) fn ordered_object(pairs: Vec<(String, Value)>) -> Value {
Value::Array(vec![
Value::String("__ordered__".into()),
Value::Array(
pairs
.into_iter()
.map(|(k, v)| serde_json::json!({"__k": k, "__v": v}))
.collect(),
),
])
}
fn is_ordered(value: &Value) -> Option<&Vec<Value>> {
let arr = value.as_array()?;
if arr.len() == 2 && arr[0].as_str() == Some("__ordered__") {
arr[1].as_array()
} else {
None
}
}
pub(crate) fn pretty_ordered(value: &Value, depth: usize) -> String {
let pad = |d: usize| " ".repeat(d);
if let Some(pairs) = is_ordered(value) {
if pairs.is_empty() {
return "{}".into();
}
let inner: Vec<String> = pairs
.iter()
.map(|p| {
format!(
"{}{}: {}",
pad(depth + 1),
serde_json::to_string(p["__k"].as_str().unwrap_or("")).unwrap(),
pretty_ordered(&p["__v"], depth + 1)
)
})
.collect();
return format!("{{\n{}\n{}}}", inner.join(",\n"), pad(depth));
}
match value {
Value::Array(items) if items.is_empty() => "[]".into(),
Value::Array(items) => {
let inner: Vec<String> = items
.iter()
.map(|v| format!("{}{}", pad(depth + 1), pretty_ordered(v, depth + 1)))
.collect();
format!("[\n{}\n{}]", inner.join(",\n"), pad(depth))
}
Value::Object(o) if o.is_empty() => "{}".into(),
Value::Object(o) => {
let inner: Vec<String> = o
.iter()
.map(|(k, v)| {
format!(
"{}{}: {}",
pad(depth + 1),
serde_json::to_string(k).unwrap(),
pretty_ordered(v, depth + 1)
)
})
.collect();
format!("{{\n{}\n{}}}", inner.join(",\n"), pad(depth))
}
Value::Number(n) => {
if let Some(f) = n.as_f64() {
if n.is_f64() && f.fract() == 0.0 && f.abs() < 1e21 {
return format!("{}", f as i64);
}
}
n.to_string()
}
other => serde_json::to_string(other).unwrap(),
}
}
pub fn encode_subscriptions_file(
profile: &Profile,
vault: Option<&BTreeMap<String, String>>,
) -> String {
let mut out = Map::new();
for (n, s) in &profile.subscriptions {
out.insert(n.clone(), encode_subscription(s, vault));
}
format!("{}\n", pretty_ordered(&Value::Object(out), 0))
}
pub fn encode_access_file(profile: &Profile) -> String {
json_to_yaml(&encode_access(&profile.access))
}
const EXECUTIONS_DDL: &str = "CREATE TABLE IF NOT EXISTS executions (
id TEXT PRIMARY KEY, job_id TEXT NOT NULL, source TEXT NOT NULL, process_id TEXT NOT NULL, pid INTEGER NOT NULL,
process_started_at INTEGER, status TEXT NOT NULL CHECK(status IN ('claimed','running','completed','failed','unknown')),
claimed_at TEXT NOT NULL, started_at TEXT, finished_at TEXT, error TEXT);
CREATE INDEX IF NOT EXISTS idx_executions_job_claimed ON executions(job_id, claimed_at DESC, id DESC);
CREATE INDEX IF NOT EXISTS idx_executions_status_claimed ON executions(status, claimed_at DESC, id DESC);";
const OBLIGATIONS_DDL: &str = "CREATE TABLE IF NOT EXISTS delivery_obligations (
obligation_id TEXT PRIMARY KEY, session_key TEXT NOT NULL, platform TEXT NOT NULL, chat_id TEXT NOT NULL, thread_id TEXT,
content TEXT NOT NULL, state TEXT NOT NULL, attempts INTEGER NOT NULL DEFAULT 0, created_at REAL NOT NULL, updated_at REAL NOT NULL,
owner_pid INTEGER, owner_started_at INTEGER, last_error TEXT, adapter_profile TEXT);";
const BINDINGS_DDL: &str = "CREATE TABLE IF NOT EXISTS bindings (
slot TEXT PRIMARY KEY,
platform TEXT NOT NULL, chat_type TEXT NOT NULL, chat_id TEXT, thread_id TEXT, participant_id TEXT,
worker_harness TEXT NOT NULL, worker_session_id TEXT, worker_locator TEXT,
started_at TEXT NOT NULL, last_activity_at TEXT NOT NULL, ended_at TEXT, end_reason TEXT,
handoff_to TEXT, handoff_state TEXT, handoff_error TEXT, recurrence_job_id TEXT, residue_json TEXT);";
pub fn write_executions(path: &Path, fires: &[crate::orchestration::Fire]) -> Result<()> {
let insert = format!(
"insert into executions ({}) values ({})",
EXECUTION_COLUMNS.join(", "),
EXECUTION_COLUMNS
.iter()
.map(|_| "?")
.collect::<Vec<_>>()
.join(",")
);
let rows: Vec<Vec<Param>> = fires
.iter()
.map(|f| encode_fire_row(f).iter().map(Param::from).collect())
.collect();
write_table(path, EXECUTIONS_DDL, &insert, &rows)
}
fn write_state(path: &Path, profile: &Profile) -> Result<()> {
let cols = [
"slot",
"platform",
"chat_type",
"chat_id",
"thread_id",
"participant_id",
"worker_harness",
"worker_session_id",
"worker_locator",
"started_at",
"last_activity_at",
"ended_at",
"end_reason",
"handoff_to",
"handoff_state",
"handoff_error",
"recurrence_job_id",
"residue_json",
];
let insert = format!(
"insert into bindings ({}) values ({})",
cols.join(", "),
cols.iter().map(|_| "?").collect::<Vec<_>>().join(",")
);
let s = |v: &Option<String>| v.clone().map(Param::Text).unwrap_or(Param::Null);
let rows: Vec<Vec<Param>> = profile
.bindings
.iter()
.map(|(slot, b)| {
vec![
Param::Text(slot.clone()),
Param::Text(b.key.platform.clone().unwrap_or_default()),
Param::Text(b.key.kind.clone().unwrap_or_default()),
Param::Text(b.key.chat_id.clone().unwrap_or_default()),
Param::Text(b.key.thread_id.clone().unwrap_or_default()),
Param::Text(b.key.participant_id.clone().unwrap_or_default()),
Param::Text(b.worker.harness.as_str().into()),
s(&b.worker.session_id.clone().filter(|v| !v.is_empty())),
s(&b.worker.locator),
Param::Text(b.started_at.clone().unwrap_or_default()),
Param::Text(b.last_activity_at.clone().unwrap_or_default()),
s(&b.ended_at),
b.end_reason
.map(|r| Param::Text(r.as_str().into()))
.unwrap_or(Param::Null),
s(&b.handoff.as_ref().and_then(|h| h.to.clone())),
b.handoff
.as_ref()
.map(|h| Param::Text(h.state.clone()))
.unwrap_or(Param::Null),
s(&b.handoff.as_ref().and_then(|h| h.error.clone())),
s(&b.recurrence.as_ref().map(|r| r.job_id.clone())),
if b.residue.is_empty() {
Param::Null
} else {
Param::Text(serde_json::to_string(&b.residue).unwrap())
},
]
})
.collect();
write_table(
path,
&format!("{BINDINGS_DDL}\n{OBLIGATIONS_DDL}"),
&insert,
&rows,
)?;
let insert = format!(
"insert into delivery_obligations ({}) values ({})",
OBLIGATION_COLUMNS.join(", "),
OBLIGATION_COLUMNS
.iter()
.map(|_| "?")
.collect::<Vec<_>>()
.join(",")
);
let rows: Vec<Vec<Param>> = profile
.obligations
.iter()
.map(|o| encode_obligation_row(o).iter().map(Param::from).collect())
.collect();
write_table(path, "", &insert, &rows)
}
fn write_if_changed(
meta: &mut ProfileIo,
dir: &Path,
rel: &str,
record: &Value,
render: impl FnOnce() -> Option<String>,
) -> Result<bool> {
let snap = canonical_json(record);
let path = dir.join(rel);
let reuse = meta.flavor == Flavor::Orchestrator && meta.snapshot.get(rel) == Some(&snap);
if reuse && path.exists() {
return Ok(false);
}
if reuse {
if let Some(raw) = meta.raw.get(rel).cloned() {
write_atomic(&path, &raw)?;
return Ok(true);
}
}
let Some(text) = render() else {
return Ok(false);
};
write_atomic(&path, &text)?;
meta.raw.insert(rel.into(), text);
meta.snapshot.insert(rel.into(), snap);
meta.flavor = Flavor::Orchestrator; Ok(true)
}
pub fn save_home(loaded: &mut LoadedHome, root: Option<&Path>) -> Result<()> {
let root = root
.map(Path::to_path_buf)
.unwrap_or_else(|| loaded.orchestration.root.clone());
fs::create_dir_all(&root)?;
let names: Vec<String> = loaded.orchestration.profiles.keys().cloned().collect();
for name in names {
let dir = if name == "default" {
root.clone()
} else {
root.join("profiles").join(&name)
};
fs::create_dir_all(dir.join("cron"))?;
let profile = loaded.orchestration.profiles[&name].clone();
let meta = loaded
.io
.entry(name.clone())
.or_insert_with(|| ProfileIo::new(Flavor::Orchestrator));
save_profile_dir(&profile, meta, &dir, &loaded.vault)?;
}
Ok(())
}
fn save_profile_dir(
profile: &Profile,
meta: &mut ProfileIo,
dir: &Path,
vault: &BTreeMap<String, String>,
) -> Result<()> {
let cfg_record = config_record(profile);
write_if_changed(meta, dir, "config.yaml", &cfg_record, || {
Some(encode_config(
profile,
None,
Some(vault),
Flavor::Orchestrator,
))
})?;
if let Some(persona) = &profile.persona {
let text = persona.text.clone().unwrap_or_default();
write_if_changed(
meta,
dir,
"AGENTS.md",
&serde_json::to_value(&profile.persona).unwrap(),
|| Some(text),
)?;
if !dir.join("CLAUDE.md").exists() {
write_atomic(&dir.join("CLAUDE.md"), "@AGENTS.md\n")?;
}
}
let jobs_record: Vec<Value> = profile
.jobs
.values()
.map(|j| serde_json::to_value(j).unwrap())
.collect();
let had_jobs = meta.raw.contains_key("cron/jobs.json");
let form = meta.jobs_form.clone();
write_if_changed(
meta,
dir,
"cron/jobs.json",
&Value::Array(jobs_record),
|| {
if profile.jobs.is_empty() && !had_jobs {
None
} else {
let stub = ProfileIo {
jobs_form: form.clone(),
..ProfileIo::new(Flavor::Orchestrator)
};
Some(encode_jobs_file(profile, Some(&stub)))
}
},
)?;
let subs_record: Vec<Value> = profile
.subscriptions
.values()
.map(|s| serde_json::to_value(s).unwrap())
.collect();
let had_subs = meta.raw.contains_key("webhook_subscriptions.json");
write_if_changed(
meta,
dir,
"webhook_subscriptions.json",
&Value::Array(subs_record),
|| {
if profile.subscriptions.is_empty() && !had_subs {
None
} else {
Some(encode_subscriptions_file(profile, None))
}
},
)?;
let a = &profile.access;
let access_empty = a.allowlist.is_empty()
&& a.admins.is_empty()
&& a.pending_pairings.is_empty()
&& a.policy.is_empty()
&& a.pairing_ttl_minutes.is_none();
let had_access = meta.raw.contains_key("access.yaml");
write_if_changed(
meta,
dir,
"access.yaml",
&serde_json::to_value(a).unwrap(),
|| {
if access_empty && !had_access {
None
} else {
Some(encode_access_file(profile))
}
},
)?;
let fires_snap = canonical_json(&serde_json::to_value(&profile.fires).unwrap());
let exec_path = dir.join("cron/executions.db");
if (meta.snapshot.get("cron/executions.db") != Some(&fires_snap) || !exec_path.exists())
&& (!profile.fires.is_empty() || exec_path.exists())
{
let tmp = exec_path.with_file_name(format!("executions.db.tmp-{}", std::process::id()));
let _ = fs::remove_file(&tmp);
write_executions(&tmp, &profile.fires)?;
fs::rename(&tmp, &exec_path)?;
meta.snapshot
.insert("cron/executions.db".into(), fires_snap);
}
let state_snap = canonical_json(&state_record(profile));
let state_path = dir.join("state.db");
if (meta.snapshot.get("state.db") != Some(&state_snap) || !state_path.exists())
&& (!profile.bindings.is_empty() || !profile.obligations.is_empty() || state_path.exists())
{
let tmp = state_path.with_file_name(format!("state.db.tmp-{}", std::process::id()));
let _ = fs::remove_file(&tmp);
write_state(&tmp, profile)?;
fs::rename(&tmp, &state_path)?;
meta.snapshot.insert("state.db".into(), state_snap);
}
let mut refs: Vec<String> = Vec::new();
for ch in profile.channels.values() {
for r in ch.credentials.values() {
if let crate::ontology::SecretRef::Dotenv(n) = r {
refs.push(n.clone());
}
}
}
for s in profile.subscriptions.values() {
if let Some(crate::ontology::SecretRef::Dotenv(n)) = &s.secret {
refs.push(n.clone());
}
}
if let Some(w) = &profile.worker {
for v in w.env.values() {
if let crate::orchestration::EnvValue::Secret(crate::ontology::SecretRef::Dotenv(n)) = v
{
refs.push(n.clone());
}
}
}
let mut entries: BTreeMap<String, String> = BTreeMap::new();
for r in refs {
if let Some(v) = vault.get(&r) {
entries.insert(r, v.clone());
}
}
if !entries.is_empty() {
let existing = meta.raw.get(".env").cloned();
let mut merged = existing.as_deref().map(parse_dotenv).unwrap_or_default();
for (k, v) in entries {
merged.insert(k, v);
}
let text = render_dotenv(&merged);
if existing.as_deref() != Some(text.as_str()) {
write_atomic(&dir.join(".env"), &text)?;
meta.raw.insert(".env".into(), text);
}
}
Ok(())
}
pub fn copy_unmodeled(profile: &Profile, meta: &ProfileIo, dest: &Path) -> Result<()> {
let Some(src) = &meta.source_dir else {
return Ok(());
};
carry_unmodeled(&profile.residue.files, src, dest)?;
Ok(())
}
pub fn carry_unmodeled(files: &[String], src: &Path, dest: &Path) -> Result<Vec<String>> {
let mut carried = Vec::new();
for rel in files {
let from = src.join(rel);
if !from.is_file() {
continue;
}
let to = dest.join(rel);
if let Some(parent) = to.parent() {
fs::create_dir_all(parent)?;
}
fs::copy(&from, &to)?;
carried.push(rel.clone());
}
Ok(carried)
}