supercode_interchange/orchestration/codec/

folder.rs

//! The folder codec: a home folder ⇄ [`Orchestration`]. The root is the
//! `default` profile; `profiles/<name>/` are the named ones; a profile folder
//! is a complete home (Hermes's rule). Two flavors of the same layout:
//!
//! * `Orchestrator` — ours: the O-blocks in `config.yaml` decode strictly,
//!   `AGENTS.md` is the persona, `state.db` holds our `bindings` table.
//! * `Hermes` — a Hermes home: `SOUL.md` is the persona, `state.db`'s
//!   `sessions` rows become bindings (the multiplexed store partitioned by
//!   `profile_name`), and the file's bytes may never be reused for our folder
//!   (they may carry inline credentials).
//!
//! Files we own are re-emitted byte for byte when their record is unchanged
//! and canonically when it changed; files we do not model are never touched.

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_channel, decode_fire_row, decode_job, decode_obligation_row, decode_route,
    decode_subscription, decode_worker, encode_channel, encode_fire_row, encode_job,
    encode_obligation_folder_extras, encode_obligation_row, encode_route, encode_subscription,
    load_error, surface_key_string, EXECUTION_COLUMNS, FOLDER_OBLIGATION_EXTRA_COLUMNS,
    OBLIGATION_COLUMNS,
};
use super::dotenv::{parse_dotenv, render_dotenv};
use super::sqlite::{read_rows, replace_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::{
    Access, AccessPolicy, ChannelConfig, Orchestration, PersonaRef, Profile, ProfileResidue,
};
use crate::Result;

/// The files our folder owns.
pub const OWNED_FILES: &[&str] = &[
    "config.yaml",
    "AGENTS.md",
    "CLAUDE.md",
    ".env",
    "cron/jobs.json",
    "cron/executions.db",
    "webhook_subscriptions.json",
    "state.db",
];

const CONFIG_O_KEYS: &[&str] = &["worker"];

/// Which harness's layout a folder was read as.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Flavor {
    /// Our own folder.
    Orchestrator,
    /// A Hermes home.
    Hermes,
}

/// The form `cron/jobs.json` was read in, remembered for the emit.
#[derive(Debug, Clone, PartialEq)]
pub struct JobsForm {
    /// `{"jobs": [...], ...}` (true) or a bare array (false).
    pub object: bool,
    /// The envelope's other keys (`updated_at`, …), verbatim.
    pub extras: Map<String, Value>,
}

/// Per-profile bookkeeping: what a decompile needs to reuse bytes and to know
/// what "unchanged" means. Never part of the orchestration.
#[derive(Debug, Clone)]
pub struct ProfileIo {
    /// Raw text of every owned file read.
    pub raw: BTreeMap<String, String>,
    /// Canonical JSON of each artifact's record as loaded.
    pub snapshot: BTreeMap<String, String>,
    /// Where the profile was read from.
    pub source_dir: Option<PathBuf>,
    /// The flavor it was read as; only our own bytes may be written back verbatim.
    pub flavor: Flavor,
    /// The jobs file's form.
    pub jobs_form: Option<JobsForm>,
    /// Whether `profile_routes` sat at the top level (else under `gateway`).
    pub routes_at_top: bool,
    /// A Hermes profile that borrowed its partition from the root store.
    pub borrowed_from: Option<PathBuf>,
    /// The named profiles that borrowed from this root store.
    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(),
        }
    }
}

/// A loaded home.
#[derive(Debug, Clone)]
pub struct LoadedHome {
    /// The orchestration.
    pub orchestration: Orchestration,
    /// Secret values by `.env` key; never in the orchestration.
    pub vault: BTreeMap<String, String>,
    /// Per-profile bookkeeping.
    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()
}

/// A persona record for `text`: `AGENTS.md` at the profile root, content hash first.
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
    })
}

