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