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