/// The strings PyYAML (YAML 1.1, what Hermes reads with) resolves as booleans. `serde_yaml` (YAML 1.2)
/// writes them bare, so a Hermes `approvals.mode: 'off'` came back as `off` and Hermes read `false`.
const YAML11_BOOLS: [&str; 18] = [
    "yes", "Yes", "YES", "no", "No", "NO", "true", "True", "TRUE", "false", "False", "FALSE", "on",
    "On", "ON", "off", "Off", "OFF",
];
const YAML11_MARK: &str = "__supercode_yaml11_string__";

fn mark_yaml11_strings(value: &Value) -> Value {
    match value {
        Value::String(s) if YAML11_BOOLS.contains(&s.as_str()) => {
            Value::String(format!("{YAML11_MARK}{s}"))
        }
        Value::Array(items) => Value::Array(items.iter().map(mark_yaml11_strings).collect()),
        Value::Object(map) => Value::Object(
            map.iter()
                .map(|(k, v)| (k.clone(), mark_yaml11_strings(v)))
                .collect(),
        ),
        other => other.clone(),
    }
}

fn json_to_yaml(value: &Value) -> String {
    if value.as_object().is_some_and(|m| m.is_empty()) {
        return String::new();
    }
    // a string YAML 1.1 reads as a boolean is written single-quoted, as Hermes writes it
    let y: serde_yaml::Value =
        serde_json::from_value(mark_yaml11_strings(value)).unwrap_or(serde_yaml::Value::Null);
    let mut out = serde_yaml::to_string(&y).unwrap_or_default();
    for word in YAML11_BOOLS {
        out = out.replace(&format!("{YAML11_MARK}{word}"), &format!("'{word}'"));
    }
    out
}

/// The empty profile.
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(),
        jobs: BTreeMap::new(),
        subscriptions: BTreeMap::new(),
        access: Default::default(),
        bindings: BTreeMap::new(),
        fires: Vec::new(),
        obligations: Vec::new(),
        residue: ProfileResidue::default(),
    }
}

/// The record `config.yaml`'s bytes are the serialization of.
pub fn config_record(profile: &Profile) -> Value {
    serde_json::json!({
        "worker": profile.worker,
        "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 })
}

/// Load a home folder.
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)
}

/// A Hermes-shaped ledger writes a fire's session and delivery nowhere. The
/// session is the one minted `cron_<job>_<instant>` inside the fire's window
/// (its compression tip, when the conversation rolled over); the delivery is
/// the obligation that session's key — else the job's surface — raised in the
/// window. `Fire.session_id` / `Fire.obligation_id` are derived here, the
/// obligation's source points back, and the encoders never write either
/// (`docs/ONTOLOGY.md` §2.7).
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()
        };
        // the obligations that store holds: a shared root store was
        // partitioned across the profiles lending from it
        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()]
            }
        };
        // an obligation that already names its fire (our folder persists the
        // source) is the link; the window join below is for stores that do not
        let mut by_fire: BTreeMap<String, (String, String)> = BTreeMap::new();
        for (holder, profile) in &loaded.orchestration.profiles {
            for o in &profile.obligations {
                if let crate::orchestration::ObligationSource::Fire { fire_id } = &o.source {
                    by_fire
                        .entry(fire_id.clone())
                        .or_insert_with(|| (holder.clone(), o.id.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() {
            if let Some(known) = by_fire.get(&fire.id) {
                // the session, when our ledger persisted it, is already on the fire
                links.push((index, fire.session_id.clone(), Some(known.clone())));
                continue;
            }
            let session_id = fire_session(&sessions, fire).or_else(|| fire.session_id.clone());
            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 };
                }
            }
        }
    }
    // the links are derived: the records the stores hold are unchanged
    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(())
}

/// The platform surface a job's fires deliver to, when the job names one an
/// obligation could carry: `origin` (the creating conversation) and the
/// explicit `<platform>:<chat>` form. `local` and `home` deliver nowhere a
/// platform ledger sees.
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,
    }
}

/// The session a Hermes fire ran in: the `cron_<job>_<YYYYMMDD_HHMMSS>` id
/// minted inside [claimed, finished] — the earliest while the fire still
/// runs, the latest once it finished — followed to its compression tip.
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)
}

