Skip to main content

supercode_interchange/orchestration/codec/
folder.rs

1//! The folder codec: a home folder ⇄ [`Orchestration`]. The root is the
2//! `default` profile; `profiles/<name>/` are the named ones; a profile folder
3//! is a complete home (Hermes's rule). Two flavors of the same layout:
4//!
5//! * `Orchestrator` — ours: the O-blocks in `config.yaml` decode strictly,
6//!   `AGENTS.md` is the persona, `state.db` holds our `bindings` table.
7//! * `Hermes` — a Hermes home: `SOUL.md` is the persona, `state.db`'s
8//!   `sessions` rows become bindings (the multiplexed store partitioned by
9//!   `profile_name`), and the file's bytes may never be reused for our folder
10//!   (they may carry inline credentials).
11//!
12//! Files we own are re-emitted byte for byte when their record is unchanged
13//! and canonically when it changed; files we do not model are never touched.
14
15use std::collections::{BTreeMap, BTreeSet};
16use std::fs;
17use std::path::{Path, PathBuf};
18
19use serde_json::{Map, Value};
20use sha2::{Digest, Sha256};
21
22use super::canonical::canonical_json;
23use super::decode::{
24    decode_access, decode_binding_row, decode_channel, decode_expiry, decode_fire_row, decode_home,
25    decode_job, decode_obligation_row, decode_route, decode_subscription, decode_worker,
26    encode_access, encode_channel, encode_fire_row, encode_job, encode_obligation_folder_extras,
27    encode_obligation_row, encode_route, encode_subscription, encode_surface_key, load_error,
28    surface_key_string, EXECUTION_COLUMNS, FOLDER_OBLIGATION_EXTRA_COLUMNS, OBLIGATION_COLUMNS,
29};
30use super::dotenv::{parse_dotenv, render_dotenv};
31use super::sqlite::{read_rows, table_exists, write_table, Param};
32use crate::ontology::{
33    hermes_cron_job_id, parse_hermes_session_key, Binding, EndReason, Handoff, HarnessId,
34    HermesSessionRow, Recurrence, Residue, SurfaceKey, Trigger, Worker,
35};
36use crate::orchestration::{ExpiryPolicy, Orchestration, PersonaRef, Profile, ProfileResidue};
37use crate::Result;
38
39/// The files our folder owns.
40pub const OWNED_FILES: &[&str] = &[
41    "config.yaml",
42    "AGENTS.md",
43    "CLAUDE.md",
44    "access.yaml",
45    ".env",
46    "cron/jobs.json",
47    "cron/executions.db",
48    "webhook_subscriptions.json",
49    "state.db",
50];
51
52const CONFIG_O_KEYS: &[&str] = &["worker", "expiry", "home"];
53
54/// Which harness's layout a folder was read as.
55#[derive(Debug, Clone, Copy, PartialEq, Eq)]
56pub enum Flavor {
57    /// Our own folder.
58    Orchestrator,
59    /// A Hermes home.
60    Hermes,
61}
62
63/// The form `cron/jobs.json` was read in, remembered for the emit.
64#[derive(Debug, Clone, PartialEq)]
65pub struct JobsForm {
66    /// `{"jobs": [...], ...}` (true) or a bare array (false).
67    pub object: bool,
68    /// The envelope's other keys (`updated_at`, …), verbatim.
69    pub extras: Map<String, Value>,
70}
71
72/// Per-profile bookkeeping: what a decompile needs to reuse bytes and to know
73/// what "unchanged" means. Never part of the orchestration.
74#[derive(Debug, Clone)]
75pub struct ProfileIo {
76    /// Raw text of every owned file read.
77    pub raw: BTreeMap<String, String>,
78    /// Canonical JSON of each artifact's record as loaded.
79    pub snapshot: BTreeMap<String, String>,
80    /// Where the profile was read from.
81    pub source_dir: Option<PathBuf>,
82    /// The flavor it was read as; only our own bytes may be written back verbatim.
83    pub flavor: Flavor,
84    /// The jobs file's form.
85    pub jobs_form: Option<JobsForm>,
86    /// Whether `profile_routes` sat at the top level (else under `gateway`).
87    pub routes_at_top: bool,
88    /// A Hermes profile that borrowed its partition from the root store.
89    pub borrowed_from: Option<PathBuf>,
90    /// The named profiles that borrowed from this root store.
91    pub lenders: Vec<String>,
92}
93
94impl ProfileIo {
95    fn new(flavor: Flavor) -> Self {
96        Self {
97            raw: BTreeMap::new(),
98            snapshot: BTreeMap::new(),
99            source_dir: None,
100            flavor,
101            jobs_form: None,
102            routes_at_top: false,
103            borrowed_from: None,
104            lenders: Vec::new(),
105        }
106    }
107}
108
109/// A loaded home.
110#[derive(Debug, Clone)]
111pub struct LoadedHome {
112    /// The orchestration.
113    pub orchestration: Orchestration,
114    /// Secret values by `.env` key; never in the orchestration.
115    pub vault: BTreeMap<String, String>,
116    /// Per-profile bookkeeping.
117    pub io: BTreeMap<String, ProfileIo>,
118}
119
120fn sha256_hex(text: &str) -> String {
121    let mut h = Sha256::new();
122    h.update(text.as_bytes());
123    h.finalize().iter().map(|b| format!("{b:02x}")).collect()
124}
125
126/// A persona record for `text`: `AGENTS.md` at the profile root, content hash first.
127pub fn persona_ref(text: &str) -> PersonaRef {
128    PersonaRef {
129        path: "AGENTS.md".into(),
130        text: Some(text.to_string()),
131        sha256: sha256_hex(text),
132    }
133}
134
135fn read_text(dir: &Path, rel: &str) -> Result<Option<String>> {
136    let p = dir.join(rel);
137    if !p.is_file() {
138        return Ok(None);
139    }
140    Ok(Some(fs::read_to_string(&p)?))
141}
142
143fn yaml_to_json(file: &str, text: &str) -> Result<Value> {
144    let value: serde_yaml::Value =
145        serde_yaml::from_str(text).map_err(|e| load_error(file, "", format!("YAML: {e}")))?;
146    let json: Value =
147        serde_json::to_value(value).map_err(|e| load_error(file, "", format!("YAML: {e}")))?;
148    Ok(if json.is_null() {
149        Value::Object(Map::new())
150    } else {
151        json
152    })
153}
154
155fn json_to_yaml(value: &Value) -> String {
156    let y: serde_yaml::Value =
157        serde_json::from_value(value.clone()).unwrap_or(serde_yaml::Value::Null);
158    match value.as_object() {
159        Some(m) if m.is_empty() => String::new(),
160        _ => serde_yaml::to_string(&y).unwrap_or_default(),
161    }
162}
163
164/// The empty profile.
165pub fn empty_profile(name: &str, dir: &Path) -> Profile {
166    Profile {
167        name: name.to_string(),
168        dir: dir.to_path_buf(),
169        worker: None,
170        persona: None,
171        channels: BTreeMap::new(),
172        routes: Vec::new(),
173        expiry: ExpiryPolicy::default(),
174        home: None,
175        jobs: BTreeMap::new(),
176        subscriptions: BTreeMap::new(),
177        access: Default::default(),
178        bindings: BTreeMap::new(),
179        fires: Vec::new(),
180        obligations: Vec::new(),
181        residue: ProfileResidue::default(),
182    }
183}
184
185/// The record `config.yaml`'s bytes are the serialization of.
186pub fn config_record(profile: &Profile) -> Value {
187    serde_json::json!({
188        "worker": profile.worker, "expiry": profile.expiry, "home": profile.home.as_ref().map(encode_surface_key),
189        "routes": profile.routes, "channels": profile.channels, "residue": profile.residue.config,
190    })
191}
192
193fn state_record(profile: &Profile) -> Value {
194    serde_json::json!({ "bindings": profile.bindings, "obligations": profile.obligations })
195}
196
197/// Load a home folder.
198pub fn load_home(root: &Path, flavor: Flavor) -> Result<LoadedHome> {
199    if !root.is_dir() {
200        return Err(load_error(
201            &root.display().to_string(),
202            "",
203            "not a directory",
204        ));
205    }
206    let mut vault = BTreeMap::new();
207    let mut io = BTreeMap::new();
208    let mut profiles = BTreeMap::new();
209    let (default, default_io) = load_profile_dir("default", root, flavor, &mut vault)?;
210    profiles.insert("default".to_string(), default);
211    io.insert("default".to_string(), default_io);
212    let profiles_dir = root.join("profiles");
213    if profiles_dir.is_dir() {
214        let mut names: Vec<String> = fs::read_dir(&profiles_dir)?
215            .flatten()
216            .filter(|e| e.path().is_dir())
217            .filter_map(|e| e.file_name().into_string().ok())
218            .filter(|n| n != "node_modules" && !n.starts_with('.'))
219            .collect();
220        names.sort();
221        for name in names {
222            let dir = profiles_dir.join(&name);
223            if name == "default" {
224                return Err(load_error(
225                    &dir.display().to_string(),
226                    "",
227                    "\"default\" is the root folder, not a named profile",
228                ));
229            }
230            let (profile, meta) = load_profile_dir(&name, &dir, flavor, &mut vault)?;
231            profiles.insert(name.clone(), profile);
232            io.insert(name, meta);
233        }
234    }
235    let mut loaded = LoadedHome {
236        orchestration: Orchestration {
237            root: root.to_path_buf(),
238            profiles,
239        },
240        vault,
241        io,
242    };
243    if flavor == Flavor::Hermes {
244        partition_shared_store(&mut loaded, root)?;
245    }
246    link_fires(&mut loaded, root)?;
247    Ok(loaded)
248}
249
250/// A Hermes-shaped ledger writes a fire's session and delivery nowhere. The
251/// session is the one minted `cron_<job>_<instant>` inside the fire's window
252/// (its compression tip, when the conversation rolled over); the delivery is
253/// the obligation that session's key — else the job's surface — raised in the
254/// window. `Fire.session_id` / `Fire.obligation_id` are derived here, the
255/// obligation's source points back, and the encoders never write either
256/// (`docs/ONTOLOGY.md` §2.7).
257fn link_fires(loaded: &mut LoadedHome, root: &Path) -> Result<()> {
258    let names: Vec<String> = loaded.orchestration.profiles.keys().cloned().collect();
259    let root_store = root.join("state.db");
260    for name in &names {
261        let own = loaded.orchestration.profiles[name].dir.join("state.db");
262        let store = if own.is_file() {
263            own
264        } else {
265            root_store.clone()
266        };
267        let sessions: Vec<Map<String, Value>> = if table_exists(&store, "sessions") {
268            read_rows(
269                &store,
270                "select id, started_at, end_reason, parent_session_id, session_key from sessions",
271                &[],
272            )?
273            .unwrap_or_default()
274        } else {
275            Vec::new()
276        };
277        // the obligations that store holds: a shared root store was
278        // partitioned across the profiles lending from it
279        let holders: Vec<String> = {
280            let io = &loaded.io[name];
281            if io.borrowed_from.is_some() || !io.lenders.is_empty() {
282                let mut v = vec!["default".to_string()];
283                v.extend(loaded.io["default"].lenders.iter().cloned());
284                v
285            } else {
286                vec![name.clone()]
287            }
288        };
289        // an obligation that already names its fire (our folder persists the
290        // source) is the link; the window join below is for stores that do not
291        let mut by_fire: BTreeMap<String, (String, String)> = BTreeMap::new();
292        for (holder, profile) in &loaded.orchestration.profiles {
293            for o in &profile.obligations {
294                if let crate::orchestration::ObligationSource::Fire { fire_id } = &o.source {
295                    by_fire
296                        .entry(fire_id.clone())
297                        .or_insert_with(|| (holder.clone(), o.id.clone()));
298                }
299            }
300        }
301        let mut links: Vec<(usize, Option<String>, Option<(String, String)>)> = Vec::new();
302        let profile = &loaded.orchestration.profiles[name];
303        for (index, fire) in profile.fires.iter().enumerate() {
304            if let Some(known) = by_fire.get(&fire.id) {
305                // the session, when our ledger persisted it, is already on the fire
306                links.push((index, fire.session_id.clone(), Some(known.clone())));
307                continue;
308            }
309            let session_id = fire_session(&sessions, fire).or_else(|| fire.session_id.clone());
310            let session_key = session_id.as_deref().and_then(|id| {
311                sessions
312                    .iter()
313                    .find(|row| row.get("id").and_then(Value::as_str) == Some(id))
314                    .and_then(|row| row.get("session_key"))
315                    .and_then(Value::as_str)
316                    .filter(|k| !k.is_empty())
317                    .map(str::to_string)
318            });
319            let surface = profile.jobs.get(&fire.job_id).and_then(job_surface);
320            let (Some(from), to) = (
321                iso_epoch(&fire.claimed_at),
322                fire.finished_at
323                    .as_deref()
324                    .and_then(iso_epoch)
325                    .unwrap_or(f64::MAX),
326            ) else {
327                links.push((index, session_id, None));
328                continue;
329            };
330            let in_window = |o: &&crate::orchestration::Obligation| {
331                o.created_at
332                    .parse::<f64>()
333                    .is_ok_and(|at| at >= from && at <= to)
334            };
335            let latest = |mut found: Vec<(&String, &crate::orchestration::Obligation)>| {
336                found.sort_by(|a, b| {
337                    let at = |o: &crate::orchestration::Obligation| {
338                        o.created_at.parse::<f64>().unwrap_or(0.0)
339                    };
340                    at(b.1)
341                        .partial_cmp(&at(a.1))
342                        .unwrap_or(std::cmp::Ordering::Equal)
343                });
344                found
345                    .first()
346                    .map(|(holder, o)| ((*holder).clone(), o.id.clone()))
347            };
348            let candidates = |pick: &dyn Fn(&crate::orchestration::Obligation) -> bool| {
349                holders
350                    .iter()
351                    .flat_map(|h| {
352                        loaded.orchestration.profiles[h]
353                            .obligations
354                            .iter()
355                            .filter(in_window)
356                            .filter(|o| pick(o))
357                            .map(move |o| (h, o))
358                    })
359                    .collect::<Vec<_>>()
360            };
361            let obligation = match &session_key {
362                Some(key) => latest(candidates(&|o| o.session_key.as_deref() == Some(key))),
363                None => None,
364            }
365            .or_else(|| {
366                let (platform, chat_id) = surface.as_ref()?;
367                latest(candidates(&|o| {
368                    o.target.platform.as_deref() == Some(platform)
369                        && o.target.chat_id.as_deref() == Some(chat_id)
370                }))
371            });
372            links.push((index, session_id, obligation));
373        }
374        for (index, session_id, obligation) in links {
375            let fire_id = {
376                let fire = &mut loaded.orchestration.profiles.get_mut(name).unwrap().fires[index];
377                fire.session_id = session_id;
378                fire.obligation_id = obligation.as_ref().map(|(_, id)| id.clone());
379                fire.id.clone()
380            };
381            if let Some((holder, obligation_id)) = obligation {
382                if let Some(o) = loaded
383                    .orchestration
384                    .profiles
385                    .get_mut(&holder)
386                    .and_then(|p| p.obligations.iter_mut().find(|o| o.id == obligation_id))
387                {
388                    o.source = crate::orchestration::ObligationSource::Fire { fire_id };
389                }
390            }
391        }
392    }
393    // the links are derived: the records the stores hold are unchanged
394    for name in &names {
395        let profile = &loaded.orchestration.profiles[name];
396        let io = loaded.io.get_mut(name).unwrap();
397        io.snapshot.insert(
398            "cron/executions.db".into(),
399            canonical_json(&serde_json::to_value(&profile.fires).unwrap()),
400        );
401        io.snapshot
402            .insert("state.db".into(), canonical_json(&state_record(profile)));
403    }
404    Ok(())
405}
406
407/// The platform surface a job's fires deliver to, when the job names one an
408/// obligation could carry: `origin` (the creating conversation) and the
409/// explicit `<platform>:<chat>` form. `local` and `home` deliver nowhere a
410/// platform ledger sees.
411fn job_surface(job: &crate::orchestration::Job) -> Option<(String, String)> {
412    use crate::orchestration::Target;
413    match &job.deliver {
414        Target::Origin => {
415            let origin = job.origin.as_ref()?;
416            Some((origin.platform.clone(), origin.chat_id.clone()?))
417        }
418        Target::Explicit {
419            platform, chat_id, ..
420        } => Some((platform.clone(), chat_id.clone()?)),
421        Target::Home | Target::Local => None,
422    }
423}
424
425/// The session a Hermes fire ran in: the `cron_<job>_<YYYYMMDD_HHMMSS>` id
426/// minted inside [claimed, finished] — the earliest while the fire still
427/// runs, the latest once it finished — followed to its compression tip.
428fn fire_session(
429    sessions: &[Map<String, Value>],
430    fire: &crate::orchestration::Fire,
431) -> Option<String> {
432    let claimed = instant_key(&fire.claimed_at)?;
433    let finished = fire.finished_at.as_deref().and_then(instant_key);
434    let candidates = sessions.iter().filter_map(|row| {
435        let id = row.get("id").and_then(Value::as_str)?;
436        let key = cron_session_instant(id, &fire.job_id)?;
437        (key >= claimed && finished.is_none_or(|f| key <= f)).then(|| (key, id.to_string()))
438    });
439    let chosen = match finished {
440        Some(_) => candidates.max_by_key(|(key, _)| *key),
441        None => candidates.min_by_key(|(key, _)| *key),
442    }?;
443    let mut current = chosen.1;
444    for _ in 0..32 {
445        let row = sessions
446            .iter()
447            .find(|row| row.get("id").and_then(Value::as_str) == Some(current.as_str()));
448        let compressed = row
449            .and_then(|row| row.get("end_reason"))
450            .and_then(Value::as_str)
451            == Some("compression");
452        if !compressed {
453            return Some(current);
454        }
455        let next = sessions
456            .iter()
457            .filter(|row| {
458                row.get("parent_session_id").and_then(Value::as_str) == Some(current.as_str())
459            })
460            .max_by(|a, b| {
461                let at = |r: &Map<String, Value>| {
462                    r.get("started_at").and_then(Value::as_f64).unwrap_or(0.0)
463                };
464                at(a)
465                    .partial_cmp(&at(b))
466                    .unwrap_or(std::cmp::Ordering::Equal)
467                    .then_with(|| {
468                        let id = |r: &Map<String, Value>| {
469                            r.get("id")
470                                .and_then(Value::as_str)
471                                .unwrap_or("")
472                                .to_string()
473                        };
474                        id(a).cmp(&id(b))
475                    })
476            })
477            .and_then(|row| row.get("id").and_then(Value::as_str).map(str::to_string));
478        match next {
479            None => return Some(current),
480            Some(next) => current = next,
481        }
482    }
483    Some(current)
484}
485
486/// `YYYYMMDDHHMMSS` of an ISO instant, for ordering against a cron session id.
487fn instant_key(iso: &str) -> Option<u64> {
488    let digits: String = iso
489        .chars()
490        .take_while(|c| *c != '+' && *c != 'Z')
491        .filter(char::is_ascii_digit)
492        .collect();
493    (digits.len() >= 14).then(|| digits[..14].parse().ok())?
494}
495
496/// `YYYYMMDDHHMMSS` of a `cron_<job>_<YYYYMMDD_HHMMSS>` session id, when it is `job`'s.
497fn cron_session_instant(session_id: &str, job_id: &str) -> Option<u64> {
498    if hermes_cron_job_id(session_id).as_deref() != Some(job_id) {
499        return None;
500    }
501    let stamp = session_id.rsplit_once('_')?;
502    let date = stamp.0.rsplit_once('_')?.1;
503    format!("{date}{}", stamp.1).parse().ok()
504}
505
506/// Seconds since the epoch of an ISO-8601 instant with `Z` or `±HH:MM`.
507pub(crate) fn iso_epoch(iso: &str) -> Option<f64> {
508    let (instant, offset) = if let Some(instant) = iso.strip_suffix('Z') {
509        (instant, 0.0)
510    } else {
511        let time_at = iso.find('T')?;
512        let sign_at = iso[time_at..].find(['+', '-']).map(|i| i + time_at)?;
513        let (instant, offset) = iso.split_at(sign_at);
514        let (hours, minutes) = offset[1..].split_once(':')?;
515        let seconds = hours.parse::<f64>().ok()? * 3_600.0 + minutes.parse::<f64>().ok()? * 60.0;
516        (
517            instant,
518            if offset.starts_with('-') {
519                -seconds
520            } else {
521                seconds
522            },
523        )
524    };
525    let (date, time) = instant.split_once('T')?;
526    let mut date = date.splitn(3, '-');
527    let year: i64 = date.next()?.parse().ok()?;
528    let month: i64 = date.next()?.parse().ok()?;
529    let day: i64 = date.next()?.parse().ok()?;
530    let mut clock = time.splitn(3, ':');
531    let hour: i64 = clock.next()?.parse().ok()?;
532    let minute: i64 = clock.next()?.parse().ok()?;
533    let seconds: f64 = clock.next()?.parse().ok()?;
534    let year = year - i64::from(month <= 2);
535    let era = year.div_euclid(400);
536    let yoe = year - era * 400;
537    let doy = (153 * (if month > 2 { month - 3 } else { month + 9 }) + 2) / 5 + day - 1;
538    let doe = yoe * 365 + yoe / 4 - yoe / 100 + doy;
539    let days = era * 146_097 + doe - 719_468;
540    Some((days * 86_400 + hour * 3_600 + minute * 60) as f64 + seconds - offset)
541}
542
543/// Hermes's multiplexed gateway keeps every profile's rows in ONE store,
544/// partitioned by `profile_name` (sessions) and by the session key's profile
545/// segment (obligations). A named profile without its own `state.db` borrows
546/// its partition from the root store; the root profile keeps the rest.
547fn partition_shared_store(loaded: &mut LoadedHome, root: &Path) -> Result<()> {
548    let root_path = root.join("state.db");
549    if !root_path.is_file() {
550        return Ok(());
551    }
552    let names: Vec<String> = loaded
553        .orchestration
554        .profiles
555        .keys()
556        .filter(|n| *n != "default")
557        .cloned()
558        .collect();
559    for name in names {
560        let has_own = loaded.orchestration.profiles[&name]
561            .dir
562            .join("state.db")
563            .is_file();
564        if has_own {
565            continue;
566        }
567        loaded.io.get_mut(&name).unwrap().borrowed_from = Some(root_path.clone());
568        loaded
569            .io
570            .get_mut("default")
571            .unwrap()
572            .lenders
573            .push(name.clone());
574        if table_exists(&root_path, "sessions") {
575            let rows = read_rows(
576                &root_path,
577                "select * from sessions where profile_name = ?1 order by started_at, id",
578                &[&name],
579            )?
580            .unwrap_or_default();
581            for row in rows {
582                if let Some(b) = binding_from_hermes_session(&root_path, &row, &name) {
583                    loaded
584                        .orchestration
585                        .profiles
586                        .get_mut(&name)
587                        .unwrap()
588                        .bindings
589                        .insert(surface_key_string(&b.key), b);
590                }
591            }
592        }
593        // obligations addressed to this profile's surfaces move out of the root partition
594        let root_profile = loaded.orchestration.profiles.get_mut("default").unwrap();
595        let (mine, rest): (Vec<_>, Vec<_>) = root_profile.obligations.drain(..).partition(|o| {
596            o.session_key
597                .as_deref()
598                .and_then(parse_hermes_session_key)
599                .and_then(|(_, p)| p)
600                .as_deref()
601                == Some(name.as_str())
602        });
603        root_profile.obligations = rest;
604        loaded
605            .orchestration
606            .profiles
607            .get_mut(&name)
608            .unwrap()
609            .obligations = mine;
610        let snap = canonical_json(&state_record(&loaded.orchestration.profiles[&name]));
611        loaded
612            .io
613            .get_mut(&name)
614            .unwrap()
615            .snapshot
616            .insert("state.db".into(), snap);
617    }
618    let snap = canonical_json(&state_record(&loaded.orchestration.profiles["default"]));
619    loaded
620        .io
621        .get_mut("default")
622        .unwrap()
623        .snapshot
624        .insert("state.db".into(), snap);
625    Ok(())
626}
627
628const HERMES_SESSION_MAPPED: &[&str] = &[
629    "id",
630    "source",
631    "session_key",
632    "chat_id",
633    "chat_type",
634    "thread_id",
635    "user_id",
636    "profile_name",
637    "handoff_state",
638    "handoff_platform",
639    "handoff_error",
640    "started_at",
641    "ended_at",
642    "end_reason",
643];
644
645/// A Hermes `sessions` row is a binding when it has a surface (session_key) or is a cron fire.
646pub fn binding_from_hermes_session(
647    file: &Path,
648    row: &Map<String, Value>,
649    profile_name: &str,
650) -> Option<Binding> {
651    let text = |k: &str| {
652        row.get(k)
653            .and_then(|v| match v {
654                Value::String(s) => Some(s.clone()),
655                Value::Number(n) => Some(n.to_string()),
656                _ => None,
657            })
658            .filter(|s| !s.is_empty())
659    };
660    let num = |k: &str| row.get(k).and_then(Value::as_f64);
661    let id = text("id")?;
662    let source = text("source");
663    // The surface may live only in `session_key` (api_server conversations
664    // carry no chat_type/chat_id columns — ORC-8 finding).
665    let parsed = text("session_key").and_then(|k| parse_hermes_session_key(&k));
666    let chat_type = text("chat_type").or_else(|| parsed.as_ref().and_then(|(k, _)| k.kind.clone()));
667    let key = if text("session_key").is_some()
668        && chat_type
669            .as_deref()
670            .is_some_and(|c| super::decode::CHAT_TYPES.contains(&c))
671    {
672        SurfaceKey {
673            key: None,
674            platform: source
675                .clone()
676                .or_else(|| parsed.as_ref().and_then(|(k, _)| k.platform.clone())),
677            kind: chat_type,
678            chat_id: text("chat_id")
679                .or_else(|| parsed.as_ref().and_then(|(k, _)| k.chat_id.clone())),
680            thread_id: text("thread_id")
681                .or_else(|| parsed.as_ref().and_then(|(k, _)| k.thread_id.clone())),
682            participant_id: parsed.as_ref().and_then(|(k, _)| k.participant_id.clone()),
683        }
684    } else if source.as_deref() == Some("cron") {
685        let job = crate::ontology::hermes_cron_job_id(&id).unwrap_or_else(|| id.clone());
686        SurfaceKey {
687            key: None,
688            platform: Some("cron".into()),
689            kind: Some("dm".into()),
690            chat_id: Some(job),
691            thread_id: None,
692            participant_id: None,
693        }
694    } else {
695        return None;
696    };
697    if let Some(p) = text("profile_name") {
698        if p != profile_name && !(profile_name == "default" && p == "main") {
699            return None; // a foreign partition is not this profile's binding
700        }
701    }
702    let iso = |v: Option<f64>| v.map(epoch_iso);
703    let end_word = text("end_reason");
704    let end_reason = end_word.as_deref().and_then(EndReason::parse);
705    let mut residue = Residue::default();
706    for (k, v) in row {
707        if !HERMES_SESSION_MAPPED.contains(&k.as_str()) && !v.is_null() {
708            residue.keep(k.clone(), v.clone());
709        }
710    }
711    if let (Some(word), None) = (&end_word, end_reason) {
712        residue.keep("end_reason", Value::String(word.clone()));
713    }
714    if let Some(u) = text("user_id") {
715        residue.keep("user_id", Value::String(u));
716    }
717    let recurrence = if key.platform.as_deref() == Some("cron") {
718        key.chat_id.clone().map(|job_id| Recurrence {
719            job_id,
720            kind: "cron".into(),
721        })
722    } else {
723        None
724    };
725    Some(Binding {
726        trigger: match (recurrence.is_some(), source.as_deref()) {
727            (true, _) => Trigger::Cron,
728            (_, Some(s)) => crate::ontology::hermes_trigger_for_source(s),
729            _ => Trigger::Unknown,
730        },
731        key,
732        profile: None,
733        worker: Worker {
734            harness: HarnessId::new(HarnessId::HERMES),
735            session_id: Some(id),
736            locator: Some(file.display().to_string()),
737        },
738        recurrence,
739        handoff: text("handoff_state").map(|state| Handoff {
740            to: text("handoff_platform"),
741            state,
742            error: text("handoff_error"),
743        }),
744        started_at: iso(num("started_at")),
745        last_activity_at: iso(num("ended_at").or_else(|| num("started_at"))),
746        ended_at: iso(num("ended_at")),
747        end_reason,
748        residue,
749    })
750}
751
752/// `new Date(seconds * 1000).toISOString()`.
753fn epoch_iso(seconds: f64) -> String {
754    let row = HermesSessionRow {
755        started_at: Some(seconds),
756        ..Default::default()
757    };
758    Binding::from_hermes_row(&row, None)
759        .started_at
760        .unwrap_or_default()
761}
762
763fn load_profile_dir(
764    name: &str,
765    dir: &Path,
766    flavor: Flavor,
767    vault: &mut BTreeMap<String, String>,
768) -> Result<(Profile, ProfileIo)> {
769    let mut profile = empty_profile(name, dir);
770    let mut config_normalized = false;
771    let mut subs_normalized = false;
772    let mut meta = ProfileIo::new(flavor);
773    meta.source_dir = Some(dir.to_path_buf());
774    let remember = |meta: &mut ProfileIo, rel: &str, raw: Option<String>, record: &Value| {
775        if let Some(raw) = raw {
776            meta.raw.insert(rel.to_string(), raw);
777        }
778        meta.snapshot
779            .insert(rel.to_string(), canonical_json(record));
780    };
781
782    // .env → vault (values), never into the model.
783    if let Some(env) = read_text(dir, ".env")? {
784        for (k, v) in parse_dotenv(&env) {
785            vault.insert(k, v);
786        }
787        meta.raw.insert(".env".into(), env);
788    }
789
790    // config.yaml
791    let cfg_file = dir.join("config.yaml").display().to_string();
792    let cfg_text = read_text(dir, "config.yaml")?;
793    let cfg = match &cfg_text {
794        Some(text) => yaml_to_json(&cfg_file, text)?,
795        None => Value::Object(Map::new()),
796    };
797    let cfg_map = cfg
798        .as_object()
799        .ok_or_else(|| load_error(&cfg_file, "", "expected a mapping"))?;
800    profile.worker = decode_worker(&cfg_file, cfg_map.get("worker"))?;
801    profile.expiry = decode_expiry(&cfg_file, cfg_map.get("expiry"))?;
802    profile.home = decode_home(&cfg_file, cfg_map.get("home"))?;
803    let gateway = cfg_map.get("gateway").and_then(Value::as_object);
804    let routes_raw: Vec<Value> = match cfg_map.get("profile_routes").and_then(Value::as_array) {
805        Some(a) => {
806            meta.routes_at_top = true;
807            a.clone()
808        }
809        None => gateway
810            .and_then(|g| g.get("profile_routes"))
811            .and_then(Value::as_array)
812            .cloned()
813            .unwrap_or_default(),
814    };
815    for (i, r) in routes_raw.iter().enumerate() {
816        profile.routes.push(decode_route(&cfg_file, i, r)?);
817    }
818    if let Some(platforms) = cfg_map.get("platforms") {
819        let map = platforms
820            .as_object()
821            .ok_or_else(|| load_error(&cfg_file, "platforms", "expected a map"))?;
822        let known: BTreeSet<String> = vault.keys().cloned().collect();
823        for (platform, raw) in map {
824            profile.channels.insert(
825                platform.clone(),
826                decode_channel(&cfg_file, platform, raw, vault)?,
827            );
828        }
829        config_normalized =
830            flavor == Flavor::Orchestrator && vault.keys().any(|k| !known.contains(k));
831    }
832    // Everything else in config.yaml is Hermes's and rides as residue.
833    for (k, v) in cfg_map {
834        if CONFIG_O_KEYS.contains(&k.as_str()) || k == "platforms" || k == "profile_routes" {
835            continue;
836        }
837        if k == "gateway" {
838            let mut g = v.as_object().cloned().unwrap_or_default();
839            g.remove("profile_routes");
840            if !g.is_empty() {
841                profile
842                    .residue
843                    .config
844                    .insert("gateway".into(), Value::Object(g));
845            }
846            continue;
847        }
848        profile.residue.config.insert(k.clone(), v.clone());
849    }
850    remember(&mut meta, "config.yaml", cfg_text, &config_record(&profile));
851    if config_normalized {
852        // our flavor: a credential VALUE was pasted inline and has moved to the
853        // vault, so the bytes on disk are not this record's and the next save
854        // re-emits the file with the ref (and `.env` receives the value). A
855        // Hermes home keeps values inline by design; its bytes stay reusable.
856        meta.raw.remove("config.yaml");
857        meta.snapshot.remove("config.yaml");
858    }
859
860    // persona
861    let persona_file = if flavor == Flavor::Hermes {
862        "SOUL.md"
863    } else {
864        "AGENTS.md"
865    };
866    let persona_text = read_text(dir, persona_file)?;
867    profile.persona = persona_text.as_ref().map(|t| PersonaRef {
868        path: "AGENTS.md".into(),
869        text: Some(t.clone()),
870        sha256: sha256_hex(t),
871    });
872    remember(
873        &mut meta,
874        persona_file,
875        persona_text,
876        &serde_json::to_value(&profile.persona).unwrap(),
877    );
878
879    // jobs — three forms load; the file's own form is remembered for the emit
880    let jobs_file = dir.join("cron/jobs.json").display().to_string();
881    let jobs_text = read_text(dir, "cron/jobs.json")?;
882    if let Some(text) = &jobs_text {
883        let parsed: Value = serde_json::from_str(text)
884            .map_err(|e| load_error(&jobs_file, "", format!("JSON: {e}")))?;
885        let arr: Vec<Value> = match &parsed {
886            Value::Array(a) => {
887                meta.jobs_form = Some(JobsForm {
888                    object: false,
889                    extras: Map::new(),
890                });
891                a.clone()
892            }
893            Value::Object(o) => match o.get("jobs") {
894                Some(Value::Array(a)) => {
895                    let mut extras = o.clone();
896                    extras.remove("jobs");
897                    meta.jobs_form = Some(JobsForm {
898                        object: true,
899                        extras,
900                    });
901                    a.clone()
902                }
903                Some(Value::Object(m)) => {
904                    let mut extras = o.clone();
905                    extras.remove("jobs");
906                    meta.jobs_form = Some(JobsForm {
907                        object: true,
908                        extras,
909                    });
910                    m.iter()
911                        .map(|(id, j)| {
912                            let mut j = j.as_object().cloned().unwrap_or_default();
913                            j.insert("id".into(), Value::String(id.clone()));
914                            Value::Object(j)
915                        })
916                        .collect()
917                }
918                _ => {
919                    return Err(load_error(
920                        &jobs_file,
921                        "",
922                        "expected an array of jobs or {\"jobs\": [...]}",
923                    ))
924                }
925            },
926            _ => {
927                return Err(load_error(
928                    &jobs_file,
929                    "",
930                    "expected an array of jobs or {\"jobs\": [...]}",
931                ))
932            }
933        };
934        for raw in &arr {
935            let job = decode_job(&jobs_file, raw)?;
936            if profile.jobs.contains_key(&job.id) {
937                return Err(load_error(&jobs_file, &job.id, "duplicate job id"));
938            }
939            profile.jobs.insert(job.id.clone(), job);
940        }
941    }
942    let jobs_record: Vec<Value> = profile
943        .jobs
944        .values()
945        .map(|j| serde_json::to_value(j).unwrap())
946        .collect();
947    remember(
948        &mut meta,
949        "cron/jobs.json",
950        jobs_text,
951        &Value::Array(jobs_record),
952    );
953
954    // fires
955    let exec_path = dir.join("cron/executions.db");
956    if table_exists(&exec_path, "executions") {
957        for row in read_rows(
958            &exec_path,
959            "select * from executions order by claimed_at, id",
960            &[],
961        )?
962        .unwrap_or_default()
963        {
964            profile
965                .fires
966                .push(decode_fire_row(&exec_path.display().to_string(), &row)?);
967        }
968    }
969    remember(
970        &mut meta,
971        "cron/executions.db",
972        None,
973        &serde_json::to_value(&profile.fires).unwrap(),
974    );
975
976    // subscriptions
977    let subs_file = dir.join("webhook_subscriptions.json").display().to_string();
978    let subs_text = read_text(dir, "webhook_subscriptions.json")?;
979    if let Some(text) = &subs_text {
980        let parsed: Value = serde_json::from_str(text)
981            .map_err(|e| load_error(&subs_file, "", format!("JSON: {e}")))?;
982        let map = parsed
983            .as_object()
984            .ok_or_else(|| load_error(&subs_file, "", "expected a map"))?;
985        let known: BTreeSet<String> = vault.keys().cloned().collect();
986        for (n, raw) in map {
987            profile
988                .subscriptions
989                .insert(n.clone(), decode_subscription(&subs_file, n, raw, vault)?);
990        }
991        subs_normalized =
992            flavor == Flavor::Orchestrator && vault.keys().any(|k| !known.contains(k));
993    }
994    let subs_record: Vec<Value> = profile
995        .subscriptions
996        .values()
997        .map(|s| serde_json::to_value(s).unwrap())
998        .collect();
999    remember(
1000        &mut meta,
1001        "webhook_subscriptions.json",
1002        subs_text,
1003        &Value::Array(subs_record),
1004    );
1005    if subs_normalized {
1006        meta.raw.remove("webhook_subscriptions.json");
1007        meta.snapshot.remove("webhook_subscriptions.json");
1008    }
1009
1010    // access
1011    let access_file = dir.join("access.yaml").display().to_string();
1012    let access_text = read_text(dir, "access.yaml")?;
1013    let access_raw = match &access_text {
1014        Some(t) => Some(yaml_to_json(&access_file, t)?),
1015        None => None,
1016    };
1017    profile.access = decode_access(&access_file, access_raw.as_ref())?;
1018    remember(
1019        &mut meta,
1020        "access.yaml",
1021        access_text,
1022        &serde_json::to_value(&profile.access).unwrap(),
1023    );
1024
1025    // state.db: bindings + obligations
1026    let state_path = dir.join("state.db");
1027    if table_exists(&state_path, "delivery_obligations") {
1028        for row in read_rows(
1029            &state_path,
1030            "select * from delivery_obligations order by created_at, obligation_id",
1031            &[],
1032        )?
1033        .unwrap_or_default()
1034        {
1035            profile.obligations.push(decode_obligation_row(
1036                &state_path.display().to_string(),
1037                &row,
1038            )?);
1039        }
1040    }
1041    if flavor == Flavor::Orchestrator {
1042        if table_exists(&state_path, "bindings") {
1043            for row in read_rows(
1044                &state_path,
1045                "select * from bindings order by started_at, slot",
1046                &[],
1047            )?
1048            .unwrap_or_default()
1049            {
1050                let b = decode_binding_row(&state_path.display().to_string(), &row)?;
1051                let slot = row
1052                    .get("slot")
1053                    .and_then(Value::as_str)
1054                    .map(str::to_string)
1055                    .unwrap_or_else(|| surface_key_string(&b.key));
1056                profile.bindings.insert(slot, b);
1057            }
1058        }
1059    } else if table_exists(&state_path, "sessions") {
1060        for row in read_rows(
1061            &state_path,
1062            "select * from sessions order by started_at, id",
1063            &[],
1064        )?
1065        .unwrap_or_default()
1066        {
1067            if let Some(b) = binding_from_hermes_session(&state_path, &row, name) {
1068                profile.bindings.insert(surface_key_string(&b.key), b);
1069            }
1070        }
1071    }
1072    remember(&mut meta, "state.db", None, &state_record(&profile));
1073
1074    // unmodeled files: carried by path, copied verbatim on export; never held in memory.
1075    profile.residue.files = list_unmodeled(dir, flavor)?;
1076    Ok((profile, meta))
1077}
1078
1079fn list_unmodeled(dir: &Path, flavor: Flavor) -> Result<Vec<String>> {
1080    let mut owned: Vec<&str> = OWNED_FILES.to_vec();
1081    if flavor == Flavor::Hermes {
1082        owned.push("SOUL.md");
1083        owned.retain(|f| !["AGENTS.md", "CLAUDE.md", "access.yaml"].contains(f));
1084    }
1085    let runtime_artifacts = ["orchestrator.lock", "orchestrator.sock", "service"];
1086    let mut out = Vec::new();
1087    fn walk(
1088        base: &Path,
1089        d: &Path,
1090        owned: &[&str],
1091        runtime: &[&str],
1092        out: &mut Vec<String>,
1093    ) -> Result<()> {
1094        let mut entries: Vec<_> = fs::read_dir(d)?.flatten().collect();
1095        entries.sort_by_key(|e| e.file_name());
1096        for entry in entries {
1097            let p = entry.path();
1098            let rel = p
1099                .strip_prefix(base)
1100                .unwrap_or(&p)
1101                .to_string_lossy()
1102                .replace('\\', "/");
1103            let name = entry.file_name().to_string_lossy().into_owned();
1104            if rel == "profiles"
1105                || name == "node_modules"
1106                || name == ".git"
1107                || rel.starts_with("state.db")
1108                || rel.starts_with("cron/executions.db")
1109            {
1110                continue;
1111            }
1112            if runtime.contains(&rel.as_str()) || regex_tmp(&name) {
1113                continue;
1114            }
1115            let st = fs::symlink_metadata(&p)?;
1116            if st.is_dir() {
1117                walk(base, &p, owned, runtime, out)?;
1118                continue;
1119            }
1120            if !st.is_file() {
1121                continue;
1122            }
1123            if owned.contains(&rel.as_str()) {
1124                continue;
1125            }
1126            out.push(rel);
1127        }
1128        Ok(())
1129    }
1130    walk(dir, dir, &owned, &runtime_artifacts, &mut out)?;
1131    Ok(out)
1132}
1133
1134fn regex_tmp(name: &str) -> bool {
1135    // `<file>.tmp-<pid>`
1136    name.rsplit_once(".tmp-")
1137        .is_some_and(|(_, pid)| !pid.is_empty() && pid.chars().all(|c| c.is_ascii_digit()))
1138}
1139
1140// ---------------------------------------------------------------- save
1141
1142fn write_atomic(path: &Path, text: &str) -> Result<()> {
1143    if let Some(parent) = path.parent() {
1144        fs::create_dir_all(parent)?;
1145    }
1146    let tmp = path.with_file_name(format!(
1147        "{}.tmp-{}",
1148        path.file_name().unwrap().to_string_lossy(),
1149        std::process::id()
1150    ));
1151    fs::write(&tmp, text)?;
1152    fs::rename(&tmp, path)?;
1153    Ok(())
1154}
1155
1156/// Canonical config.yaml: O-blocks + Hermes-shaped blocks + residue.
1157pub fn encode_config(
1158    profile: &Profile,
1159    meta: Option<&ProfileIo>,
1160    vault: Option<&BTreeMap<String, String>>,
1161    flavor: Flavor,
1162) -> String {
1163    let mut out = Map::new();
1164    for (k, v) in &profile.residue.config {
1165        if k != "gateway" {
1166            out.insert(k.clone(), v.clone());
1167        }
1168    }
1169    if let Some(w) = &profile.worker {
1170        let mut wm = Map::new();
1171        wm.insert("harness".into(), Value::String(w.harness.as_str().into()));
1172        if let Some(m) = &w.model {
1173            wm.insert("model".into(), Value::String(m.clone()));
1174        }
1175        if let Some(p) = &w.preset {
1176            wm.insert("preset".into(), Value::String(p.clone()));
1177        }
1178        if w.cwd != "." {
1179            wm.insert("cwd".into(), Value::String(w.cwd.clone()));
1180        }
1181        if !w.env.is_empty() {
1182            wm.insert("env".into(), serde_json::to_value(&w.env).unwrap());
1183        }
1184        if w.permission.timeout_seconds != 300
1185            || w.permission.default != crate::orchestration::PermissionDefault::Deny
1186        {
1187            wm.insert(
1188                "permission".into(),
1189                serde_json::to_value(&w.permission).unwrap(),
1190            );
1191        }
1192        out.insert("worker".into(), Value::Object(wm));
1193    }
1194    if flavor == Flavor::Orchestrator || profile.expiry != ExpiryPolicy::default() {
1195        out.insert(
1196            "expiry".into(),
1197            serde_json::to_value(&profile.expiry).unwrap(),
1198        );
1199    }
1200    if let Some(h) = &profile.home {
1201        out.insert("home".into(), encode_surface_key(h));
1202    }
1203    let mut gateway = profile
1204        .residue
1205        .config
1206        .get("gateway")
1207        .and_then(Value::as_object)
1208        .cloned()
1209        .unwrap_or_default();
1210    let routes: Vec<Value> = profile.routes.iter().map(encode_route).collect();
1211    if meta.is_some_and(|m| m.routes_at_top) {
1212        if !routes.is_empty() {
1213            out.insert("profile_routes".into(), Value::Array(routes));
1214        }
1215    } else if !routes.is_empty() {
1216        gateway.insert("profile_routes".into(), Value::Array(routes));
1217    }
1218    if !gateway.is_empty() {
1219        out.insert("gateway".into(), Value::Object(gateway));
1220    }
1221    let mut platforms = Map::new();
1222    for (p, ch) in &profile.channels {
1223        platforms.insert(
1224            p.clone(),
1225            encode_channel(
1226                ch,
1227                if flavor == Flavor::Hermes {
1228                    vault
1229                } else {
1230                    None
1231                },
1232            ),
1233        );
1234    }
1235    if !platforms.is_empty() {
1236        out.insert("platforms".into(), Value::Object(platforms));
1237    }
1238    json_to_yaml(&Value::Object(out))
1239}
1240
1241/// The pin's canonical form is `{"jobs": [...]}`; a file read as a bare array keeps that form.
1242pub fn encode_jobs_file(profile: &Profile, meta: Option<&ProfileIo>) -> String {
1243    let jobs: Vec<Value> = profile
1244        .jobs
1245        .values()
1246        .map(|j| ordered_object(encode_job(j)))
1247        .collect();
1248    let form = meta.and_then(|m| m.jobs_form.clone()).unwrap_or(JobsForm {
1249        object: true,
1250        extras: Map::new(),
1251    });
1252    let body = if form.object {
1253        let mut pairs = vec![("jobs".to_string(), Value::Array(jobs))];
1254        pairs.extend(form.extras.iter().map(|(k, v)| (k.clone(), v.clone())));
1255        ordered_object(pairs)
1256    } else {
1257        Value::Array(jobs)
1258    };
1259    format!("{}\n", pretty_ordered(&body, 0))
1260}
1261
1262/// An object whose key order is the given one, regardless of features: wrapped
1263/// as a marker array the ordered printer understands.
1264pub(crate) fn ordered_object(pairs: Vec<(String, Value)>) -> Value {
1265    // serde_json::Map may sort; keep the order in a side channel: encode as
1266    // [{"__k": k, "__v": v}, ...] under a private tag.
1267    Value::Array(vec![
1268        Value::String("__ordered__".into()),
1269        Value::Array(
1270            pairs
1271                .into_iter()
1272                .map(|(k, v)| serde_json::json!({"__k": k, "__v": v}))
1273                .collect(),
1274        ),
1275    ])
1276}
1277
1278fn is_ordered(value: &Value) -> Option<&Vec<Value>> {
1279    let arr = value.as_array()?;
1280    if arr.len() == 2 && arr[0].as_str() == Some("__ordered__") {
1281        arr[1].as_array()
1282    } else {
1283        None
1284    }
1285}
1286
1287/// `JSON.stringify(value, null, 2)` with ordered objects honoured and plain objects sorted.
1288pub(crate) fn pretty_ordered(value: &Value, depth: usize) -> String {
1289    let pad = |d: usize| "  ".repeat(d);
1290    if let Some(pairs) = is_ordered(value) {
1291        if pairs.is_empty() {
1292            return "{}".into();
1293        }
1294        let inner: Vec<String> = pairs
1295            .iter()
1296            .map(|p| {
1297                format!(
1298                    "{}{}: {}",
1299                    pad(depth + 1),
1300                    serde_json::to_string(p["__k"].as_str().unwrap_or("")).unwrap(),
1301                    pretty_ordered(&p["__v"], depth + 1)
1302                )
1303            })
1304            .collect();
1305        return format!("{{\n{}\n{}}}", inner.join(",\n"), pad(depth));
1306    }
1307    match value {
1308        Value::Array(items) if items.is_empty() => "[]".into(),
1309        Value::Array(items) => {
1310            let inner: Vec<String> = items
1311                .iter()
1312                .map(|v| format!("{}{}", pad(depth + 1), pretty_ordered(v, depth + 1)))
1313                .collect();
1314            format!("[\n{}\n{}]", inner.join(",\n"), pad(depth))
1315        }
1316        Value::Object(o) if o.is_empty() => "{}".into(),
1317        Value::Object(o) => {
1318            let inner: Vec<String> = o
1319                .iter()
1320                .map(|(k, v)| {
1321                    format!(
1322                        "{}{}: {}",
1323                        pad(depth + 1),
1324                        serde_json::to_string(k).unwrap(),
1325                        pretty_ordered(v, depth + 1)
1326                    )
1327                })
1328                .collect();
1329            format!("{{\n{}\n{}}}", inner.join(",\n"), pad(depth))
1330        }
1331        Value::Number(n) => {
1332            if let Some(f) = n.as_f64() {
1333                if n.is_f64() && f.fract() == 0.0 && f.abs() < 1e21 {
1334                    return format!("{}", f as i64);
1335                }
1336            }
1337            n.to_string()
1338        }
1339        other => serde_json::to_string(other).unwrap(),
1340    }
1341}
1342
1343/// `webhook_subscriptions.json`; secrets inlined only for a Hermes export.
1344pub fn encode_subscriptions_file(
1345    profile: &Profile,
1346    vault: Option<&BTreeMap<String, String>>,
1347) -> String {
1348    let mut out = Map::new();
1349    for (n, s) in &profile.subscriptions {
1350        out.insert(n.clone(), encode_subscription(s, vault));
1351    }
1352    format!("{}\n", pretty_ordered(&Value::Object(out), 0))
1353}
1354
1355/// `access.yaml`.
1356pub fn encode_access_file(profile: &Profile) -> String {
1357    json_to_yaml(&encode_access(&profile.access))
1358}
1359
1360const EXECUTIONS_DDL: &str = "CREATE TABLE IF NOT EXISTS executions (
1361  id TEXT PRIMARY KEY, job_id TEXT NOT NULL, source TEXT NOT NULL, process_id TEXT NOT NULL, pid INTEGER NOT NULL,
1362  process_started_at INTEGER, status TEXT NOT NULL CHECK(status IN ('claimed','running','completed','failed','unknown')),
1363  claimed_at TEXT NOT NULL, started_at TEXT, finished_at TEXT, error TEXT);
1364CREATE INDEX IF NOT EXISTS idx_executions_job_claimed ON executions(job_id, claimed_at DESC, id DESC);
1365CREATE INDEX IF NOT EXISTS idx_executions_status_claimed ON executions(status, claimed_at DESC, id DESC);";
1366/// Our own ledger: Hermes's shape plus `residue_json`, the columns another
1367/// harness's run log has and Hermes's does not (an OpenClaw run's delivery
1368/// status, session key, summary, model, token counts, sequence — ORC-12
1369/// finding 5), so a run migrated through our folder loses none of them.
1370const FOLDER_EXECUTIONS_DDL: &str = "CREATE TABLE IF NOT EXISTS executions (
1371  id TEXT PRIMARY KEY, job_id TEXT NOT NULL, source TEXT NOT NULL, process_id TEXT NOT NULL, pid INTEGER NOT NULL,
1372  process_started_at INTEGER, status TEXT NOT NULL CHECK(status IN ('claimed','running','completed','failed','unknown')),
1373  claimed_at TEXT NOT NULL, started_at TEXT, finished_at TEXT, error TEXT, residue_json TEXT,
1374  session_id TEXT, obligation_id TEXT);
1375CREATE INDEX IF NOT EXISTS idx_executions_job_claimed ON executions(job_id, claimed_at DESC, id DESC);
1376CREATE INDEX IF NOT EXISTS idx_executions_status_claimed ON executions(status, claimed_at DESC, id DESC);";
1377
1378const OBLIGATIONS_DDL: &str = "CREATE TABLE IF NOT EXISTS delivery_obligations (
1379  obligation_id TEXT PRIMARY KEY, session_key TEXT NOT NULL, platform TEXT NOT NULL, chat_id TEXT NOT NULL, thread_id TEXT,
1380  content TEXT NOT NULL, state TEXT NOT NULL, attempts INTEGER NOT NULL DEFAULT 0, created_at REAL NOT NULL, updated_at REAL NOT NULL,
1381  owner_pid INTEGER, owner_started_at INTEGER, last_error TEXT, adapter_profile TEXT,
1382  posted_message_id TEXT, source_json TEXT);";
1383
1384const BINDINGS_DDL: &str = "CREATE TABLE IF NOT EXISTS bindings (
1385  slot TEXT PRIMARY KEY,
1386  platform TEXT NOT NULL, chat_type TEXT NOT NULL, chat_id TEXT, thread_id TEXT, participant_id TEXT,
1387  worker_harness TEXT NOT NULL, worker_session_id TEXT, worker_locator TEXT,
1388  started_at TEXT NOT NULL, last_activity_at TEXT NOT NULL, ended_at TEXT, end_reason TEXT,
1389  handoff_to TEXT, handoff_state TEXT, handoff_error TEXT, recurrence_job_id TEXT, residue_json TEXT);";
1390
1391/// The executions ledger, written fresh.
1392pub fn write_executions(path: &Path, fires: &[crate::orchestration::Fire]) -> Result<()> {
1393    write_executions_shaped(path, fires, false)
1394}
1395
1396/// `ours`: our folder's ledger carries every fire's residue; a Hermes home's
1397/// ledger is Hermes's shape exactly.
1398pub fn write_executions_shaped(
1399    path: &Path,
1400    fires: &[crate::orchestration::Fire],
1401    ours: bool,
1402) -> Result<()> {
1403    let mut cols: Vec<&str> = EXECUTION_COLUMNS.to_vec();
1404    if ours {
1405        // our ledger also keeps the two links Hermes's derives: a fire's
1406        // session and the delivery it produced
1407        cols.extend(["residue_json", "session_id", "obligation_id"]);
1408    }
1409    let insert = format!(
1410        "insert into executions ({}) values ({})",
1411        cols.join(", "),
1412        cols.iter().map(|_| "?").collect::<Vec<_>>().join(",")
1413    );
1414    let rows: Vec<Vec<Param>> = fires
1415        .iter()
1416        .map(|f| {
1417            let mut row: Vec<Param> = encode_fire_row(f).iter().map(Param::from).collect();
1418            if ours {
1419                // what the Hermes columns already hold stays out of the json
1420                let rest: serde_json::Map<String, Value> = f
1421                    .residue
1422                    .0
1423                    .iter()
1424                    .filter(|(k, _)| {
1425                        !["source", "process_id", "pid", "process_started_at"].contains(&k.as_str())
1426                    })
1427                    .map(|(k, v)| (k.clone(), v.clone()))
1428                    .collect();
1429                row.push(if rest.is_empty() {
1430                    Param::Null
1431                } else {
1432                    Param::Text(serde_json::to_string(&rest).unwrap())
1433                });
1434                let opt = |v: &Option<String>| v.clone().map(Param::Text).unwrap_or(Param::Null);
1435                row.push(opt(&f.session_id));
1436                row.push(opt(&f.obligation_id));
1437            }
1438            row
1439        })
1440        .collect();
1441    write_table(
1442        path,
1443        if ours {
1444            FOLDER_EXECUTIONS_DDL
1445        } else {
1446            EXECUTIONS_DDL
1447        },
1448        &insert,
1449        &rows,
1450    )
1451}
1452
1453fn write_state(path: &Path, profile: &Profile) -> Result<()> {
1454    let cols = [
1455        "slot",
1456        "platform",
1457        "chat_type",
1458        "chat_id",
1459        "thread_id",
1460        "participant_id",
1461        "worker_harness",
1462        "worker_session_id",
1463        "worker_locator",
1464        "started_at",
1465        "last_activity_at",
1466        "ended_at",
1467        "end_reason",
1468        "handoff_to",
1469        "handoff_state",
1470        "handoff_error",
1471        "recurrence_job_id",
1472        "residue_json",
1473    ];
1474    let insert = format!(
1475        "insert into bindings ({}) values ({})",
1476        cols.join(", "),
1477        cols.iter().map(|_| "?").collect::<Vec<_>>().join(",")
1478    );
1479    let s = |v: &Option<String>| v.clone().map(Param::Text).unwrap_or(Param::Null);
1480    let rows: Vec<Vec<Param>> = profile
1481        .bindings
1482        .iter()
1483        .map(|(slot, b)| {
1484            vec![
1485                Param::Text(slot.clone()),
1486                Param::Text(b.key.platform.clone().unwrap_or_default()),
1487                Param::Text(b.key.kind.clone().unwrap_or_default()),
1488                Param::Text(b.key.chat_id.clone().unwrap_or_default()),
1489                Param::Text(b.key.thread_id.clone().unwrap_or_default()),
1490                Param::Text(b.key.participant_id.clone().unwrap_or_default()),
1491                Param::Text(b.worker.harness.as_str().into()),
1492                s(&b.worker.session_id.clone().filter(|v| !v.is_empty())),
1493                s(&b.worker.locator),
1494                Param::Text(b.started_at.clone().unwrap_or_default()),
1495                Param::Text(b.last_activity_at.clone().unwrap_or_default()),
1496                s(&b.ended_at),
1497                b.end_reason
1498                    .map(|r| Param::Text(r.as_str().into()))
1499                    .unwrap_or(Param::Null),
1500                s(&b.handoff.as_ref().and_then(|h| h.to.clone())),
1501                b.handoff
1502                    .as_ref()
1503                    .map(|h| Param::Text(h.state.clone()))
1504                    .unwrap_or(Param::Null),
1505                s(&b.handoff.as_ref().and_then(|h| h.error.clone())),
1506                s(&b.recurrence.as_ref().map(|r| r.job_id.clone())),
1507                if b.residue.is_empty() {
1508                    Param::Null
1509                } else {
1510                    Param::Text(serde_json::to_string(&b.residue).unwrap())
1511                },
1512            ]
1513        })
1514        .collect();
1515    write_table(
1516        path,
1517        &format!("{BINDINGS_DDL}\n{OBLIGATIONS_DDL}"),
1518        &insert,
1519        &rows,
1520    )?;
1521    // Hermes's columns, then ours (the sent message's handle, the source)
1522    let cols: Vec<&str> = OBLIGATION_COLUMNS
1523        .iter()
1524        .chain(FOLDER_OBLIGATION_EXTRA_COLUMNS.iter())
1525        .copied()
1526        .collect();
1527    let insert = format!(
1528        "insert into delivery_obligations ({}) values ({})",
1529        cols.join(", "),
1530        cols.iter().map(|_| "?").collect::<Vec<_>>().join(",")
1531    );
1532    let rows: Vec<Vec<Param>> = profile
1533        .obligations
1534        .iter()
1535        .map(|o| {
1536            encode_obligation_row(o)
1537                .iter()
1538                .chain(encode_obligation_folder_extras(o).iter())
1539                .map(Param::from)
1540                .collect()
1541        })
1542        .collect();
1543    write_table(path, "", &insert, &rows)
1544}
1545
1546fn write_if_changed(
1547    meta: &mut ProfileIo,
1548    dir: &Path,
1549    rel: &str,
1550    record: &Value,
1551    render: impl FnOnce() -> Option<String>,
1552) -> Result<bool> {
1553    let snap = canonical_json(record);
1554    let path = dir.join(rel);
1555    // Raw bytes are reused only for a file read from OUR folder: another
1556    // harness's bytes may carry inline secrets our folder must never hold.
1557    let reuse = meta.flavor == Flavor::Orchestrator && meta.snapshot.get(rel) == Some(&snap);
1558    if reuse && path.exists() {
1559        return Ok(false);
1560    }
1561    if reuse {
1562        if let Some(raw) = meta.raw.get(rel).cloned() {
1563            write_atomic(&path, &raw)?;
1564            return Ok(true);
1565        }
1566    }
1567    let Some(text) = render() else {
1568        return Ok(false);
1569    };
1570    write_atomic(&path, &text)?;
1571    meta.raw.insert(rel.into(), text);
1572    meta.snapshot.insert(rel.into(), snap);
1573    meta.flavor = Flavor::Orchestrator; // from here on the bytes on disk are ours
1574    Ok(true)
1575}
1576
1577/// Save OUR folder. Never deletes a file it does not own; never writes a secret outside `.env`.
1578pub fn save_home(loaded: &mut LoadedHome, root: Option<&Path>) -> Result<()> {
1579    let root = root
1580        .map(Path::to_path_buf)
1581        .unwrap_or_else(|| loaded.orchestration.root.clone());
1582    fs::create_dir_all(&root)?;
1583    let names: Vec<String> = loaded.orchestration.profiles.keys().cloned().collect();
1584    for name in names {
1585        let dir = if name == "default" {
1586            root.clone()
1587        } else {
1588            root.join("profiles").join(&name)
1589        };
1590        fs::create_dir_all(dir.join("cron"))?;
1591        let profile = loaded.orchestration.profiles[&name].clone();
1592        let meta = loaded
1593            .io
1594            .entry(name.clone())
1595            .or_insert_with(|| ProfileIo::new(Flavor::Orchestrator));
1596        save_profile_dir(&profile, meta, &dir, &loaded.vault)?;
1597    }
1598    Ok(())
1599}
1600
1601fn save_profile_dir(
1602    profile: &Profile,
1603    meta: &mut ProfileIo,
1604    dir: &Path,
1605    vault: &BTreeMap<String, String>,
1606) -> Result<()> {
1607    let cfg_record = config_record(profile);
1608    write_if_changed(meta, dir, "config.yaml", &cfg_record, || {
1609        Some(encode_config(
1610            profile,
1611            None,
1612            Some(vault),
1613            Flavor::Orchestrator,
1614        ))
1615    })?;
1616    if let Some(persona) = &profile.persona {
1617        let text = persona.text.clone().unwrap_or_default();
1618        write_if_changed(
1619            meta,
1620            dir,
1621            "AGENTS.md",
1622            &serde_json::to_value(&profile.persona).unwrap(),
1623            || Some(text),
1624        )?;
1625        if !dir.join("CLAUDE.md").exists() {
1626            write_atomic(&dir.join("CLAUDE.md"), "@AGENTS.md\n")?;
1627        }
1628    }
1629    let jobs_record: Vec<Value> = profile
1630        .jobs
1631        .values()
1632        .map(|j| serde_json::to_value(j).unwrap())
1633        .collect();
1634    let had_jobs = meta.raw.contains_key("cron/jobs.json");
1635    let form = meta.jobs_form.clone();
1636    write_if_changed(
1637        meta,
1638        dir,
1639        "cron/jobs.json",
1640        &Value::Array(jobs_record),
1641        || {
1642            if profile.jobs.is_empty() && !had_jobs {
1643                None
1644            } else {
1645                let stub = ProfileIo {
1646                    jobs_form: form.clone(),
1647                    ..ProfileIo::new(Flavor::Orchestrator)
1648                };
1649                Some(encode_jobs_file(profile, Some(&stub)))
1650            }
1651        },
1652    )?;
1653    let subs_record: Vec<Value> = profile
1654        .subscriptions
1655        .values()
1656        .map(|s| serde_json::to_value(s).unwrap())
1657        .collect();
1658    let had_subs = meta.raw.contains_key("webhook_subscriptions.json");
1659    write_if_changed(
1660        meta,
1661        dir,
1662        "webhook_subscriptions.json",
1663        &Value::Array(subs_record),
1664        || {
1665            if profile.subscriptions.is_empty() && !had_subs {
1666                None
1667            } else {
1668                Some(encode_subscriptions_file(profile, None))
1669            }
1670        },
1671    )?;
1672    let a = &profile.access;
1673    let access_empty = a.allowlist.is_empty()
1674        && a.admins.is_empty()
1675        && a.pending_pairings.is_empty()
1676        && a.policy.is_empty()
1677        && a.pairing_ttl_minutes.is_none();
1678    let had_access = meta.raw.contains_key("access.yaml");
1679    write_if_changed(
1680        meta,
1681        dir,
1682        "access.yaml",
1683        &serde_json::to_value(a).unwrap(),
1684        || {
1685            if access_empty && !had_access {
1686                None
1687            } else {
1688                Some(encode_access_file(profile))
1689            }
1690        },
1691    )?;
1692
1693    // sqlite: rewrite only when the decoded rows changed
1694    let fires_snap = canonical_json(&serde_json::to_value(&profile.fires).unwrap());
1695    let exec_path = dir.join("cron/executions.db");
1696    if (meta.snapshot.get("cron/executions.db") != Some(&fires_snap) || !exec_path.exists())
1697        && (!profile.fires.is_empty() || exec_path.exists())
1698    {
1699        let tmp = exec_path.with_file_name(format!("executions.db.tmp-{}", std::process::id()));
1700        let _ = fs::remove_file(&tmp);
1701        write_executions_shaped(&tmp, &profile.fires, true)?;
1702        fs::rename(&tmp, &exec_path)?;
1703        meta.snapshot
1704            .insert("cron/executions.db".into(), fires_snap);
1705    }
1706    let state_snap = canonical_json(&state_record(profile));
1707    let state_path = dir.join("state.db");
1708    if (meta.snapshot.get("state.db") != Some(&state_snap) || !state_path.exists())
1709        && (!profile.bindings.is_empty() || !profile.obligations.is_empty() || state_path.exists())
1710    {
1711        let tmp = state_path.with_file_name(format!("state.db.tmp-{}", std::process::id()));
1712        let _ = fs::remove_file(&tmp);
1713        write_state(&tmp, profile)?;
1714        fs::rename(&tmp, &state_path)?;
1715        meta.snapshot.insert("state.db".into(), state_snap);
1716    }
1717
1718    // .env: every ref this profile uses
1719    let mut refs: Vec<String> = Vec::new();
1720    for ch in profile.channels.values() {
1721        for r in ch.credentials.values() {
1722            if let crate::ontology::SecretRef::Dotenv(n) = r {
1723                refs.push(n.clone());
1724            }
1725        }
1726    }
1727    for s in profile.subscriptions.values() {
1728        if let Some(crate::ontology::SecretRef::Dotenv(n)) = &s.secret {
1729            refs.push(n.clone());
1730        }
1731    }
1732    if let Some(w) = &profile.worker {
1733        for v in w.env.values() {
1734            if let crate::orchestration::EnvValue::Secret(crate::ontology::SecretRef::Dotenv(n)) = v
1735            {
1736                refs.push(n.clone());
1737            }
1738        }
1739    }
1740    let mut entries: BTreeMap<String, String> = BTreeMap::new();
1741    for r in refs {
1742        if let Some(v) = vault.get(&r) {
1743            entries.insert(r, v.clone());
1744        }
1745    }
1746    if !entries.is_empty() {
1747        let existing = meta.raw.get(".env").cloned();
1748        let mut merged = existing.as_deref().map(parse_dotenv).unwrap_or_default();
1749        for (k, v) in entries {
1750            merged.insert(k, v);
1751        }
1752        let text = render_dotenv(&merged);
1753        if existing.as_deref() != Some(text.as_str()) {
1754            write_atomic(&dir.join(".env"), &text)?;
1755            meta.raw.insert(".env".into(), text);
1756        }
1757    }
1758    Ok(())
1759}
1760
1761/// Copy every unmodeled file of a profile from its source dir into `dest`, byte for byte.
1762pub fn copy_unmodeled(profile: &Profile, meta: &ProfileIo, dest: &Path) -> Result<()> {
1763    let Some(src) = &meta.source_dir else {
1764        return Ok(());
1765    };
1766    carry_unmodeled(&profile.residue.files, src, dest)?;
1767    Ok(())
1768}
1769
1770/// Copy the named unmodeled files from `src` into `dest`, byte for byte,
1771/// answering the relative paths carried. A file the source no longer holds is
1772/// skipped: the list is what was seen at load, the copy is what is there now.
1773pub fn carry_unmodeled(files: &[String], src: &Path, dest: &Path) -> Result<Vec<String>> {
1774    let mut carried = Vec::new();
1775    for rel in files {
1776        let from = src.join(rel);
1777        if !from.is_file() {
1778            continue;
1779        }
1780        let to = dest.join(rel);
1781        if let Some(parent) = to.parent() {
1782            fs::create_dir_all(parent)?;
1783        }
1784        fs::copy(&from, &to)?;
1785        carried.push(rel.clone());
1786    }
1787    Ok(carried)
1788}