1use 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_channel, decode_fire_row, decode_job, decode_obligation_row, decode_route,
25 decode_subscription, decode_worker, encode_channel, encode_fire_row, encode_job,
26 encode_obligation_folder_extras, encode_obligation_row, encode_route, encode_subscription,
27 load_error, surface_key_string, EXECUTION_COLUMNS, FOLDER_OBLIGATION_EXTRA_COLUMNS,
28 OBLIGATION_COLUMNS,
29};
30use super::dotenv::{parse_dotenv, render_dotenv};
31use super::sqlite::{read_rows, replace_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, SecretRef, SurfaceKey, Trigger, Worker,
35};
36use crate::orchestration::{
37 Access, AccessPolicy, ChannelConfig, Orchestration, PersonaRef, Profile, ProfileResidue,
38};
39use crate::Result;
40
41pub const OWNED_FILES: &[&str] = &[
43 "config.yaml",
44 "AGENTS.md",
45 "CLAUDE.md",
46 ".env",
47 "cron/jobs.json",
48 "cron/executions.db",
49 "webhook_subscriptions.json",
50 "state.db",
51];
52
53const CONFIG_O_KEYS: &[&str] = &["worker"];
54
55#[derive(Debug, Clone, Copy, PartialEq, Eq)]
57pub enum Flavor {
58 Orchestrator,
60 Hermes,
62}
63
64#[derive(Debug, Clone, PartialEq)]
66pub struct JobsForm {
67 pub object: bool,
69 pub extras: Map<String, Value>,
71}
72
73#[derive(Debug, Clone)]
76pub struct ProfileIo {
77 pub raw: BTreeMap<String, String>,
79 pub snapshot: BTreeMap<String, String>,
81 pub source_dir: Option<PathBuf>,
83 pub flavor: Flavor,
85 pub jobs_form: Option<JobsForm>,
87 pub routes_at_top: bool,
89 pub borrowed_from: Option<PathBuf>,
91 pub lenders: Vec<String>,
93}
94
95impl ProfileIo {
96 fn new(flavor: Flavor) -> Self {
97 Self {
98 raw: BTreeMap::new(),
99 snapshot: BTreeMap::new(),
100 source_dir: None,
101 flavor,
102 jobs_form: None,
103 routes_at_top: false,
104 borrowed_from: None,
105 lenders: Vec::new(),
106 }
107 }
108}
109
110#[derive(Debug, Clone)]
112pub struct LoadedHome {
113 pub orchestration: Orchestration,
115 pub vault: BTreeMap<String, String>,
117 pub io: BTreeMap<String, ProfileIo>,
119}
120
121fn sha256_hex(text: &str) -> String {
122 let mut h = Sha256::new();
123 h.update(text.as_bytes());
124 h.finalize().iter().map(|b| format!("{b:02x}")).collect()
125}
126
127pub fn persona_ref(text: &str) -> PersonaRef {
129 PersonaRef {
130 path: "AGENTS.md".into(),
131 text: Some(text.to_string()),
132 sha256: sha256_hex(text),
133 }
134}
135
136fn read_text(dir: &Path, rel: &str) -> Result<Option<String>> {
137 let p = dir.join(rel);
138 if !p.is_file() {
139 return Ok(None);
140 }
141 Ok(Some(fs::read_to_string(&p)?))
142}
143
144fn yaml_to_json(file: &str, text: &str) -> Result<Value> {
145 let value: serde_yaml::Value =
146 serde_yaml::from_str(text).map_err(|e| load_error(file, "", format!("YAML: {e}")))?;
147 let json: Value =
148 serde_json::to_value(value).map_err(|e| load_error(file, "", format!("YAML: {e}")))?;
149 Ok(if json.is_null() {
150 Value::Object(Map::new())
151 } else {
152 json
153 })
154}
155
156const YAML11_BOOLS: [&str; 18] = [
159 "yes", "Yes", "YES", "no", "No", "NO", "true", "True", "TRUE", "false", "False", "FALSE", "on",
160 "On", "ON", "off", "Off", "OFF",
161];
162const YAML11_MARK: &str = "__supercode_yaml11_string__";
163
164fn mark_yaml11_strings(value: &Value) -> Value {
165 match value {
166 Value::String(s) if YAML11_BOOLS.contains(&s.as_str()) => {
167 Value::String(format!("{YAML11_MARK}{s}"))
168 }
169 Value::Array(items) => Value::Array(items.iter().map(mark_yaml11_strings).collect()),
170 Value::Object(map) => Value::Object(
171 map.iter()
172 .map(|(k, v)| (k.clone(), mark_yaml11_strings(v)))
173 .collect(),
174 ),
175 other => other.clone(),
176 }
177}
178
179fn json_to_yaml(value: &Value) -> String {
180 if value.as_object().is_some_and(|m| m.is_empty()) {
181 return String::new();
182 }
183 let y: serde_yaml::Value =
185 serde_json::from_value(mark_yaml11_strings(value)).unwrap_or(serde_yaml::Value::Null);
186 let mut out = serde_yaml::to_string(&y).unwrap_or_default();
187 for word in YAML11_BOOLS {
188 out = out.replace(&format!("{YAML11_MARK}{word}"), &format!("'{word}'"));
189 }
190 out
191}
192
193pub fn empty_profile(name: &str, dir: &Path) -> Profile {
195 Profile {
196 name: name.to_string(),
197 dir: dir.to_path_buf(),
198 worker: None,
199 persona: None,
200 channels: BTreeMap::new(),
201 routes: Vec::new(),
202 jobs: BTreeMap::new(),
203 subscriptions: BTreeMap::new(),
204 access: Default::default(),
205 bindings: BTreeMap::new(),
206 fires: Vec::new(),
207 obligations: Vec::new(),
208 residue: ProfileResidue::default(),
209 }
210}
211
212pub fn config_record(profile: &Profile) -> Value {
214 serde_json::json!({
215 "worker": profile.worker,
216 "routes": profile.routes, "channels": profile.channels, "residue": profile.residue.config,
217 })
218}
219
220fn state_record(profile: &Profile) -> Value {
221 serde_json::json!({ "bindings": profile.bindings, "obligations": profile.obligations })
222}
223
224pub fn load_home(root: &Path, flavor: Flavor) -> Result<LoadedHome> {
226 if !root.is_dir() {
227 return Err(load_error(
228 &root.display().to_string(),
229 "",
230 "not a directory",
231 ));
232 }
233 let mut vault = BTreeMap::new();
234 let mut io = BTreeMap::new();
235 let mut profiles = BTreeMap::new();
236 let (default, default_io) = load_profile_dir("default", root, flavor, &mut vault)?;
237 profiles.insert("default".to_string(), default);
238 io.insert("default".to_string(), default_io);
239 let profiles_dir = root.join("profiles");
240 if profiles_dir.is_dir() {
241 let mut names: Vec<String> = fs::read_dir(&profiles_dir)?
242 .flatten()
243 .filter(|e| e.path().is_dir())
244 .filter_map(|e| e.file_name().into_string().ok())
245 .filter(|n| n != "node_modules" && !n.starts_with('.'))
246 .collect();
247 names.sort();
248 for name in names {
249 let dir = profiles_dir.join(&name);
250 if name == "default" {
251 return Err(load_error(
252 &dir.display().to_string(),
253 "",
254 "\"default\" is the root folder, not a named profile",
255 ));
256 }
257 let (profile, meta) = load_profile_dir(&name, &dir, flavor, &mut vault)?;
258 profiles.insert(name.clone(), profile);
259 io.insert(name, meta);
260 }
261 }
262 let mut loaded = LoadedHome {
263 orchestration: Orchestration {
264 root: root.to_path_buf(),
265 profiles,
266 },
267 vault,
268 io,
269 };
270 if flavor == Flavor::Hermes {
271 partition_shared_store(&mut loaded, root)?;
272 }
273 link_fires(&mut loaded, root)?;
274 Ok(loaded)
275}
276
277fn link_fires(loaded: &mut LoadedHome, root: &Path) -> Result<()> {
285 let names: Vec<String> = loaded.orchestration.profiles.keys().cloned().collect();
286 let root_store = root.join("state.db");
287 for name in &names {
288 let own = loaded.orchestration.profiles[name].dir.join("state.db");
289 let store = if own.is_file() {
290 own
291 } else {
292 root_store.clone()
293 };
294 let sessions: Vec<Map<String, Value>> = if table_exists(&store, "sessions") {
295 read_rows(
296 &store,
297 "select id, started_at, end_reason, parent_session_id, session_key from sessions",
298 &[],
299 )?
300 .unwrap_or_default()
301 } else {
302 Vec::new()
303 };
304 let holders: Vec<String> = {
307 let io = &loaded.io[name];
308 if io.borrowed_from.is_some() || !io.lenders.is_empty() {
309 let mut v = vec!["default".to_string()];
310 v.extend(loaded.io["default"].lenders.iter().cloned());
311 v
312 } else {
313 vec![name.clone()]
314 }
315 };
316 let mut by_fire: BTreeMap<String, (String, String)> = BTreeMap::new();
319 for (holder, profile) in &loaded.orchestration.profiles {
320 for o in &profile.obligations {
321 if let crate::orchestration::ObligationSource::Fire { fire_id } = &o.source {
322 by_fire
323 .entry(fire_id.clone())
324 .or_insert_with(|| (holder.clone(), o.id.clone()));
325 }
326 }
327 }
328 let mut links: Vec<(usize, Option<String>, Option<(String, String)>)> = Vec::new();
329 let profile = &loaded.orchestration.profiles[name];
330 for (index, fire) in profile.fires.iter().enumerate() {
331 if let Some(known) = by_fire.get(&fire.id) {
332 links.push((index, fire.session_id.clone(), Some(known.clone())));
334 continue;
335 }
336 let session_id = fire_session(&sessions, fire).or_else(|| fire.session_id.clone());
337 let session_key = session_id.as_deref().and_then(|id| {
338 sessions
339 .iter()
340 .find(|row| row.get("id").and_then(Value::as_str) == Some(id))
341 .and_then(|row| row.get("session_key"))
342 .and_then(Value::as_str)
343 .filter(|k| !k.is_empty())
344 .map(str::to_string)
345 });
346 let surface = profile.jobs.get(&fire.job_id).and_then(job_surface);
347 let (Some(from), to) = (
348 iso_epoch(&fire.claimed_at),
349 fire.finished_at
350 .as_deref()
351 .and_then(iso_epoch)
352 .unwrap_or(f64::MAX),
353 ) else {
354 links.push((index, session_id, None));
355 continue;
356 };
357 let in_window = |o: &&crate::orchestration::Obligation| {
358 o.created_at
359 .parse::<f64>()
360 .is_ok_and(|at| at >= from && at <= to)
361 };
362 let latest = |mut found: Vec<(&String, &crate::orchestration::Obligation)>| {
363 found.sort_by(|a, b| {
364 let at = |o: &crate::orchestration::Obligation| {
365 o.created_at.parse::<f64>().unwrap_or(0.0)
366 };
367 at(b.1)
368 .partial_cmp(&at(a.1))
369 .unwrap_or(std::cmp::Ordering::Equal)
370 });
371 found
372 .first()
373 .map(|(holder, o)| ((*holder).clone(), o.id.clone()))
374 };
375 let candidates = |pick: &dyn Fn(&crate::orchestration::Obligation) -> bool| {
376 holders
377 .iter()
378 .flat_map(|h| {
379 loaded.orchestration.profiles[h]
380 .obligations
381 .iter()
382 .filter(in_window)
383 .filter(|o| pick(o))
384 .map(move |o| (h, o))
385 })
386 .collect::<Vec<_>>()
387 };
388 let obligation = match &session_key {
389 Some(key) => latest(candidates(&|o| o.session_key.as_deref() == Some(key))),
390 None => None,
391 }
392 .or_else(|| {
393 let (platform, chat_id) = surface.as_ref()?;
394 latest(candidates(&|o| {
395 o.target.platform.as_deref() == Some(platform)
396 && o.target.chat_id.as_deref() == Some(chat_id)
397 }))
398 });
399 links.push((index, session_id, obligation));
400 }
401 for (index, session_id, obligation) in links {
402 let fire_id = {
403 let fire = &mut loaded.orchestration.profiles.get_mut(name).unwrap().fires[index];
404 fire.session_id = session_id;
405 fire.obligation_id = obligation.as_ref().map(|(_, id)| id.clone());
406 fire.id.clone()
407 };
408 if let Some((holder, obligation_id)) = obligation {
409 if let Some(o) = loaded
410 .orchestration
411 .profiles
412 .get_mut(&holder)
413 .and_then(|p| p.obligations.iter_mut().find(|o| o.id == obligation_id))
414 {
415 o.source = crate::orchestration::ObligationSource::Fire { fire_id };
416 }
417 }
418 }
419 }
420 for name in &names {
422 let profile = &loaded.orchestration.profiles[name];
423 let io = loaded.io.get_mut(name).unwrap();
424 io.snapshot.insert(
425 "cron/executions.db".into(),
426 canonical_json(&serde_json::to_value(&profile.fires).unwrap()),
427 );
428 io.snapshot
429 .insert("state.db".into(), canonical_json(&state_record(profile)));
430 }
431 Ok(())
432}
433
434fn job_surface(job: &crate::orchestration::Job) -> Option<(String, String)> {
439 use crate::orchestration::Target;
440 match &job.deliver {
441 Target::Origin => {
442 let origin = job.origin.as_ref()?;
443 Some((origin.platform.clone(), origin.chat_id.clone()?))
444 }
445 Target::Explicit {
446 platform, chat_id, ..
447 } => Some((platform.clone(), chat_id.clone()?)),
448 Target::Home | Target::Local => None,
449 }
450}
451
452fn fire_session(
456 sessions: &[Map<String, Value>],
457 fire: &crate::orchestration::Fire,
458) -> Option<String> {
459 let claimed = instant_key(&fire.claimed_at)?;
460 let finished = fire.finished_at.as_deref().and_then(instant_key);
461 let candidates = sessions.iter().filter_map(|row| {
462 let id = row.get("id").and_then(Value::as_str)?;
463 let key = cron_session_instant(id, &fire.job_id)?;
464 (key >= claimed && finished.is_none_or(|f| key <= f)).then(|| (key, id.to_string()))
465 });
466 let chosen = match finished {
467 Some(_) => candidates.max_by_key(|(key, _)| *key),
468 None => candidates.min_by_key(|(key, _)| *key),
469 }?;
470 let mut current = chosen.1;
471 for _ in 0..32 {
472 let row = sessions
473 .iter()
474 .find(|row| row.get("id").and_then(Value::as_str) == Some(current.as_str()));
475 let compressed = row
476 .and_then(|row| row.get("end_reason"))
477 .and_then(Value::as_str)
478 == Some("compression");
479 if !compressed {
480 return Some(current);
481 }
482 let next = sessions
483 .iter()
484 .filter(|row| {
485 row.get("parent_session_id").and_then(Value::as_str) == Some(current.as_str())
486 })
487 .max_by(|a, b| {
488 let at = |r: &Map<String, Value>| {
489 r.get("started_at").and_then(Value::as_f64).unwrap_or(0.0)
490 };
491 at(a)
492 .partial_cmp(&at(b))
493 .unwrap_or(std::cmp::Ordering::Equal)
494 .then_with(|| {
495 let id = |r: &Map<String, Value>| {
496 r.get("id")
497 .and_then(Value::as_str)
498 .unwrap_or("")
499 .to_string()
500 };
501 id(a).cmp(&id(b))
502 })
503 })
504 .and_then(|row| row.get("id").and_then(Value::as_str).map(str::to_string));
505 match next {
506 None => return Some(current),
507 Some(next) => current = next,
508 }
509 }
510 Some(current)
511}
512
513fn instant_key(iso: &str) -> Option<u64> {
515 let digits: String = iso
516 .chars()
517 .take_while(|c| *c != '+' && *c != 'Z')
518 .filter(char::is_ascii_digit)
519 .collect();
520 (digits.len() >= 14).then(|| digits[..14].parse().ok())?
521}
522
523fn cron_session_instant(session_id: &str, job_id: &str) -> Option<u64> {
525 if hermes_cron_job_id(session_id).as_deref() != Some(job_id) {
526 return None;
527 }
528 let stamp = session_id.rsplit_once('_')?;
529 let date = stamp.0.rsplit_once('_')?.1;
530 format!("{date}{}", stamp.1).parse().ok()
531}
532
533pub(crate) fn iso_epoch(iso: &str) -> Option<f64> {
535 let (instant, offset) = if let Some(instant) = iso.strip_suffix('Z') {
536 (instant, 0.0)
537 } else {
538 let time_at = iso.find('T')?;
539 let sign_at = iso[time_at..].find(['+', '-']).map(|i| i + time_at)?;
540 let (instant, offset) = iso.split_at(sign_at);
541 let (hours, minutes) = offset[1..].split_once(':')?;
542 let seconds = hours.parse::<f64>().ok()? * 3_600.0 + minutes.parse::<f64>().ok()? * 60.0;
543 (
544 instant,
545 if offset.starts_with('-') {
546 -seconds
547 } else {
548 seconds
549 },
550 )
551 };
552 let (date, time) = instant.split_once('T')?;
553 let mut date = date.splitn(3, '-');
554 let year: i64 = date.next()?.parse().ok()?;
555 let month: i64 = date.next()?.parse().ok()?;
556 let day: i64 = date.next()?.parse().ok()?;
557 let mut clock = time.splitn(3, ':');
558 let hour: i64 = clock.next()?.parse().ok()?;
559 let minute: i64 = clock.next()?.parse().ok()?;
560 let seconds: f64 = clock.next()?.parse().ok()?;
561 let year = year - i64::from(month <= 2);
562 let era = year.div_euclid(400);
563 let yoe = year - era * 400;
564 let doy = (153 * (if month > 2 { month - 3 } else { month + 9 }) + 2) / 5 + day - 1;
565 let doe = yoe * 365 + yoe / 4 - yoe / 100 + doy;
566 let days = era * 146_097 + doe - 719_468;
567 Some((days * 86_400 + hour * 3_600 + minute * 60) as f64 + seconds - offset)
568}
569
570fn partition_shared_store(loaded: &mut LoadedHome, root: &Path) -> Result<()> {
575 let root_path = root.join("state.db");
576 if !root_path.is_file() {
577 return Ok(());
578 }
579 let names: Vec<String> = loaded
580 .orchestration
581 .profiles
582 .keys()
583 .filter(|n| *n != "default")
584 .cloned()
585 .collect();
586 for name in names {
587 let has_own = loaded.orchestration.profiles[&name]
588 .dir
589 .join("state.db")
590 .is_file();
591 if has_own {
592 continue;
593 }
594 loaded.io.get_mut(&name).unwrap().borrowed_from = Some(root_path.clone());
595 loaded
596 .io
597 .get_mut("default")
598 .unwrap()
599 .lenders
600 .push(name.clone());
601 if table_exists(&root_path, "sessions") {
602 let rows = read_rows(
603 &root_path,
604 "select * from sessions where profile_name = ?1 order by started_at, id",
605 &[&name],
606 )?
607 .unwrap_or_default();
608 for row in rows {
609 if let Some(b) = binding_from_hermes_session(&root_path, &row, &name) {
610 loaded
611 .orchestration
612 .profiles
613 .get_mut(&name)
614 .unwrap()
615 .bindings
616 .insert(surface_key_string(&b.key), b);
617 }
618 }
619 }
620 let root_profile = loaded.orchestration.profiles.get_mut("default").unwrap();
622 let (mine, rest): (Vec<_>, Vec<_>) = root_profile.obligations.drain(..).partition(|o| {
623 o.session_key
624 .as_deref()
625 .and_then(parse_hermes_session_key)
626 .and_then(|(_, p)| p)
627 .as_deref()
628 == Some(name.as_str())
629 });
630 root_profile.obligations = rest;
631 loaded
632 .orchestration
633 .profiles
634 .get_mut(&name)
635 .unwrap()
636 .obligations = mine;
637 let snap = canonical_json(&state_record(&loaded.orchestration.profiles[&name]));
638 loaded
639 .io
640 .get_mut(&name)
641 .unwrap()
642 .snapshot
643 .insert("state.db".into(), snap);
644 }
645 let snap = canonical_json(&state_record(&loaded.orchestration.profiles["default"]));
646 loaded
647 .io
648 .get_mut("default")
649 .unwrap()
650 .snapshot
651 .insert("state.db".into(), snap);
652 Ok(())
653}
654
655const HERMES_SESSION_MAPPED: &[&str] = &[
656 "id",
657 "source",
658 "session_key",
659 "chat_id",
660 "chat_type",
661 "thread_id",
662 "user_id",
663 "profile_name",
664 "handoff_state",
665 "handoff_platform",
666 "handoff_error",
667 "started_at",
668 "ended_at",
669 "end_reason",
670];
671
672fn routing_scope(dir: &Path) -> String {
674 let sessions = dir.join("sessions");
675 fs::canonicalize(&sessions)
676 .or_else(|_| fs::canonicalize(dir).map(|d| d.join("sessions")))
677 .unwrap_or(sessions)
678 .display()
679 .to_string()
680}
681
682fn routing_entry(profile_name: &str, slot: &str, b: &Binding) -> (String, Value) {
685 let ns = if profile_name == "default" {
686 ""
687 } else {
688 profile_name
689 };
690 let key = b
691 .key
692 .key
693 .clone()
694 .filter(|k| k.starts_with("agent:"))
695 .unwrap_or_else(|| crate::ontology::render_hermes_session_key(ns, &b.key));
696 let entry = serde_json::json!({
697 "session_key": key,
698 "session_id": b.worker.session_id.clone().unwrap_or_else(|| slot.to_string()),
699 "created_at": b.started_at,
700 "updated_at": b.last_activity_at.clone().or_else(|| b.started_at.clone()),
701 "display_name": Value::Null,
702 "platform": b.key.platform,
703 "chat_type": b.key.kind.clone().unwrap_or_else(|| "dm".into()),
704 "origin": {
705 "platform": b.key.platform,
706 "chat_id": b.key.chat_id,
707 "chat_type": b.key.kind,
708 "thread_id": b.key.thread_id,
709 "user_id": b.key.participant_id,
710 },
711 "metadata": { "supercode": { "slot": slot, "binding": b } },
712 "input_tokens": 0, "output_tokens": 0, "cache_read_tokens": 0, "cache_write_tokens": 0,
713 "total_tokens": 0, "last_prompt_tokens": 0, "estimated_cost_usd": 0.0, "cost_status": "unknown",
714 "was_auto_reset": false, "auto_reset_reason": Value::Null, "reset_had_activity": false,
715 });
716 (key, entry)
717}
718
719fn binding_from_routing(file: &Path, row: &Map<String, Value>) -> Option<(String, Binding)> {
722 let entry: Value = serde_json::from_str(row.get("entry_json")?.as_str()?).ok()?;
723 if let Some(sc) = entry.pointer("/metadata/supercode") {
724 let b: Binding = serde_json::from_value(sc.get("binding")?.clone()).ok()?;
725 let slot = sc
726 .get("slot")
727 .and_then(Value::as_str)
728 .map(str::to_string)
729 .unwrap_or_else(|| surface_key_string(&b.key));
730 return Some((slot, b));
731 }
732 let session_key = entry.get("session_key").and_then(Value::as_str)?;
733 let (mut key, _) = parse_hermes_session_key(session_key)?;
734 key.key = None;
735 let text = |k: &str| entry.get(k).and_then(Value::as_str).map(str::to_string);
736 let b = Binding {
737 key: key.clone(),
738 profile: None,
739 worker: Worker {
740 harness: HarnessId::new(HarnessId::HERMES),
741 session_id: text("session_id"),
742 locator: Some(file.display().to_string()),
743 },
744 trigger: Trigger::Unknown,
745 recurrence: None,
746 handoff: None,
747 started_at: text("created_at"),
748 last_activity_at: text("updated_at"),
749 ended_at: None,
750 end_reason: None,
751 residue: Residue::default(),
752 };
753 Some((surface_key_string(&key), b))
754}
755
756pub fn binding_from_hermes_session(
758 file: &Path,
759 row: &Map<String, Value>,
760 profile_name: &str,
761) -> Option<Binding> {
762 let text = |k: &str| {
763 row.get(k)
764 .and_then(|v| match v {
765 Value::String(s) => Some(s.clone()),
766 Value::Number(n) => Some(n.to_string()),
767 _ => None,
768 })
769 .filter(|s| !s.is_empty())
770 };
771 let num = |k: &str| row.get(k).and_then(Value::as_f64);
772 let id = text("id")?;
773 let source = text("source");
774 let parsed = text("session_key").and_then(|k| parse_hermes_session_key(&k));
777 let chat_type = text("chat_type").or_else(|| parsed.as_ref().and_then(|(k, _)| k.kind.clone()));
778 let key = if text("session_key").is_some()
779 && chat_type
780 .as_deref()
781 .is_some_and(|c| super::decode::CHAT_TYPES.contains(&c))
782 {
783 SurfaceKey {
784 key: None,
785 platform: source
786 .clone()
787 .or_else(|| parsed.as_ref().and_then(|(k, _)| k.platform.clone())),
788 kind: chat_type,
789 chat_id: text("chat_id")
790 .or_else(|| parsed.as_ref().and_then(|(k, _)| k.chat_id.clone())),
791 thread_id: text("thread_id")
792 .or_else(|| parsed.as_ref().and_then(|(k, _)| k.thread_id.clone())),
793 participant_id: parsed.as_ref().and_then(|(k, _)| k.participant_id.clone()),
794 }
795 } else if source.as_deref() == Some("cron") {
796 let job = crate::ontology::hermes_cron_job_id(&id).unwrap_or_else(|| id.clone());
797 SurfaceKey {
798 key: None,
799 platform: Some("cron".into()),
800 kind: Some("dm".into()),
801 chat_id: Some(job),
802 thread_id: None,
803 participant_id: None,
804 }
805 } else {
806 return None;
807 };
808 if let Some(p) = text("profile_name") {
809 if p != profile_name && !(profile_name == "default" && p == "main") {
810 return None; }
812 }
813 let iso = |v: Option<f64>| v.map(epoch_iso);
814 let end_word = text("end_reason");
815 let end_reason = end_word.as_deref().and_then(EndReason::parse);
816 let mut residue = Residue::default();
817 for (k, v) in row {
818 if !HERMES_SESSION_MAPPED.contains(&k.as_str()) && !v.is_null() {
819 residue.keep(k.clone(), v.clone());
820 }
821 }
822 if let (Some(word), None) = (&end_word, end_reason) {
823 residue.keep("end_reason", Value::String(word.clone()));
824 }
825 if let Some(u) = text("user_id") {
826 residue.keep("user_id", Value::String(u));
827 }
828 let recurrence = if key.platform.as_deref() == Some("cron") {
829 key.chat_id.clone().map(|job_id| Recurrence {
830 job_id,
831 kind: "cron".into(),
832 })
833 } else {
834 None
835 };
836 Some(Binding {
837 trigger: match (recurrence.is_some(), source.as_deref()) {
838 (true, _) => Trigger::Cron,
839 (_, Some(s)) => crate::ontology::hermes_trigger_for_source(s),
840 _ => Trigger::Unknown,
841 },
842 key,
843 profile: None,
844 worker: Worker {
845 harness: HarnessId::new(HarnessId::HERMES),
846 session_id: Some(id),
847 locator: Some(file.display().to_string()),
848 },
849 recurrence,
850 handoff: text("handoff_state").map(|state| Handoff {
851 to: text("handoff_platform"),
852 state,
853 error: text("handoff_error"),
854 }),
855 started_at: iso(num("started_at")),
856 last_activity_at: iso(num("ended_at").or_else(|| num("started_at"))),
857 ended_at: iso(num("ended_at")),
858 end_reason,
859 residue,
860 })
861}
862
863fn epoch_iso(seconds: f64) -> String {
865 let row = HermesSessionRow {
866 started_at: Some(seconds),
867 ..Default::default()
868 };
869 Binding::from_hermes_row(&row, None)
870 .started_at
871 .unwrap_or_default()
872}
873
874fn load_profile_dir(
875 name: &str,
876 dir: &Path,
877 flavor: Flavor,
878 vault: &mut BTreeMap<String, String>,
879) -> Result<(Profile, ProfileIo)> {
880 let mut profile = empty_profile(name, dir);
881 let mut config_normalized = false;
882 let mut subs_normalized = false;
883 let mut meta = ProfileIo::new(flavor);
884 meta.source_dir = Some(dir.to_path_buf());
885 let remember = |meta: &mut ProfileIo, rel: &str, raw: Option<String>, record: &Value| {
886 if let Some(raw) = raw {
887 meta.raw.insert(rel.to_string(), raw);
888 }
889 meta.snapshot
890 .insert(rel.to_string(), canonical_json(record));
891 };
892
893 if let Some(env) = read_text(dir, ".env")? {
895 for (k, v) in parse_dotenv(&env) {
896 vault.insert(k, v);
897 }
898 meta.raw.insert(".env".into(), env);
899 }
900
901 let cfg_file = dir.join("config.yaml").display().to_string();
903 let cfg_text = read_text(dir, "config.yaml")?;
904 let cfg = match &cfg_text {
905 Some(text) => yaml_to_json(&cfg_file, text)?,
906 None => Value::Object(Map::new()),
907 };
908 let cfg_map = cfg
909 .as_object()
910 .ok_or_else(|| load_error(&cfg_file, "", "expected a mapping"))?;
911 profile.worker = decode_worker(&cfg_file, cfg_map.get("worker"))?;
912 let gateway = cfg_map.get("gateway").and_then(Value::as_object);
913 let routes_raw: Vec<Value> = match cfg_map.get("profile_routes").and_then(Value::as_array) {
914 Some(a) => {
915 meta.routes_at_top = true;
916 a.clone()
917 }
918 None => gateway
919 .and_then(|g| g.get("profile_routes"))
920 .and_then(Value::as_array)
921 .cloned()
922 .unwrap_or_default(),
923 };
924 for (i, r) in routes_raw.iter().enumerate() {
925 profile.routes.push(decode_route(&cfg_file, i, r)?);
926 }
927 if let Some(platforms) = cfg_map.get("platforms") {
928 let map = platforms
929 .as_object()
930 .ok_or_else(|| load_error(&cfg_file, "platforms", "expected a map"))?;
931 let known: BTreeSet<String> = vault.keys().cloned().collect();
932 for (platform, raw) in map {
933 profile.channels.insert(
934 platform.clone(),
935 decode_channel(&cfg_file, platform, raw, vault)?,
936 );
937 }
938 config_normalized =
939 flavor == Flavor::Orchestrator && vault.keys().any(|k| !known.contains(k));
940 }
941 {
947 let own: BTreeMap<String, String> = meta
948 .raw
949 .get(".env")
950 .map(|text| parse_dotenv(text).into_iter().collect())
951 .unwrap_or_default();
952 let set = |k: &str| own.get(k).filter(|v| !v.trim().is_empty()).cloned();
953 for (platform, token, home) in [
954 ("discord", "DISCORD_BOT_TOKEN", "DISCORD_HOME_CHANNEL"),
955 ("telegram", "TELEGRAM_BOT_TOKEN", "TELEGRAM_HOME_CHANNEL"),
956 ("slack", "SLACK_BOT_TOKEN", "SLACK_HOME_CHANNEL"),
957 ] {
958 if profile.channels.contains_key(platform) || set(token).is_none() {
959 continue;
960 }
961 let mut extra = BTreeMap::new();
962 extra.insert("from_env".to_string(), Value::Bool(true));
963 if let Some(chat) = set(home) {
964 let mut hc = Map::new();
965 hc.insert("platform".into(), Value::String(platform.into()));
966 hc.insert("chat_id".into(), Value::String(chat));
967 hc.insert(
968 "name".into(),
969 Value::String(set(&format!("{home}_NAME")).unwrap_or_else(|| "Home".into())),
970 );
971 if let Some(thread) = set(&format!("{home}_THREAD_ID")) {
972 hc.insert("thread_id".into(), Value::String(thread));
973 }
974 extra.insert("home_channel".to_string(), Value::Object(hc));
975 }
976 let mut credentials = BTreeMap::new();
977 credentials.insert("token".to_string(), SecretRef::Dotenv(token.to_string()));
978 profile.channels.insert(
979 platform.to_string(),
980 ChannelConfig {
981 platform: platform.to_string(),
982 enabled: true,
983 credentials,
984 extra,
985 },
986 );
987 }
988 }
989 for (k, v) in cfg_map {
991 if CONFIG_O_KEYS.contains(&k.as_str()) || k == "platforms" || k == "profile_routes" {
992 continue;
993 }
994 if k == "gateway" {
995 let mut g = v.as_object().cloned().unwrap_or_default();
996 g.remove("profile_routes");
997 if !g.is_empty() {
998 profile
999 .residue
1000 .config
1001 .insert("gateway".into(), Value::Object(g));
1002 }
1003 continue;
1004 }
1005 profile.residue.config.insert(k.clone(), v.clone());
1006 }
1007 remember(&mut meta, "config.yaml", cfg_text, &config_record(&profile));
1008 if config_normalized {
1009 meta.raw.remove("config.yaml");
1014 meta.snapshot.remove("config.yaml");
1015 }
1016
1017 let persona_file = if flavor == Flavor::Hermes {
1019 "SOUL.md"
1020 } else {
1021 "AGENTS.md"
1022 };
1023 let persona_text = read_text(dir, persona_file)?;
1024 profile.persona = persona_text.as_ref().map(|t| PersonaRef {
1025 path: "AGENTS.md".into(),
1026 text: Some(t.clone()),
1027 sha256: sha256_hex(t),
1028 });
1029 remember(
1030 &mut meta,
1031 persona_file,
1032 persona_text,
1033 &serde_json::to_value(&profile.persona).unwrap(),
1034 );
1035
1036 let jobs_file = dir.join("cron/jobs.json").display().to_string();
1038 let jobs_text = read_text(dir, "cron/jobs.json")?;
1039 if let Some(text) = &jobs_text {
1040 let parsed: Value = serde_json::from_str(text)
1041 .map_err(|e| load_error(&jobs_file, "", format!("JSON: {e}")))?;
1042 let arr: Vec<Value> = match &parsed {
1043 Value::Array(a) => {
1044 meta.jobs_form = Some(JobsForm {
1045 object: false,
1046 extras: Map::new(),
1047 });
1048 a.clone()
1049 }
1050 Value::Object(o) => match o.get("jobs") {
1051 Some(Value::Array(a)) => {
1052 let mut extras = o.clone();
1053 extras.remove("jobs");
1054 meta.jobs_form = Some(JobsForm {
1055 object: true,
1056 extras,
1057 });
1058 a.clone()
1059 }
1060 Some(Value::Object(m)) => {
1061 let mut extras = o.clone();
1062 extras.remove("jobs");
1063 meta.jobs_form = Some(JobsForm {
1064 object: true,
1065 extras,
1066 });
1067 m.iter()
1068 .map(|(id, j)| {
1069 let mut j = j.as_object().cloned().unwrap_or_default();
1070 j.insert("id".into(), Value::String(id.clone()));
1071 Value::Object(j)
1072 })
1073 .collect()
1074 }
1075 _ => {
1076 return Err(load_error(
1077 &jobs_file,
1078 "",
1079 "expected an array of jobs or {\"jobs\": [...]}",
1080 ))
1081 }
1082 },
1083 _ => {
1084 return Err(load_error(
1085 &jobs_file,
1086 "",
1087 "expected an array of jobs or {\"jobs\": [...]}",
1088 ))
1089 }
1090 };
1091 for raw in &arr {
1092 let job = decode_job(&jobs_file, raw)?;
1093 if profile.jobs.contains_key(&job.id) {
1094 return Err(load_error(&jobs_file, &job.id, "duplicate job id"));
1095 }
1096 profile.jobs.insert(job.id.clone(), job);
1097 }
1098 }
1099 let jobs_record: Vec<Value> = profile
1100 .jobs
1101 .values()
1102 .map(|j| serde_json::to_value(j).unwrap())
1103 .collect();
1104 remember(
1105 &mut meta,
1106 "cron/jobs.json",
1107 jobs_text,
1108 &Value::Array(jobs_record),
1109 );
1110
1111 let exec_path = dir.join("cron/executions.db");
1113 if table_exists(&exec_path, "executions") {
1114 for row in read_rows(
1115 &exec_path,
1116 "select * from executions order by claimed_at, id",
1117 &[],
1118 )?
1119 .unwrap_or_default()
1120 {
1121 profile
1122 .fires
1123 .push(decode_fire_row(&exec_path.display().to_string(), &row)?);
1124 }
1125 }
1126 remember(
1127 &mut meta,
1128 "cron/executions.db",
1129 None,
1130 &serde_json::to_value(&profile.fires).unwrap(),
1131 );
1132
1133 let subs_file = dir.join("webhook_subscriptions.json").display().to_string();
1135 let subs_text = read_text(dir, "webhook_subscriptions.json")?;
1136 if let Some(text) = &subs_text {
1137 let parsed: Value = serde_json::from_str(text)
1138 .map_err(|e| load_error(&subs_file, "", format!("JSON: {e}")))?;
1139 let map = parsed
1140 .as_object()
1141 .ok_or_else(|| load_error(&subs_file, "", "expected a map"))?;
1142 let known: BTreeSet<String> = vault.keys().cloned().collect();
1143 for (n, raw) in map {
1144 profile
1145 .subscriptions
1146 .insert(n.clone(), decode_subscription(&subs_file, n, raw, vault)?);
1147 }
1148 subs_normalized =
1149 flavor == Flavor::Orchestrator && vault.keys().any(|k| !known.contains(k));
1150 }
1151 let subs_record: Vec<Value> = profile
1152 .subscriptions
1153 .values()
1154 .map(|s| serde_json::to_value(s).unwrap())
1155 .collect();
1156 remember(
1157 &mut meta,
1158 "webhook_subscriptions.json",
1159 subs_text,
1160 &Value::Array(subs_record),
1161 );
1162 if subs_normalized {
1163 meta.raw.remove("webhook_subscriptions.json");
1164 meta.snapshot.remove("webhook_subscriptions.json");
1165 }
1166
1167 let own_env = meta
1169 .raw
1170 .get(".env")
1171 .map(|t| parse_dotenv(t))
1172 .unwrap_or_default();
1173 profile.access = hermes_access(dir, &own_env, cfg_map.get("platforms"));
1174
1175 let state_path = dir.join("state.db");
1177 if table_exists(&state_path, "delivery_obligations") {
1178 for row in read_rows(
1179 &state_path,
1180 "select * from delivery_obligations order by created_at, obligation_id",
1181 &[],
1182 )?
1183 .unwrap_or_default()
1184 {
1185 profile.obligations.push(decode_obligation_row(
1186 &state_path.display().to_string(),
1187 &row,
1188 )?);
1189 }
1190 }
1191 if flavor == Flavor::Orchestrator {
1192 if table_exists(&state_path, "gateway_routing") {
1195 let scope = routing_scope(dir);
1196 for row in read_rows(
1197 &state_path,
1198 "select session_key, entry_json from gateway_routing where scope = ? order by updated_at",
1199 &[&scope],
1200 )?
1201 .unwrap_or_default()
1202 {
1203 if let Some((slot, b)) = binding_from_routing(&state_path, &row) {
1204 profile.bindings.insert(slot, b);
1205 }
1206 }
1207 }
1208 } else if table_exists(&state_path, "sessions") {
1209 for row in read_rows(
1210 &state_path,
1211 "select * from sessions order by started_at, id",
1212 &[],
1213 )?
1214 .unwrap_or_default()
1215 {
1216 if let Some(b) = binding_from_hermes_session(&state_path, &row, name) {
1217 profile.bindings.insert(surface_key_string(&b.key), b);
1218 }
1219 }
1220 }
1221 remember(&mut meta, "state.db", None, &state_record(&profile));
1222
1223 profile.residue.files = list_unmodeled(dir, flavor)?;
1225 Ok((profile, meta))
1226}
1227
1228fn list_unmodeled(dir: &Path, flavor: Flavor) -> Result<Vec<String>> {
1229 let mut owned: Vec<&str> = OWNED_FILES.to_vec();
1230 if flavor == Flavor::Hermes {
1231 owned.push("SOUL.md");
1232 owned.retain(|f| !["AGENTS.md", "CLAUDE.md"].contains(f));
1233 }
1234 let runtime_artifacts = ["orchestrator.lock", "orchestrator.sock", "service"];
1235 let mut out = Vec::new();
1236 fn walk(
1237 base: &Path,
1238 d: &Path,
1239 owned: &[&str],
1240 runtime: &[&str],
1241 out: &mut Vec<String>,
1242 ) -> Result<()> {
1243 let entries = match fs::read_dir(d) {
1244 Ok(entries) => entries,
1245 Err(error) if d != base && error.kind() == std::io::ErrorKind::NotFound => {
1247 return Ok(());
1248 }
1249 Err(error) => return Err(error.into()),
1250 };
1251 let mut entries = entries.collect::<std::io::Result<Vec<_>>>()?;
1252 entries.sort_by_key(|e| e.file_name());
1253 for entry in entries {
1254 let p = entry.path();
1255 let rel = p
1256 .strip_prefix(base)
1257 .unwrap_or(&p)
1258 .to_string_lossy()
1259 .replace('\\', "/");
1260 let name = entry.file_name().to_string_lossy().into_owned();
1261 if rel == "profiles"
1262 || name == "node_modules"
1263 || name == ".git"
1264 || rel.starts_with("state.db")
1265 || rel.starts_with("cron/executions.db")
1266 {
1267 continue;
1268 }
1269 if runtime.contains(&rel.as_str()) || regex_tmp(&name) {
1270 continue;
1271 }
1272 let st = match fs::symlink_metadata(&p) {
1273 Ok(metadata) => metadata,
1274 Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue,
1276 Err(error) => return Err(error.into()),
1277 };
1278 if st.is_dir() {
1279 walk(base, &p, owned, runtime, out)?;
1280 continue;
1281 }
1282 if !st.is_file() {
1283 continue;
1284 }
1285 if owned.contains(&rel.as_str()) {
1286 continue;
1287 }
1288 out.push(rel);
1289 }
1290 Ok(())
1291 }
1292 walk(dir, dir, &owned, &runtime_artifacts, &mut out)?;
1293 Ok(out)
1294}
1295
1296fn regex_tmp(name: &str) -> bool {
1297 name.rsplit_once(".tmp-")
1299 .is_some_and(|(_, pid)| !pid.is_empty() && pid.chars().all(|c| c.is_ascii_digit()))
1300}
1301
1302fn write_atomic(path: &Path, text: &str) -> Result<()> {
1305 if let Some(parent) = path.parent() {
1306 fs::create_dir_all(parent)?;
1307 }
1308 let tmp = path.with_file_name(format!(
1309 "{}.tmp-{}",
1310 path.file_name().unwrap().to_string_lossy(),
1311 std::process::id()
1312 ));
1313 write_with_mode(&tmp, text, target_mode(path))?;
1314 fs::rename(&tmp, path)?;
1315 Ok(())
1316}
1317
1318#[cfg(unix)]
1322fn target_mode(path: &Path) -> Option<u32> {
1323 use std::os::unix::fs::PermissionsExt;
1324 let current = fs::metadata(path)
1325 .ok()
1326 .map(|m| m.permissions().mode() & 0o777);
1327 if path.file_name().is_some_and(|name| name == ".env") {
1328 return Some(current.unwrap_or(0o600) & 0o600);
1329 }
1330 current
1331}
1332
1333#[cfg(not(unix))]
1334fn target_mode(_path: &Path) -> Option<u32> {
1335 None
1336}
1337
1338#[cfg(unix)]
1339fn write_with_mode(path: &Path, text: &str, mode: Option<u32>) -> Result<()> {
1340 use std::io::Write;
1341 use std::os::unix::fs::{OpenOptionsExt, PermissionsExt};
1342 let mut options = fs::OpenOptions::new();
1343 options.write(true).create(true).truncate(true);
1344 if let Some(mode) = mode {
1345 options.mode(mode);
1346 }
1347 let mut file = options.open(path)?;
1348 if let Some(mode) = mode {
1350 file.set_permissions(fs::Permissions::from_mode(mode))?;
1351 }
1352 file.write_all(text.as_bytes())?;
1353 Ok(())
1354}
1355
1356#[cfg(not(unix))]
1357fn write_with_mode(path: &Path, text: &str, _mode: Option<u32>) -> Result<()> {
1358 fs::write(path, text)?;
1359 Ok(())
1360}
1361
1362pub fn encode_config(
1364 profile: &Profile,
1365 meta: Option<&ProfileIo>,
1366 vault: Option<&BTreeMap<String, String>>,
1367 flavor: Flavor,
1368) -> String {
1369 let mut out = Map::new();
1370 for (k, v) in &profile.residue.config {
1371 if k != "gateway" {
1372 out.insert(k.clone(), v.clone());
1373 }
1374 }
1375 if let Some(w) = &profile.worker {
1376 let mut wm = Map::new();
1377 wm.insert("harness".into(), Value::String(w.harness.as_str().into()));
1378 if let Some(m) = &w.model {
1379 wm.insert("model".into(), Value::String(m.clone()));
1380 }
1381 if let Some(p) = &w.preset {
1382 wm.insert("preset".into(), Value::String(p.clone()));
1383 }
1384 if w.cwd != "." {
1385 wm.insert("cwd".into(), Value::String(w.cwd.clone()));
1386 }
1387 if !w.env.is_empty() {
1388 let env: Map<String, Value> = w
1389 .env
1390 .iter()
1391 .map(|(k, v)| {
1392 let value = match v {
1393 crate::orchestration::EnvValue::Literal(s) => Value::String(s.clone()),
1394 crate::orchestration::EnvValue::Secret(r) => super::decode::render_ref(r),
1395 };
1396 (k.clone(), value)
1397 })
1398 .collect();
1399 wm.insert("env".into(), Value::Object(env));
1400 }
1401 if w.permission.timeout_seconds != 300
1402 || w.permission.default != crate::orchestration::PermissionDefault::Deny
1403 || w.permission.unattended != crate::orchestration::PermissionUnattended::Deny
1404 {
1405 wm.insert(
1406 "permission".into(),
1407 serde_json::to_value(&w.permission).unwrap(),
1408 );
1409 }
1410 out.insert("worker".into(), Value::Object(wm));
1411 }
1412 let mut gateway = profile
1413 .residue
1414 .config
1415 .get("gateway")
1416 .and_then(Value::as_object)
1417 .cloned()
1418 .unwrap_or_default();
1419 let routes: Vec<Value> = profile.routes.iter().map(encode_route).collect();
1420 if meta.is_some_and(|m| m.routes_at_top) {
1421 if !routes.is_empty() {
1422 out.insert("profile_routes".into(), Value::Array(routes));
1423 }
1424 } else if !routes.is_empty() {
1425 gateway.insert("profile_routes".into(), Value::Array(routes));
1426 }
1427 if !gateway.is_empty() {
1428 out.insert("gateway".into(), Value::Object(gateway));
1429 }
1430 let mut platforms = Map::new();
1431 for (p, ch) in &profile.channels {
1432 if ch.extra.get("from_env") == Some(&Value::Bool(true)) {
1434 continue;
1435 }
1436 platforms.insert(
1437 p.clone(),
1438 encode_channel(
1439 ch,
1440 if flavor == Flavor::Hermes {
1441 vault
1442 } else {
1443 None
1444 },
1445 ),
1446 );
1447 }
1448 if !platforms.is_empty() {
1449 out.insert("platforms".into(), Value::Object(platforms));
1450 }
1451 json_to_yaml(&Value::Object(out))
1452}
1453
1454pub fn encode_jobs_file(profile: &Profile, meta: Option<&ProfileIo>) -> String {
1456 let jobs: Vec<Value> = profile
1457 .jobs
1458 .values()
1459 .map(|j| ordered_object(encode_job(j)))
1460 .collect();
1461 let form = meta.and_then(|m| m.jobs_form.clone()).unwrap_or(JobsForm {
1462 object: true,
1463 extras: Map::new(),
1464 });
1465 let body = if form.object {
1466 let mut pairs = vec![("jobs".to_string(), Value::Array(jobs))];
1467 pairs.extend(form.extras.iter().map(|(k, v)| (k.clone(), v.clone())));
1468 ordered_object(pairs)
1469 } else {
1470 Value::Array(jobs)
1471 };
1472 format!("{}\n", pretty_ordered(&body, 0))
1473}
1474
1475pub(crate) fn ordered_object(pairs: Vec<(String, Value)>) -> Value {
1478 Value::Array(vec![
1481 Value::String("__ordered__".into()),
1482 Value::Array(
1483 pairs
1484 .into_iter()
1485 .map(|(k, v)| serde_json::json!({"__k": k, "__v": v}))
1486 .collect(),
1487 ),
1488 ])
1489}
1490
1491fn is_ordered(value: &Value) -> Option<&Vec<Value>> {
1492 let arr = value.as_array()?;
1493 if arr.len() == 2 && arr[0].as_str() == Some("__ordered__") {
1494 arr[1].as_array()
1495 } else {
1496 None
1497 }
1498}
1499
1500pub(crate) fn pretty_ordered(value: &Value, depth: usize) -> String {
1502 let pad = |d: usize| " ".repeat(d);
1503 if let Some(pairs) = is_ordered(value) {
1504 if pairs.is_empty() {
1505 return "{}".into();
1506 }
1507 let inner: Vec<String> = pairs
1508 .iter()
1509 .map(|p| {
1510 format!(
1511 "{}{}: {}",
1512 pad(depth + 1),
1513 serde_json::to_string(p["__k"].as_str().unwrap_or("")).unwrap(),
1514 pretty_ordered(&p["__v"], depth + 1)
1515 )
1516 })
1517 .collect();
1518 return format!("{{\n{}\n{}}}", inner.join(",\n"), pad(depth));
1519 }
1520 match value {
1521 Value::Array(items) if items.is_empty() => "[]".into(),
1522 Value::Array(items) => {
1523 let inner: Vec<String> = items
1524 .iter()
1525 .map(|v| format!("{}{}", pad(depth + 1), pretty_ordered(v, depth + 1)))
1526 .collect();
1527 format!("[\n{}\n{}]", inner.join(",\n"), pad(depth))
1528 }
1529 Value::Object(o) if o.is_empty() => "{}".into(),
1530 Value::Object(o) => {
1531 let inner: Vec<String> = o
1532 .iter()
1533 .map(|(k, v)| {
1534 format!(
1535 "{}{}: {}",
1536 pad(depth + 1),
1537 serde_json::to_string(k).unwrap(),
1538 pretty_ordered(v, depth + 1)
1539 )
1540 })
1541 .collect();
1542 format!("{{\n{}\n{}}}", inner.join(",\n"), pad(depth))
1543 }
1544 Value::Number(n) => {
1545 if let Some(f) = n.as_f64() {
1546 if n.is_f64() && f.fract() == 0.0 && f.abs() < 1e21 {
1547 return format!("{}", f as i64);
1548 }
1549 }
1550 n.to_string()
1551 }
1552 other => serde_json::to_string(other).unwrap(),
1553 }
1554}
1555
1556pub fn encode_subscriptions_file(
1558 profile: &Profile,
1559 vault: Option<&BTreeMap<String, String>>,
1560) -> String {
1561 let mut out = Map::new();
1562 for (n, s) in &profile.subscriptions {
1563 out.insert(n.clone(), encode_subscription(s, vault));
1564 }
1565 format!("{}\n", pretty_ordered(&Value::Object(out), 0))
1566}
1567
1568fn hermes_access(dir: &Path, env: &BTreeMap<String, String>, platforms: Option<&Value>) -> Access {
1574 let mut access = Access::default();
1575 let truthy = |v: &str| {
1576 matches!(
1577 v.trim().to_ascii_lowercase().as_str(),
1578 "1" | "true" | "yes" | "on"
1579 )
1580 };
1581 let split = |v: &str| {
1582 v.split(',')
1583 .map(|s| s.trim().to_string())
1584 .filter(|s| !s.is_empty())
1585 .collect::<Vec<_>>()
1586 };
1587 let global_open = env
1588 .get("GATEWAY_ALLOW_ALL_USERS")
1589 .is_some_and(|v| truthy(v));
1590 for (k, v) in env {
1591 if let Some(p) = k.strip_suffix("_ALLOWED_USERS") {
1592 access
1593 .allowlist
1594 .entry(p.to_ascii_lowercase())
1595 .or_default()
1596 .extend(split(v));
1597 } else if let Some(p) = k.strip_suffix("_ALLOW_ALL_USERS") {
1598 if truthy(v) {
1599 access
1600 .policy
1601 .insert(p.to_ascii_lowercase(), AccessPolicy::Open);
1602 }
1603 }
1604 }
1605 for sub in ["platforms/pairing", "pairing"] {
1606 let Ok(entries) = fs::read_dir(dir.join(sub)) else {
1607 continue;
1608 };
1609 for entry in entries.flatten() {
1610 let name = entry.file_name().to_string_lossy().into_owned();
1611 let Some(platform) = name.strip_suffix("-approved.json") else {
1612 continue;
1613 };
1614 let Ok(text) = fs::read_to_string(entry.path()) else {
1615 continue;
1616 };
1617 if let Ok(Value::Object(map)) = serde_json::from_str::<Value>(&text) {
1618 access
1619 .allowlist
1620 .entry(platform.to_string())
1621 .or_default()
1622 .extend(map.keys().cloned());
1623 }
1624 }
1625 }
1626 if let Some(Value::Object(map)) = platforms {
1627 for (platform, block) in map {
1628 if global_open {
1629 access.policy.insert(platform.clone(), AccessPolicy::Open);
1630 }
1631 for key in ["allow_admin_from", "group_allow_admin_from"] {
1632 let found = block
1633 .get("extra")
1634 .and_then(|e| e.get(key))
1635 .or_else(|| block.get(key));
1636 let ids = match found {
1637 Some(Value::String(s)) => split(s),
1638 Some(Value::Array(items)) => items
1639 .iter()
1640 .filter_map(|v| {
1641 v.as_str()
1642 .map(str::to_string)
1643 .or_else(|| v.as_i64().map(|n| n.to_string()))
1644 })
1645 .collect(),
1646 _ => Vec::new(),
1647 };
1648 access
1649 .admins
1650 .entry(platform.clone())
1651 .or_default()
1652 .extend(ids);
1653 }
1654 }
1655 }
1656 for list in access
1657 .allowlist
1658 .values_mut()
1659 .chain(access.admins.values_mut())
1660 {
1661 list.sort();
1662 list.dedup();
1663 }
1664 access.allowlist.retain(|_, v| !v.is_empty());
1665 access.admins.retain(|_, v| !v.is_empty());
1666 access
1667}
1668
1669const EXECUTIONS_DDL: &str = "CREATE TABLE IF NOT EXISTS executions (
1670 id TEXT PRIMARY KEY, job_id TEXT NOT NULL, source TEXT NOT NULL, process_id TEXT NOT NULL, pid INTEGER NOT NULL,
1671 process_started_at INTEGER, status TEXT NOT NULL CHECK(status IN ('claimed','running','completed','failed','unknown')),
1672 claimed_at TEXT NOT NULL, started_at TEXT, finished_at TEXT, error TEXT);
1673CREATE INDEX IF NOT EXISTS idx_executions_job_claimed ON executions(job_id, claimed_at DESC, id DESC);
1674CREATE INDEX IF NOT EXISTS idx_executions_status_claimed ON executions(status, claimed_at DESC, id DESC);";
1675const FOLDER_EXECUTIONS_DDL: &str = "CREATE TABLE IF NOT EXISTS executions (
1680 id TEXT PRIMARY KEY, job_id TEXT NOT NULL, source TEXT NOT NULL, process_id TEXT NOT NULL, pid INTEGER NOT NULL,
1681 process_started_at INTEGER, status TEXT NOT NULL CHECK(status IN ('claimed','running','completed','failed','unknown')),
1682 claimed_at TEXT NOT NULL, started_at TEXT, finished_at TEXT, error TEXT, residue_json TEXT,
1683 session_id TEXT, obligation_id TEXT);
1684CREATE INDEX IF NOT EXISTS idx_executions_job_claimed ON executions(job_id, claimed_at DESC, id DESC);
1685CREATE INDEX IF NOT EXISTS idx_executions_status_claimed ON executions(status, claimed_at DESC, id DESC);";
1686
1687const OBLIGATIONS_DDL: &str = "CREATE TABLE IF NOT EXISTS delivery_obligations (
1688 obligation_id TEXT PRIMARY KEY, session_key TEXT NOT NULL, platform TEXT NOT NULL, chat_id TEXT NOT NULL, thread_id TEXT,
1689 content TEXT NOT NULL, state TEXT NOT NULL, attempts INTEGER NOT NULL DEFAULT 0, created_at REAL NOT NULL, updated_at REAL NOT NULL,
1690 owner_pid INTEGER, owner_started_at INTEGER, last_error TEXT, adapter_profile TEXT,
1691 posted_message_id TEXT, source_json TEXT);";
1692
1693pub fn write_executions(path: &Path, fires: &[crate::orchestration::Fire]) -> Result<()> {
1695 write_executions_shaped(path, fires, false)
1696}
1697
1698pub fn write_executions_shaped(
1701 path: &Path,
1702 fires: &[crate::orchestration::Fire],
1703 ours: bool,
1704) -> Result<()> {
1705 let mut cols: Vec<&str> = EXECUTION_COLUMNS.to_vec();
1706 if ours {
1707 cols.extend(["residue_json", "session_id", "obligation_id"]);
1710 }
1711 let insert = format!(
1712 "insert into executions ({}) values ({})",
1713 cols.join(", "),
1714 cols.iter().map(|_| "?").collect::<Vec<_>>().join(",")
1715 );
1716 let rows: Vec<Vec<Param>> = fires
1717 .iter()
1718 .map(|f| {
1719 let mut row: Vec<Param> = encode_fire_row(f).iter().map(Param::from).collect();
1720 if ours {
1721 let rest: serde_json::Map<String, Value> = f
1723 .residue
1724 .0
1725 .iter()
1726 .filter(|(k, _)| {
1727 !["source", "process_id", "pid", "process_started_at"].contains(&k.as_str())
1728 })
1729 .map(|(k, v)| (k.clone(), v.clone()))
1730 .collect();
1731 row.push(if rest.is_empty() {
1732 Param::Null
1733 } else {
1734 Param::Text(serde_json::to_string(&rest).unwrap())
1735 });
1736 let opt = |v: &Option<String>| v.clone().map(Param::Text).unwrap_or(Param::Null);
1737 row.push(opt(&f.session_id));
1738 row.push(opt(&f.obligation_id));
1739 }
1740 row
1741 })
1742 .collect();
1743 write_table(
1744 path,
1745 if ours {
1746 FOLDER_EXECUTIONS_DDL
1747 } else {
1748 EXECUTIONS_DDL
1749 },
1750 &insert,
1751 &rows,
1752 )
1753}
1754
1755const ROUTING_DDL: &str = "CREATE TABLE IF NOT EXISTS gateway_routing (
1757 scope TEXT NOT NULL DEFAULT '', session_key TEXT NOT NULL, entry_json TEXT NOT NULL, updated_at REAL NOT NULL,
1758 PRIMARY KEY (scope, session_key));";
1759
1760fn write_state(dir: &Path, path: &Path, profile: &Profile) -> Result<()> {
1763 fs::create_dir_all(dir.join("sessions"))?;
1764 let scope = routing_scope(dir);
1765 let now = std::time::SystemTime::now()
1766 .duration_since(std::time::UNIX_EPOCH)
1767 .map(|d| d.as_secs_f64())
1768 .unwrap_or(0.0);
1769 let rank = |b: &Binding| {
1772 (
1773 b.ended_at.is_none(),
1774 b.last_activity_at
1775 .clone()
1776 .or_else(|| b.started_at.clone())
1777 .unwrap_or_default(),
1778 )
1779 };
1780 let mut chosen: BTreeMap<String, (&Binding, Value)> = BTreeMap::new();
1781 for (slot, b) in &profile.bindings {
1782 let (key, entry) = routing_entry(&profile.name, slot, b);
1783 if chosen
1784 .get(&key)
1785 .is_none_or(|(held, _)| rank(b) > rank(held))
1786 {
1787 chosen.insert(key, (b, entry));
1788 }
1789 }
1790 let mut mirror = Map::new();
1791 let rows: Vec<Vec<Param>> = chosen
1792 .into_iter()
1793 .map(|(key, (_, entry))| {
1794 mirror.insert(key.clone(), entry.clone());
1795 vec![
1796 Param::Text(scope.clone()),
1797 Param::Text(key),
1798 Param::Text(serde_json::to_string(&entry).unwrap()),
1799 Param::Real(now),
1800 ]
1801 })
1802 .collect();
1803 replace_rows(
1804 path,
1805 ROUTING_DDL,
1806 "gateway_routing",
1807 &[],
1808 ("delete from gateway_routing where scope = ?", &[Param::Text(scope)]),
1809 "insert into gateway_routing (scope, session_key, entry_json, updated_at) values (?, ?, ?, ?)",
1810 &rows,
1811 )?;
1812 let mirror_file = dir.join("sessions/sessions.json");
1813 let tmp = mirror_file.with_file_name(format!("sessions.json.tmp-{}", std::process::id()));
1814 fs::write(
1815 &tmp,
1816 serde_json::to_string_pretty(&Value::Object(mirror)).unwrap(),
1817 )?;
1818 fs::rename(&tmp, &mirror_file)?;
1819 let cols: Vec<&str> = OBLIGATION_COLUMNS
1821 .iter()
1822 .chain(FOLDER_OBLIGATION_EXTRA_COLUMNS.iter())
1823 .copied()
1824 .collect();
1825 let extras: Vec<(&str, &str)> = FOLDER_OBLIGATION_EXTRA_COLUMNS
1826 .iter()
1827 .map(|c| (*c, "TEXT"))
1828 .collect();
1829 let insert = format!(
1830 "insert into delivery_obligations ({}) values ({})",
1831 cols.join(", "),
1832 cols.iter().map(|_| "?").collect::<Vec<_>>().join(",")
1833 );
1834 let rows: Vec<Vec<Param>> = profile
1835 .obligations
1836 .iter()
1837 .map(|o| {
1838 encode_obligation_row(o)
1839 .iter()
1840 .chain(encode_obligation_folder_extras(o).iter())
1841 .map(Param::from)
1842 .collect()
1843 })
1844 .collect();
1845 replace_rows(
1846 path,
1847 OBLIGATIONS_DDL,
1848 "delivery_obligations",
1849 &extras,
1850 ("delete from delivery_obligations", &[]),
1851 &insert,
1852 &rows,
1853 )
1854}
1855
1856fn write_if_changed(
1857 meta: &mut ProfileIo,
1858 dir: &Path,
1859 rel: &str,
1860 record: &Value,
1861 render: impl FnOnce() -> Option<String>,
1862) -> Result<bool> {
1863 let snap = canonical_json(record);
1864 let path = dir.join(rel);
1865 let reuse = meta.flavor == Flavor::Orchestrator && meta.snapshot.get(rel) == Some(&snap);
1868 if reuse && path.exists() {
1869 return Ok(false);
1870 }
1871 if reuse {
1872 if let Some(raw) = meta.raw.get(rel).cloned() {
1873 write_atomic(&path, &raw)?;
1874 return Ok(true);
1875 }
1876 }
1877 let Some(text) = render() else {
1878 return Ok(false);
1879 };
1880 write_atomic(&path, &text)?;
1881 meta.raw.insert(rel.into(), text);
1882 meta.snapshot.insert(rel.into(), snap);
1883 meta.flavor = Flavor::Orchestrator; Ok(true)
1885}
1886
1887pub fn save_home(loaded: &mut LoadedHome, root: Option<&Path>) -> Result<()> {
1889 let root = root
1890 .map(Path::to_path_buf)
1891 .unwrap_or_else(|| loaded.orchestration.root.clone());
1892 fs::create_dir_all(&root)?;
1893 let names: Vec<String> = loaded.orchestration.profiles.keys().cloned().collect();
1894 for name in names {
1895 let dir = if name == "default" {
1896 root.clone()
1897 } else {
1898 root.join("profiles").join(&name)
1899 };
1900 fs::create_dir_all(dir.join("cron"))?;
1901 let profile = loaded.orchestration.profiles[&name].clone();
1902 let meta = loaded
1903 .io
1904 .entry(name.clone())
1905 .or_insert_with(|| ProfileIo::new(Flavor::Orchestrator));
1906 save_profile_dir(&profile, meta, &dir, &loaded.vault)?;
1907 }
1908 Ok(())
1909}
1910
1911fn save_profile_dir(
1912 profile: &Profile,
1913 meta: &mut ProfileIo,
1914 dir: &Path,
1915 vault: &BTreeMap<String, String>,
1916) -> Result<()> {
1917 let cfg_record = config_record(profile);
1918 write_if_changed(meta, dir, "config.yaml", &cfg_record, || {
1919 Some(encode_config(
1920 profile,
1921 None,
1922 Some(vault),
1923 Flavor::Orchestrator,
1924 ))
1925 })?;
1926 if let Some(persona) = &profile.persona {
1927 let text = persona.text.clone().unwrap_or_default();
1928 write_if_changed(
1929 meta,
1930 dir,
1931 "AGENTS.md",
1932 &serde_json::to_value(&profile.persona).unwrap(),
1933 || Some(text),
1934 )?;
1935 if !dir.join("CLAUDE.md").exists() {
1936 write_atomic(&dir.join("CLAUDE.md"), "@AGENTS.md\n")?;
1937 }
1938 }
1939 let jobs_record: Vec<Value> = profile
1940 .jobs
1941 .values()
1942 .map(|j| serde_json::to_value(j).unwrap())
1943 .collect();
1944 let had_jobs = meta.raw.contains_key("cron/jobs.json");
1945 let form = meta.jobs_form.clone();
1946 write_if_changed(
1947 meta,
1948 dir,
1949 "cron/jobs.json",
1950 &Value::Array(jobs_record),
1951 || {
1952 if profile.jobs.is_empty() && !had_jobs {
1953 None
1954 } else {
1955 let stub = ProfileIo {
1956 jobs_form: form.clone(),
1957 ..ProfileIo::new(Flavor::Orchestrator)
1958 };
1959 Some(encode_jobs_file(profile, Some(&stub)))
1960 }
1961 },
1962 )?;
1963 let subs_record: Vec<Value> = profile
1964 .subscriptions
1965 .values()
1966 .map(|s| serde_json::to_value(s).unwrap())
1967 .collect();
1968 let had_subs = meta.raw.contains_key("webhook_subscriptions.json");
1969 write_if_changed(
1970 meta,
1971 dir,
1972 "webhook_subscriptions.json",
1973 &Value::Array(subs_record),
1974 || {
1975 if profile.subscriptions.is_empty() && !had_subs {
1976 None
1977 } else {
1978 Some(encode_subscriptions_file(profile, None))
1979 }
1980 },
1981 )?;
1982 let fires_snap = canonical_json(&serde_json::to_value(&profile.fires).unwrap());
1984 let exec_path = dir.join("cron/executions.db");
1985 if (meta.snapshot.get("cron/executions.db") != Some(&fires_snap) || !exec_path.exists())
1986 && (!profile.fires.is_empty() || exec_path.exists())
1987 {
1988 let tmp = exec_path.with_file_name(format!("executions.db.tmp-{}", std::process::id()));
1989 let _ = fs::remove_file(&tmp);
1990 write_executions_shaped(&tmp, &profile.fires, true)?;
1991 fs::rename(&tmp, &exec_path)?;
1992 meta.snapshot
1993 .insert("cron/executions.db".into(), fires_snap);
1994 }
1995 let state_snap = canonical_json(&state_record(profile));
1996 let state_path = dir.join("state.db");
1997 if (meta.snapshot.get("state.db") != Some(&state_snap) || !state_path.exists())
1998 && (!profile.bindings.is_empty() || !profile.obligations.is_empty() || state_path.exists())
1999 {
2000 write_state(dir, &state_path, profile)?;
2002 meta.snapshot.insert("state.db".into(), state_snap);
2003 }
2004
2005 let mut refs: Vec<String> = Vec::new();
2007 for ch in profile.channels.values() {
2008 for r in ch.credentials.values() {
2009 if let crate::ontology::SecretRef::Dotenv(n) = r {
2010 refs.push(n.clone());
2011 }
2012 }
2013 }
2014 for s in profile.subscriptions.values() {
2015 if let Some(crate::ontology::SecretRef::Dotenv(n)) = &s.secret {
2016 refs.push(n.clone());
2017 }
2018 }
2019 if let Some(w) = &profile.worker {
2020 for v in w.env.values() {
2021 if let crate::orchestration::EnvValue::Secret(crate::ontology::SecretRef::Dotenv(n)) = v
2022 {
2023 refs.push(n.clone());
2024 }
2025 }
2026 }
2027 let mut entries: BTreeMap<String, String> = BTreeMap::new();
2028 for r in refs {
2029 if let Some(v) = vault.get(&r) {
2030 entries.insert(r, v.clone());
2031 }
2032 }
2033 if !entries.is_empty() {
2034 let existing = meta.raw.get(".env").cloned();
2035 let mut merged = existing.as_deref().map(parse_dotenv).unwrap_or_default();
2036 for (k, v) in entries {
2037 merged.insert(k, v);
2038 }
2039 let text = render_dotenv(&merged);
2040 if existing.as_deref() != Some(text.as_str()) {
2041 write_atomic(&dir.join(".env"), &text)?;
2042 meta.raw.insert(".env".into(), text);
2043 }
2044 }
2045 Ok(())
2046}
2047
2048pub fn copy_unmodeled(profile: &Profile, meta: &ProfileIo, dest: &Path) -> Result<()> {
2050 let Some(src) = &meta.source_dir else {
2051 return Ok(());
2052 };
2053 carry_unmodeled(&profile.residue.files, src, dest)?;
2054 Ok(())
2055}
2056
2057pub fn carry_unmodeled(files: &[String], src: &Path, dest: &Path) -> Result<Vec<String>> {
2061 let mut carried = Vec::new();
2062 for rel in files {
2063 let from = src.join(rel);
2064 if !from.is_file() {
2065 continue;
2066 }
2067 let to = dest.join(rel);
2068 if let Some(parent) = to.parent() {
2069 fs::create_dir_all(parent)?;
2070 }
2071 fs::copy(&from, &to)?;
2072 carried.push(rel.clone());
2073 }
2074 Ok(carried)
2075}