/// `YYYYMMDDHHMMSS` of an ISO instant, for ordering against a cron session id.
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())?
}

/// `YYYYMMDDHHMMSS` of a `cron_<job>_<YYYYMMDD_HHMMSS>` session id, when it is `job`'s.
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()
}

/// Seconds since the epoch of an ISO-8601 instant with `Z` or `±HH:MM`.
pub(crate) 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)
}

/// Hermes's multiplexed gateway keeps every profile's rows in ONE store,
/// partitioned by `profile_name` (sessions) and by the session key's profile
/// segment (obligations). A named profile without its own `state.db` borrows
/// its partition from the root store; the root profile keeps the rest.
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);
                }
            }
        }
        // obligations addressed to this profile's surfaces move out of the root partition
        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",
];

/// The `gateway_routing` scope Hermes's session store uses for a home: its `sessions` folder, resolved.
fn routing_scope(dir: &Path) -> String {
    let sessions = dir.join("sessions");
    fs::canonicalize(&sessions)
        .or_else(|_| fs::canonicalize(dir).map(|d| d.join("sessions")))
        .unwrap_or(sessions)
        .display()
        .to_string()
}

/// A binding as Hermes's `SessionEntry` (`gateway/session.py`): its session key, session id, times, origin,
/// platform and chat type, with the binding itself in `metadata.supercode` so it reads back whole.
fn routing_entry(profile_name: &str, slot: &str, b: &Binding) -> (String, Value) {
    let ns = if profile_name == "default" {
        ""
    } else {
        profile_name
    };
    let key = b
        .key
        .key
        .clone()
        .filter(|k| k.starts_with("agent:"))
        .unwrap_or_else(|| crate::ontology::render_hermes_session_key(ns, &b.key));
    let entry = serde_json::json!({
        "session_key": key,
        "session_id": b.worker.session_id.clone().unwrap_or_else(|| slot.to_string()),
        "created_at": b.started_at,
        "updated_at": b.last_activity_at.clone().or_else(|| b.started_at.clone()),
        "display_name": Value::Null,
        "platform": b.key.platform,
        "chat_type": b.key.kind.clone().unwrap_or_else(|| "dm".into()),
        "origin": {
            "platform": b.key.platform,
            "chat_id": b.key.chat_id,
            "chat_type": b.key.kind,
            "thread_id": b.key.thread_id,
            "user_id": b.key.participant_id,
        },
        "metadata": { "supercode": { "slot": slot, "binding": b } },
        "input_tokens": 0, "output_tokens": 0, "cache_read_tokens": 0, "cache_write_tokens": 0,
        "total_tokens": 0, "last_prompt_tokens": 0, "estimated_cost_usd": 0.0, "cost_status": "unknown",
        "was_auto_reset": false, "auto_reset_reason": Value::Null, "reset_had_activity": false,
    });
    (key, entry)
}

/// One `gateway_routing` row as a binding: whole from `metadata.supercode` when the orchestrator wrote it,
/// else from the entry Hermes wrote (a Hermes session, its worker Hermes itself).
fn binding_from_routing(file: &Path, row: &Map<String, Value>) -> Option<(String, Binding)> {
    let entry: Value = serde_json::from_str(row.get("entry_json")?.as_str()?).ok()?;
    if let Some(sc) = entry.pointer("/metadata/supercode") {
        let b: Binding = serde_json::from_value(sc.get("binding")?.clone()).ok()?;
        let slot = sc
            .get("slot")
            .and_then(Value::as_str)
            .map(str::to_string)
            .unwrap_or_else(|| surface_key_string(&b.key));
        return Some((slot, b));
    }
    let session_key = entry.get("session_key").and_then(Value::as_str)?;
    let (mut key, _) = parse_hermes_session_key(session_key)?;
    key.key = None;
    let text = |k: &str| entry.get(k).and_then(Value::as_str).map(str::to_string);
    let b = Binding {
        key: key.clone(),
        profile: None,
        worker: Worker {
            harness: HarnessId::new(HarnessId::HERMES),
            session_id: text("session_id"),
            locator: Some(file.display().to_string()),
        },
        trigger: Trigger::Unknown,
        recurrence: None,
        handoff: None,
        started_at: text("created_at"),
        last_activity_at: text("updated_at"),
        ended_at: None,
        end_reason: None,
        residue: Residue::default(),
    };
    Some((surface_key_string(&key), b))
}

