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 hermes_cron_job_id, parse_hermes_session_key, Binding, EndReason, Handoff, HarnessId,
34 HermesSessionRow, Recurrence, Residue, SurfaceKey, Trigger, Worker,
35};
36use crate::orchestration::{ExpiryPolicy, Orchestration, PersonaRef, Profile, ProfileResidue};
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 orchestration: Orchestration,
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 orchestration: Orchestration {
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 link_fires(&mut loaded, root)?;
247 Ok(loaded)
248}
249
250fn link_fires(loaded: &mut LoadedHome, root: &Path) -> Result<()> {
258 let names: Vec<String> = loaded.orchestration.profiles.keys().cloned().collect();
259 let root_store = root.join("state.db");
260 for name in &names {
261 let own = loaded.orchestration.profiles[name].dir.join("state.db");
262 let store = if own.is_file() {
263 own
264 } else {
265 root_store.clone()
266 };
267 let sessions: Vec<Map<String, Value>> = if table_exists(&store, "sessions") {
268 read_rows(
269 &store,
270 "select id, started_at, end_reason, parent_session_id, session_key from sessions",
271 &[],
272 )?
273 .unwrap_or_default()
274 } else {
275 Vec::new()
276 };
277 let holders: Vec<String> = {
280 let io = &loaded.io[name];
281 if io.borrowed_from.is_some() || !io.lenders.is_empty() {
282 let mut v = vec!["default".to_string()];
283 v.extend(loaded.io["default"].lenders.iter().cloned());
284 v
285 } else {
286 vec![name.clone()]
287 }
288 };
289 let mut links: Vec<(usize, Option<String>, Option<(String, String)>)> = Vec::new();
290 let profile = &loaded.orchestration.profiles[name];
291 for (index, fire) in profile.fires.iter().enumerate() {
292 let session_id = fire_session(&sessions, fire);
293 let session_key = session_id.as_deref().and_then(|id| {
294 sessions
295 .iter()
296 .find(|row| row.get("id").and_then(Value::as_str) == Some(id))
297 .and_then(|row| row.get("session_key"))
298 .and_then(Value::as_str)
299 .filter(|k| !k.is_empty())
300 .map(str::to_string)
301 });
302 let surface = profile.jobs.get(&fire.job_id).and_then(job_surface);
303 let (Some(from), to) = (
304 iso_epoch(&fire.claimed_at),
305 fire.finished_at
306 .as_deref()
307 .and_then(iso_epoch)
308 .unwrap_or(f64::MAX),
309 ) else {
310 links.push((index, session_id, None));
311 continue;
312 };
313 let in_window = |o: &&crate::orchestration::Obligation| {
314 o.created_at
315 .parse::<f64>()
316 .is_ok_and(|at| at >= from && at <= to)
317 };
318 let latest = |mut found: Vec<(&String, &crate::orchestration::Obligation)>| {
319 found.sort_by(|a, b| {
320 let at = |o: &crate::orchestration::Obligation| {
321 o.created_at.parse::<f64>().unwrap_or(0.0)
322 };
323 at(b.1)
324 .partial_cmp(&at(a.1))
325 .unwrap_or(std::cmp::Ordering::Equal)
326 });
327 found
328 .first()
329 .map(|(holder, o)| ((*holder).clone(), o.id.clone()))
330 };
331 let candidates = |pick: &dyn Fn(&crate::orchestration::Obligation) -> bool| {
332 holders
333 .iter()
334 .flat_map(|h| {
335 loaded.orchestration.profiles[h]
336 .obligations
337 .iter()
338 .filter(in_window)
339 .filter(|o| pick(o))
340 .map(move |o| (h, o))
341 })
342 .collect::<Vec<_>>()
343 };
344 let obligation = match &session_key {
345 Some(key) => latest(candidates(&|o| o.session_key.as_deref() == Some(key))),
346 None => None,
347 }
348 .or_else(|| {
349 let (platform, chat_id) = surface.as_ref()?;
350 latest(candidates(&|o| {
351 o.target.platform.as_deref() == Some(platform)
352 && o.target.chat_id.as_deref() == Some(chat_id)
353 }))
354 });
355 links.push((index, session_id, obligation));
356 }
357 for (index, session_id, obligation) in links {
358 let fire_id = {
359 let fire = &mut loaded.orchestration.profiles.get_mut(name).unwrap().fires[index];
360 fire.session_id = session_id;
361 fire.obligation_id = obligation.as_ref().map(|(_, id)| id.clone());
362 fire.id.clone()
363 };
364 if let Some((holder, obligation_id)) = obligation {
365 if let Some(o) = loaded
366 .orchestration
367 .profiles
368 .get_mut(&holder)
369 .and_then(|p| p.obligations.iter_mut().find(|o| o.id == obligation_id))
370 {
371 o.source = crate::orchestration::ObligationSource::Fire { fire_id };
372 }
373 }
374 }
375 }
376 for name in &names {
378 let profile = &loaded.orchestration.profiles[name];
379 let io = loaded.io.get_mut(name).unwrap();
380 io.snapshot.insert(
381 "cron/executions.db".into(),
382 canonical_json(&serde_json::to_value(&profile.fires).unwrap()),
383 );
384 io.snapshot
385 .insert("state.db".into(), canonical_json(&state_record(profile)));
386 }
387 Ok(())
388}
389
390fn job_surface(job: &crate::orchestration::Job) -> Option<(String, String)> {
395 use crate::orchestration::Target;
396 match &job.deliver {
397 Target::Origin => {
398 let origin = job.origin.as_ref()?;
399 Some((origin.platform.clone(), origin.chat_id.clone()?))
400 }
401 Target::Explicit {
402 platform, chat_id, ..
403 } => Some((platform.clone(), chat_id.clone()?)),
404 Target::Home | Target::Local => None,
405 }
406}
407
408fn fire_session(
412 sessions: &[Map<String, Value>],
413 fire: &crate::orchestration::Fire,
414) -> Option<String> {
415 let claimed = instant_key(&fire.claimed_at)?;
416 let finished = fire.finished_at.as_deref().and_then(instant_key);
417 let candidates = sessions.iter().filter_map(|row| {
418 let id = row.get("id").and_then(Value::as_str)?;
419 let key = cron_session_instant(id, &fire.job_id)?;
420 (key >= claimed && finished.is_none_or(|f| key <= f)).then(|| (key, id.to_string()))
421 });
422 let chosen = match finished {
423 Some(_) => candidates.max_by_key(|(key, _)| *key),
424 None => candidates.min_by_key(|(key, _)| *key),
425 }?;
426 let mut current = chosen.1;
427 for _ in 0..32 {
428 let row = sessions
429 .iter()
430 .find(|row| row.get("id").and_then(Value::as_str) == Some(current.as_str()));
431 let compressed = row
432 .and_then(|row| row.get("end_reason"))
433 .and_then(Value::as_str)
434 == Some("compression");
435 if !compressed {
436 return Some(current);
437 }
438 let next = sessions
439 .iter()
440 .filter(|row| {
441 row.get("parent_session_id").and_then(Value::as_str) == Some(current.as_str())
442 })
443 .max_by(|a, b| {
444 let at = |r: &Map<String, Value>| {
445 r.get("started_at").and_then(Value::as_f64).unwrap_or(0.0)
446 };
447 at(a)
448 .partial_cmp(&at(b))
449 .unwrap_or(std::cmp::Ordering::Equal)
450 .then_with(|| {
451 let id = |r: &Map<String, Value>| {
452 r.get("id")
453 .and_then(Value::as_str)
454 .unwrap_or("")
455 .to_string()
456 };
457 id(a).cmp(&id(b))
458 })
459 })
460 .and_then(|row| row.get("id").and_then(Value::as_str).map(str::to_string));
461 match next {
462 None => return Some(current),
463 Some(next) => current = next,
464 }
465 }
466 Some(current)
467}
468
469fn instant_key(iso: &str) -> Option<u64> {
471 let digits: String = iso
472 .chars()
473 .take_while(|c| *c != '+' && *c != 'Z')
474 .filter(char::is_ascii_digit)
475 .collect();
476 (digits.len() >= 14).then(|| digits[..14].parse().ok())?
477}
478
479fn cron_session_instant(session_id: &str, job_id: &str) -> Option<u64> {
481 if hermes_cron_job_id(session_id).as_deref() != Some(job_id) {
482 return None;
483 }
484 let stamp = session_id.rsplit_once('_')?;
485 let date = stamp.0.rsplit_once('_')?.1;
486 format!("{date}{}", stamp.1).parse().ok()
487}
488
489fn iso_epoch(iso: &str) -> Option<f64> {
491 let (instant, offset) = if let Some(instant) = iso.strip_suffix('Z') {
492 (instant, 0.0)
493 } else {
494 let time_at = iso.find('T')?;
495 let sign_at = iso[time_at..].find(['+', '-']).map(|i| i + time_at)?;
496 let (instant, offset) = iso.split_at(sign_at);
497 let (hours, minutes) = offset[1..].split_once(':')?;
498 let seconds = hours.parse::<f64>().ok()? * 3_600.0 + minutes.parse::<f64>().ok()? * 60.0;
499 (
500 instant,
501 if offset.starts_with('-') {
502 -seconds
503 } else {
504 seconds
505 },
506 )
507 };
508 let (date, time) = instant.split_once('T')?;
509 let mut date = date.splitn(3, '-');
510 let year: i64 = date.next()?.parse().ok()?;
511 let month: i64 = date.next()?.parse().ok()?;
512 let day: i64 = date.next()?.parse().ok()?;
513 let mut clock = time.splitn(3, ':');
514 let hour: i64 = clock.next()?.parse().ok()?;
515 let minute: i64 = clock.next()?.parse().ok()?;
516 let seconds: f64 = clock.next()?.parse().ok()?;
517 let year = year - i64::from(month <= 2);
518 let era = year.div_euclid(400);
519 let yoe = year - era * 400;
520 let doy = (153 * (if month > 2 { month - 3 } else { month + 9 }) + 2) / 5 + day - 1;
521 let doe = yoe * 365 + yoe / 4 - yoe / 100 + doy;
522 let days = era * 146_097 + doe - 719_468;
523 Some((days * 86_400 + hour * 3_600 + minute * 60) as f64 + seconds - offset)
524}
525
526fn partition_shared_store(loaded: &mut LoadedHome, root: &Path) -> Result<()> {
531 let root_path = root.join("state.db");
532 if !root_path.is_file() {
533 return Ok(());
534 }
535 let names: Vec<String> = loaded
536 .orchestration
537 .profiles
538 .keys()
539 .filter(|n| *n != "default")
540 .cloned()
541 .collect();
542 for name in names {
543 let has_own = loaded.orchestration.profiles[&name]
544 .dir
545 .join("state.db")
546 .is_file();
547 if has_own {
548 continue;
549 }
550 loaded.io.get_mut(&name).unwrap().borrowed_from = Some(root_path.clone());
551 loaded
552 .io
553 .get_mut("default")
554 .unwrap()
555 .lenders
556 .push(name.clone());
557 if table_exists(&root_path, "sessions") {
558 let rows = read_rows(
559 &root_path,
560 "select * from sessions where profile_name = ?1 order by started_at, id",
561 &[&name],
562 )?
563 .unwrap_or_default();
564 for row in rows {
565 if let Some(b) = binding_from_hermes_session(&root_path, &row, &name) {
566 loaded
567 .orchestration
568 .profiles
569 .get_mut(&name)
570 .unwrap()
571 .bindings
572 .insert(surface_key_string(&b.key), b);
573 }
574 }
575 }
576 let root_profile = loaded.orchestration.profiles.get_mut("default").unwrap();
578 let (mine, rest): (Vec<_>, Vec<_>) = root_profile.obligations.drain(..).partition(|o| {
579 o.session_key
580 .as_deref()
581 .and_then(parse_hermes_session_key)
582 .and_then(|(_, p)| p)
583 .as_deref()
584 == Some(name.as_str())
585 });
586 root_profile.obligations = rest;
587 loaded
588 .orchestration
589 .profiles
590 .get_mut(&name)
591 .unwrap()
592 .obligations = mine;
593 let snap = canonical_json(&state_record(&loaded.orchestration.profiles[&name]));
594 loaded
595 .io
596 .get_mut(&name)
597 .unwrap()
598 .snapshot
599 .insert("state.db".into(), snap);
600 }
601 let snap = canonical_json(&state_record(&loaded.orchestration.profiles["default"]));
602 loaded
603 .io
604 .get_mut("default")
605 .unwrap()
606 .snapshot
607 .insert("state.db".into(), snap);
608 Ok(())
609}
610
611const HERMES_SESSION_MAPPED: &[&str] = &[
612 "id",
613 "source",
614 "session_key",
615 "chat_id",
616 "chat_type",
617 "thread_id",
618 "user_id",
619 "profile_name",
620 "handoff_state",
621 "handoff_platform",
622 "handoff_error",
623 "started_at",
624 "ended_at",
625 "end_reason",
626];
627
628pub fn binding_from_hermes_session(
630 file: &Path,
631 row: &Map<String, Value>,
632 profile_name: &str,
633) -> Option<Binding> {
634 let text = |k: &str| {
635 row.get(k)
636 .and_then(|v| match v {
637 Value::String(s) => Some(s.clone()),
638 Value::Number(n) => Some(n.to_string()),
639 _ => None,
640 })
641 .filter(|s| !s.is_empty())
642 };
643 let num = |k: &str| row.get(k).and_then(Value::as_f64);
644 let id = text("id")?;
645 let source = text("source");
646 let parsed = text("session_key").and_then(|k| parse_hermes_session_key(&k));
649 let chat_type = text("chat_type").or_else(|| parsed.as_ref().and_then(|(k, _)| k.kind.clone()));
650 let key = if text("session_key").is_some()
651 && chat_type
652 .as_deref()
653 .is_some_and(|c| super::decode::CHAT_TYPES.contains(&c))
654 {
655 SurfaceKey {
656 key: None,
657 platform: source
658 .clone()
659 .or_else(|| parsed.as_ref().and_then(|(k, _)| k.platform.clone())),
660 kind: chat_type,
661 chat_id: text("chat_id")
662 .or_else(|| parsed.as_ref().and_then(|(k, _)| k.chat_id.clone())),
663 thread_id: text("thread_id")
664 .or_else(|| parsed.as_ref().and_then(|(k, _)| k.thread_id.clone())),
665 participant_id: parsed.as_ref().and_then(|(k, _)| k.participant_id.clone()),
666 }
667 } else if source.as_deref() == Some("cron") {
668 let job = crate::ontology::hermes_cron_job_id(&id).unwrap_or_else(|| id.clone());
669 SurfaceKey {
670 key: None,
671 platform: Some("cron".into()),
672 kind: Some("dm".into()),
673 chat_id: Some(job),
674 thread_id: None,
675 participant_id: None,
676 }
677 } else {
678 return None;
679 };
680 if let Some(p) = text("profile_name") {
681 if p != profile_name && !(profile_name == "default" && p == "main") {
682 return None; }
684 }
685 let iso = |v: Option<f64>| v.map(epoch_iso);
686 let end_word = text("end_reason");
687 let end_reason = end_word.as_deref().and_then(EndReason::parse);
688 let mut residue = Residue::default();
689 for (k, v) in row {
690 if !HERMES_SESSION_MAPPED.contains(&k.as_str()) && !v.is_null() {
691 residue.keep(k.clone(), v.clone());
692 }
693 }
694 if let (Some(word), None) = (&end_word, end_reason) {
695 residue.keep("end_reason", Value::String(word.clone()));
696 }
697 if let Some(u) = text("user_id") {
698 residue.keep("user_id", Value::String(u));
699 }
700 let recurrence = if key.platform.as_deref() == Some("cron") {
701 key.chat_id.clone().map(|job_id| Recurrence {
702 job_id,
703 kind: "cron".into(),
704 })
705 } else {
706 None
707 };
708 Some(Binding {
709 trigger: match (recurrence.is_some(), source.as_deref()) {
710 (true, _) => Trigger::Cron,
711 (_, Some(s)) => crate::ontology::hermes_trigger_for_source(s),
712 _ => Trigger::Unknown,
713 },
714 key,
715 profile: None,
716 worker: Worker {
717 harness: HarnessId::new(HarnessId::HERMES),
718 session_id: Some(id),
719 locator: Some(file.display().to_string()),
720 },
721 recurrence,
722 handoff: text("handoff_state").map(|state| Handoff {
723 to: text("handoff_platform"),
724 state,
725 error: text("handoff_error"),
726 }),
727 started_at: iso(num("started_at")),
728 last_activity_at: iso(num("ended_at").or_else(|| num("started_at"))),
729 ended_at: iso(num("ended_at")),
730 end_reason,
731 residue,
732 })
733}
734
735fn epoch_iso(seconds: f64) -> String {
737 let row = HermesSessionRow {
738 started_at: Some(seconds),
739 ..Default::default()
740 };
741 Binding::from_hermes_row(&row, None)
742 .started_at
743 .unwrap_or_default()
744}
745
746fn load_profile_dir(
747 name: &str,
748 dir: &Path,
749 flavor: Flavor,
750 vault: &mut BTreeMap<String, String>,
751) -> Result<(Profile, ProfileIo)> {
752 let mut profile = empty_profile(name, dir);
753 let mut config_normalized = false;
754 let mut subs_normalized = false;
755 let mut meta = ProfileIo::new(flavor);
756 meta.source_dir = Some(dir.to_path_buf());
757 let remember = |meta: &mut ProfileIo, rel: &str, raw: Option<String>, record: &Value| {
758 if let Some(raw) = raw {
759 meta.raw.insert(rel.to_string(), raw);
760 }
761 meta.snapshot
762 .insert(rel.to_string(), canonical_json(record));
763 };
764
765 if let Some(env) = read_text(dir, ".env")? {
767 for (k, v) in parse_dotenv(&env) {
768 vault.insert(k, v);
769 }
770 meta.raw.insert(".env".into(), env);
771 }
772
773 let cfg_file = dir.join("config.yaml").display().to_string();
775 let cfg_text = read_text(dir, "config.yaml")?;
776 let cfg = match &cfg_text {
777 Some(text) => yaml_to_json(&cfg_file, text)?,
778 None => Value::Object(Map::new()),
779 };
780 let cfg_map = cfg
781 .as_object()
782 .ok_or_else(|| load_error(&cfg_file, "", "expected a mapping"))?;
783 profile.worker = decode_worker(&cfg_file, cfg_map.get("worker"))?;
784 profile.expiry = decode_expiry(&cfg_file, cfg_map.get("expiry"))?;
785 profile.home = decode_home(&cfg_file, cfg_map.get("home"))?;
786 let gateway = cfg_map.get("gateway").and_then(Value::as_object);
787 let routes_raw: Vec<Value> = match cfg_map.get("profile_routes").and_then(Value::as_array) {
788 Some(a) => {
789 meta.routes_at_top = true;
790 a.clone()
791 }
792 None => gateway
793 .and_then(|g| g.get("profile_routes"))
794 .and_then(Value::as_array)
795 .cloned()
796 .unwrap_or_default(),
797 };
798 for (i, r) in routes_raw.iter().enumerate() {
799 profile.routes.push(decode_route(&cfg_file, i, r)?);
800 }
801 if let Some(platforms) = cfg_map.get("platforms") {
802 let map = platforms
803 .as_object()
804 .ok_or_else(|| load_error(&cfg_file, "platforms", "expected a map"))?;
805 let known: BTreeSet<String> = vault.keys().cloned().collect();
806 for (platform, raw) in map {
807 profile.channels.insert(
808 platform.clone(),
809 decode_channel(&cfg_file, platform, raw, vault)?,
810 );
811 }
812 config_normalized =
813 flavor == Flavor::Orchestrator && vault.keys().any(|k| !known.contains(k));
814 }
815 for (k, v) in cfg_map {
817 if CONFIG_O_KEYS.contains(&k.as_str()) || k == "platforms" || k == "profile_routes" {
818 continue;
819 }
820 if k == "gateway" {
821 let mut g = v.as_object().cloned().unwrap_or_default();
822 g.remove("profile_routes");
823 if !g.is_empty() {
824 profile
825 .residue
826 .config
827 .insert("gateway".into(), Value::Object(g));
828 }
829 continue;
830 }
831 profile.residue.config.insert(k.clone(), v.clone());
832 }
833 remember(&mut meta, "config.yaml", cfg_text, &config_record(&profile));
834 if config_normalized {
835 meta.raw.remove("config.yaml");
840 meta.snapshot.remove("config.yaml");
841 }
842
843 let persona_file = if flavor == Flavor::Hermes {
845 "SOUL.md"
846 } else {
847 "AGENTS.md"
848 };
849 let persona_text = read_text(dir, persona_file)?;
850 profile.persona = persona_text.as_ref().map(|t| PersonaRef {
851 path: "AGENTS.md".into(),
852 text: Some(t.clone()),
853 sha256: sha256_hex(t),
854 });
855 remember(
856 &mut meta,
857 persona_file,
858 persona_text,
859 &serde_json::to_value(&profile.persona).unwrap(),
860 );
861
862 let jobs_file = dir.join("cron/jobs.json").display().to_string();
864 let jobs_text = read_text(dir, "cron/jobs.json")?;
865 if let Some(text) = &jobs_text {
866 let parsed: Value = serde_json::from_str(text)
867 .map_err(|e| load_error(&jobs_file, "", format!("JSON: {e}")))?;
868 let arr: Vec<Value> = match &parsed {
869 Value::Array(a) => {
870 meta.jobs_form = Some(JobsForm {
871 object: false,
872 extras: Map::new(),
873 });
874 a.clone()
875 }
876 Value::Object(o) => match o.get("jobs") {
877 Some(Value::Array(a)) => {
878 let mut extras = o.clone();
879 extras.remove("jobs");
880 meta.jobs_form = Some(JobsForm {
881 object: true,
882 extras,
883 });
884 a.clone()
885 }
886 Some(Value::Object(m)) => {
887 let mut extras = o.clone();
888 extras.remove("jobs");
889 meta.jobs_form = Some(JobsForm {
890 object: true,
891 extras,
892 });
893 m.iter()
894 .map(|(id, j)| {
895 let mut j = j.as_object().cloned().unwrap_or_default();
896 j.insert("id".into(), Value::String(id.clone()));
897 Value::Object(j)
898 })
899 .collect()
900 }
901 _ => {
902 return Err(load_error(
903 &jobs_file,
904 "",
905 "expected an array of jobs or {\"jobs\": [...]}",
906 ))
907 }
908 },
909 _ => {
910 return Err(load_error(
911 &jobs_file,
912 "",
913 "expected an array of jobs or {\"jobs\": [...]}",
914 ))
915 }
916 };
917 for raw in &arr {
918 let job = decode_job(&jobs_file, raw)?;
919 if profile.jobs.contains_key(&job.id) {
920 return Err(load_error(&jobs_file, &job.id, "duplicate job id"));
921 }
922 profile.jobs.insert(job.id.clone(), job);
923 }
924 }
925 let jobs_record: Vec<Value> = profile
926 .jobs
927 .values()
928 .map(|j| serde_json::to_value(j).unwrap())
929 .collect();
930 remember(
931 &mut meta,
932 "cron/jobs.json",
933 jobs_text,
934 &Value::Array(jobs_record),
935 );
936
937 let exec_path = dir.join("cron/executions.db");
939 if table_exists(&exec_path, "executions") {
940 for row in read_rows(
941 &exec_path,
942 "select * from executions order by claimed_at, id",
943 &[],
944 )?
945 .unwrap_or_default()
946 {
947 profile
948 .fires
949 .push(decode_fire_row(&exec_path.display().to_string(), &row)?);
950 }
951 }
952 remember(
953 &mut meta,
954 "cron/executions.db",
955 None,
956 &serde_json::to_value(&profile.fires).unwrap(),
957 );
958
959 let subs_file = dir.join("webhook_subscriptions.json").display().to_string();
961 let subs_text = read_text(dir, "webhook_subscriptions.json")?;
962 if let Some(text) = &subs_text {
963 let parsed: Value = serde_json::from_str(text)
964 .map_err(|e| load_error(&subs_file, "", format!("JSON: {e}")))?;
965 let map = parsed
966 .as_object()
967 .ok_or_else(|| load_error(&subs_file, "", "expected a map"))?;
968 let known: BTreeSet<String> = vault.keys().cloned().collect();
969 for (n, raw) in map {
970 profile
971 .subscriptions
972 .insert(n.clone(), decode_subscription(&subs_file, n, raw, vault)?);
973 }
974 subs_normalized =
975 flavor == Flavor::Orchestrator && vault.keys().any(|k| !known.contains(k));
976 }
977 let subs_record: Vec<Value> = profile
978 .subscriptions
979 .values()
980 .map(|s| serde_json::to_value(s).unwrap())
981 .collect();
982 remember(
983 &mut meta,
984 "webhook_subscriptions.json",
985 subs_text,
986 &Value::Array(subs_record),
987 );
988 if subs_normalized {
989 meta.raw.remove("webhook_subscriptions.json");
990 meta.snapshot.remove("webhook_subscriptions.json");
991 }
992
993 let access_file = dir.join("access.yaml").display().to_string();
995 let access_text = read_text(dir, "access.yaml")?;
996 let access_raw = match &access_text {
997 Some(t) => Some(yaml_to_json(&access_file, t)?),
998 None => None,
999 };
1000 profile.access = decode_access(&access_file, access_raw.as_ref())?;
1001 remember(
1002 &mut meta,
1003 "access.yaml",
1004 access_text,
1005 &serde_json::to_value(&profile.access).unwrap(),
1006 );
1007
1008 let state_path = dir.join("state.db");
1010 if table_exists(&state_path, "delivery_obligations") {
1011 for row in read_rows(
1012 &state_path,
1013 "select * from delivery_obligations order by created_at, obligation_id",
1014 &[],
1015 )?
1016 .unwrap_or_default()
1017 {
1018 profile.obligations.push(decode_obligation_row(
1019 &state_path.display().to_string(),
1020 &row,
1021 )?);
1022 }
1023 }
1024 if flavor == Flavor::Orchestrator {
1025 if table_exists(&state_path, "bindings") {
1026 for row in read_rows(
1027 &state_path,
1028 "select * from bindings order by started_at, slot",
1029 &[],
1030 )?
1031 .unwrap_or_default()
1032 {
1033 let b = decode_binding_row(&state_path.display().to_string(), &row)?;
1034 let slot = row
1035 .get("slot")
1036 .and_then(Value::as_str)
1037 .map(str::to_string)
1038 .unwrap_or_else(|| surface_key_string(&b.key));
1039 profile.bindings.insert(slot, b);
1040 }
1041 }
1042 } else if table_exists(&state_path, "sessions") {
1043 for row in read_rows(
1044 &state_path,
1045 "select * from sessions order by started_at, id",
1046 &[],
1047 )?
1048 .unwrap_or_default()
1049 {
1050 if let Some(b) = binding_from_hermes_session(&state_path, &row, name) {
1051 profile.bindings.insert(surface_key_string(&b.key), b);
1052 }
1053 }
1054 }
1055 remember(&mut meta, "state.db", None, &state_record(&profile));
1056
1057 profile.residue.files = list_unmodeled(dir, flavor)?;
1059 Ok((profile, meta))
1060}
1061
1062fn list_unmodeled(dir: &Path, flavor: Flavor) -> Result<Vec<String>> {
1063 let mut owned: Vec<&str> = OWNED_FILES.to_vec();
1064 if flavor == Flavor::Hermes {
1065 owned.push("SOUL.md");
1066 owned.retain(|f| !["AGENTS.md", "CLAUDE.md", "access.yaml"].contains(f));
1067 }
1068 let runtime_artifacts = ["orchestrator.lock", "orchestrator.sock", "service"];
1069 let mut out = Vec::new();
1070 fn walk(
1071 base: &Path,
1072 d: &Path,
1073 owned: &[&str],
1074 runtime: &[&str],
1075 out: &mut Vec<String>,
1076 ) -> Result<()> {
1077 let mut entries: Vec<_> = fs::read_dir(d)?.flatten().collect();
1078 entries.sort_by_key(|e| e.file_name());
1079 for entry in entries {
1080 let p = entry.path();
1081 let rel = p
1082 .strip_prefix(base)
1083 .unwrap_or(&p)
1084 .to_string_lossy()
1085 .replace('\\', "/");
1086 let name = entry.file_name().to_string_lossy().into_owned();
1087 if rel == "profiles"
1088 || name == "node_modules"
1089 || name == ".git"
1090 || rel.starts_with("state.db")
1091 || rel.starts_with("cron/executions.db")
1092 {
1093 continue;
1094 }
1095 if runtime.contains(&rel.as_str()) || regex_tmp(&name) {
1096 continue;
1097 }
1098 let st = fs::symlink_metadata(&p)?;
1099 if st.is_dir() {
1100 walk(base, &p, owned, runtime, out)?;
1101 continue;
1102 }
1103 if !st.is_file() {
1104 continue;
1105 }
1106 if owned.contains(&rel.as_str()) {
1107 continue;
1108 }
1109 out.push(rel);
1110 }
1111 Ok(())
1112 }
1113 walk(dir, dir, &owned, &runtime_artifacts, &mut out)?;
1114 Ok(out)
1115}
1116
1117fn regex_tmp(name: &str) -> bool {
1118 name.rsplit_once(".tmp-")
1120 .is_some_and(|(_, pid)| !pid.is_empty() && pid.chars().all(|c| c.is_ascii_digit()))
1121}
1122
1123fn write_atomic(path: &Path, text: &str) -> Result<()> {
1126 if let Some(parent) = path.parent() {
1127 fs::create_dir_all(parent)?;
1128 }
1129 let tmp = path.with_file_name(format!(
1130 "{}.tmp-{}",
1131 path.file_name().unwrap().to_string_lossy(),
1132 std::process::id()
1133 ));
1134 fs::write(&tmp, text)?;
1135 fs::rename(&tmp, path)?;
1136 Ok(())
1137}
1138
1139pub fn encode_config(
1141 profile: &Profile,
1142 meta: Option<&ProfileIo>,
1143 vault: Option<&BTreeMap<String, String>>,
1144 flavor: Flavor,
1145) -> String {
1146 let mut out = Map::new();
1147 for (k, v) in &profile.residue.config {
1148 if k != "gateway" {
1149 out.insert(k.clone(), v.clone());
1150 }
1151 }
1152 if let Some(w) = &profile.worker {
1153 let mut wm = Map::new();
1154 wm.insert("harness".into(), Value::String(w.harness.as_str().into()));
1155 if let Some(m) = &w.model {
1156 wm.insert("model".into(), Value::String(m.clone()));
1157 }
1158 if let Some(p) = &w.preset {
1159 wm.insert("preset".into(), Value::String(p.clone()));
1160 }
1161 if w.cwd != "." {
1162 wm.insert("cwd".into(), Value::String(w.cwd.clone()));
1163 }
1164 if !w.env.is_empty() {
1165 wm.insert("env".into(), serde_json::to_value(&w.env).unwrap());
1166 }
1167 if w.permission.timeout_seconds != 300
1168 || w.permission.default != crate::orchestration::PermissionDefault::Deny
1169 {
1170 wm.insert(
1171 "permission".into(),
1172 serde_json::to_value(&w.permission).unwrap(),
1173 );
1174 }
1175 out.insert("worker".into(), Value::Object(wm));
1176 }
1177 if flavor == Flavor::Orchestrator || profile.expiry != ExpiryPolicy::default() {
1178 out.insert(
1179 "expiry".into(),
1180 serde_json::to_value(&profile.expiry).unwrap(),
1181 );
1182 }
1183 if let Some(h) = &profile.home {
1184 out.insert("home".into(), encode_surface_key(h));
1185 }
1186 let mut gateway = profile
1187 .residue
1188 .config
1189 .get("gateway")
1190 .and_then(Value::as_object)
1191 .cloned()
1192 .unwrap_or_default();
1193 let routes: Vec<Value> = profile.routes.iter().map(encode_route).collect();
1194 if meta.is_some_and(|m| m.routes_at_top) {
1195 if !routes.is_empty() {
1196 out.insert("profile_routes".into(), Value::Array(routes));
1197 }
1198 } else if !routes.is_empty() {
1199 gateway.insert("profile_routes".into(), Value::Array(routes));
1200 }
1201 if !gateway.is_empty() {
1202 out.insert("gateway".into(), Value::Object(gateway));
1203 }
1204 let mut platforms = Map::new();
1205 for (p, ch) in &profile.channels {
1206 platforms.insert(
1207 p.clone(),
1208 encode_channel(
1209 ch,
1210 if flavor == Flavor::Hermes {
1211 vault
1212 } else {
1213 None
1214 },
1215 ),
1216 );
1217 }
1218 if !platforms.is_empty() {
1219 out.insert("platforms".into(), Value::Object(platforms));
1220 }
1221 json_to_yaml(&Value::Object(out))
1222}
1223
1224pub fn encode_jobs_file(profile: &Profile, meta: Option<&ProfileIo>) -> String {
1226 let jobs: Vec<Value> = profile
1227 .jobs
1228 .values()
1229 .map(|j| ordered_object(encode_job(j)))
1230 .collect();
1231 let form = meta.and_then(|m| m.jobs_form.clone()).unwrap_or(JobsForm {
1232 object: true,
1233 extras: Map::new(),
1234 });
1235 let body = if form.object {
1236 let mut pairs = vec![("jobs".to_string(), Value::Array(jobs))];
1237 pairs.extend(form.extras.iter().map(|(k, v)| (k.clone(), v.clone())));
1238 ordered_object(pairs)
1239 } else {
1240 Value::Array(jobs)
1241 };
1242 format!("{}\n", pretty_ordered(&body, 0))
1243}
1244
1245pub(crate) fn ordered_object(pairs: Vec<(String, Value)>) -> Value {
1248 Value::Array(vec![
1251 Value::String("__ordered__".into()),
1252 Value::Array(
1253 pairs
1254 .into_iter()
1255 .map(|(k, v)| serde_json::json!({"__k": k, "__v": v}))
1256 .collect(),
1257 ),
1258 ])
1259}
1260
1261fn is_ordered(value: &Value) -> Option<&Vec<Value>> {
1262 let arr = value.as_array()?;
1263 if arr.len() == 2 && arr[0].as_str() == Some("__ordered__") {
1264 arr[1].as_array()
1265 } else {
1266 None
1267 }
1268}
1269
1270pub(crate) fn pretty_ordered(value: &Value, depth: usize) -> String {
1272 let pad = |d: usize| " ".repeat(d);
1273 if let Some(pairs) = is_ordered(value) {
1274 if pairs.is_empty() {
1275 return "{}".into();
1276 }
1277 let inner: Vec<String> = pairs
1278 .iter()
1279 .map(|p| {
1280 format!(
1281 "{}{}: {}",
1282 pad(depth + 1),
1283 serde_json::to_string(p["__k"].as_str().unwrap_or("")).unwrap(),
1284 pretty_ordered(&p["__v"], depth + 1)
1285 )
1286 })
1287 .collect();
1288 return format!("{{\n{}\n{}}}", inner.join(",\n"), pad(depth));
1289 }
1290 match value {
1291 Value::Array(items) if items.is_empty() => "[]".into(),
1292 Value::Array(items) => {
1293 let inner: Vec<String> = items
1294 .iter()
1295 .map(|v| format!("{}{}", pad(depth + 1), pretty_ordered(v, depth + 1)))
1296 .collect();
1297 format!("[\n{}\n{}]", inner.join(",\n"), pad(depth))
1298 }
1299 Value::Object(o) if o.is_empty() => "{}".into(),
1300 Value::Object(o) => {
1301 let inner: Vec<String> = o
1302 .iter()
1303 .map(|(k, v)| {
1304 format!(
1305 "{}{}: {}",
1306 pad(depth + 1),
1307 serde_json::to_string(k).unwrap(),
1308 pretty_ordered(v, depth + 1)
1309 )
1310 })
1311 .collect();
1312 format!("{{\n{}\n{}}}", inner.join(",\n"), pad(depth))
1313 }
1314 Value::Number(n) => {
1315 if let Some(f) = n.as_f64() {
1316 if n.is_f64() && f.fract() == 0.0 && f.abs() < 1e21 {
1317 return format!("{}", f as i64);
1318 }
1319 }
1320 n.to_string()
1321 }
1322 other => serde_json::to_string(other).unwrap(),
1323 }
1324}
1325
1326pub fn encode_subscriptions_file(
1328 profile: &Profile,
1329 vault: Option<&BTreeMap<String, String>>,
1330) -> String {
1331 let mut out = Map::new();
1332 for (n, s) in &profile.subscriptions {
1333 out.insert(n.clone(), encode_subscription(s, vault));
1334 }
1335 format!("{}\n", pretty_ordered(&Value::Object(out), 0))
1336}
1337
1338pub fn encode_access_file(profile: &Profile) -> String {
1340 json_to_yaml(&encode_access(&profile.access))
1341}
1342
1343const EXECUTIONS_DDL: &str = "CREATE TABLE IF NOT EXISTS executions (
1344 id TEXT PRIMARY KEY, job_id TEXT NOT NULL, source TEXT NOT NULL, process_id TEXT NOT NULL, pid INTEGER NOT NULL,
1345 process_started_at INTEGER, status TEXT NOT NULL CHECK(status IN ('claimed','running','completed','failed','unknown')),
1346 claimed_at TEXT NOT NULL, started_at TEXT, finished_at TEXT, error TEXT);
1347CREATE INDEX IF NOT EXISTS idx_executions_job_claimed ON executions(job_id, claimed_at DESC, id DESC);
1348CREATE INDEX IF NOT EXISTS idx_executions_status_claimed ON executions(status, claimed_at DESC, id DESC);";
1349
1350const OBLIGATIONS_DDL: &str = "CREATE TABLE IF NOT EXISTS delivery_obligations (
1351 obligation_id TEXT PRIMARY KEY, session_key TEXT NOT NULL, platform TEXT NOT NULL, chat_id TEXT NOT NULL, thread_id TEXT,
1352 content TEXT NOT NULL, state TEXT NOT NULL, attempts INTEGER NOT NULL DEFAULT 0, created_at REAL NOT NULL, updated_at REAL NOT NULL,
1353 owner_pid INTEGER, owner_started_at INTEGER, last_error TEXT, adapter_profile TEXT);";
1354
1355const BINDINGS_DDL: &str = "CREATE TABLE IF NOT EXISTS bindings (
1356 slot TEXT PRIMARY KEY,
1357 platform TEXT NOT NULL, chat_type TEXT NOT NULL, chat_id TEXT, thread_id TEXT, participant_id TEXT,
1358 worker_harness TEXT NOT NULL, worker_session_id TEXT, worker_locator TEXT,
1359 started_at TEXT NOT NULL, last_activity_at TEXT NOT NULL, ended_at TEXT, end_reason TEXT,
1360 handoff_to TEXT, handoff_state TEXT, handoff_error TEXT, recurrence_job_id TEXT, residue_json TEXT);";
1361
1362pub fn write_executions(path: &Path, fires: &[crate::orchestration::Fire]) -> Result<()> {
1364 let insert = format!(
1365 "insert into executions ({}) values ({})",
1366 EXECUTION_COLUMNS.join(", "),
1367 EXECUTION_COLUMNS
1368 .iter()
1369 .map(|_| "?")
1370 .collect::<Vec<_>>()
1371 .join(",")
1372 );
1373 let rows: Vec<Vec<Param>> = fires
1374 .iter()
1375 .map(|f| encode_fire_row(f).iter().map(Param::from).collect())
1376 .collect();
1377 write_table(path, EXECUTIONS_DDL, &insert, &rows)
1378}
1379
1380fn write_state(path: &Path, profile: &Profile) -> Result<()> {
1381 let cols = [
1382 "slot",
1383 "platform",
1384 "chat_type",
1385 "chat_id",
1386 "thread_id",
1387 "participant_id",
1388 "worker_harness",
1389 "worker_session_id",
1390 "worker_locator",
1391 "started_at",
1392 "last_activity_at",
1393 "ended_at",
1394 "end_reason",
1395 "handoff_to",
1396 "handoff_state",
1397 "handoff_error",
1398 "recurrence_job_id",
1399 "residue_json",
1400 ];
1401 let insert = format!(
1402 "insert into bindings ({}) values ({})",
1403 cols.join(", "),
1404 cols.iter().map(|_| "?").collect::<Vec<_>>().join(",")
1405 );
1406 let s = |v: &Option<String>| v.clone().map(Param::Text).unwrap_or(Param::Null);
1407 let rows: Vec<Vec<Param>> = profile
1408 .bindings
1409 .iter()
1410 .map(|(slot, b)| {
1411 vec![
1412 Param::Text(slot.clone()),
1413 Param::Text(b.key.platform.clone().unwrap_or_default()),
1414 Param::Text(b.key.kind.clone().unwrap_or_default()),
1415 Param::Text(b.key.chat_id.clone().unwrap_or_default()),
1416 Param::Text(b.key.thread_id.clone().unwrap_or_default()),
1417 Param::Text(b.key.participant_id.clone().unwrap_or_default()),
1418 Param::Text(b.worker.harness.as_str().into()),
1419 s(&b.worker.session_id.clone().filter(|v| !v.is_empty())),
1420 s(&b.worker.locator),
1421 Param::Text(b.started_at.clone().unwrap_or_default()),
1422 Param::Text(b.last_activity_at.clone().unwrap_or_default()),
1423 s(&b.ended_at),
1424 b.end_reason
1425 .map(|r| Param::Text(r.as_str().into()))
1426 .unwrap_or(Param::Null),
1427 s(&b.handoff.as_ref().and_then(|h| h.to.clone())),
1428 b.handoff
1429 .as_ref()
1430 .map(|h| Param::Text(h.state.clone()))
1431 .unwrap_or(Param::Null),
1432 s(&b.handoff.as_ref().and_then(|h| h.error.clone())),
1433 s(&b.recurrence.as_ref().map(|r| r.job_id.clone())),
1434 if b.residue.is_empty() {
1435 Param::Null
1436 } else {
1437 Param::Text(serde_json::to_string(&b.residue).unwrap())
1438 },
1439 ]
1440 })
1441 .collect();
1442 write_table(
1443 path,
1444 &format!("{BINDINGS_DDL}\n{OBLIGATIONS_DDL}"),
1445 &insert,
1446 &rows,
1447 )?;
1448 let insert = format!(
1449 "insert into delivery_obligations ({}) values ({})",
1450 OBLIGATION_COLUMNS.join(", "),
1451 OBLIGATION_COLUMNS
1452 .iter()
1453 .map(|_| "?")
1454 .collect::<Vec<_>>()
1455 .join(",")
1456 );
1457 let rows: Vec<Vec<Param>> = profile
1458 .obligations
1459 .iter()
1460 .map(|o| encode_obligation_row(o).iter().map(Param::from).collect())
1461 .collect();
1462 write_table(path, "", &insert, &rows)
1463}
1464
1465fn write_if_changed(
1466 meta: &mut ProfileIo,
1467 dir: &Path,
1468 rel: &str,
1469 record: &Value,
1470 render: impl FnOnce() -> Option<String>,
1471) -> Result<bool> {
1472 let snap = canonical_json(record);
1473 let path = dir.join(rel);
1474 let reuse = meta.flavor == Flavor::Orchestrator && meta.snapshot.get(rel) == Some(&snap);
1477 if reuse && path.exists() {
1478 return Ok(false);
1479 }
1480 if reuse {
1481 if let Some(raw) = meta.raw.get(rel).cloned() {
1482 write_atomic(&path, &raw)?;
1483 return Ok(true);
1484 }
1485 }
1486 let Some(text) = render() else {
1487 return Ok(false);
1488 };
1489 write_atomic(&path, &text)?;
1490 meta.raw.insert(rel.into(), text);
1491 meta.snapshot.insert(rel.into(), snap);
1492 meta.flavor = Flavor::Orchestrator; Ok(true)
1494}
1495
1496pub fn save_home(loaded: &mut LoadedHome, root: Option<&Path>) -> Result<()> {
1498 let root = root
1499 .map(Path::to_path_buf)
1500 .unwrap_or_else(|| loaded.orchestration.root.clone());
1501 fs::create_dir_all(&root)?;
1502 let names: Vec<String> = loaded.orchestration.profiles.keys().cloned().collect();
1503 for name in names {
1504 let dir = if name == "default" {
1505 root.clone()
1506 } else {
1507 root.join("profiles").join(&name)
1508 };
1509 fs::create_dir_all(dir.join("cron"))?;
1510 let profile = loaded.orchestration.profiles[&name].clone();
1511 let meta = loaded
1512 .io
1513 .entry(name.clone())
1514 .or_insert_with(|| ProfileIo::new(Flavor::Orchestrator));
1515 save_profile_dir(&profile, meta, &dir, &loaded.vault)?;
1516 }
1517 Ok(())
1518}
1519
1520fn save_profile_dir(
1521 profile: &Profile,
1522 meta: &mut ProfileIo,
1523 dir: &Path,
1524 vault: &BTreeMap<String, String>,
1525) -> Result<()> {
1526 let cfg_record = config_record(profile);
1527 write_if_changed(meta, dir, "config.yaml", &cfg_record, || {
1528 Some(encode_config(
1529 profile,
1530 None,
1531 Some(vault),
1532 Flavor::Orchestrator,
1533 ))
1534 })?;
1535 if let Some(persona) = &profile.persona {
1536 let text = persona.text.clone().unwrap_or_default();
1537 write_if_changed(
1538 meta,
1539 dir,
1540 "AGENTS.md",
1541 &serde_json::to_value(&profile.persona).unwrap(),
1542 || Some(text),
1543 )?;
1544 if !dir.join("CLAUDE.md").exists() {
1545 write_atomic(&dir.join("CLAUDE.md"), "@AGENTS.md\n")?;
1546 }
1547 }
1548 let jobs_record: Vec<Value> = profile
1549 .jobs
1550 .values()
1551 .map(|j| serde_json::to_value(j).unwrap())
1552 .collect();
1553 let had_jobs = meta.raw.contains_key("cron/jobs.json");
1554 let form = meta.jobs_form.clone();
1555 write_if_changed(
1556 meta,
1557 dir,
1558 "cron/jobs.json",
1559 &Value::Array(jobs_record),
1560 || {
1561 if profile.jobs.is_empty() && !had_jobs {
1562 None
1563 } else {
1564 let stub = ProfileIo {
1565 jobs_form: form.clone(),
1566 ..ProfileIo::new(Flavor::Orchestrator)
1567 };
1568 Some(encode_jobs_file(profile, Some(&stub)))
1569 }
1570 },
1571 )?;
1572 let subs_record: Vec<Value> = profile
1573 .subscriptions
1574 .values()
1575 .map(|s| serde_json::to_value(s).unwrap())
1576 .collect();
1577 let had_subs = meta.raw.contains_key("webhook_subscriptions.json");
1578 write_if_changed(
1579 meta,
1580 dir,
1581 "webhook_subscriptions.json",
1582 &Value::Array(subs_record),
1583 || {
1584 if profile.subscriptions.is_empty() && !had_subs {
1585 None
1586 } else {
1587 Some(encode_subscriptions_file(profile, None))
1588 }
1589 },
1590 )?;
1591 let a = &profile.access;
1592 let access_empty = a.allowlist.is_empty()
1593 && a.admins.is_empty()
1594 && a.pending_pairings.is_empty()
1595 && a.policy.is_empty()
1596 && a.pairing_ttl_minutes.is_none();
1597 let had_access = meta.raw.contains_key("access.yaml");
1598 write_if_changed(
1599 meta,
1600 dir,
1601 "access.yaml",
1602 &serde_json::to_value(a).unwrap(),
1603 || {
1604 if access_empty && !had_access {
1605 None
1606 } else {
1607 Some(encode_access_file(profile))
1608 }
1609 },
1610 )?;
1611
1612 let fires_snap = canonical_json(&serde_json::to_value(&profile.fires).unwrap());
1614 let exec_path = dir.join("cron/executions.db");
1615 if (meta.snapshot.get("cron/executions.db") != Some(&fires_snap) || !exec_path.exists())
1616 && (!profile.fires.is_empty() || exec_path.exists())
1617 {
1618 let tmp = exec_path.with_file_name(format!("executions.db.tmp-{}", std::process::id()));
1619 let _ = fs::remove_file(&tmp);
1620 write_executions(&tmp, &profile.fires)?;
1621 fs::rename(&tmp, &exec_path)?;
1622 meta.snapshot
1623 .insert("cron/executions.db".into(), fires_snap);
1624 }
1625 let state_snap = canonical_json(&state_record(profile));
1626 let state_path = dir.join("state.db");
1627 if (meta.snapshot.get("state.db") != Some(&state_snap) || !state_path.exists())
1628 && (!profile.bindings.is_empty() || !profile.obligations.is_empty() || state_path.exists())
1629 {
1630 let tmp = state_path.with_file_name(format!("state.db.tmp-{}", std::process::id()));
1631 let _ = fs::remove_file(&tmp);
1632 write_state(&tmp, profile)?;
1633 fs::rename(&tmp, &state_path)?;
1634 meta.snapshot.insert("state.db".into(), state_snap);
1635 }
1636
1637 let mut refs: Vec<String> = Vec::new();
1639 for ch in profile.channels.values() {
1640 for r in ch.credentials.values() {
1641 if let crate::ontology::SecretRef::Dotenv(n) = r {
1642 refs.push(n.clone());
1643 }
1644 }
1645 }
1646 for s in profile.subscriptions.values() {
1647 if let Some(crate::ontology::SecretRef::Dotenv(n)) = &s.secret {
1648 refs.push(n.clone());
1649 }
1650 }
1651 if let Some(w) = &profile.worker {
1652 for v in w.env.values() {
1653 if let crate::orchestration::EnvValue::Secret(crate::ontology::SecretRef::Dotenv(n)) = v
1654 {
1655 refs.push(n.clone());
1656 }
1657 }
1658 }
1659 let mut entries: BTreeMap<String, String> = BTreeMap::new();
1660 for r in refs {
1661 if let Some(v) = vault.get(&r) {
1662 entries.insert(r, v.clone());
1663 }
1664 }
1665 if !entries.is_empty() {
1666 let existing = meta.raw.get(".env").cloned();
1667 let mut merged = existing.as_deref().map(parse_dotenv).unwrap_or_default();
1668 for (k, v) in entries {
1669 merged.insert(k, v);
1670 }
1671 let text = render_dotenv(&merged);
1672 if existing.as_deref() != Some(text.as_str()) {
1673 write_atomic(&dir.join(".env"), &text)?;
1674 meta.raw.insert(".env".into(), text);
1675 }
1676 }
1677 Ok(())
1678}
1679
1680pub fn copy_unmodeled(profile: &Profile, meta: &ProfileIo, dest: &Path) -> Result<()> {
1682 let Some(src) = &meta.source_dir else {
1683 return Ok(());
1684 };
1685 carry_unmodeled(&profile.residue.files, src, dest)?;
1686 Ok(())
1687}
1688
1689pub fn carry_unmodeled(files: &[String], src: &Path, dest: &Path) -> Result<Vec<String>> {
1693 let mut carried = Vec::new();
1694 for rel in files {
1695 let from = src.join(rel);
1696 if !from.is_file() {
1697 continue;
1698 }
1699 let to = dest.join(rel);
1700 if let Some(parent) = to.parent() {
1701 fs::create_dir_all(parent)?;
1702 }
1703 fs::copy(&from, &to)?;
1704 carried.push(rel.clone());
1705 }
1706 Ok(carried)
1707}