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_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_folder_extras,
    encode_obligation_row, encode_route, encode_subscription, encode_surface_key, 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, table_exists, write_table, Param};
use crate::ontology::{
    hermes_cron_job_id, parse_hermes_session_key, Binding, EndReason, Handoff, HarnessId,
    HermesSessionRow, Recurrence, Residue, SurfaceKey, Trigger, Worker,
};
use crate::orchestration::{ExpiryPolicy, Orchestration, PersonaRef, Profile, ProfileResidue};
use crate::Result;

/// The files our folder owns.
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"];

/// 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
    })
}

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(),
    }
}

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

/// The record `config.yaml`'s bytes are the serialization of.
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 })
}

/// 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",
];

/// 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"))?;
    profile.expiry = decode_expiry(&cfg_file, cfg_map.get("expiry"))?;
    profile.home = decode_home(&cfg_file, cfg_map.get("home"))?;
    let gateway = cfg_map.get("gateway").and_then(Value::as_object);
    let routes_raw: Vec<Value> = match cfg_map.get("profile_routes").and_then(Value::as_array) {
        Some(a) => {
            meta.routes_at_top = true;
            a.clone()
        }
        None => gateway
            .and_then(|g| g.get("profile_routes"))
            .and_then(Value::as_array)
            .cloned()
            .unwrap_or_default(),
    };
    for (i, r) in routes_raw.iter().enumerate() {
        profile.routes.push(decode_route(&cfg_file, i, r)?);
    }
    if let Some(platforms) = cfg_map.get("platforms") {
        let map = platforms
            .as_object()
            .ok_or_else(|| load_error(&cfg_file, "platforms", "expected a map"))?;
        let known: BTreeSet<String> = vault.keys().cloned().collect();
        for (platform, raw) in map {
            profile.channels.insert(
                platform.clone(),
                decode_channel(&cfg_file, platform, raw, vault)?,
            );
        }
        config_normalized =
            flavor == Flavor::Orchestrator && vault.keys().any(|k| !known.contains(k));
    }
    // 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
    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(),
    );

    // 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 {
        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));

    // 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", "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 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() {
            wm.insert("env".into(), serde_json::to_value(&w.env).unwrap());
        }
        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));
    }
    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))
}

/// 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))
}

/// `access.yaml`.
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);";
/// 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);";

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);";

/// 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,
    )
}

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,
    )?;
    // Hermes's columns, then ours (the sent message's handle, the source)
    let cols: Vec<&str> = OBLIGATION_COLUMNS
        .iter()
        .chain(FOLDER_OBLIGATION_EXTRA_COLUMNS.iter())
        .copied()
        .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();
    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);
    // 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))
            }
        },
    )?;
    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))
            }
        },
    )?;

    // 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())
    {
        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);
    }

    // .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)
}