/// A Hermes `sessions` row is a binding when it has a surface (session_key) or is a cron fire.
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");
    // The surface may live only in `session_key` (api_server conversations
    // carry no chat_type/chat_id columns — ORC-8 finding).
    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; // a foreign partition is not this profile's binding
        }
    }
    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,
    })
}

/// `new Date(seconds * 1000).toISOString()`.
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));
    };

    // .env → vault (values), never into the model.
    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);
    }

    // config.yaml
    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"))?;
    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));
    }
    // Hermes turns a messaging platform on from its token in the profile's own `.env` alone
    // (`gateway/config.py` `_apply_env_overrides`), with its home channel from `<PLATFORM>_HOME_CHANNEL`.
    // Such a channel is the environment's: it carries `extra.from_env`, and the writer never turns it
    // into a `platforms:` block, so the home's `config.yaml` stays as its owner wrote it. Both flavors: the
    // orchestrator's own folder is a Hermes home.
    {
        let own: BTreeMap<String, String> = meta
            .raw
            .get(".env")
            .map(|text| parse_dotenv(text).into_iter().collect())
            .unwrap_or_default();
        let set = |k: &str| own.get(k).filter(|v| !v.trim().is_empty()).cloned();
        for (platform, token, home) in [
            ("discord", "DISCORD_BOT_TOKEN", "DISCORD_HOME_CHANNEL"),
            ("telegram", "TELEGRAM_BOT_TOKEN", "TELEGRAM_HOME_CHANNEL"),
            ("slack", "SLACK_BOT_TOKEN", "SLACK_HOME_CHANNEL"),
        ] {
            if profile.channels.contains_key(platform) || set(token).is_none() {
                continue;
            }
            let mut extra = BTreeMap::new();
            extra.insert("from_env".to_string(), Value::Bool(true));
            if let Some(chat) = set(home) {
                let mut hc = Map::new();
                hc.insert("platform".into(), Value::String(platform.into()));
                hc.insert("chat_id".into(), Value::String(chat));
                hc.insert(
                    "name".into(),
                    Value::String(set(&format!("{home}_NAME")).unwrap_or_else(|| "Home".into())),
                );
                if let Some(thread) = set(&format!("{home}_THREAD_ID")) {
                    hc.insert("thread_id".into(), Value::String(thread));
                }
                extra.insert("home_channel".to_string(), Value::Object(hc));
            }
            // no credentials: the adapter then reads Hermes's own variables from the profile's `.env`, as Hermes does
            profile.channels.insert(
                platform.to_string(),
                ChannelConfig {
                    platform: platform.to_string(),
                    enabled: true,
                    credentials: BTreeMap::new(),
                    extra,
                },
            );
        }
    }
    // Everything else in config.yaml is Hermes's and rides as residue.
    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 {
        // our flavor: a credential VALUE was pasted inline and has moved to the
        // vault, so the bytes on disk are not this record's and the next save
        // re-emits the file with the ref (and `.env` receives the value). A
        // Hermes home keeps values inline by design; its bytes stay reusable.
        meta.raw.remove("config.yaml");
        meta.snapshot.remove("config.yaml");
    }

    // persona
    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(),
    );

    // jobs — three forms load; the file's own form is remembered for the emit
    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),
    );

    // fires
    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(),
    );

    // subscriptions
    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");
    }

    // access: Hermes's own stores (docs/plans/hermes-compat.md row 8), this profile's `.env` only
    let own_env = meta
        .raw
        .get(".env")
        .map(|t| parse_dotenv(t))
        .unwrap_or_default();
    profile.access = hermes_access(dir, &own_env, cfg_map.get("platforms"));

    // state.db: bindings + obligations
    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 {
        // Hermes's session entries (docs/plans/hermes-compat.md row 10): this home's rows of
        // `gateway_routing`, whoever wrote them
        if table_exists(&state_path, "gateway_routing") {
            let scope = routing_scope(dir);
            for row in read_rows(
                &state_path,
                "select session_key, entry_json from gateway_routing where scope = ? order by updated_at",
                &[&scope],
            )?
            .unwrap_or_default()
            {
                if let Some((slot, b)) = binding_from_routing(&state_path, &row) {
                    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));

    // unmodeled files: carried by path, copied verbatim on export; never held in memory.
    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"].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 entries = match fs::read_dir(d) {
            Ok(entries) => entries,
            // A live harness can remove an inventoried child directory between reads.
            Err(error) if d != base && error.kind() == std::io::ErrorKind::NotFound => {
                return Ok(());
            }
            Err(error) => return Err(error.into()),
        };
        let mut entries = entries.collect::<std::io::Result<Vec<_>>>()?;
        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 = match fs::symlink_metadata(&p) {
                Ok(metadata) => metadata,
                // SQLite sidecars and other live residue can disappear after listing.
                Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue,
                Err(error) => return Err(error.into()),
            };
            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 {
    // `<file>.tmp-<pid>`
    name.rsplit_once(".tmp-")
        .is_some_and(|(_, pid)| !pid.is_empty() && pid.chars().all(|c| c.is_ascii_digit()))
}

// ---------------------------------------------------------------- save

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()
    ));
    write_with_mode(&tmp, text, target_mode(path))?;
    fs::rename(&tmp, path)?;
    Ok(())
}

