supercode_interchange/orchestration/codec/

hermes.rs

//! A Hermes home written from the orchestration (`hermes.mjs::toHermes`). Tiers per
//! file: byte for `cron/jobs.json`, `cron/executions.db`,
//! `webhook_subscriptions.json` (inline secrets re-inlined from the vault),
//! `config.yaml` when unchanged, every unmodeled file, and `state.db` when
//! bindings/obligations are UNCHANGED since import (copied). A FRESH
//! destination gets a store born at the fixture's schema (22) carrying the
//! bindings as `sessions` rows and the obligations as `delivery_obligations`
//! rows — Hermes migrates it on open; the transcripts live in the worker's
//! store and are not carried (semantic). A LIVE destination store is never
//! written: that first write into a shared WAL single-writer store is
//! UNI-18's, refused by name. `state.db` with
//! changed rows is REFUSED: the Hermes session-store write path is behind
//! UNI-22. Semantic for `SOUL.md` ⇄ `AGENTS.md` and a re-rendered
//! `config.yaml` (the O-blocks ride as top-level keys — ORC-1 F5).

use std::fs;
use std::path::Path;

use serde_json::Value;

use super::canonical::canonical_json;
use super::decode::{encode_obligation_row, OBLIGATION_COLUMNS};
use super::folder::{
    config_record, copy_unmodeled, encode_config, encode_jobs_file, encode_subscriptions_file,
    write_executions, Flavor, LoadedHome, ProfileIo,
};
use super::openclaw::ms_from_iso;
use super::sqlite::{write_table, Param};
use crate::ontology::{
    hermes_source_for_binding, render_hermes_session_key, ArtifactFidelity, Binding, Fidelity,
};
use crate::orchestration::Profile;

/// Hermes's `state.db` schema as the fixture home carries it (`SCHEMA_VERSION`
/// 22), without the FTS virtual tables and their triggers: Hermes creates
/// those itself on open (`hermes_state_common.py`, `CREATE VIRTUAL TABLE IF
/// NOT EXISTS`), and its main schema "advances freely on open (so future
/// migrations always land)".
const HERMES_STATE_V22: &str = include_str!("hermes_state_v22.sql");
const HERMES_STATE_V22_VERSION: i64 = 22;

/// The `sessions` columns a binding fills; every other column keeps Hermes's
/// own default.
const SESSION_COLUMNS: &[&str] = &[
    "id",
    "source",
    "user_id",
    "session_key",
    "chat_id",
    "chat_type",
    "thread_id",
    "expiry_finalized",
    "started_at",
    "ended_at",
    "end_reason",
    "handoff_state",
    "handoff_platform",
    "handoff_error",
    "profile_name",
];

fn epoch_seconds(iso: Option<&str>) -> Option<f64> {
    ms_from_iso(iso).map(|ms| ms as f64 / 1000.0)
}

/// A binding as a Hermes `sessions` row (inverse of `Binding::from_hermes_row`).
fn hermes_session_row(profile: &str, slot: &str, b: &Binding) -> Vec<Param> {
    let text = |v: &Option<String>| v.clone().map(Param::Text).unwrap_or(Param::Null);
    let hermes_profile = if profile == "default" {
        "main"
    } else {
        profile
    };
    let started = epoch_seconds(b.started_at.as_deref())
        .or_else(|| epoch_seconds(b.last_activity_at.as_deref()))
        .unwrap_or(0.0);
    vec![
        Param::Text(
            b.worker
                .session_id
                .clone()
                .unwrap_or_else(|| slot.to_string()),
        ),
        Param::Text(hermes_source_for_binding(b)),
        text(&b.key.participant_id),
        Param::Text(
            b.key
                .key
                .clone()
                .unwrap_or_else(|| render_hermes_session_key(hermes_profile, &b.key)),
        ),
        text(&b.key.chat_id),
        text(&b.key.kind),
        text(&b.key.thread_id),
        Param::Int(i64::from(b.ended_at.is_some())),
        Param::Real(started),
        epoch_seconds(b.ended_at.as_deref())
            .map(Param::Real)
            .unwrap_or(Param::Null),
        b.end_reason
            .map(|r| Param::Text(r.as_str().into()))
            .unwrap_or(Param::Null),
        text(&b.handoff.as_ref().map(|h| h.state.clone())),
        text(&b.handoff.as_ref().and_then(|h| h.to.clone())),
        text(&b.handoff.as_ref().and_then(|h| h.error.clone())),
        if profile == "default" {
            Param::Null
        } else {
            Param::Text(profile.into())
        },
    ]
}

