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_access, decode_binding_row, decode_channel, decode_expiry, decode_fire_row, decode_home,
25 decode_job, decode_obligation_row, decode_route, decode_subscription, decode_worker,
26 encode_access, encode_channel, encode_fire_row, encode_job, encode_obligation_row,
27 encode_route, encode_subscription, encode_surface_key, load_error, surface_key_string,
28 EXECUTION_COLUMNS, OBLIGATION_COLUMNS,
29};
30use super::dotenv::{parse_dotenv, render_dotenv};
31use super::sqlite::{read_rows, table_exists, write_table, Param};
32use crate::ontology::{
33 parse_hermes_session_key, Binding, EndReason, Handoff, HarnessId, HermesSessionRow, Recurrence,
34 Residue, SurfaceKey, Trigger, Worker,
35};
36use crate::world::{ExpiryPolicy, PersonaRef, Profile, ProfileResidue, World};
37use crate::Result;
38
39pub const OWNED_FILES: &[&str] = &[
41 "config.yaml",
42 "AGENTS.md",
43 "CLAUDE.md",
44 "access.yaml",
45 ".env",
46 "cron/jobs.json",
47 "cron/executions.db",
48 "webhook_subscriptions.json",
49 "state.db",
50];
51
52const CONFIG_O_KEYS: &[&str] = &["worker", "expiry", "home"];
53
54#[derive(Debug, Clone, Copy, PartialEq, Eq)]
56pub enum Flavor {
57 Orchestrator,
59 Hermes,
61}
62
63#[derive(Debug, Clone, PartialEq)]
65pub struct JobsForm {
66 pub object: bool,
68 pub extras: Map<String, Value>,
70}
71
72#[derive(Debug, Clone)]
75pub struct ProfileIo {
76 pub raw: BTreeMap<String, String>,
78 pub snapshot: BTreeMap<String, String>,
80 pub source_dir: Option<PathBuf>,
82 pub flavor: Flavor,
84 pub jobs_form: Option<JobsForm>,
86 pub routes_at_top: bool,
88 pub borrowed_from: Option<PathBuf>,
90 pub lenders: Vec<String>,
92}
93
94impl ProfileIo {
95 fn new(flavor: Flavor) -> Self {
96 Self {
97 raw: BTreeMap::new(),
98 snapshot: BTreeMap::new(),
99 source_dir: None,
100 flavor,
101 jobs_form: None,
102 routes_at_top: false,
103 borrowed_from: None,
104 lenders: Vec::new(),
105 }
106 }
107}
108
109#[derive(Debug, Clone)]
111pub struct LoadedHome {
112 pub world: World,
114 pub vault: BTreeMap<String, String>,
116 pub io: BTreeMap<String, ProfileIo>,
118}
119
120fn sha256_hex(text: &str) -> String {
121 let mut h = Sha256::new();
122 h.update(text.as_bytes());
123 h.finalize().iter().map(|b| format!("{b:02x}")).collect()
124}
125
126pub fn persona_ref(text: &str) -> PersonaRef {
128 PersonaRef {
129 path: "AGENTS.md".into(),
130 text: Some(text.to_string()),
131 sha256: sha256_hex(text),
132 }
133}
134
135fn read_text(dir: &Path, rel: &str) -> Result<Option<String>> {
136 let p = dir.join(rel);
137 if !p.is_file() {
138 return Ok(None);
139 }
140 Ok(Some(fs::read_to_string(&p)?))
141}
142
143fn yaml_to_json(file: &str, text: &str) -> Result<Value> {
144 let value: serde_yaml::Value =
145 serde_yaml::from_str(text).map_err(|e| load_error(file, "", format!("YAML: {e}")))?;
146 let json: Value =
147 serde_json::to_value(value).map_err(|e| load_error(file, "", format!("YAML: {e}")))?;
148 Ok(if json.is_null() {
149 Value::Object(Map::new())
150 } else {
151 json
152 })
153}
154
155fn json_to_yaml(value: &Value) -> String {
156 let y: serde_yaml::Value =
157 serde_json::from_value(value.clone()).unwrap_or(serde_yaml::Value::Null);
158 match value.as_object() {
159 Some(m) if m.is_empty() => String::new(),
160 _ => serde_yaml::to_string(&y).unwrap_or_default(),
161 }
162}
163
164pub fn empty_profile(name: &str, dir: &Path) -> Profile {
166 Profile {
167 name: name.to_string(),
168 dir: dir.to_path_buf(),
169 worker: None,
170 persona: None,
171 channels: BTreeMap::new(),
172 routes: Vec::new(),
173 expiry: ExpiryPolicy::default(),
174 home: None,
175 jobs: BTreeMap::new(),
176 subscriptions: BTreeMap::new(),
177 access: Default::default(),
178 bindings: BTreeMap::new(),
179 fires: Vec::new(),
180 obligations: Vec::new(),
181 residue: ProfileResidue::default(),
182 }
183}
184
185pub fn config_record(profile: &Profile) -> Value {
187 serde_json::json!({
188 "worker": profile.worker, "expiry": profile.expiry, "home": profile.home.as_ref().map(encode_surface_key),
189 "routes": profile.routes, "channels": profile.channels, "residue": profile.residue.config,
190 })
191}
192
193fn state_record(profile: &Profile) -> Value {
194 serde_json::json!({ "bindings": profile.bindings, "obligations": profile.obligations })
195}
196
197pub fn load_home(root: &Path, flavor: Flavor) -> Result<LoadedHome> {
199 if !root.is_dir() {
200 return Err(load_error(
201 &root.display().to_string(),
202 "",
203 "not a directory",
204 ));
205 }
206 let mut vault = BTreeMap::new();
207 let mut io = BTreeMap::new();
208 let mut profiles = BTreeMap::new();
209 let (default, default_io) = load_profile_dir("default", root, flavor, &mut vault)?;
210 profiles.insert("default".to_string(), default);
211 io.insert("default".to_string(), default_io);
212 let profiles_dir = root.join("profiles");
213 if profiles_dir.is_dir() {
214 let mut names: Vec<String> = fs::read_dir(&profiles_dir)?
215 .flatten()
216 .filter(|e| e.path().is_dir())
217 .filter_map(|e| e.file_name().into_string().ok())
218 .filter(|n| n != "node_modules" && !n.starts_with('.'))
219 .collect();
220 names.sort();
221 for name in names {
222 let dir = profiles_dir.join(&name);
223 if name == "default" {
224 return Err(load_error(
225 &dir.display().to_string(),
226 "",
227 "\"default\" is the root folder, not a named profile",
228 ));
229 }
230 let (profile, meta) = load_profile_dir(&name, &dir, flavor, &mut vault)?;
231 profiles.insert(name.clone(), profile);
232 io.insert(name, meta);
233 }
234 }
235 let mut loaded = LoadedHome {
236 world: World {
237 root: root.to_path_buf(),
238 profiles,
239 },
240 vault,
241 io,
242 };
243 if flavor == Flavor::Hermes {
244 partition_shared_store(&mut loaded, root)?;
245 }
246 Ok(loaded)
247}
248
249fn partition_shared_store(loaded: &mut LoadedHome, root: &Path) -> Result<()> {
254 let root_path = root.join("state.db");
255 if !root_path.is_file() {
256 return Ok(());
257 }
258 let names: Vec<String> = loaded
259 .world
260 .profiles
261 .keys()
262 .filter(|n| *n != "default")
263 .cloned()
264 .collect();
265 for name in names {
266 let has_own = loaded.world.profiles[&name].dir.join("state.db").is_file();
267 if has_own {
268 continue;
269 }
270 loaded.io.get_mut(&name).unwrap().borrowed_from = Some(root_path.clone());
271 loaded
272 .io
273 .get_mut("default")
274 .unwrap()
275 .lenders
276 .push(name.clone());
277 if table_exists(&root_path, "sessions") {
278 let rows = read_rows(
279 &root_path,
280 "select * from sessions where profile_name = ?1 order by started_at, id",
281 &[&name],
282 )?
283 .unwrap_or_default();
284 for row in rows {
285 if let Some(b) = binding_from_hermes_session(&root_path, &row, &name) {
286 loaded
287 .world
288 .profiles
289 .get_mut(&name)
290 .unwrap()
291 .bindings
292 .insert(surface_key_string(&b.key), b);
293 }
294 }
295 }
296 let root_profile = loaded.world.profiles.get_mut("default").unwrap();
298 let (mine, rest): (Vec<_>, Vec<_>) = root_profile.obligations.drain(..).partition(|o| {
299 o.session_key
300 .as_deref()
301 .and_then(parse_hermes_session_key)
302 .and_then(|(_, p)| p)
303 .as_deref()
304 == Some(name.as_str())
305 });
306 root_profile.obligations = rest;
307 loaded.world.profiles.get_mut(&name).unwrap().obligations = mine;
308 let snap = canonical_json(&state_record(&loaded.world.profiles[&name]));
309 loaded
310 .io
311 .get_mut(&name)
312 .unwrap()
313 .snapshot
314 .insert("state.db".into(), snap);
315 }
316 let snap = canonical_json(&state_record(&loaded.world.profiles["default"]));
317 loaded
318 .io
319 .get_mut("default")
320 .unwrap()
321 .snapshot
322 .insert("state.db".into(), snap);
323 Ok(())
324}
325
326const HERMES_SESSION_MAPPED: &[&str] = &[
327 "id",
328 "source",
329 "session_key",
330 "chat_id",
331 "chat_type",
332 "thread_id",
333 "user_id",
334 "profile_name",
335 "handoff_state",
336 "handoff_platform",
337 "handoff_error",
338 "started_at",
339 "ended_at",
340 "end_reason",
341];
342
343pub fn binding_from_hermes_session(
345 file: &Path,
346 row: &Map<String, Value>,
347 profile_name: &str,
348) -> Option<Binding> {
349 let text = |k: &str| {
350 row.get(k)
351 .and_then(|v| match v {
352 Value::String(s) => Some(s.clone()),
353 Value::Number(n) => Some(n.to_string()),
354 _ => None,
355 })
356 .filter(|s| !s.is_empty())
357 };
358 let num = |k: &str| row.get(k).and_then(Value::as_f64);
359 let id = text("id")?;
360 let source = text("source");
361 let parsed = text("session_key").and_then(|k| parse_hermes_session_key(&k));
364 let chat_type = text("chat_type").or_else(|| parsed.as_ref().and_then(|(k, _)| k.kind.clone()));
365 let key = if text("session_key").is_some()
366 && chat_type
367 .as_deref()
368 .is_some_and(|c| super::decode::CHAT_TYPES.contains(&c))
369 {
370 SurfaceKey {
371 key: None,
372 platform: source
373 .clone()
374 .or_else(|| parsed.as_ref().and_then(|(k, _)| k.platform.clone())),
375 kind: chat_type,
376 chat_id: text("chat_id")
377 .or_else(|| parsed.as_ref().and_then(|(k, _)| k.chat_id.clone())),
378 thread_id: text("thread_id")
379 .or_else(|| parsed.as_ref().and_then(|(k, _)| k.thread_id.clone())),
380 participant_id: parsed.as_ref().and_then(|(k, _)| k.participant_id.clone()),
381 }
382 } else if source.as_deref() == Some("cron") {
383 let job = crate::ontology::hermes_cron_job_id(&id).unwrap_or_else(|| id.clone());
384 SurfaceKey {
385 key: None,
386 platform: Some("cron".into()),
387 kind: Some("dm".into()),
388 chat_id: Some(job),
389 thread_id: None,
390 participant_id: None,
391 }
392 } else {
393 return None;
394 };
395 if let Some(p) = text("profile_name") {
396 if p != profile_name && !(profile_name == "default" && p == "main") {
397 return None; }
399 }
400 let iso = |v: Option<f64>| v.map(epoch_iso);
401 let end_word = text("end_reason");
402 let end_reason = end_word.as_deref().and_then(EndReason::parse);
403 let mut residue = Residue::default();
404 for (k, v) in row {
405 if !HERMES_SESSION_MAPPED.contains(&k.as_str()) && !v.is_null() {
406 residue.keep(k.clone(), v.clone());
407 }
408 }
409 if let (Some(word), None) = (&end_word, end_reason) {
410 residue.keep("end_reason", Value::String(word.clone()));
411 }
412 if let Some(u) = text("user_id") {
413 residue.keep("user_id", Value::String(u));
414 }
415 let recurrence = if key.platform.as_deref() == Some("cron") {
416 key.chat_id.clone().map(|job_id| Recurrence {
417 job_id,
418 kind: "cron".into(),
419 })
420 } else {
421 None
422 };
423 Some(Binding {
424 trigger: match (recurrence.is_some(), source.as_deref()) {
425 (true, _) => Trigger::Cron,
426 (_, Some(s)) => crate::ontology::hermes_trigger_for_source(s),
427 _ => Trigger::Unknown,
428 },
429 key,
430 profile: None,
431 worker: Worker {
432 harness: HarnessId::new(HarnessId::HERMES),
433 session_id: Some(id),
434 locator: Some(file.display().to_string()),
435 },
436 recurrence,
437 handoff: text("handoff_state").map(|state| Handoff {
438 to: text("handoff_platform"),
439 state,
440 error: text("handoff_error"),
441 }),
442 started_at: iso(num("started_at")),
443 last_activity_at: iso(num("ended_at").or_else(|| num("started_at"))),
444 ended_at: iso(num("ended_at")),
445 end_reason,
446 residue,
447 })
448}
449
450fn epoch_iso(seconds: f64) -> String {
452 let row = HermesSessionRow {
453 started_at: Some(seconds),
454 ..Default::default()
455 };
456 Binding::from_hermes_row(&row, None)
457 .started_at
458 .unwrap_or_default()
459}
460
461fn refuse_inline_credentials(
468 file: &str,
469 flavor: Flavor,
470 known: &BTreeSet<String>,
471 vault: &BTreeMap<String, String>,
472) -> Result<()> {
473 if flavor != Flavor::Orchestrator {
474 return Ok(());
475 }
476 let inline: Vec<&str> = vault
477 .keys()
478 .filter(|k| !known.contains(*k))
479 .map(String::as_str)
480 .collect();
481 if inline.is_empty() {
482 return Ok(());
483 }
484 Err(load_error(
485 file,
486 "",
487 format!(
488 "a credential value lives in .env, not here: put {} in .env and reference it as {{dotenv: NAME}}",
489 inline.join(", ")
490 ),
491 ))
492}
493
494fn load_profile_dir(
495 name: &str,
496 dir: &Path,
497 flavor: Flavor,
498 vault: &mut BTreeMap<String, String>,
499) -> Result<(Profile, ProfileIo)> {
500 let mut profile = empty_profile(name, dir);
501 let mut meta = ProfileIo::new(flavor);
502 meta.source_dir = Some(dir.to_path_buf());
503 let remember = |meta: &mut ProfileIo, rel: &str, raw: Option<String>, record: &Value| {
504 if let Some(raw) = raw {
505 meta.raw.insert(rel.to_string(), raw);
506 }
507 meta.snapshot
508 .insert(rel.to_string(), canonical_json(record));
509 };
510
511 if let Some(env) = read_text(dir, ".env")? {
513 for (k, v) in parse_dotenv(&env) {
514 vault.insert(k, v);
515 }
516 meta.raw.insert(".env".into(), env);
517 }
518
519 let cfg_file = dir.join("config.yaml").display().to_string();
521 let cfg_text = read_text(dir, "config.yaml")?;
522 let cfg = match &cfg_text {
523 Some(text) => yaml_to_json(&cfg_file, text)?,
524 None => Value::Object(Map::new()),
525 };
526 let cfg_map = cfg
527 .as_object()
528 .ok_or_else(|| load_error(&cfg_file, "", "expected a mapping"))?;
529 profile.worker = decode_worker(&cfg_file, cfg_map.get("worker"))?;
530 profile.expiry = decode_expiry(&cfg_file, cfg_map.get("expiry"))?;
531 profile.home = decode_home(&cfg_file, cfg_map.get("home"))?;
532 let gateway = cfg_map.get("gateway").and_then(Value::as_object);
533 let routes_raw: Vec<Value> = match cfg_map.get("profile_routes").and_then(Value::as_array) {
534 Some(a) => {
535 meta.routes_at_top = true;
536 a.clone()
537 }
538 None => gateway
539 .and_then(|g| g.get("profile_routes"))
540 .and_then(Value::as_array)
541 .cloned()
542 .unwrap_or_default(),
543 };
544 for (i, r) in routes_raw.iter().enumerate() {
545 profile.routes.push(decode_route(&cfg_file, i, r)?);
546 }
547 if let Some(platforms) = cfg_map.get("platforms") {
548 let map = platforms
549 .as_object()
550 .ok_or_else(|| load_error(&cfg_file, "platforms", "expected a map"))?;
551 let known: BTreeSet<String> = vault.keys().cloned().collect();
552 for (platform, raw) in map {
553 profile.channels.insert(
554 platform.clone(),
555 decode_channel(&cfg_file, platform, raw, vault)?,
556 );
557 }
558 refuse_inline_credentials(&cfg_file, flavor, &known, vault)?;
559 }
560 for (k, v) in cfg_map {
562 if CONFIG_O_KEYS.contains(&k.as_str()) || k == "platforms" || k == "profile_routes" {
563 continue;
564 }
565 if k == "gateway" {
566 let mut g = v.as_object().cloned().unwrap_or_default();
567 g.remove("profile_routes");
568 if !g.is_empty() {
569 profile
570 .residue
571 .config
572 .insert("gateway".into(), Value::Object(g));
573 }
574 continue;
575 }
576 profile.residue.config.insert(k.clone(), v.clone());
577 }
578 remember(&mut meta, "config.yaml", cfg_text, &config_record(&profile));
579
580 let persona_file = if flavor == Flavor::Hermes {
582 "SOUL.md"
583 } else {
584 "AGENTS.md"
585 };
586 let persona_text = read_text(dir, persona_file)?;
587 profile.persona = persona_text.as_ref().map(|t| PersonaRef {
588 path: "AGENTS.md".into(),
589 text: Some(t.clone()),
590 sha256: sha256_hex(t),
591 });
592 remember(
593 &mut meta,
594 persona_file,
595 persona_text,
596 &serde_json::to_value(&profile.persona).unwrap(),
597 );
598
599 let jobs_file = dir.join("cron/jobs.json").display().to_string();
601 let jobs_text = read_text(dir, "cron/jobs.json")?;
602 if let Some(text) = &jobs_text {
603 let parsed: Value = serde_json::from_str(text)
604 .map_err(|e| load_error(&jobs_file, "", format!("JSON: {e}")))?;
605 let arr: Vec<Value> = match &parsed {
606 Value::Array(a) => {
607 meta.jobs_form = Some(JobsForm {
608 object: false,
609 extras: Map::new(),
610 });
611 a.clone()
612 }
613 Value::Object(o) => match o.get("jobs") {
614 Some(Value::Array(a)) => {
615 let mut extras = o.clone();
616 extras.remove("jobs");
617 meta.jobs_form = Some(JobsForm {
618 object: true,
619 extras,
620 });
621 a.clone()
622 }
623 Some(Value::Object(m)) => {
624 let mut extras = o.clone();
625 extras.remove("jobs");
626 meta.jobs_form = Some(JobsForm {
627 object: true,
628 extras,
629 });
630 m.iter()
631 .map(|(id, j)| {
632 let mut j = j.as_object().cloned().unwrap_or_default();
633 j.insert("id".into(), Value::String(id.clone()));
634 Value::Object(j)
635 })
636 .collect()
637 }
638 _ => {
639 return Err(load_error(
640 &jobs_file,
641 "",
642 "expected an array of jobs or {\"jobs\": [...]}",
643 ))
644 }
645 },
646 _ => {
647 return Err(load_error(
648 &jobs_file,
649 "",
650 "expected an array of jobs or {\"jobs\": [...]}",
651 ))
652 }
653 };
654 for raw in &arr {
655 let job = decode_job(&jobs_file, raw)?;
656 if profile.jobs.contains_key(&job.id) {
657 return Err(load_error(&jobs_file, &job.id, "duplicate job id"));
658 }
659 profile.jobs.insert(job.id.clone(), job);
660 }
661 }
662 let jobs_record: Vec<Value> = profile
663 .jobs
664 .values()
665 .map(|j| serde_json::to_value(j).unwrap())
666 .collect();
667 remember(
668 &mut meta,
669 "cron/jobs.json",
670 jobs_text,
671 &Value::Array(jobs_record),
672 );
673
674 let exec_path = dir.join("cron/executions.db");
676 if table_exists(&exec_path, "executions") {
677 for row in read_rows(
678 &exec_path,
679 "select * from executions order by claimed_at, id",
680 &[],
681 )?
682 .unwrap_or_default()
683 {
684 profile
685 .fires
686 .push(decode_fire_row(&exec_path.display().to_string(), &row)?);
687 }
688 }
689 remember(
690 &mut meta,
691 "cron/executions.db",
692 None,
693 &serde_json::to_value(&profile.fires).unwrap(),
694 );
695
696 let subs_file = dir.join("webhook_subscriptions.json").display().to_string();
698 let subs_text = read_text(dir, "webhook_subscriptions.json")?;
699 if let Some(text) = &subs_text {
700 let parsed: Value = serde_json::from_str(text)
701 .map_err(|e| load_error(&subs_file, "", format!("JSON: {e}")))?;
702 let map = parsed
703 .as_object()
704 .ok_or_else(|| load_error(&subs_file, "", "expected a map"))?;
705 let known: BTreeSet<String> = vault.keys().cloned().collect();
706 for (n, raw) in map {
707 profile
708 .subscriptions
709 .insert(n.clone(), decode_subscription(&subs_file, n, raw, vault)?);
710 }
711 refuse_inline_credentials(&subs_file, flavor, &known, vault)?;
712 }
713 let subs_record: Vec<Value> = profile
714 .subscriptions
715 .values()
716 .map(|s| serde_json::to_value(s).unwrap())
717 .collect();
718 remember(
719 &mut meta,
720 "webhook_subscriptions.json",
721 subs_text,
722 &Value::Array(subs_record),
723 );
724
725 let access_file = dir.join("access.yaml").display().to_string();
727 let access_text = read_text(dir, "access.yaml")?;
728 let access_raw = match &access_text {
729 Some(t) => Some(yaml_to_json(&access_file, t)?),
730 None => None,
731 };
732 profile.access = decode_access(&access_file, access_raw.as_ref())?;
733 remember(
734 &mut meta,
735 "access.yaml",
736 access_text,
737 &serde_json::to_value(&profile.access).unwrap(),
738 );
739
740 let state_path = dir.join("state.db");
742 if table_exists(&state_path, "delivery_obligations") {
743 for row in read_rows(
744 &state_path,
745 "select * from delivery_obligations order by created_at, obligation_id",
746 &[],
747 )?
748 .unwrap_or_default()
749 {
750 profile.obligations.push(decode_obligation_row(
751 &state_path.display().to_string(),
752 &row,
753 )?);
754 }
755 }
756 if flavor == Flavor::Orchestrator {
757 if table_exists(&state_path, "bindings") {
758 for row in read_rows(
759 &state_path,
760 "select * from bindings order by started_at, slot",
761 &[],
762 )?
763 .unwrap_or_default()
764 {
765 let b = decode_binding_row(&state_path.display().to_string(), &row)?;
766 let slot = row
767 .get("slot")
768 .and_then(Value::as_str)
769 .map(str::to_string)
770 .unwrap_or_else(|| surface_key_string(&b.key));
771 profile.bindings.insert(slot, b);
772 }
773 }
774 } else if table_exists(&state_path, "sessions") {
775 for row in read_rows(
776 &state_path,
777 "select * from sessions order by started_at, id",
778 &[],
779 )?
780 .unwrap_or_default()
781 {
782 if let Some(b) = binding_from_hermes_session(&state_path, &row, name) {
783 profile.bindings.insert(surface_key_string(&b.key), b);
784 }
785 }
786 }
787 remember(&mut meta, "state.db", None, &state_record(&profile));
788
789 profile.residue.files = list_unmodeled(dir, flavor)?;
791 Ok((profile, meta))
792}
793
794fn list_unmodeled(dir: &Path, flavor: Flavor) -> Result<Vec<String>> {
795 let mut owned: Vec<&str> = OWNED_FILES.to_vec();
796 if flavor == Flavor::Hermes {
797 owned.push("SOUL.md");
798 owned.retain(|f| !["AGENTS.md", "CLAUDE.md", "access.yaml"].contains(f));
799 }
800 let runtime_artifacts = ["orchestrator.lock", "orchestrator.sock", "service"];
801 let mut out = Vec::new();
802 fn walk(
803 base: &Path,
804 d: &Path,
805 owned: &[&str],
806 runtime: &[&str],
807 out: &mut Vec<String>,
808 ) -> Result<()> {
809 let mut entries: Vec<_> = fs::read_dir(d)?.flatten().collect();
810 entries.sort_by_key(|e| e.file_name());
811 for entry in entries {
812 let p = entry.path();
813 let rel = p
814 .strip_prefix(base)
815 .unwrap_or(&p)
816 .to_string_lossy()
817 .replace('\\', "/");
818 let name = entry.file_name().to_string_lossy().into_owned();
819 if rel == "profiles"
820 || name == "node_modules"
821 || name == ".git"
822 || rel.starts_with("state.db")
823 || rel.starts_with("cron/executions.db")
824 {
825 continue;
826 }
827 if runtime.contains(&rel.as_str()) || regex_tmp(&name) {
828 continue;
829 }
830 let st = fs::symlink_metadata(&p)?;
831 if st.is_dir() {
832 walk(base, &p, owned, runtime, out)?;
833 continue;
834 }
835 if !st.is_file() {
836 continue;
837 }
838 if owned.contains(&rel.as_str()) {
839 continue;
840 }
841 out.push(rel);
842 }
843 Ok(())
844 }
845 walk(dir, dir, &owned, &runtime_artifacts, &mut out)?;
846 Ok(out)
847}
848
849fn regex_tmp(name: &str) -> bool {
850 name.rsplit_once(".tmp-")
852 .is_some_and(|(_, pid)| !pid.is_empty() && pid.chars().all(|c| c.is_ascii_digit()))
853}
854
855fn write_atomic(path: &Path, text: &str) -> Result<()> {
858 if let Some(parent) = path.parent() {
859 fs::create_dir_all(parent)?;
860 }
861 let tmp = path.with_file_name(format!(
862 "{}.tmp-{}",
863 path.file_name().unwrap().to_string_lossy(),
864 std::process::id()
865 ));
866 fs::write(&tmp, text)?;
867 fs::rename(&tmp, path)?;
868 Ok(())
869}
870
871pub fn encode_config(
873 profile: &Profile,
874 meta: Option<&ProfileIo>,
875 vault: Option<&BTreeMap<String, String>>,
876 flavor: Flavor,
877) -> String {
878 let mut out = Map::new();
879 for (k, v) in &profile.residue.config {
880 if k != "gateway" {
881 out.insert(k.clone(), v.clone());
882 }
883 }
884 if let Some(w) = &profile.worker {
885 let mut wm = Map::new();
886 wm.insert("harness".into(), Value::String(w.harness.as_str().into()));
887 if let Some(m) = &w.model {
888 wm.insert("model".into(), Value::String(m.clone()));
889 }
890 if let Some(p) = &w.preset {
891 wm.insert("preset".into(), Value::String(p.clone()));
892 }
893 if w.cwd != "." {
894 wm.insert("cwd".into(), Value::String(w.cwd.clone()));
895 }
896 if !w.env.is_empty() {
897 wm.insert("env".into(), serde_json::to_value(&w.env).unwrap());
898 }
899 if w.permission.timeout_seconds != 300
900 || w.permission.default != crate::world::PermissionDefault::Deny
901 {
902 wm.insert(
903 "permission".into(),
904 serde_json::to_value(&w.permission).unwrap(),
905 );
906 }
907 out.insert("worker".into(), Value::Object(wm));
908 }
909 if flavor == Flavor::Orchestrator || profile.expiry != ExpiryPolicy::default() {
910 out.insert(
911 "expiry".into(),
912 serde_json::to_value(&profile.expiry).unwrap(),
913 );
914 }
915 if let Some(h) = &profile.home {
916 out.insert("home".into(), encode_surface_key(h));
917 }
918 let mut gateway = profile
919 .residue
920 .config
921 .get("gateway")
922 .and_then(Value::as_object)
923 .cloned()
924 .unwrap_or_default();
925 let routes: Vec<Value> = profile.routes.iter().map(encode_route).collect();
926 if meta.is_some_and(|m| m.routes_at_top) {
927 if !routes.is_empty() {
928 out.insert("profile_routes".into(), Value::Array(routes));
929 }
930 } else if !routes.is_empty() {
931 gateway.insert("profile_routes".into(), Value::Array(routes));
932 }
933 if !gateway.is_empty() {
934 out.insert("gateway".into(), Value::Object(gateway));
935 }
936 let mut platforms = Map::new();
937 for (p, ch) in &profile.channels {
938 platforms.insert(
939 p.clone(),
940 encode_channel(
941 ch,
942 if flavor == Flavor::Hermes {
943 vault
944 } else {
945 None
946 },
947 ),
948 );
949 }
950 if !platforms.is_empty() {
951 out.insert("platforms".into(), Value::Object(platforms));
952 }
953 json_to_yaml(&Value::Object(out))
954}
955
956pub fn encode_jobs_file(profile: &Profile, meta: Option<&ProfileIo>) -> String {
958 let jobs: Vec<Value> = profile
959 .jobs
960 .values()
961 .map(|j| ordered_object(encode_job(j)))
962 .collect();
963 let form = meta.and_then(|m| m.jobs_form.clone()).unwrap_or(JobsForm {
964 object: true,
965 extras: Map::new(),
966 });
967 let body = if form.object {
968 let mut pairs = vec![("jobs".to_string(), Value::Array(jobs))];
969 pairs.extend(form.extras.iter().map(|(k, v)| (k.clone(), v.clone())));
970 ordered_object(pairs)
971 } else {
972 Value::Array(jobs)
973 };
974 format!("{}\n", pretty_ordered(&body, 0))
975}
976
977pub(crate) fn ordered_object(pairs: Vec<(String, Value)>) -> Value {
980 Value::Array(vec![
983 Value::String("__ordered__".into()),
984 Value::Array(
985 pairs
986 .into_iter()
987 .map(|(k, v)| serde_json::json!({"__k": k, "__v": v}))
988 .collect(),
989 ),
990 ])
991}
992
993fn is_ordered(value: &Value) -> Option<&Vec<Value>> {
994 let arr = value.as_array()?;
995 if arr.len() == 2 && arr[0].as_str() == Some("__ordered__") {
996 arr[1].as_array()
997 } else {
998 None
999 }
1000}
1001
1002pub(crate) fn pretty_ordered(value: &Value, depth: usize) -> String {
1004 let pad = |d: usize| " ".repeat(d);
1005 if let Some(pairs) = is_ordered(value) {
1006 if pairs.is_empty() {
1007 return "{}".into();
1008 }
1009 let inner: Vec<String> = pairs
1010 .iter()
1011 .map(|p| {
1012 format!(
1013 "{}{}: {}",
1014 pad(depth + 1),
1015 serde_json::to_string(p["__k"].as_str().unwrap_or("")).unwrap(),
1016 pretty_ordered(&p["__v"], depth + 1)
1017 )
1018 })
1019 .collect();
1020 return format!("{{\n{}\n{}}}", inner.join(",\n"), pad(depth));
1021 }
1022 match value {
1023 Value::Array(items) if items.is_empty() => "[]".into(),
1024 Value::Array(items) => {
1025 let inner: Vec<String> = items
1026 .iter()
1027 .map(|v| format!("{}{}", pad(depth + 1), pretty_ordered(v, depth + 1)))
1028 .collect();
1029 format!("[\n{}\n{}]", inner.join(",\n"), pad(depth))
1030 }
1031 Value::Object(o) if o.is_empty() => "{}".into(),
1032 Value::Object(o) => {
1033 let inner: Vec<String> = o
1034 .iter()
1035 .map(|(k, v)| {
1036 format!(
1037 "{}{}: {}",
1038 pad(depth + 1),
1039 serde_json::to_string(k).unwrap(),
1040 pretty_ordered(v, depth + 1)
1041 )
1042 })
1043 .collect();
1044 format!("{{\n{}\n{}}}", inner.join(",\n"), pad(depth))
1045 }
1046 Value::Number(n) => {
1047 if let Some(f) = n.as_f64() {
1048 if n.is_f64() && f.fract() == 0.0 && f.abs() < 1e21 {
1049 return format!("{}", f as i64);
1050 }
1051 }
1052 n.to_string()
1053 }
1054 other => serde_json::to_string(other).unwrap(),
1055 }
1056}
1057
1058pub fn encode_subscriptions_file(
1060 profile: &Profile,
1061 vault: Option<&BTreeMap<String, String>>,
1062) -> String {
1063 let mut out = Map::new();
1064 for (n, s) in &profile.subscriptions {
1065 out.insert(n.clone(), encode_subscription(s, vault));
1066 }
1067 format!("{}\n", pretty_ordered(&Value::Object(out), 0))
1068}
1069
1070pub fn encode_access_file(profile: &Profile) -> String {
1072 json_to_yaml(&encode_access(&profile.access))
1073}
1074
1075const EXECUTIONS_DDL: &str = "CREATE TABLE IF NOT EXISTS executions (
1076 id TEXT PRIMARY KEY, job_id TEXT NOT NULL, source TEXT NOT NULL, process_id TEXT NOT NULL, pid INTEGER NOT NULL,
1077 process_started_at INTEGER, status TEXT NOT NULL CHECK(status IN ('claimed','running','completed','failed','unknown')),
1078 claimed_at TEXT NOT NULL, started_at TEXT, finished_at TEXT, error TEXT);
1079CREATE INDEX IF NOT EXISTS idx_executions_job_claimed ON executions(job_id, claimed_at DESC, id DESC);
1080CREATE INDEX IF NOT EXISTS idx_executions_status_claimed ON executions(status, claimed_at DESC, id DESC);";
1081
1082const OBLIGATIONS_DDL: &str = "CREATE TABLE IF NOT EXISTS delivery_obligations (
1083 obligation_id TEXT PRIMARY KEY, session_key TEXT NOT NULL, platform TEXT NOT NULL, chat_id TEXT NOT NULL, thread_id TEXT,
1084 content TEXT NOT NULL, state TEXT NOT NULL, attempts INTEGER NOT NULL DEFAULT 0, created_at REAL NOT NULL, updated_at REAL NOT NULL,
1085 owner_pid INTEGER, owner_started_at INTEGER, last_error TEXT, adapter_profile TEXT);";
1086
1087const BINDINGS_DDL: &str = "CREATE TABLE IF NOT EXISTS bindings (
1088 slot TEXT PRIMARY KEY,
1089 platform TEXT NOT NULL, chat_type TEXT NOT NULL, chat_id TEXT, thread_id TEXT, participant_id TEXT,
1090 worker_harness TEXT NOT NULL, worker_session_id TEXT, worker_locator TEXT,
1091 started_at TEXT NOT NULL, last_activity_at TEXT NOT NULL, ended_at TEXT, end_reason TEXT,
1092 handoff_to TEXT, handoff_state TEXT, handoff_error TEXT, recurrence_job_id TEXT, residue_json TEXT);";
1093
1094pub fn write_executions(path: &Path, fires: &[crate::world::Fire]) -> Result<()> {
1096 let insert = format!(
1097 "insert into executions ({}) values ({})",
1098 EXECUTION_COLUMNS.join(", "),
1099 EXECUTION_COLUMNS
1100 .iter()
1101 .map(|_| "?")
1102 .collect::<Vec<_>>()
1103 .join(",")
1104 );
1105 let rows: Vec<Vec<Param>> = fires
1106 .iter()
1107 .map(|f| encode_fire_row(f).iter().map(Param::from).collect())
1108 .collect();
1109 write_table(path, EXECUTIONS_DDL, &insert, &rows)
1110}
1111
1112fn write_state(path: &Path, profile: &Profile) -> Result<()> {
1113 let cols = [
1114 "slot",
1115 "platform",
1116 "chat_type",
1117 "chat_id",
1118 "thread_id",
1119 "participant_id",
1120 "worker_harness",
1121 "worker_session_id",
1122 "worker_locator",
1123 "started_at",
1124 "last_activity_at",
1125 "ended_at",
1126 "end_reason",
1127 "handoff_to",
1128 "handoff_state",
1129 "handoff_error",
1130 "recurrence_job_id",
1131 "residue_json",
1132 ];
1133 let insert = format!(
1134 "insert into bindings ({}) values ({})",
1135 cols.join(", "),
1136 cols.iter().map(|_| "?").collect::<Vec<_>>().join(",")
1137 );
1138 let s = |v: &Option<String>| v.clone().map(Param::Text).unwrap_or(Param::Null);
1139 let rows: Vec<Vec<Param>> = profile
1140 .bindings
1141 .iter()
1142 .map(|(slot, b)| {
1143 vec![
1144 Param::Text(slot.clone()),
1145 Param::Text(b.key.platform.clone().unwrap_or_default()),
1146 Param::Text(b.key.kind.clone().unwrap_or_default()),
1147 Param::Text(b.key.chat_id.clone().unwrap_or_default()),
1148 Param::Text(b.key.thread_id.clone().unwrap_or_default()),
1149 Param::Text(b.key.participant_id.clone().unwrap_or_default()),
1150 Param::Text(b.worker.harness.as_str().into()),
1151 s(&b.worker.session_id.clone().filter(|v| !v.is_empty())),
1152 s(&b.worker.locator),
1153 Param::Text(b.started_at.clone().unwrap_or_default()),
1154 Param::Text(b.last_activity_at.clone().unwrap_or_default()),
1155 s(&b.ended_at),
1156 b.end_reason
1157 .map(|r| Param::Text(r.as_str().into()))
1158 .unwrap_or(Param::Null),
1159 s(&b.handoff.as_ref().and_then(|h| h.to.clone())),
1160 b.handoff
1161 .as_ref()
1162 .map(|h| Param::Text(h.state.clone()))
1163 .unwrap_or(Param::Null),
1164 s(&b.handoff.as_ref().and_then(|h| h.error.clone())),
1165 s(&b.recurrence.as_ref().map(|r| r.job_id.clone())),
1166 if b.residue.is_empty() {
1167 Param::Null
1168 } else {
1169 Param::Text(serde_json::to_string(&b.residue).unwrap())
1170 },
1171 ]
1172 })
1173 .collect();
1174 write_table(
1175 path,
1176 &format!("{BINDINGS_DDL}\n{OBLIGATIONS_DDL}"),
1177 &insert,
1178 &rows,
1179 )?;
1180 let insert = format!(
1181 "insert into delivery_obligations ({}) values ({})",
1182 OBLIGATION_COLUMNS.join(", "),
1183 OBLIGATION_COLUMNS
1184 .iter()
1185 .map(|_| "?")
1186 .collect::<Vec<_>>()
1187 .join(",")
1188 );
1189 let rows: Vec<Vec<Param>> = profile
1190 .obligations
1191 .iter()
1192 .map(|o| encode_obligation_row(o).iter().map(Param::from).collect())
1193 .collect();
1194 write_table(path, "", &insert, &rows)
1195}
1196
1197fn write_if_changed(
1198 meta: &mut ProfileIo,
1199 dir: &Path,
1200 rel: &str,
1201 record: &Value,
1202 render: impl FnOnce() -> Option<String>,
1203) -> Result<bool> {
1204 let snap = canonical_json(record);
1205 let path = dir.join(rel);
1206 let reuse = meta.flavor == Flavor::Orchestrator && meta.snapshot.get(rel) == Some(&snap);
1209 if reuse && path.exists() {
1210 return Ok(false);
1211 }
1212 if reuse {
1213 if let Some(raw) = meta.raw.get(rel).cloned() {
1214 write_atomic(&path, &raw)?;
1215 return Ok(true);
1216 }
1217 }
1218 let Some(text) = render() else {
1219 return Ok(false);
1220 };
1221 write_atomic(&path, &text)?;
1222 meta.raw.insert(rel.into(), text);
1223 meta.snapshot.insert(rel.into(), snap);
1224 meta.flavor = Flavor::Orchestrator; Ok(true)
1226}
1227
1228pub fn save_home(loaded: &mut LoadedHome, root: Option<&Path>) -> Result<()> {
1230 let root = root
1231 .map(Path::to_path_buf)
1232 .unwrap_or_else(|| loaded.world.root.clone());
1233 fs::create_dir_all(&root)?;
1234 let names: Vec<String> = loaded.world.profiles.keys().cloned().collect();
1235 for name in names {
1236 let dir = if name == "default" {
1237 root.clone()
1238 } else {
1239 root.join("profiles").join(&name)
1240 };
1241 fs::create_dir_all(dir.join("cron"))?;
1242 let profile = loaded.world.profiles[&name].clone();
1243 let meta = loaded
1244 .io
1245 .entry(name.clone())
1246 .or_insert_with(|| ProfileIo::new(Flavor::Orchestrator));
1247 save_profile_dir(&profile, meta, &dir, &loaded.vault)?;
1248 }
1249 Ok(())
1250}
1251
1252fn save_profile_dir(
1253 profile: &Profile,
1254 meta: &mut ProfileIo,
1255 dir: &Path,
1256 vault: &BTreeMap<String, String>,
1257) -> Result<()> {
1258 let cfg_record = config_record(profile);
1259 write_if_changed(meta, dir, "config.yaml", &cfg_record, || {
1260 Some(encode_config(
1261 profile,
1262 None,
1263 Some(vault),
1264 Flavor::Orchestrator,
1265 ))
1266 })?;
1267 if let Some(persona) = &profile.persona {
1268 let text = persona.text.clone().unwrap_or_default();
1269 write_if_changed(
1270 meta,
1271 dir,
1272 "AGENTS.md",
1273 &serde_json::to_value(&profile.persona).unwrap(),
1274 || Some(text),
1275 )?;
1276 if !dir.join("CLAUDE.md").exists() {
1277 write_atomic(&dir.join("CLAUDE.md"), "@AGENTS.md\n")?;
1278 }
1279 }
1280 let jobs_record: Vec<Value> = profile
1281 .jobs
1282 .values()
1283 .map(|j| serde_json::to_value(j).unwrap())
1284 .collect();
1285 let had_jobs = meta.raw.contains_key("cron/jobs.json");
1286 let form = meta.jobs_form.clone();
1287 write_if_changed(
1288 meta,
1289 dir,
1290 "cron/jobs.json",
1291 &Value::Array(jobs_record),
1292 || {
1293 if profile.jobs.is_empty() && !had_jobs {
1294 None
1295 } else {
1296 let stub = ProfileIo {
1297 jobs_form: form.clone(),
1298 ..ProfileIo::new(Flavor::Orchestrator)
1299 };
1300 Some(encode_jobs_file(profile, Some(&stub)))
1301 }
1302 },
1303 )?;
1304 let subs_record: Vec<Value> = profile
1305 .subscriptions
1306 .values()
1307 .map(|s| serde_json::to_value(s).unwrap())
1308 .collect();
1309 let had_subs = meta.raw.contains_key("webhook_subscriptions.json");
1310 write_if_changed(
1311 meta,
1312 dir,
1313 "webhook_subscriptions.json",
1314 &Value::Array(subs_record),
1315 || {
1316 if profile.subscriptions.is_empty() && !had_subs {
1317 None
1318 } else {
1319 Some(encode_subscriptions_file(profile, None))
1320 }
1321 },
1322 )?;
1323 let a = &profile.access;
1324 let access_empty = a.allowlist.is_empty()
1325 && a.admins.is_empty()
1326 && a.pending_pairings.is_empty()
1327 && a.policy.is_empty()
1328 && a.pairing_ttl_minutes.is_none();
1329 let had_access = meta.raw.contains_key("access.yaml");
1330 write_if_changed(
1331 meta,
1332 dir,
1333 "access.yaml",
1334 &serde_json::to_value(a).unwrap(),
1335 || {
1336 if access_empty && !had_access {
1337 None
1338 } else {
1339 Some(encode_access_file(profile))
1340 }
1341 },
1342 )?;
1343
1344 let fires_snap = canonical_json(&serde_json::to_value(&profile.fires).unwrap());
1346 let exec_path = dir.join("cron/executions.db");
1347 if (meta.snapshot.get("cron/executions.db") != Some(&fires_snap) || !exec_path.exists())
1348 && (!profile.fires.is_empty() || exec_path.exists())
1349 {
1350 let tmp = exec_path.with_file_name(format!("executions.db.tmp-{}", std::process::id()));
1351 let _ = fs::remove_file(&tmp);
1352 write_executions(&tmp, &profile.fires)?;
1353 fs::rename(&tmp, &exec_path)?;
1354 meta.snapshot
1355 .insert("cron/executions.db".into(), fires_snap);
1356 }
1357 let state_snap = canonical_json(&state_record(profile));
1358 let state_path = dir.join("state.db");
1359 if (meta.snapshot.get("state.db") != Some(&state_snap) || !state_path.exists())
1360 && (!profile.bindings.is_empty() || !profile.obligations.is_empty() || state_path.exists())
1361 {
1362 let tmp = state_path.with_file_name(format!("state.db.tmp-{}", std::process::id()));
1363 let _ = fs::remove_file(&tmp);
1364 write_state(&tmp, profile)?;
1365 fs::rename(&tmp, &state_path)?;
1366 meta.snapshot.insert("state.db".into(), state_snap);
1367 }
1368
1369 let mut refs: Vec<String> = Vec::new();
1371 for ch in profile.channels.values() {
1372 for r in ch.credentials.values() {
1373 if let crate::ontology::SecretRef::Dotenv(n) = r {
1374 refs.push(n.clone());
1375 }
1376 }
1377 }
1378 for s in profile.subscriptions.values() {
1379 if let Some(crate::ontology::SecretRef::Dotenv(n)) = &s.secret {
1380 refs.push(n.clone());
1381 }
1382 }
1383 if let Some(w) = &profile.worker {
1384 for v in w.env.values() {
1385 if let crate::world::EnvValue::Secret(crate::ontology::SecretRef::Dotenv(n)) = v {
1386 refs.push(n.clone());
1387 }
1388 }
1389 }
1390 let mut entries: BTreeMap<String, String> = BTreeMap::new();
1391 for r in refs {
1392 if let Some(v) = vault.get(&r) {
1393 entries.insert(r, v.clone());
1394 }
1395 }
1396 if !entries.is_empty() {
1397 let existing = meta.raw.get(".env").cloned();
1398 let mut merged = existing.as_deref().map(parse_dotenv).unwrap_or_default();
1399 for (k, v) in entries {
1400 merged.insert(k, v);
1401 }
1402 let text = render_dotenv(&merged);
1403 if existing.as_deref() != Some(text.as_str()) {
1404 write_atomic(&dir.join(".env"), &text)?;
1405 meta.raw.insert(".env".into(), text);
1406 }
1407 }
1408 Ok(())
1409}
1410
1411pub fn copy_unmodeled(profile: &Profile, meta: &ProfileIo, dest: &Path) -> Result<()> {
1413 let Some(src) = &meta.source_dir else {
1414 return Ok(());
1415 };
1416 carry_unmodeled(&profile.residue.files, src, dest)?;
1417 Ok(())
1418}
1419
1420pub fn carry_unmodeled(files: &[String], src: &Path, dest: &Path) -> Result<Vec<String>> {
1424 let mut carried = Vec::new();
1425 for rel in files {
1426 let from = src.join(rel);
1427 if !from.is_file() {
1428 continue;
1429 }
1430 let to = dest.join(rel);
1431 if let Some(parent) = to.parent() {
1432 fs::create_dir_all(parent)?;
1433 }
1434 fs::copy(&from, &to)?;
1435 carried.push(rel.clone());
1436 }
1437 Ok(carried)
1438}