/// The mode the replaced file keeps: its own when it exists, and never wider than owner-only for the
/// vault (`.env`), whose values are credentials. A rename over a file does not carry the old file's
/// mode, so the temporary file is created with it.
#[cfg(unix)]
fn target_mode(path: &Path) -> Option<u32> {
    use std::os::unix::fs::PermissionsExt;
    let current = fs::metadata(path)
        .ok()
        .map(|m| m.permissions().mode() & 0o777);
    if path.file_name().is_some_and(|name| name == ".env") {
        return Some(current.unwrap_or(0o600) & 0o600);
    }
    current
}

#[cfg(not(unix))]
fn target_mode(_path: &Path) -> Option<u32> {
    None
}

#[cfg(unix)]
fn write_with_mode(path: &Path, text: &str, mode: Option<u32>) -> Result<()> {
    use std::io::Write;
    use std::os::unix::fs::{OpenOptionsExt, PermissionsExt};
    let mut options = fs::OpenOptions::new();
    options.write(true).create(true).truncate(true);
    if let Some(mode) = mode {
        options.mode(mode);
    }
    let mut file = options.open(path)?;
    // `mode` at creation is filtered by the umask; the explicit set makes it exact.
    if let Some(mode) = mode {
        file.set_permissions(fs::Permissions::from_mode(mode))?;
    }
    file.write_all(text.as_bytes())?;
    Ok(())
}

#[cfg(not(unix))]
fn write_with_mode(path: &Path, text: &str, _mode: Option<u32>) -> Result<()> {
    fs::write(path, text)?;
    Ok(())
}

/// Canonical config.yaml: O-blocks + Hermes-shaped blocks + residue.
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() {
            let env: Map<String, Value> = w
                .env
                .iter()
                .map(|(k, v)| {
                    let value = match v {
                        crate::orchestration::EnvValue::Literal(s) => Value::String(s.clone()),
                        crate::orchestration::EnvValue::Secret(r) => super::decode::render_ref(r),
                    };
                    (k.clone(), value)
                })
                .collect();
            wm.insert("env".into(), Value::Object(env));
        }
        if w.permission.timeout_seconds != 300
            || w.permission.default != crate::orchestration::PermissionDefault::Deny
            || w.permission.unattended != crate::orchestration::PermissionUnattended::Deny
        {
            wm.insert(
                "permission".into(),
                serde_json::to_value(&w.permission).unwrap(),
            );
        }
        out.insert("worker".into(), Value::Object(wm));
    }
    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 {
        // a channel Hermes enables from the profile's `.env` stays there (the loader's `from_env`)
        if ch.extra.get("from_env") == Some(&Value::Bool(true)) {
            continue;
        }
        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))
}