/// A profile whose bindings carry the latest handoff record of their session, as Hermes's columns hold it:
/// the destination conversation's platform, the state word and the error.
fn with_handoffs(profile: &Profile, o: &crate::orchestration::Orchestration) -> Profile {
    let mut out = profile.clone();
    for b in out.bindings.values_mut() {
        let Some(id) = b.worker.session_id.clone() else {
            continue;
        };
        let session = format!("{}:{}", b.worker.harness.as_str(), id);
        if let Some(record) = o.handoffs.iter().rev().find(|h| h.session == session) {
            b.handoff = Some(crate::ontology::Handoff {
                to: o
                    .conversations
                    .get(&record.to)
                    .and_then(|c| c.surface.as_ref())
                    .and_then(|s| s.platform.clone()),
                state: match record.state {
                    crate::orchestration::HandoffState::Pending => "pending",
                    crate::orchestration::HandoffState::Done => "done",
                    crate::orchestration::HandoffState::Failed => "failed",
                }
                .into(),
                error: record.error.clone(),
            });
        }
    }
    out
}

/// A fresh Hermes store at `path`: the schema Hermes migrates on open, every
/// profile's bindings as sessions rows (a satellite profile's under its
/// `profile_name`, the way Hermes's own multiplexed gateway keeps them in the
/// ROOT store), every profile's obligations as delivery rows.
fn write_fresh_store(profiles: &[(&str, &Profile)], path: &Path) -> Result<()> {
    let placeholders = |n: usize| std::iter::repeat_n("?", n).collect::<Vec<_>>().join(",");
    write_table(
        path,
        HERMES_STATE_V22,
        "insert into schema_version (version) values (?)",
        &[vec![Param::Int(HERMES_STATE_V22_VERSION)]],
    )?;
    let sessions: Vec<Vec<Param>> = profiles
        .iter()
        .flat_map(|(name, profile)| {
            profile
                .bindings
                .iter()
                .map(move |(slot, b)| hermes_session_row(name, slot, b))
        })
        .collect();
    write_table(
        path,
        "",
        &format!(
            "insert into sessions ({}) values ({})",
            SESSION_COLUMNS.join(", "),
            placeholders(SESSION_COLUMNS.len())
        ),
        &sessions,
    )?;
    let obligations: Vec<Vec<Param>> = profiles
        .iter()
        .flat_map(|(_, profile)| profile.obligations.iter())
        .map(|o| encode_obligation_row(o).iter().map(Param::from).collect())
        .collect();
    write_table(
        path,
        "",
        &format!(
            "insert into delivery_obligations ({}) values ({})",
            OBLIGATION_COLUMNS.join(", "),
            placeholders(OBLIGATION_COLUMNS.len())
        ),
        &obligations,
    )
}
use crate::Result;

/// A write the codec would not guess at.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Refusal {
    /// The file, relative to the destination home.
    pub file: String,
    /// Why, naming the gate.
    pub reason: String,
}

/// What a Hermes decompile did.
#[derive(Debug, Clone, Default)]
pub struct HermesReport {
    /// Every artifact written, with its tier.
    pub written: Vec<ArtifactFidelity>,
    /// Every write refused.
    pub refused: Vec<Refusal>,
    /// What the export folded or dropped, listed before anything was written and kept beside it.
    pub loss: super::loss::LossReport,
}

fn write_atomic(path: &Path, text: &str) -> Result<()> {
    if let Some(parent) = path.parent() {
        fs::create_dir_all(parent)?;
    }
    let tmp = path.with_file_name(format!(
        "{}.tmp-{}",
        path.file_name().unwrap().to_string_lossy(),
        std::process::id()
    ));
    fs::write(&tmp, text)?;
    fs::rename(&tmp, path)?;
    Ok(())
}

/// Load a Hermes home into the orchestration.
pub fn from_hermes(home: &Path) -> Result<LoadedHome> {
    super::folder::load_home(home, Flavor::Hermes)
}

