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::{
parse_hermes_session_key, Binding, EndReason, Handoff, HarnessId, HermesSessionRow, Recurrence,
Residue, SurfaceKey, Trigger, Worker,
};
use crate::world::{ExpiryPolicy, PersonaRef, Profile, ProfileResidue, World};
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 world: World,
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 {
world: World {
root: root.to_path_buf(),
profiles,
},
vault,
io,
};
if flavor == Flavor::Hermes {
partition_shared_store(&mut loaded, root)?;
}
Ok(loaded)
}
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
.world
.profiles
.keys()
.filter(|n| *n != "default")
.cloned()
.collect();
for name in names {
let has_own = loaded.world.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
.world
.profiles
.get_mut(&name)
.unwrap()
.bindings
.insert(surface_key_string(&b.key), b);
}
}
}
let root_profile = loaded.world.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.world.profiles.get_mut(&name).unwrap().obligations = mine;
let snap = canonical_json(&state_record(&loaded.world.profiles[&name]));
loaded
.io
.get_mut(&name)
.unwrap()
.snapshot
.insert("state.db".into(), snap);
}
let snap = canonical_json(&state_record(&loaded.world.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 refuse_inline_credentials(
file: &str,
flavor: Flavor,
known: &BTreeSet<String>,
vault: &BTreeMap<String, String>,
) -> Result<()> {
if flavor != Flavor::Orchestrator {
return Ok(());
}
let inline: Vec<&str> = vault
.keys()
.filter(|k| !known.contains(*k))
.map(String::as_str)
.collect();
if inline.is_empty() {
return Ok(());
}
Err(load_error(
file,
"",
format!(
"a credential value lives in .env, not here: put {} in .env and reference it as {{dotenv: NAME}}",
inline.join(", ")
),
))
}
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 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)?,
);
}
refuse_inline_credentials(&cfg_file, flavor, &known, vault)?;
}
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));
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)?);
}
refuse_inline_credentials(&subs_file, flavor, &known, vault)?;
}
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),
);
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::world::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::world::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.world.root.clone());
fs::create_dir_all(&root)?;
let names: Vec<String> = loaded.world.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.world.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::world::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)
}