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 load_profile_dir(
462 name: &str,
463 dir: &Path,
464 flavor: Flavor,
465 vault: &mut BTreeMap<String, String>,
466) -> Result<(Profile, ProfileIo)> {
467 let mut profile = empty_profile(name, dir);
468 let mut config_normalized = false;
469 let mut subs_normalized = false;
470 let mut meta = ProfileIo::new(flavor);
471 meta.source_dir = Some(dir.to_path_buf());
472 let remember = |meta: &mut ProfileIo, rel: &str, raw: Option<String>, record: &Value| {
473 if let Some(raw) = raw {
474 meta.raw.insert(rel.to_string(), raw);
475 }
476 meta.snapshot
477 .insert(rel.to_string(), canonical_json(record));
478 };
479
480 if let Some(env) = read_text(dir, ".env")? {
482 for (k, v) in parse_dotenv(&env) {
483 vault.insert(k, v);
484 }
485 meta.raw.insert(".env".into(), env);
486 }
487
488 let cfg_file = dir.join("config.yaml").display().to_string();
490 let cfg_text = read_text(dir, "config.yaml")?;
491 let cfg = match &cfg_text {
492 Some(text) => yaml_to_json(&cfg_file, text)?,
493 None => Value::Object(Map::new()),
494 };
495 let cfg_map = cfg
496 .as_object()
497 .ok_or_else(|| load_error(&cfg_file, "", "expected a mapping"))?;
498 profile.worker = decode_worker(&cfg_file, cfg_map.get("worker"))?;
499 profile.expiry = decode_expiry(&cfg_file, cfg_map.get("expiry"))?;
500 profile.home = decode_home(&cfg_file, cfg_map.get("home"))?;
501 let gateway = cfg_map.get("gateway").and_then(Value::as_object);
502 let routes_raw: Vec<Value> = match cfg_map.get("profile_routes").and_then(Value::as_array) {
503 Some(a) => {
504 meta.routes_at_top = true;
505 a.clone()
506 }
507 None => gateway
508 .and_then(|g| g.get("profile_routes"))
509 .and_then(Value::as_array)
510 .cloned()
511 .unwrap_or_default(),
512 };
513 for (i, r) in routes_raw.iter().enumerate() {
514 profile.routes.push(decode_route(&cfg_file, i, r)?);
515 }
516 if let Some(platforms) = cfg_map.get("platforms") {
517 let map = platforms
518 .as_object()
519 .ok_or_else(|| load_error(&cfg_file, "platforms", "expected a map"))?;
520 let known: BTreeSet<String> = vault.keys().cloned().collect();
521 for (platform, raw) in map {
522 profile.channels.insert(
523 platform.clone(),
524 decode_channel(&cfg_file, platform, raw, vault)?,
525 );
526 }
527 config_normalized =
528 flavor == Flavor::Orchestrator && vault.keys().any(|k| !known.contains(k));
529 }
530 for (k, v) in cfg_map {
532 if CONFIG_O_KEYS.contains(&k.as_str()) || k == "platforms" || k == "profile_routes" {
533 continue;
534 }
535 if k == "gateway" {
536 let mut g = v.as_object().cloned().unwrap_or_default();
537 g.remove("profile_routes");
538 if !g.is_empty() {
539 profile
540 .residue
541 .config
542 .insert("gateway".into(), Value::Object(g));
543 }
544 continue;
545 }
546 profile.residue.config.insert(k.clone(), v.clone());
547 }
548 remember(&mut meta, "config.yaml", cfg_text, &config_record(&profile));
549 if config_normalized {
550 meta.raw.remove("config.yaml");
555 meta.snapshot.remove("config.yaml");
556 }
557
558 let persona_file = if flavor == Flavor::Hermes {
560 "SOUL.md"
561 } else {
562 "AGENTS.md"
563 };
564 let persona_text = read_text(dir, persona_file)?;
565 profile.persona = persona_text.as_ref().map(|t| PersonaRef {
566 path: "AGENTS.md".into(),
567 text: Some(t.clone()),
568 sha256: sha256_hex(t),
569 });
570 remember(
571 &mut meta,
572 persona_file,
573 persona_text,
574 &serde_json::to_value(&profile.persona).unwrap(),
575 );
576
577 let jobs_file = dir.join("cron/jobs.json").display().to_string();
579 let jobs_text = read_text(dir, "cron/jobs.json")?;
580 if let Some(text) = &jobs_text {
581 let parsed: Value = serde_json::from_str(text)
582 .map_err(|e| load_error(&jobs_file, "", format!("JSON: {e}")))?;
583 let arr: Vec<Value> = match &parsed {
584 Value::Array(a) => {
585 meta.jobs_form = Some(JobsForm {
586 object: false,
587 extras: Map::new(),
588 });
589 a.clone()
590 }
591 Value::Object(o) => match o.get("jobs") {
592 Some(Value::Array(a)) => {
593 let mut extras = o.clone();
594 extras.remove("jobs");
595 meta.jobs_form = Some(JobsForm {
596 object: true,
597 extras,
598 });
599 a.clone()
600 }
601 Some(Value::Object(m)) => {
602 let mut extras = o.clone();
603 extras.remove("jobs");
604 meta.jobs_form = Some(JobsForm {
605 object: true,
606 extras,
607 });
608 m.iter()
609 .map(|(id, j)| {
610 let mut j = j.as_object().cloned().unwrap_or_default();
611 j.insert("id".into(), Value::String(id.clone()));
612 Value::Object(j)
613 })
614 .collect()
615 }
616 _ => {
617 return Err(load_error(
618 &jobs_file,
619 "",
620 "expected an array of jobs or {\"jobs\": [...]}",
621 ))
622 }
623 },
624 _ => {
625 return Err(load_error(
626 &jobs_file,
627 "",
628 "expected an array of jobs or {\"jobs\": [...]}",
629 ))
630 }
631 };
632 for raw in &arr {
633 let job = decode_job(&jobs_file, raw)?;
634 if profile.jobs.contains_key(&job.id) {
635 return Err(load_error(&jobs_file, &job.id, "duplicate job id"));
636 }
637 profile.jobs.insert(job.id.clone(), job);
638 }
639 }
640 let jobs_record: Vec<Value> = profile
641 .jobs
642 .values()
643 .map(|j| serde_json::to_value(j).unwrap())
644 .collect();
645 remember(
646 &mut meta,
647 "cron/jobs.json",
648 jobs_text,
649 &Value::Array(jobs_record),
650 );
651
652 let exec_path = dir.join("cron/executions.db");
654 if table_exists(&exec_path, "executions") {
655 for row in read_rows(
656 &exec_path,
657 "select * from executions order by claimed_at, id",
658 &[],
659 )?
660 .unwrap_or_default()
661 {
662 profile
663 .fires
664 .push(decode_fire_row(&exec_path.display().to_string(), &row)?);
665 }
666 }
667 remember(
668 &mut meta,
669 "cron/executions.db",
670 None,
671 &serde_json::to_value(&profile.fires).unwrap(),
672 );
673
674 let subs_file = dir.join("webhook_subscriptions.json").display().to_string();
676 let subs_text = read_text(dir, "webhook_subscriptions.json")?;
677 if let Some(text) = &subs_text {
678 let parsed: Value = serde_json::from_str(text)
679 .map_err(|e| load_error(&subs_file, "", format!("JSON: {e}")))?;
680 let map = parsed
681 .as_object()
682 .ok_or_else(|| load_error(&subs_file, "", "expected a map"))?;
683 let known: BTreeSet<String> = vault.keys().cloned().collect();
684 for (n, raw) in map {
685 profile
686 .subscriptions
687 .insert(n.clone(), decode_subscription(&subs_file, n, raw, vault)?);
688 }
689 subs_normalized =
690 flavor == Flavor::Orchestrator && vault.keys().any(|k| !known.contains(k));
691 }
692 let subs_record: Vec<Value> = profile
693 .subscriptions
694 .values()
695 .map(|s| serde_json::to_value(s).unwrap())
696 .collect();
697 remember(
698 &mut meta,
699 "webhook_subscriptions.json",
700 subs_text,
701 &Value::Array(subs_record),
702 );
703 if subs_normalized {
704 meta.raw.remove("webhook_subscriptions.json");
705 meta.snapshot.remove("webhook_subscriptions.json");
706 }
707
708 let access_file = dir.join("access.yaml").display().to_string();
710 let access_text = read_text(dir, "access.yaml")?;
711 let access_raw = match &access_text {
712 Some(t) => Some(yaml_to_json(&access_file, t)?),
713 None => None,
714 };
715 profile.access = decode_access(&access_file, access_raw.as_ref())?;
716 remember(
717 &mut meta,
718 "access.yaml",
719 access_text,
720 &serde_json::to_value(&profile.access).unwrap(),
721 );
722
723 let state_path = dir.join("state.db");
725 if table_exists(&state_path, "delivery_obligations") {
726 for row in read_rows(
727 &state_path,
728 "select * from delivery_obligations order by created_at, obligation_id",
729 &[],
730 )?
731 .unwrap_or_default()
732 {
733 profile.obligations.push(decode_obligation_row(
734 &state_path.display().to_string(),
735 &row,
736 )?);
737 }
738 }
739 if flavor == Flavor::Orchestrator {
740 if table_exists(&state_path, "bindings") {
741 for row in read_rows(
742 &state_path,
743 "select * from bindings order by started_at, slot",
744 &[],
745 )?
746 .unwrap_or_default()
747 {
748 let b = decode_binding_row(&state_path.display().to_string(), &row)?;
749 let slot = row
750 .get("slot")
751 .and_then(Value::as_str)
752 .map(str::to_string)
753 .unwrap_or_else(|| surface_key_string(&b.key));
754 profile.bindings.insert(slot, b);
755 }
756 }
757 } else if table_exists(&state_path, "sessions") {
758 for row in read_rows(
759 &state_path,
760 "select * from sessions order by started_at, id",
761 &[],
762 )?
763 .unwrap_or_default()
764 {
765 if let Some(b) = binding_from_hermes_session(&state_path, &row, name) {
766 profile.bindings.insert(surface_key_string(&b.key), b);
767 }
768 }
769 }
770 remember(&mut meta, "state.db", None, &state_record(&profile));
771
772 profile.residue.files = list_unmodeled(dir, flavor)?;
774 Ok((profile, meta))
775}
776
777fn list_unmodeled(dir: &Path, flavor: Flavor) -> Result<Vec<String>> {
778 let mut owned: Vec<&str> = OWNED_FILES.to_vec();
779 if flavor == Flavor::Hermes {
780 owned.push("SOUL.md");
781 owned.retain(|f| !["AGENTS.md", "CLAUDE.md", "access.yaml"].contains(f));
782 }
783 let runtime_artifacts = ["orchestrator.lock", "orchestrator.sock", "service"];
784 let mut out = Vec::new();
785 fn walk(
786 base: &Path,
787 d: &Path,
788 owned: &[&str],
789 runtime: &[&str],
790 out: &mut Vec<String>,
791 ) -> Result<()> {
792 let mut entries: Vec<_> = fs::read_dir(d)?.flatten().collect();
793 entries.sort_by_key(|e| e.file_name());
794 for entry in entries {
795 let p = entry.path();
796 let rel = p
797 .strip_prefix(base)
798 .unwrap_or(&p)
799 .to_string_lossy()
800 .replace('\\', "/");
801 let name = entry.file_name().to_string_lossy().into_owned();
802 if rel == "profiles"
803 || name == "node_modules"
804 || name == ".git"
805 || rel.starts_with("state.db")
806 || rel.starts_with("cron/executions.db")
807 {
808 continue;
809 }
810 if runtime.contains(&rel.as_str()) || regex_tmp(&name) {
811 continue;
812 }
813 let st = fs::symlink_metadata(&p)?;
814 if st.is_dir() {
815 walk(base, &p, owned, runtime, out)?;
816 continue;
817 }
818 if !st.is_file() {
819 continue;
820 }
821 if owned.contains(&rel.as_str()) {
822 continue;
823 }
824 out.push(rel);
825 }
826 Ok(())
827 }
828 walk(dir, dir, &owned, &runtime_artifacts, &mut out)?;
829 Ok(out)
830}
831
832fn regex_tmp(name: &str) -> bool {
833 name.rsplit_once(".tmp-")
835 .is_some_and(|(_, pid)| !pid.is_empty() && pid.chars().all(|c| c.is_ascii_digit()))
836}
837
838fn write_atomic(path: &Path, text: &str) -> Result<()> {
841 if let Some(parent) = path.parent() {
842 fs::create_dir_all(parent)?;
843 }
844 let tmp = path.with_file_name(format!(
845 "{}.tmp-{}",
846 path.file_name().unwrap().to_string_lossy(),
847 std::process::id()
848 ));
849 fs::write(&tmp, text)?;
850 fs::rename(&tmp, path)?;
851 Ok(())
852}
853
854pub fn encode_config(
856 profile: &Profile,
857 meta: Option<&ProfileIo>,
858 vault: Option<&BTreeMap<String, String>>,
859 flavor: Flavor,
860) -> String {
861 let mut out = Map::new();
862 for (k, v) in &profile.residue.config {
863 if k != "gateway" {
864 out.insert(k.clone(), v.clone());
865 }
866 }
867 if let Some(w) = &profile.worker {
868 let mut wm = Map::new();
869 wm.insert("harness".into(), Value::String(w.harness.as_str().into()));
870 if let Some(m) = &w.model {
871 wm.insert("model".into(), Value::String(m.clone()));
872 }
873 if let Some(p) = &w.preset {
874 wm.insert("preset".into(), Value::String(p.clone()));
875 }
876 if w.cwd != "." {
877 wm.insert("cwd".into(), Value::String(w.cwd.clone()));
878 }
879 if !w.env.is_empty() {
880 wm.insert("env".into(), serde_json::to_value(&w.env).unwrap());
881 }
882 if w.permission.timeout_seconds != 300
883 || w.permission.default != crate::world::PermissionDefault::Deny
884 {
885 wm.insert(
886 "permission".into(),
887 serde_json::to_value(&w.permission).unwrap(),
888 );
889 }
890 out.insert("worker".into(), Value::Object(wm));
891 }
892 if flavor == Flavor::Orchestrator || profile.expiry != ExpiryPolicy::default() {
893 out.insert(
894 "expiry".into(),
895 serde_json::to_value(&profile.expiry).unwrap(),
896 );
897 }
898 if let Some(h) = &profile.home {
899 out.insert("home".into(), encode_surface_key(h));
900 }
901 let mut gateway = profile
902 .residue
903 .config
904 .get("gateway")
905 .and_then(Value::as_object)
906 .cloned()
907 .unwrap_or_default();
908 let routes: Vec<Value> = profile.routes.iter().map(encode_route).collect();
909 if meta.is_some_and(|m| m.routes_at_top) {
910 if !routes.is_empty() {
911 out.insert("profile_routes".into(), Value::Array(routes));
912 }
913 } else if !routes.is_empty() {
914 gateway.insert("profile_routes".into(), Value::Array(routes));
915 }
916 if !gateway.is_empty() {
917 out.insert("gateway".into(), Value::Object(gateway));
918 }
919 let mut platforms = Map::new();
920 for (p, ch) in &profile.channels {
921 platforms.insert(
922 p.clone(),
923 encode_channel(
924 ch,
925 if flavor == Flavor::Hermes {
926 vault
927 } else {
928 None
929 },
930 ),
931 );
932 }
933 if !platforms.is_empty() {
934 out.insert("platforms".into(), Value::Object(platforms));
935 }
936 json_to_yaml(&Value::Object(out))
937}
938
939pub fn encode_jobs_file(profile: &Profile, meta: Option<&ProfileIo>) -> String {
941 let jobs: Vec<Value> = profile
942 .jobs
943 .values()
944 .map(|j| ordered_object(encode_job(j)))
945 .collect();
946 let form = meta.and_then(|m| m.jobs_form.clone()).unwrap_or(JobsForm {
947 object: true,
948 extras: Map::new(),
949 });
950 let body = if form.object {
951 let mut pairs = vec![("jobs".to_string(), Value::Array(jobs))];
952 pairs.extend(form.extras.iter().map(|(k, v)| (k.clone(), v.clone())));
953 ordered_object(pairs)
954 } else {
955 Value::Array(jobs)
956 };
957 format!("{}\n", pretty_ordered(&body, 0))
958}
959
960pub(crate) fn ordered_object(pairs: Vec<(String, Value)>) -> Value {
963 Value::Array(vec![
966 Value::String("__ordered__".into()),
967 Value::Array(
968 pairs
969 .into_iter()
970 .map(|(k, v)| serde_json::json!({"__k": k, "__v": v}))
971 .collect(),
972 ),
973 ])
974}
975
976fn is_ordered(value: &Value) -> Option<&Vec<Value>> {
977 let arr = value.as_array()?;
978 if arr.len() == 2 && arr[0].as_str() == Some("__ordered__") {
979 arr[1].as_array()
980 } else {
981 None
982 }
983}
984
985pub(crate) fn pretty_ordered(value: &Value, depth: usize) -> String {
987 let pad = |d: usize| " ".repeat(d);
988 if let Some(pairs) = is_ordered(value) {
989 if pairs.is_empty() {
990 return "{}".into();
991 }
992 let inner: Vec<String> = pairs
993 .iter()
994 .map(|p| {
995 format!(
996 "{}{}: {}",
997 pad(depth + 1),
998 serde_json::to_string(p["__k"].as_str().unwrap_or("")).unwrap(),
999 pretty_ordered(&p["__v"], depth + 1)
1000 )
1001 })
1002 .collect();
1003 return format!("{{\n{}\n{}}}", inner.join(",\n"), pad(depth));
1004 }
1005 match value {
1006 Value::Array(items) if items.is_empty() => "[]".into(),
1007 Value::Array(items) => {
1008 let inner: Vec<String> = items
1009 .iter()
1010 .map(|v| format!("{}{}", pad(depth + 1), pretty_ordered(v, depth + 1)))
1011 .collect();
1012 format!("[\n{}\n{}]", inner.join(",\n"), pad(depth))
1013 }
1014 Value::Object(o) if o.is_empty() => "{}".into(),
1015 Value::Object(o) => {
1016 let inner: Vec<String> = o
1017 .iter()
1018 .map(|(k, v)| {
1019 format!(
1020 "{}{}: {}",
1021 pad(depth + 1),
1022 serde_json::to_string(k).unwrap(),
1023 pretty_ordered(v, depth + 1)
1024 )
1025 })
1026 .collect();
1027 format!("{{\n{}\n{}}}", inner.join(",\n"), pad(depth))
1028 }
1029 Value::Number(n) => {
1030 if let Some(f) = n.as_f64() {
1031 if n.is_f64() && f.fract() == 0.0 && f.abs() < 1e21 {
1032 return format!("{}", f as i64);
1033 }
1034 }
1035 n.to_string()
1036 }
1037 other => serde_json::to_string(other).unwrap(),
1038 }
1039}
1040
1041pub fn encode_subscriptions_file(
1043 profile: &Profile,
1044 vault: Option<&BTreeMap<String, String>>,
1045) -> String {
1046 let mut out = Map::new();
1047 for (n, s) in &profile.subscriptions {
1048 out.insert(n.clone(), encode_subscription(s, vault));
1049 }
1050 format!("{}\n", pretty_ordered(&Value::Object(out), 0))
1051}
1052
1053pub fn encode_access_file(profile: &Profile) -> String {
1055 json_to_yaml(&encode_access(&profile.access))
1056}
1057
1058const EXECUTIONS_DDL: &str = "CREATE TABLE IF NOT EXISTS executions (
1059 id TEXT PRIMARY KEY, job_id TEXT NOT NULL, source TEXT NOT NULL, process_id TEXT NOT NULL, pid INTEGER NOT NULL,
1060 process_started_at INTEGER, status TEXT NOT NULL CHECK(status IN ('claimed','running','completed','failed','unknown')),
1061 claimed_at TEXT NOT NULL, started_at TEXT, finished_at TEXT, error TEXT);
1062CREATE INDEX IF NOT EXISTS idx_executions_job_claimed ON executions(job_id, claimed_at DESC, id DESC);
1063CREATE INDEX IF NOT EXISTS idx_executions_status_claimed ON executions(status, claimed_at DESC, id DESC);";
1064
1065const OBLIGATIONS_DDL: &str = "CREATE TABLE IF NOT EXISTS delivery_obligations (
1066 obligation_id TEXT PRIMARY KEY, session_key TEXT NOT NULL, platform TEXT NOT NULL, chat_id TEXT NOT NULL, thread_id TEXT,
1067 content TEXT NOT NULL, state TEXT NOT NULL, attempts INTEGER NOT NULL DEFAULT 0, created_at REAL NOT NULL, updated_at REAL NOT NULL,
1068 owner_pid INTEGER, owner_started_at INTEGER, last_error TEXT, adapter_profile TEXT);";
1069
1070const BINDINGS_DDL: &str = "CREATE TABLE IF NOT EXISTS bindings (
1071 slot TEXT PRIMARY KEY,
1072 platform TEXT NOT NULL, chat_type TEXT NOT NULL, chat_id TEXT, thread_id TEXT, participant_id TEXT,
1073 worker_harness TEXT NOT NULL, worker_session_id TEXT, worker_locator TEXT,
1074 started_at TEXT NOT NULL, last_activity_at TEXT NOT NULL, ended_at TEXT, end_reason TEXT,
1075 handoff_to TEXT, handoff_state TEXT, handoff_error TEXT, recurrence_job_id TEXT, residue_json TEXT);";
1076
1077pub fn write_executions(path: &Path, fires: &[crate::world::Fire]) -> Result<()> {
1079 let insert = format!(
1080 "insert into executions ({}) values ({})",
1081 EXECUTION_COLUMNS.join(", "),
1082 EXECUTION_COLUMNS
1083 .iter()
1084 .map(|_| "?")
1085 .collect::<Vec<_>>()
1086 .join(",")
1087 );
1088 let rows: Vec<Vec<Param>> = fires
1089 .iter()
1090 .map(|f| encode_fire_row(f).iter().map(Param::from).collect())
1091 .collect();
1092 write_table(path, EXECUTIONS_DDL, &insert, &rows)
1093}
1094
1095fn write_state(path: &Path, profile: &Profile) -> Result<()> {
1096 let cols = [
1097 "slot",
1098 "platform",
1099 "chat_type",
1100 "chat_id",
1101 "thread_id",
1102 "participant_id",
1103 "worker_harness",
1104 "worker_session_id",
1105 "worker_locator",
1106 "started_at",
1107 "last_activity_at",
1108 "ended_at",
1109 "end_reason",
1110 "handoff_to",
1111 "handoff_state",
1112 "handoff_error",
1113 "recurrence_job_id",
1114 "residue_json",
1115 ];
1116 let insert = format!(
1117 "insert into bindings ({}) values ({})",
1118 cols.join(", "),
1119 cols.iter().map(|_| "?").collect::<Vec<_>>().join(",")
1120 );
1121 let s = |v: &Option<String>| v.clone().map(Param::Text).unwrap_or(Param::Null);
1122 let rows: Vec<Vec<Param>> = profile
1123 .bindings
1124 .iter()
1125 .map(|(slot, b)| {
1126 vec![
1127 Param::Text(slot.clone()),
1128 Param::Text(b.key.platform.clone().unwrap_or_default()),
1129 Param::Text(b.key.kind.clone().unwrap_or_default()),
1130 Param::Text(b.key.chat_id.clone().unwrap_or_default()),
1131 Param::Text(b.key.thread_id.clone().unwrap_or_default()),
1132 Param::Text(b.key.participant_id.clone().unwrap_or_default()),
1133 Param::Text(b.worker.harness.as_str().into()),
1134 s(&b.worker.session_id.clone().filter(|v| !v.is_empty())),
1135 s(&b.worker.locator),
1136 Param::Text(b.started_at.clone().unwrap_or_default()),
1137 Param::Text(b.last_activity_at.clone().unwrap_or_default()),
1138 s(&b.ended_at),
1139 b.end_reason
1140 .map(|r| Param::Text(r.as_str().into()))
1141 .unwrap_or(Param::Null),
1142 s(&b.handoff.as_ref().and_then(|h| h.to.clone())),
1143 b.handoff
1144 .as_ref()
1145 .map(|h| Param::Text(h.state.clone()))
1146 .unwrap_or(Param::Null),
1147 s(&b.handoff.as_ref().and_then(|h| h.error.clone())),
1148 s(&b.recurrence.as_ref().map(|r| r.job_id.clone())),
1149 if b.residue.is_empty() {
1150 Param::Null
1151 } else {
1152 Param::Text(serde_json::to_string(&b.residue).unwrap())
1153 },
1154 ]
1155 })
1156 .collect();
1157 write_table(
1158 path,
1159 &format!("{BINDINGS_DDL}\n{OBLIGATIONS_DDL}"),
1160 &insert,
1161 &rows,
1162 )?;
1163 let insert = format!(
1164 "insert into delivery_obligations ({}) values ({})",
1165 OBLIGATION_COLUMNS.join(", "),
1166 OBLIGATION_COLUMNS
1167 .iter()
1168 .map(|_| "?")
1169 .collect::<Vec<_>>()
1170 .join(",")
1171 );
1172 let rows: Vec<Vec<Param>> = profile
1173 .obligations
1174 .iter()
1175 .map(|o| encode_obligation_row(o).iter().map(Param::from).collect())
1176 .collect();
1177 write_table(path, "", &insert, &rows)
1178}
1179
1180fn write_if_changed(
1181 meta: &mut ProfileIo,
1182 dir: &Path,
1183 rel: &str,
1184 record: &Value,
1185 render: impl FnOnce() -> Option<String>,
1186) -> Result<bool> {
1187 let snap = canonical_json(record);
1188 let path = dir.join(rel);
1189 let reuse = meta.flavor == Flavor::Orchestrator && meta.snapshot.get(rel) == Some(&snap);
1192 if reuse && path.exists() {
1193 return Ok(false);
1194 }
1195 if reuse {
1196 if let Some(raw) = meta.raw.get(rel).cloned() {
1197 write_atomic(&path, &raw)?;
1198 return Ok(true);
1199 }
1200 }
1201 let Some(text) = render() else {
1202 return Ok(false);
1203 };
1204 write_atomic(&path, &text)?;
1205 meta.raw.insert(rel.into(), text);
1206 meta.snapshot.insert(rel.into(), snap);
1207 meta.flavor = Flavor::Orchestrator; Ok(true)
1209}
1210
1211pub fn save_home(loaded: &mut LoadedHome, root: Option<&Path>) -> Result<()> {
1213 let root = root
1214 .map(Path::to_path_buf)
1215 .unwrap_or_else(|| loaded.world.root.clone());
1216 fs::create_dir_all(&root)?;
1217 let names: Vec<String> = loaded.world.profiles.keys().cloned().collect();
1218 for name in names {
1219 let dir = if name == "default" {
1220 root.clone()
1221 } else {
1222 root.join("profiles").join(&name)
1223 };
1224 fs::create_dir_all(dir.join("cron"))?;
1225 let profile = loaded.world.profiles[&name].clone();
1226 let meta = loaded
1227 .io
1228 .entry(name.clone())
1229 .or_insert_with(|| ProfileIo::new(Flavor::Orchestrator));
1230 save_profile_dir(&profile, meta, &dir, &loaded.vault)?;
1231 }
1232 Ok(())
1233}
1234
1235fn save_profile_dir(
1236 profile: &Profile,
1237 meta: &mut ProfileIo,
1238 dir: &Path,
1239 vault: &BTreeMap<String, String>,
1240) -> Result<()> {
1241 let cfg_record = config_record(profile);
1242 write_if_changed(meta, dir, "config.yaml", &cfg_record, || {
1243 Some(encode_config(
1244 profile,
1245 None,
1246 Some(vault),
1247 Flavor::Orchestrator,
1248 ))
1249 })?;
1250 if let Some(persona) = &profile.persona {
1251 let text = persona.text.clone().unwrap_or_default();
1252 write_if_changed(
1253 meta,
1254 dir,
1255 "AGENTS.md",
1256 &serde_json::to_value(&profile.persona).unwrap(),
1257 || Some(text),
1258 )?;
1259 if !dir.join("CLAUDE.md").exists() {
1260 write_atomic(&dir.join("CLAUDE.md"), "@AGENTS.md\n")?;
1261 }
1262 }
1263 let jobs_record: Vec<Value> = profile
1264 .jobs
1265 .values()
1266 .map(|j| serde_json::to_value(j).unwrap())
1267 .collect();
1268 let had_jobs = meta.raw.contains_key("cron/jobs.json");
1269 let form = meta.jobs_form.clone();
1270 write_if_changed(
1271 meta,
1272 dir,
1273 "cron/jobs.json",
1274 &Value::Array(jobs_record),
1275 || {
1276 if profile.jobs.is_empty() && !had_jobs {
1277 None
1278 } else {
1279 let stub = ProfileIo {
1280 jobs_form: form.clone(),
1281 ..ProfileIo::new(Flavor::Orchestrator)
1282 };
1283 Some(encode_jobs_file(profile, Some(&stub)))
1284 }
1285 },
1286 )?;
1287 let subs_record: Vec<Value> = profile
1288 .subscriptions
1289 .values()
1290 .map(|s| serde_json::to_value(s).unwrap())
1291 .collect();
1292 let had_subs = meta.raw.contains_key("webhook_subscriptions.json");
1293 write_if_changed(
1294 meta,
1295 dir,
1296 "webhook_subscriptions.json",
1297 &Value::Array(subs_record),
1298 || {
1299 if profile.subscriptions.is_empty() && !had_subs {
1300 None
1301 } else {
1302 Some(encode_subscriptions_file(profile, None))
1303 }
1304 },
1305 )?;
1306 let a = &profile.access;
1307 let access_empty = a.allowlist.is_empty()
1308 && a.admins.is_empty()
1309 && a.pending_pairings.is_empty()
1310 && a.policy.is_empty()
1311 && a.pairing_ttl_minutes.is_none();
1312 let had_access = meta.raw.contains_key("access.yaml");
1313 write_if_changed(
1314 meta,
1315 dir,
1316 "access.yaml",
1317 &serde_json::to_value(a).unwrap(),
1318 || {
1319 if access_empty && !had_access {
1320 None
1321 } else {
1322 Some(encode_access_file(profile))
1323 }
1324 },
1325 )?;
1326
1327 let fires_snap = canonical_json(&serde_json::to_value(&profile.fires).unwrap());
1329 let exec_path = dir.join("cron/executions.db");
1330 if (meta.snapshot.get("cron/executions.db") != Some(&fires_snap) || !exec_path.exists())
1331 && (!profile.fires.is_empty() || exec_path.exists())
1332 {
1333 let tmp = exec_path.with_file_name(format!("executions.db.tmp-{}", std::process::id()));
1334 let _ = fs::remove_file(&tmp);
1335 write_executions(&tmp, &profile.fires)?;
1336 fs::rename(&tmp, &exec_path)?;
1337 meta.snapshot
1338 .insert("cron/executions.db".into(), fires_snap);
1339 }
1340 let state_snap = canonical_json(&state_record(profile));
1341 let state_path = dir.join("state.db");
1342 if (meta.snapshot.get("state.db") != Some(&state_snap) || !state_path.exists())
1343 && (!profile.bindings.is_empty() || !profile.obligations.is_empty() || state_path.exists())
1344 {
1345 let tmp = state_path.with_file_name(format!("state.db.tmp-{}", std::process::id()));
1346 let _ = fs::remove_file(&tmp);
1347 write_state(&tmp, profile)?;
1348 fs::rename(&tmp, &state_path)?;
1349 meta.snapshot.insert("state.db".into(), state_snap);
1350 }
1351
1352 let mut refs: Vec<String> = Vec::new();
1354 for ch in profile.channels.values() {
1355 for r in ch.credentials.values() {
1356 if let crate::ontology::SecretRef::Dotenv(n) = r {
1357 refs.push(n.clone());
1358 }
1359 }
1360 }
1361 for s in profile.subscriptions.values() {
1362 if let Some(crate::ontology::SecretRef::Dotenv(n)) = &s.secret {
1363 refs.push(n.clone());
1364 }
1365 }
1366 if let Some(w) = &profile.worker {
1367 for v in w.env.values() {
1368 if let crate::world::EnvValue::Secret(crate::ontology::SecretRef::Dotenv(n)) = v {
1369 refs.push(n.clone());
1370 }
1371 }
1372 }
1373 let mut entries: BTreeMap<String, String> = BTreeMap::new();
1374 for r in refs {
1375 if let Some(v) = vault.get(&r) {
1376 entries.insert(r, v.clone());
1377 }
1378 }
1379 if !entries.is_empty() {
1380 let existing = meta.raw.get(".env").cloned();
1381 let mut merged = existing.as_deref().map(parse_dotenv).unwrap_or_default();
1382 for (k, v) in entries {
1383 merged.insert(k, v);
1384 }
1385 let text = render_dotenv(&merged);
1386 if existing.as_deref() != Some(text.as_str()) {
1387 write_atomic(&dir.join(".env"), &text)?;
1388 meta.raw.insert(".env".into(), text);
1389 }
1390 }
1391 Ok(())
1392}
1393
1394pub fn copy_unmodeled(profile: &Profile, meta: &ProfileIo, dest: &Path) -> Result<()> {
1396 let Some(src) = &meta.source_dir else {
1397 return Ok(());
1398 };
1399 carry_unmodeled(&profile.residue.files, src, dest)?;
1400 Ok(())
1401}
1402
1403pub fn carry_unmodeled(files: &[String], src: &Path, dest: &Path) -> Result<Vec<String>> {
1407 let mut carried = Vec::new();
1408 for rel in files {
1409 let from = src.join(rel);
1410 if !from.is_file() {
1411 continue;
1412 }
1413 let to = dest.join(rel);
1414 if let Some(parent) = to.parent() {
1415 fs::create_dir_all(parent)?;
1416 }
1417 fs::copy(&from, &to)?;
1418 carried.push(rel.clone());
1419 }
1420 Ok(carried)
1421}