/// Write a Hermes home. Byte-tier files whose records are unchanged since a
/// Hermes import are copied from the import source; everything else is
/// emitted canonically.
pub fn to_hermes(loaded: &LoadedHome, dest: &Path, only: Option<&str>) -> Result<HermesReport> {
    let mut report = HermesReport::default();
    report.loss = super::loss::loss_report(
        "hermes",
        &loaded.orchestration,
        super::loss::HERMES_FEATURES,
    );
    super::loss::write_loss_report(dest, &report.loss)?;
    let empty = ProfileIo {
        raw: Default::default(),
        snapshot: Default::default(),
        source_dir: None,
        flavor: Flavor::Orchestrator,
        jobs_form: None,
        routes_at_top: false,
        borrowed_from: None,
        lenders: Vec::new(),
    };
    let mut fresh_rows: Vec<(&str, &Profile)> = Vec::new();
    for (name, profile) in &loaded.orchestration.profiles {
        if only.is_some_and(|o| o != name) {
            continue;
        }
        let dir = if name == "default" {
            dest.to_path_buf()
        } else {
            dest.join("profiles").join(name)
        };
        fs::create_dir_all(dir.join("cron"))?;
        let meta = loaded.io.get(name).unwrap_or(&empty);
        let rel = |p: &str| {
            if name == "default" {
                p.to_string()
            } else {
                format!("profiles/{name}/{p}")
            }
        };
        let unchanged =
            |file: &str, record: &Value| meta.snapshot.get(file) == Some(&canonical_json(record));
        fn emit_file(
            written: &mut Vec<ArtifactFidelity>,
            dir: &Path,
            path: String,
            file: &str,
            text: &str,
            tier: Fidelity,
        ) -> Result<()> {
            write_atomic(&dir.join(file), text)?;
            written.push(ArtifactFidelity {
                path,
                fidelity: tier,
                loss: Vec::new(),
            });
            Ok(())
        }
        macro_rules! emit {
            ($file:expr, $text:expr, $tier:expr) => {
                emit_file(&mut report.written, &dir, rel($file), $file, $text, $tier)?
            };
        }

        // config.yaml (only when the profile has one, or has something to say)
        let cfg_record = config_record(profile);
        let cfg_empty = profile.routes.is_empty()
            && profile.channels.is_empty()
            && profile.residue.config.is_empty()
            && profile.worker.is_none();
        // Only bytes read FROM A HERMES HOME may be reused for the two files
        // that carry credentials: our folder's config.yaml holds `{dotenv}`
        // refs where Hermes reads values.
        let hermes_bytes = meta.flavor == Flavor::Hermes;
        if hermes_bytes
            && unchanged("config.yaml", &cfg_record)
            && meta.raw.contains_key("config.yaml")
        {
            emit!(
                "config.yaml",
                &meta.raw["config.yaml"],
                Fidelity::ByteLossless
            );
        } else if !cfg_empty || meta.raw.contains_key("config.yaml") {
            emit!(
                "config.yaml",
                &encode_config(
                    profile,
                    Some(meta),
                    Some(&loaded.vault),
                    Flavor::Hermes,
                    &loaded.orchestration.agents
                ),
                Fidelity::Semantic
            );
        }

        // persona: AGENTS.md -> SOUL.md
        if let Some(persona) = &profile.persona {
            let src_name = if meta.raw.contains_key("SOUL.md") {
                "SOUL.md"
            } else {
                "AGENTS.md"
            };
            let record = serde_json::to_value(&profile.persona).unwrap();
            if unchanged(src_name, &record) && meta.raw.contains_key(src_name) {
                emit!(
                    "SOUL.md",
                    &meta.raw[src_name],
                    if src_name == "SOUL.md" {
                        Fidelity::ByteLossless
                    } else {
                        Fidelity::Semantic
                    }
                );
            } else {
                emit!(
                    "SOUL.md",
                    persona.text.as_deref().unwrap_or(""),
                    Fidelity::Semantic
                );
            }
        }

        // jobs — Hermes requires an ABSOLUTE workdir; ours is relative to the
        // profile folder, so the emitted copy resolves it against the destination
        let jobs_record: Vec<Value> = profile
            .jobs
            .values()
            .map(|j| serde_json::to_value(j).unwrap())
            .collect();
        if !profile.jobs.is_empty() || meta.raw.contains_key("cron/jobs.json") {
            if unchanged("cron/jobs.json", &Value::Array(jobs_record))
                && meta.raw.contains_key("cron/jobs.json")
            {
                emit!(
                    "cron/jobs.json",
                    &meta.raw["cron/jobs.json"],
                    Fidelity::ByteLossless
                );
            } else {
                let mut view: Profile = profile.clone();
                for job in view.jobs.values_mut() {
                    if let Some(w) = &job.workdir {
                        if !w.starts_with('/') {
                            job.workdir = Some(dir.join(w).display().to_string());
                        }
                    }
                }
                emit!(
                    "cron/jobs.json",
                    &encode_jobs_file(&view, Some(meta)),
                    Fidelity::ByteLossless
                );
            }
        }

        // fires
        let src_exec = meta
            .source_dir
            .as_ref()
            .map(|d| d.join("cron/executions.db"));
        if !profile.fires.is_empty() || src_exec.as_ref().is_some_and(|p| p.exists()) {
            let target = dir.join("cron/executions.db");
            let fires_record = serde_json::to_value(&profile.fires).unwrap();
            if unchanged("cron/executions.db", &fires_record)
                && src_exec.as_ref().is_some_and(|p| p.exists())
            {
                fs::copy(src_exec.as_ref().unwrap(), &target)?;
            } else {
                let tmp =
                    target.with_file_name(format!("executions.db.tmp-{}", std::process::id()));
                let _ = fs::remove_file(&tmp);
                write_executions(&tmp, &profile.fires)?;
                fs::rename(&tmp, &target)?;
            }
            report
                .written
                .push(ArtifactFidelity::byte(rel("cron/executions.db")));
        }

        // subscriptions (secrets re-inlined: Hermes reads them from this file)
        let subs_record: Vec<Value> = profile
            .subscriptions
            .values()
            .map(|s| serde_json::to_value(s).unwrap())
            .collect();
        if !profile.subscriptions.is_empty() || meta.raw.contains_key("webhook_subscriptions.json")
        {
            if hermes_bytes
                && unchanged("webhook_subscriptions.json", &Value::Array(subs_record))
                && meta.raw.contains_key("webhook_subscriptions.json")
            {
                emit!(
                    "webhook_subscriptions.json",
                    &meta.raw["webhook_subscriptions.json"],
                    Fidelity::ByteLossless
                );
            } else {
                emit!(
                    "webhook_subscriptions.json",
                    &encode_subscriptions_file(profile, Some(&loaded.vault)),
                    Fidelity::ByteLossless
                );
            }
        }

        // state.db: copy when untouched, refuse otherwise (UNI-22)
        if meta.borrowed_from.is_none() {
            let state_record = serde_json::json!({ "bindings": profile.bindings, "obligations": profile.obligations });
            let src_state = meta.source_dir.as_ref().map(|d| d.join("state.db"));
            let lenders_unchanged = meta.lenders.iter().all(|n| {
                let p = &loaded.orchestration.profiles[n];
                loaded.io.get(n).and_then(|m| m.snapshot.get("state.db")) == Some(&canonical_json(&serde_json::json!({ "bindings": p.bindings, "obligations": p.obligations })))
            });
            let dest_store = dir.join("state.db");
            if src_state.as_ref().is_some_and(|p| p.exists()) && meta.flavor == Flavor::Hermes {
                if unchanged("state.db", &state_record) && lenders_unchanged {
                    fs::copy(src_state.as_ref().unwrap(), &dest_store)?;
                    report.written.push(ArtifactFidelity::byte(rel("state.db")));
                } else {
                    // the source store holds transcripts the orchestration does not
                    // model; the change has to be written INTO it, which is
                    // UNI-18's first shared-store write
                    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() });
                }
            } else if !profile.bindings.is_empty() || !profile.obligations.is_empty() {
                // our own rows: they go into ONE fresh root store after the
                // loop, partitioned by profile_name as Hermes keeps them
                let _ = dest_store;
                fresh_rows.push((name.as_str(), profile));
            }
        }

        // everything Hermes has that we do not model
        copy_unmodeled(profile, meta, &dir)?;
        for f in &profile.residue.files {
            report.written.push(ArtifactFidelity::byte(rel(f)));
        }
    }
    if !fresh_rows.is_empty() {
        let root_store = dest.join("state.db");
        if root_store.exists() {
            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() });
        } else {
            // the IR's handoff records become each session row's handoff_* columns (G2)
            let patched: Vec<(&str, Profile)> = fresh_rows
                .iter()
                .map(|(name, p)| (*name, with_handoffs(p, &loaded.orchestration)))
                .collect();
            let refs: Vec<(&str, &Profile)> = patched.iter().map(|(n, p)| (*n, p)).collect();
            write_fresh_store(&refs, &root_store)?;
            let bindings: usize = fresh_rows.iter().map(|(_, p)| p.bindings.len()).sum();
            let obligations: usize = fresh_rows.iter().map(|(_, p)| p.obligations.len()).sum();
            report.written.push(ArtifactFidelity::semantic(
                "state.db",
                vec![format!(
                    "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",
                    fresh_rows.len()
                )],
            ));
        }
    }
    Ok(report)
}