/// The pin's canonical form is `{"jobs": [...]}`; a file read as a bare array keeps that form.
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))
}

/// An object whose key order is the given one, regardless of features: wrapped
/// as a marker array the ordered printer understands.
pub(crate) fn ordered_object(pairs: Vec<(String, Value)>) -> Value {
    // serde_json::Map may sort; keep the order in a side channel: encode as
    // [{"__k": k, "__v": v}, ...] under a private tag.
    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
    }
}

/// `JSON.stringify(value, null, 2)` with ordered objects honoured and plain objects sorted.
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(),
    }
}

/// `webhook_subscriptions.json`; secrets inlined only for a Hermes export.
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))
}

/// Allowlists, allow-all policy and admins as Hermes keeps them: `<PLATFORM>_ALLOWED_USERS` and
/// `<PLATFORM>_ALLOW_ALL_USERS` (and `GATEWAY_ALLOW_ALL_USERS`) in the profile's own `.env`, the approved
/// pairings in `platforms/pairing/<platform>-approved.json` (and the legacy `pairing/`), and each
/// platform's `allow_admin_from` / `group_allow_admin_from` in `config.yaml`. Read, never written: a
/// change goes through Hermes's own pairing store and writers.
fn hermes_access(dir: &Path, env: &BTreeMap<String, String>, platforms: Option<&Value>) -> Access {
    let mut access = Access::default();
    let truthy = |v: &str| {
        matches!(
            v.trim().to_ascii_lowercase().as_str(),
            "1" | "true" | "yes" | "on"
        )
    };
    let split = |v: &str| {
        v.split(',')
            .map(|s| s.trim().to_string())
            .filter(|s| !s.is_empty())
            .collect::<Vec<_>>()
    };
    let global_open = env
        .get("GATEWAY_ALLOW_ALL_USERS")
        .is_some_and(|v| truthy(v));
    for (k, v) in env {
        if let Some(p) = k.strip_suffix("_ALLOWED_USERS") {
            access
                .allowlist
                .entry(p.to_ascii_lowercase())
                .or_default()
                .extend(split(v));
        } else if let Some(p) = k.strip_suffix("_ALLOW_ALL_USERS") {
            if truthy(v) {
                access
                    .policy
                    .insert(p.to_ascii_lowercase(), AccessPolicy::Open);
            }
        }
    }
    for sub in ["platforms/pairing", "pairing"] {
        let Ok(entries) = fs::read_dir(dir.join(sub)) else {
            continue;
        };
        for entry in entries.flatten() {
            let name = entry.file_name().to_string_lossy().into_owned();
            let Some(platform) = name.strip_suffix("-approved.json") else {
                continue;
            };
            let Ok(text) = fs::read_to_string(entry.path()) else {
                continue;
            };
            if let Ok(Value::Object(map)) = serde_json::from_str::<Value>(&text) {
                access
                    .allowlist
                    .entry(platform.to_string())
                    .or_default()
                    .extend(map.keys().cloned());
            }
        }
    }
    if let Some(Value::Object(map)) = platforms {
        for (platform, block) in map {
            if global_open {
                access.policy.insert(platform.clone(), AccessPolicy::Open);
            }
            for key in ["allow_admin_from", "group_allow_admin_from"] {
                let found = block
                    .get("extra")
                    .and_then(|e| e.get(key))
                    .or_else(|| block.get(key));
                let ids = match found {
                    Some(Value::String(s)) => split(s),
                    Some(Value::Array(items)) => items
                        .iter()
                        .filter_map(|v| {
                            v.as_str()
                                .map(str::to_string)
                                .or_else(|| v.as_i64().map(|n| n.to_string()))
                        })
                        .collect(),
                    _ => Vec::new(),
                };
                access
                    .admins
                    .entry(platform.clone())
                    .or_default()
                    .extend(ids);
            }
        }
    }
    for list in access
        .allowlist
        .values_mut()
        .chain(access.admins.values_mut())
    {
        list.sort();
        list.dedup();
    }
    access.allowlist.retain(|_, v| !v.is_empty());
    access.admins.retain(|_, v| !v.is_empty());
    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);";
