supercode_interchange/orchestration/codec/
hermes.rs

1//! A Hermes home written from the orchestration (`hermes.mjs::toHermes`). Tiers per
2//! file: byte for `cron/jobs.json`, `cron/executions.db`,
3//! `webhook_subscriptions.json` (inline secrets re-inlined from the vault),
4//! `config.yaml` when unchanged, every unmodeled file, and `state.db` when
5//! bindings/obligations are UNCHANGED since import (copied). A FRESH
6//! destination gets a store born at the fixture's schema (22) carrying the
7//! bindings as `sessions` rows and the obligations as `delivery_obligations`
8//! rows — Hermes migrates it on open; the transcripts live in the worker's
9//! store and are not carried (semantic). A LIVE destination store is never
10//! written: that first write into a shared WAL single-writer store is
11//! UNI-18's, refused by name. `state.db` with
12//! changed rows is REFUSED: the Hermes session-store write path is behind
13//! UNI-22. Semantic for `SOUL.md` ⇄ `AGENTS.md` and a re-rendered
14//! `config.yaml` (the O-blocks ride as top-level keys — ORC-1 F5).
15
16use std::fs;
17use std::path::Path;
18
19use serde_json::Value;
20
21use super::canonical::canonical_json;
22use super::decode::{encode_obligation_row, OBLIGATION_COLUMNS};
23use super::folder::{
24    config_record, copy_unmodeled, encode_config, encode_jobs_file, encode_subscriptions_file,
25    write_executions, Flavor, LoadedHome, ProfileIo,
26};
27use super::openclaw::ms_from_iso;
28use super::sqlite::{write_table, Param};
29use crate::ontology::{
30    hermes_source_for_binding, render_hermes_session_key, ArtifactFidelity, Binding, Fidelity,
31};
32use crate::orchestration::Profile;
33
34/// Hermes's `state.db` schema as the fixture home carries it (`SCHEMA_VERSION`
35/// 22), without the FTS virtual tables and their triggers: Hermes creates
36/// those itself on open (`hermes_state_common.py`, `CREATE VIRTUAL TABLE IF
37/// NOT EXISTS`), and its main schema "advances freely on open (so future
38/// migrations always land)".
39const HERMES_STATE_V22: &str = include_str!("hermes_state_v22.sql");
40const HERMES_STATE_V22_VERSION: i64 = 22;
41
42/// The `sessions` columns a binding fills; every other column keeps Hermes's
43/// own default.
44const SESSION_COLUMNS: &[&str] = &[
45    "id",
46    "source",
47    "user_id",
48    "session_key",
49    "chat_id",
50    "chat_type",
51    "thread_id",
52    "expiry_finalized",
53    "started_at",
54    "ended_at",
55    "end_reason",
56    "handoff_state",
57    "handoff_platform",
58    "handoff_error",
59    "profile_name",
60];
61
62fn epoch_seconds(iso: Option<&str>) -> Option<f64> {
63    ms_from_iso(iso).map(|ms| ms as f64 / 1000.0)
64}
65
66/// A binding as a Hermes `sessions` row (inverse of `Binding::from_hermes_row`).
67fn hermes_session_row(profile: &str, slot: &str, b: &Binding) -> Vec<Param> {
68    let text = |v: &Option<String>| v.clone().map(Param::Text).unwrap_or(Param::Null);
69    let hermes_profile = if profile == "default" {
70        "main"
71    } else {
72        profile
73    };
74    let started = epoch_seconds(b.started_at.as_deref())
75        .or_else(|| epoch_seconds(b.last_activity_at.as_deref()))
76        .unwrap_or(0.0);
77    vec![
78        Param::Text(
79            b.worker
80                .session_id
81                .clone()
82                .unwrap_or_else(|| slot.to_string()),
83        ),
84        Param::Text(hermes_source_for_binding(b)),
85        text(&b.key.participant_id),
86        Param::Text(
87            b.key
88                .key
89                .clone()
90                .unwrap_or_else(|| render_hermes_session_key(hermes_profile, &b.key)),
91        ),
92        text(&b.key.chat_id),
93        text(&b.key.kind),
94        text(&b.key.thread_id),
95        Param::Int(i64::from(b.ended_at.is_some())),
96        Param::Real(started),
97        epoch_seconds(b.ended_at.as_deref())
98            .map(Param::Real)
99            .unwrap_or(Param::Null),
100        b.end_reason
101            .map(|r| Param::Text(r.as_str().into()))
102            .unwrap_or(Param::Null),
103        text(&b.handoff.as_ref().map(|h| h.state.clone())),
104        text(&b.handoff.as_ref().and_then(|h| h.to.clone())),
105        text(&b.handoff.as_ref().and_then(|h| h.error.clone())),
106        if profile == "default" {
107            Param::Null
108        } else {
109            Param::Text(profile.into())
110        },
111    ]
112}
113
114/// A fresh Hermes store at `path`: the schema Hermes migrates on open, every
115/// profile's bindings as sessions rows (a satellite profile's under its
116/// `profile_name`, the way Hermes's own multiplexed gateway keeps them in the
117/// ROOT store), every profile's obligations as delivery rows.
118fn write_fresh_store(profiles: &[(&str, &Profile)], path: &Path) -> Result<()> {
119    let placeholders = |n: usize| std::iter::repeat_n("?", n).collect::<Vec<_>>().join(",");
120    write_table(
121        path,
122        HERMES_STATE_V22,
123        "insert into schema_version (version) values (?)",
124        &[vec![Param::Int(HERMES_STATE_V22_VERSION)]],
125    )?;
126    let sessions: Vec<Vec<Param>> = profiles
127        .iter()
128        .flat_map(|(name, profile)| {
129            profile
130                .bindings
131                .iter()
132                .map(move |(slot, b)| hermes_session_row(name, slot, b))
133        })
134        .collect();
135    write_table(
136        path,
137        "",
138        &format!(
139            "insert into sessions ({}) values ({})",
140            SESSION_COLUMNS.join(", "),
141            placeholders(SESSION_COLUMNS.len())
142        ),
143        &sessions,
144    )?;
145    let obligations: Vec<Vec<Param>> = profiles
146        .iter()
147        .flat_map(|(_, profile)| profile.obligations.iter())
148        .map(|o| encode_obligation_row(o).iter().map(Param::from).collect())
149        .collect();
150    write_table(
151        path,
152        "",
153        &format!(
154            "insert into delivery_obligations ({}) values ({})",
155            OBLIGATION_COLUMNS.join(", "),
156            placeholders(OBLIGATION_COLUMNS.len())
157        ),
158        &obligations,
159    )
160}
161use crate::Result;
162
163/// A write the codec would not guess at.
164#[derive(Debug, Clone, PartialEq, Eq)]
165pub struct Refusal {
166    /// The file, relative to the destination home.
167    pub file: String,
168    /// Why, naming the gate.
169    pub reason: String,
170}
171
172/// What a Hermes decompile did.
173#[derive(Debug, Clone, Default)]
174pub struct HermesReport {
175    /// Every artifact written, with its tier.
176    pub written: Vec<ArtifactFidelity>,
177    /// Every write refused.
178    pub refused: Vec<Refusal>,
179}
180
181fn write_atomic(path: &Path, text: &str) -> Result<()> {
182    if let Some(parent) = path.parent() {
183        fs::create_dir_all(parent)?;
184    }
185    let tmp = path.with_file_name(format!(
186        "{}.tmp-{}",
187        path.file_name().unwrap().to_string_lossy(),
188        std::process::id()
189    ));
190    fs::write(&tmp, text)?;
191    fs::rename(&tmp, path)?;
192    Ok(())
193}
194
195/// Load a Hermes home into the orchestration.
196pub fn from_hermes(home: &Path) -> Result<LoadedHome> {
197    super::folder::load_home(home, Flavor::Hermes)
198}
199
200/// Write a Hermes home. Byte-tier files whose records are unchanged since a
201/// Hermes import are copied from the import source; everything else is
202/// emitted canonically.
203pub fn to_hermes(loaded: &LoadedHome, dest: &Path, only: Option<&str>) -> Result<HermesReport> {
204    let mut report = HermesReport::default();
205    let empty = ProfileIo {
206        raw: Default::default(),
207        snapshot: Default::default(),
208        source_dir: None,
209        flavor: Flavor::Orchestrator,
210        jobs_form: None,
211        routes_at_top: false,
212        borrowed_from: None,
213        lenders: Vec::new(),
214    };
215    let mut fresh_rows: Vec<(&str, &Profile)> = Vec::new();
216    for (name, profile) in &loaded.orchestration.profiles {
217        if only.is_some_and(|o| o != name) {
218            continue;
219        }
220        let dir = if name == "default" {
221            dest.to_path_buf()
222        } else {
223            dest.join("profiles").join(name)
224        };
225        fs::create_dir_all(dir.join("cron"))?;
226        let meta = loaded.io.get(name).unwrap_or(&empty);
227        let rel = |p: &str| {
228            if name == "default" {
229                p.to_string()
230            } else {
231                format!("profiles/{name}/{p}")
232            }
233        };
234        let unchanged =
235            |file: &str, record: &Value| meta.snapshot.get(file) == Some(&canonical_json(record));
236        fn emit_file(
237            written: &mut Vec<ArtifactFidelity>,
238            dir: &Path,
239            path: String,
240            file: &str,
241            text: &str,
242            tier: Fidelity,
243        ) -> Result<()> {
244            write_atomic(&dir.join(file), text)?;
245            written.push(ArtifactFidelity {
246                path,
247                fidelity: tier,
248                loss: Vec::new(),
249            });
250            Ok(())
251        }
252        macro_rules! emit {
253            ($file:expr, $text:expr, $tier:expr) => {
254                emit_file(&mut report.written, &dir, rel($file), $file, $text, $tier)?
255            };
256        }
257
258        // config.yaml (only when the profile has one, or has something to say)
259        let cfg_record = config_record(profile);
260        let cfg_empty = profile.routes.is_empty()
261            && profile.channels.is_empty()
262            && profile.residue.config.is_empty()
263            && profile.worker.is_none()
264            && profile.home.is_none()
265            && profile.expiry == Default::default();
266        // Only bytes read FROM A HERMES HOME may be reused for the two files
267        // that carry credentials: our folder's config.yaml holds `{dotenv}`
268        // refs where Hermes reads values.
269        let hermes_bytes = meta.flavor == Flavor::Hermes;
270        if hermes_bytes
271            && unchanged("config.yaml", &cfg_record)
272            && meta.raw.contains_key("config.yaml")
273        {
274            emit!(
275                "config.yaml",
276                &meta.raw["config.yaml"],
277                Fidelity::ByteLossless
278            );
279        } else if !cfg_empty || meta.raw.contains_key("config.yaml") {
280            emit!(
281                "config.yaml",
282                &encode_config(profile, Some(meta), Some(&loaded.vault), Flavor::Hermes),
283                Fidelity::Semantic
284            );
285        }
286
287        // persona: AGENTS.md -> SOUL.md
288        if let Some(persona) = &profile.persona {
289            let src_name = if meta.raw.contains_key("SOUL.md") {
290                "SOUL.md"
291            } else {
292                "AGENTS.md"
293            };
294            let record = serde_json::to_value(&profile.persona).unwrap();
295            if unchanged(src_name, &record) && meta.raw.contains_key(src_name) {
296                emit!(
297                    "SOUL.md",
298                    &meta.raw[src_name],
299                    if src_name == "SOUL.md" {
300                        Fidelity::ByteLossless
301                    } else {
302                        Fidelity::Semantic
303                    }
304                );
305            } else {
306                emit!(
307                    "SOUL.md",
308                    persona.text.as_deref().unwrap_or(""),
309                    Fidelity::Semantic
310                );
311            }
312        }
313
314        // jobs — Hermes requires an ABSOLUTE workdir; ours is relative to the
315        // profile folder, so the emitted copy resolves it against the destination
316        let jobs_record: Vec<Value> = profile
317            .jobs
318            .values()
319            .map(|j| serde_json::to_value(j).unwrap())
320            .collect();
321        if !profile.jobs.is_empty() || meta.raw.contains_key("cron/jobs.json") {
322            if unchanged("cron/jobs.json", &Value::Array(jobs_record))
323                && meta.raw.contains_key("cron/jobs.json")
324            {
325                emit!(
326                    "cron/jobs.json",
327                    &meta.raw["cron/jobs.json"],
328                    Fidelity::ByteLossless
329                );
330            } else {
331                let mut view: Profile = profile.clone();
332                for job in view.jobs.values_mut() {
333                    if let Some(w) = &job.workdir {
334                        if !w.starts_with('/') {
335                            job.workdir = Some(dir.join(w).display().to_string());
336                        }
337                    }
338                }
339                emit!(
340                    "cron/jobs.json",
341                    &encode_jobs_file(&view, Some(meta)),
342                    Fidelity::ByteLossless
343                );
344            }
345        }
346
347        // fires
348        let src_exec = meta
349            .source_dir
350            .as_ref()
351            .map(|d| d.join("cron/executions.db"));
352        if !profile.fires.is_empty() || src_exec.as_ref().is_some_and(|p| p.exists()) {
353            let target = dir.join("cron/executions.db");
354            let fires_record = serde_json::to_value(&profile.fires).unwrap();
355            if unchanged("cron/executions.db", &fires_record)
356                && src_exec.as_ref().is_some_and(|p| p.exists())
357            {
358                fs::copy(src_exec.as_ref().unwrap(), &target)?;
359            } else {
360                let tmp =
361                    target.with_file_name(format!("executions.db.tmp-{}", std::process::id()));
362                let _ = fs::remove_file(&tmp);
363                write_executions(&tmp, &profile.fires)?;
364                fs::rename(&tmp, &target)?;
365            }
366            report
367                .written
368                .push(ArtifactFidelity::byte(rel("cron/executions.db")));
369        }
370
371        // subscriptions (secrets re-inlined: Hermes reads them from this file)
372        let subs_record: Vec<Value> = profile
373            .subscriptions
374            .values()
375            .map(|s| serde_json::to_value(s).unwrap())
376            .collect();
377        if !profile.subscriptions.is_empty() || meta.raw.contains_key("webhook_subscriptions.json")
378        {
379            if hermes_bytes
380                && unchanged("webhook_subscriptions.json", &Value::Array(subs_record))
381                && meta.raw.contains_key("webhook_subscriptions.json")
382            {
383                emit!(
384                    "webhook_subscriptions.json",
385                    &meta.raw["webhook_subscriptions.json"],
386                    Fidelity::ByteLossless
387                );
388            } else {
389                emit!(
390                    "webhook_subscriptions.json",
391                    &encode_subscriptions_file(profile, Some(&loaded.vault)),
392                    Fidelity::ByteLossless
393                );
394            }
395        }
396
397        // state.db: copy when untouched, refuse otherwise (UNI-22)
398        if meta.borrowed_from.is_none() {
399            let state_record = serde_json::json!({ "bindings": profile.bindings, "obligations": profile.obligations });
400            let src_state = meta.source_dir.as_ref().map(|d| d.join("state.db"));
401            let lenders_unchanged = meta.lenders.iter().all(|n| {
402                let p = &loaded.orchestration.profiles[n];
403                loaded.io.get(n).and_then(|m| m.snapshot.get("state.db")) == Some(&canonical_json(&serde_json::json!({ "bindings": p.bindings, "obligations": p.obligations })))
404            });
405            let dest_store = dir.join("state.db");
406            if src_state.as_ref().is_some_and(|p| p.exists()) && meta.flavor == Flavor::Hermes {
407                if unchanged("state.db", &state_record) && lenders_unchanged {
408                    fs::copy(src_state.as_ref().unwrap(), &dest_store)?;
409                    report.written.push(ArtifactFidelity::byte(rel("state.db")));
410                } else {
411                    // the source store holds transcripts the orchestration does not
412                    // model; the change has to be written INTO it, which is
413                    // UNI-18's first shared-store write
414                    report.refused.push(Refusal { file: rel("state.db"), reason: "bindings/obligations changed since import; writing the change into a Hermes session store is UNI-18 (the first shared-WAL write), not this codec's".into() });
415                }
416            } else if !profile.bindings.is_empty() || !profile.obligations.is_empty() {
417                // our own rows: they go into ONE fresh root store after the
418                // loop, partitioned by profile_name as Hermes keeps them
419                let _ = dest_store;
420                fresh_rows.push((name.as_str(), profile));
421            }
422        }
423
424        // everything Hermes has that we do not model
425        copy_unmodeled(profile, meta, &dir)?;
426        for f in &profile.residue.files {
427            report.written.push(ArtifactFidelity::byte(rel(f)));
428        }
429    }
430    if !fresh_rows.is_empty() {
431        let root_store = dest.join("state.db");
432        if root_store.exists() {
433            report.refused.push(Refusal { file: "state.db".into(), reason: "the destination already holds a Hermes session store; writing into a live store is UNI-18 (the first shared-WAL write), not this codec's".into() });
434        } else {
435            write_fresh_store(&fresh_rows, &root_store)?;
436            let bindings: usize = fresh_rows.iter().map(|(_, p)| p.bindings.len()).sum();
437            let obligations: usize = fresh_rows.iter().map(|(_, p)| p.obligations.len()).sum();
438            report.written.push(ArtifactFidelity::semantic(
439                "state.db",
440                vec![format!(
441                    "a fresh store at schema {HERMES_STATE_V22_VERSION} (Hermes migrates it on open): {bindings} binding(s) as sessions rows across {} profile(s) partitioned by profile_name, {obligations} obligation(s) as delivery rows; the transcripts live in the worker's store and are not carried",
442                    fresh_rows.len()
443                )],
444            ));
445        }
446    }
447    Ok(report)
448}
supercode_interchange/orchestration/codec/hermes.rs

supercode_interchange/orchestration/codec/
hermes.rs