/// Our own ledger: Hermes's shape plus `residue_json`, the columns another
/// harness's run log has and Hermes's does not (an OpenClaw run's delivery
/// status, session key, summary, model, token counts, sequence — ORC-12
/// finding 5), so a run migrated through our folder loses none of them.
const FOLDER_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, residue_json TEXT,
  session_id TEXT, obligation_id 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,
  posted_message_id TEXT, source_json TEXT);";

/// The executions ledger, written fresh.
pub fn write_executions(path: &Path, fires: &[crate::orchestration::Fire]) -> Result<()> {
    write_executions_shaped(path, fires, false)
}

/// `ours`: our folder's ledger carries every fire's residue; a Hermes home's
/// ledger is Hermes's shape exactly.
pub fn write_executions_shaped(
    path: &Path,
    fires: &[crate::orchestration::Fire],
    ours: bool,
) -> Result<()> {
    let mut cols: Vec<&str> = EXECUTION_COLUMNS.to_vec();
    if ours {
        // our ledger also keeps the two links Hermes's derives: a fire's
        // session and the delivery it produced
        cols.extend(["residue_json", "session_id", "obligation_id"]);
    }
    let insert = format!(
        "insert into executions ({}) values ({})",
        cols.join(", "),
        cols.iter().map(|_| "?").collect::<Vec<_>>().join(",")
    );
    let rows: Vec<Vec<Param>> = fires
        .iter()
        .map(|f| {
            let mut row: Vec<Param> = encode_fire_row(f).iter().map(Param::from).collect();
            if ours {
                // what the Hermes columns already hold stays out of the json
                let rest: serde_json::Map<String, Value> = f
                    .residue
                    .0
                    .iter()
                    .filter(|(k, _)| {
                        !["source", "process_id", "pid", "process_started_at"].contains(&k.as_str())
                    })
                    .map(|(k, v)| (k.clone(), v.clone()))
                    .collect();
                row.push(if rest.is_empty() {
                    Param::Null
                } else {
                    Param::Text(serde_json::to_string(&rest).unwrap())
                });
                let opt = |v: &Option<String>| v.clone().map(Param::Text).unwrap_or(Param::Null);
                row.push(opt(&f.session_id));
                row.push(opt(&f.obligation_id));
            }
            row
        })
        .collect();
    write_table(
        path,
        if ours {
            FOLDER_EXECUTIONS_DDL
        } else {
            EXECUTIONS_DDL
        },
        &insert,
        &rows,
    )
}

/// Hermes's `gateway_routing` DDL (`hermes_state_schema.py`).
const ROUTING_DDL: &str = "CREATE TABLE IF NOT EXISTS gateway_routing (
  scope TEXT NOT NULL DEFAULT '', session_key TEXT NOT NULL, entry_json TEXT NOT NULL, updated_at REAL NOT NULL,
  PRIMARY KEY (scope, session_key));";

/// The profile's session entries and delivery obligations into its `state.db`, in place, and the
/// `sessions/sessions.json` mirror Hermes also writes (`gateway.write_sessions_json`, on by default).
fn write_state(dir: &Path, path: &Path, profile: &Profile) -> Result<()> {
    fs::create_dir_all(dir.join("sessions"))?;
    let scope = routing_scope(dir);
    let now = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_secs_f64())
        .unwrap_or(0.0);
    // Hermes routes one session per key: where several bindings share a surface (a job's successive fires, a
    // conversation reset and begun again), the live one is the key's, else the most recent
    let rank = |b: &Binding| {
        (
            b.ended_at.is_none(),
            b.last_activity_at
                .clone()
                .or_else(|| b.started_at.clone())
                .unwrap_or_default(),
        )
    };
    let mut chosen: BTreeMap<String, (&Binding, Value)> = BTreeMap::new();
    for (slot, b) in &profile.bindings {
        let (key, entry) = routing_entry(&profile.name, slot, b);
        if chosen
            .get(&key)
            .is_none_or(|(held, _)| rank(b) > rank(held))
        {
            chosen.insert(key, (b, entry));
        }
    }
    let mut mirror = Map::new();
    let rows: Vec<Vec<Param>> = chosen
        .into_iter()
        .map(|(key, (_, entry))| {
            mirror.insert(key.clone(), entry.clone());
            vec![
                Param::Text(scope.clone()),
                Param::Text(key),
                Param::Text(serde_json::to_string(&entry).unwrap()),
                Param::Real(now),
            ]
        })
        .collect();
    replace_rows(
        path,
        ROUTING_DDL,
        "gateway_routing",
        &[],
        ("delete from gateway_routing where scope = ?", &[Param::Text(scope)]),
        "insert into gateway_routing (scope, session_key, entry_json, updated_at) values (?, ?, ?, ?)",
        &rows,
    )?;
    let mirror_file = dir.join("sessions/sessions.json");
    let tmp = mirror_file.with_file_name(format!("sessions.json.tmp-{}", std::process::id()));
    fs::write(
        &tmp,
        serde_json::to_string_pretty(&Value::Object(mirror)).unwrap(),
    )?;
    fs::rename(&tmp, &mirror_file)?;
    // Hermes's columns, then ours (the sent message's handle, the source), added where Hermes's table lacks them
    let cols: Vec<&str> = OBLIGATION_COLUMNS
        .iter()
        .chain(FOLDER_OBLIGATION_EXTRA_COLUMNS.iter())
        .copied()
        .collect();
    let extras: Vec<(&str, &str)> = FOLDER_OBLIGATION_EXTRA_COLUMNS
        .iter()
        .map(|c| (*c, "TEXT"))
        .collect();
    let insert = format!(
        "insert into delivery_obligations ({}) values ({})",
        cols.join(", "),
        cols.iter().map(|_| "?").collect::<Vec<_>>().join(",")
    );
    let rows: Vec<Vec<Param>> = profile
        .obligations
        .iter()
        .map(|o| {
            encode_obligation_row(o)
                .iter()
                .chain(encode_obligation_folder_extras(o).iter())
                .map(Param::from)
                .collect()
        })
        .collect();
    replace_rows(
        path,
        OBLIGATIONS_DDL,
        "delivery_obligations",
        &extras,
        ("delete from delivery_obligations", &[]),
        &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);
    // Raw bytes are reused only for a file read from OUR folder: another
    // harness's bytes may carry inline secrets our folder must never hold.
    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; // from here on the bytes on disk are ours
    Ok(true)
}

/// Save OUR folder. Never deletes a file it does not own; never writes a secret outside `.env`.
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))
            }
        },
    )?;
    // sqlite: rewrite only when the decoded rows changed
    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_shaped(&tmp, &profile.fires, true)?;
        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())
    {
        // in place: a Hermes home's state.db holds Hermes's sessions and messages beside these tables
        write_state(dir, &state_path, profile)?;
        meta.snapshot.insert("state.db".into(), state_snap);
    }

    // .env: every ref this profile uses
    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(())
}

/// Copy every unmodeled file of a profile from its source dir into `dest`, byte for byte.
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(())
}

/// Copy the named unmodeled files from `src` into `dest`, byte for byte,
/// answering the relative paths carried. A file the source no longer holds is
/// skipped: the list is what was seen at load, the copy is what is there now.
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)
}