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 mut entries: Vec<_> = fs::read_dir(d)?.flatten().collect();
1095 entries.sort_by_key(|e| e.file_name());
1096 for entry in entries {
1097 let p = entry.path();
1098 let rel = p
1099 .strip_prefix(base)
1100 .unwrap_or(&p)
1101 .to_string_lossy()
1102 .replace('\\', "/");
1103 let name = entry.file_name().to_string_lossy().into_owned();
1104 if rel == "profiles"
1105 || name == "node_modules"
1106 || name == ".git"
1107 || rel.starts_with("state.db")
1108 || rel.starts_with("cron/executions.db")
1109 {
1110 continue;
1111 }
1112 if runtime.contains(&rel.as_str()) || regex_tmp(&name) {
1113 continue;
1114 }
1115 let st = fs::symlink_metadata(&p)?;
1116 if st.is_dir() {
1117 walk(base, &p, owned, runtime, out)?;
1118 continue;
1119 }
1120 if !st.is_file() {
1121 continue;
1122 }
1123 if owned.contains(&rel.as_str()) {
1124 continue;
1125 }
1126 out.push(rel);
1127 }
1128 Ok(())
1129 }
1130 walk(dir, dir, &owned, &runtime_artifacts, &mut out)?;
1131 Ok(out)
1132}
1133
1134fn regex_tmp(name: &str) -> bool {
1135 name.rsplit_once(".tmp-")
1137 .is_some_and(|(_, pid)| !pid.is_empty() && pid.chars().all(|c| c.is_ascii_digit()))
1138}
1139
1140fn write_atomic(path: &Path, text: &str) -> Result<()> {
1143 if let Some(parent) = path.parent() {
1144 fs::create_dir_all(parent)?;
1145 }
1146 let tmp = path.with_file_name(format!(
1147 "{}.tmp-{}",
1148 path.file_name().unwrap().to_string_lossy(),
1149 std::process::id()
1150 ));
1151 fs::write(&tmp, text)?;
1152 fs::rename(&tmp, path)?;
1153 Ok(())
1154}
1155
1156pub fn encode_config(
1158 profile: &Profile,
1159 meta: Option<&ProfileIo>,
1160 vault: Option<&BTreeMap<String, String>>,
1161 flavor: Flavor,
1162) -> String {
1163 let mut out = Map::new();
1164 for (k, v) in &profile.residue.config {
1165 if k != "gateway" {
1166 out.insert(k.clone(), v.clone());
1167 }
1168 }
1169 if let Some(w) = &profile.worker {
1170 let mut wm = Map::new();
1171 wm.insert("harness".into(), Value::String(w.harness.as_str().into()));
1172 if let Some(m) = &w.model {
1173 wm.insert("model".into(), Value::String(m.clone()));
1174 }
1175 if let Some(p) = &w.preset {
1176 wm.insert("preset".into(), Value::String(p.clone()));
1177 }
1178 if w.cwd != "." {
1179 wm.insert("cwd".into(), Value::String(w.cwd.clone()));
1180 }
1181 if !w.env.is_empty() {
1182 wm.insert("env".into(), serde_json::to_value(&w.env).unwrap());
1183 }
1184 if w.permission.timeout_seconds != 300
1185 || w.permission.default != crate::orchestration::PermissionDefault::Deny
1186 {
1187 wm.insert(
1188 "permission".into(),
1189 serde_json::to_value(&w.permission).unwrap(),
1190 );
1191 }
1192 out.insert("worker".into(), Value::Object(wm));
1193 }
1194 if flavor == Flavor::Orchestrator || profile.expiry != ExpiryPolicy::default() {
1195 out.insert(
1196 "expiry".into(),
1197 serde_json::to_value(&profile.expiry).unwrap(),
1198 );
1199 }
1200 if let Some(h) = &profile.home {
1201 out.insert("home".into(), encode_surface_key(h));
1202 }
1203 let mut gateway = profile
1204 .residue
1205 .config
1206 .get("gateway")
1207 .and_then(Value::as_object)
1208 .cloned()
1209 .unwrap_or_default();
1210 let routes: Vec<Value> = profile.routes.iter().map(encode_route).collect();
1211 if meta.is_some_and(|m| m.routes_at_top) {
1212 if !routes.is_empty() {
1213 out.insert("profile_routes".into(), Value::Array(routes));
1214 }
1215 } else if !routes.is_empty() {
1216 gateway.insert("profile_routes".into(), Value::Array(routes));
1217 }
1218 if !gateway.is_empty() {
1219 out.insert("gateway".into(), Value::Object(gateway));
1220 }
1221 let mut platforms = Map::new();
1222 for (p, ch) in &profile.channels {
1223 platforms.insert(
1224 p.clone(),
1225 encode_channel(
1226 ch,
1227 if flavor == Flavor::Hermes {
1228 vault
1229 } else {
1230 None
1231 },
1232 ),
1233 );
1234 }
1235 if !platforms.is_empty() {
1236 out.insert("platforms".into(), Value::Object(platforms));
1237 }
1238 json_to_yaml(&Value::Object(out))
1239}
1240
1241pub fn encode_jobs_file(profile: &Profile, meta: Option<&ProfileIo>) -> String {
1243 let jobs: Vec<Value> = profile
1244 .jobs
1245 .values()
1246 .map(|j| ordered_object(encode_job(j)))
1247 .collect();
1248 let form = meta.and_then(|m| m.jobs_form.clone()).unwrap_or(JobsForm {
1249 object: true,
1250 extras: Map::new(),
1251 });
1252 let body = if form.object {
1253 let mut pairs = vec![("jobs".to_string(), Value::Array(jobs))];
1254 pairs.extend(form.extras.iter().map(|(k, v)| (k.clone(), v.clone())));
1255 ordered_object(pairs)
1256 } else {
1257 Value::Array(jobs)
1258 };
1259 format!("{}\n", pretty_ordered(&body, 0))
1260}
1261
1262pub(crate) fn ordered_object(pairs: Vec<(String, Value)>) -> Value {
1265 Value::Array(vec![
1268 Value::String("__ordered__".into()),
1269 Value::Array(
1270 pairs
1271 .into_iter()
1272 .map(|(k, v)| serde_json::json!({"__k": k, "__v": v}))
1273 .collect(),
1274 ),
1275 ])
1276}
1277
1278fn is_ordered(value: &Value) -> Option<&Vec<Value>> {
1279 let arr = value.as_array()?;
1280 if arr.len() == 2 && arr[0].as_str() == Some("__ordered__") {
1281 arr[1].as_array()
1282 } else {
1283 None
1284 }
1285}
1286
1287pub(crate) fn pretty_ordered(value: &Value, depth: usize) -> String {
1289 let pad = |d: usize| " ".repeat(d);
1290 if let Some(pairs) = is_ordered(value) {
1291 if pairs.is_empty() {
1292 return "{}".into();
1293 }
1294 let inner: Vec<String> = pairs
1295 .iter()
1296 .map(|p| {
1297 format!(
1298 "{}{}: {}",
1299 pad(depth + 1),
1300 serde_json::to_string(p["__k"].as_str().unwrap_or("")).unwrap(),
1301 pretty_ordered(&p["__v"], depth + 1)
1302 )
1303 })
1304 .collect();
1305 return format!("{{\n{}\n{}}}", inner.join(",\n"), pad(depth));
1306 }
1307 match value {
1308 Value::Array(items) if items.is_empty() => "[]".into(),
1309 Value::Array(items) => {
1310 let inner: Vec<String> = items
1311 .iter()
1312 .map(|v| format!("{}{}", pad(depth + 1), pretty_ordered(v, depth + 1)))
1313 .collect();
1314 format!("[\n{}\n{}]", inner.join(",\n"), pad(depth))
1315 }
1316 Value::Object(o) if o.is_empty() => "{}".into(),
1317 Value::Object(o) => {
1318 let inner: Vec<String> = o
1319 .iter()
1320 .map(|(k, v)| {
1321 format!(
1322 "{}{}: {}",
1323 pad(depth + 1),
1324 serde_json::to_string(k).unwrap(),
1325 pretty_ordered(v, depth + 1)
1326 )
1327 })
1328 .collect();
1329 format!("{{\n{}\n{}}}", inner.join(",\n"), pad(depth))
1330 }
1331 Value::Number(n) => {
1332 if let Some(f) = n.as_f64() {
1333 if n.is_f64() && f.fract() == 0.0 && f.abs() < 1e21 {
1334 return format!("{}", f as i64);
1335 }
1336 }
1337 n.to_string()
1338 }
1339 other => serde_json::to_string(other).unwrap(),
1340 }
1341}
1342
1343pub fn encode_subscriptions_file(
1345 profile: &Profile,
1346 vault: Option<&BTreeMap<String, String>>,
1347) -> String {
1348 let mut out = Map::new();
1349 for (n, s) in &profile.subscriptions {
1350 out.insert(n.clone(), encode_subscription(s, vault));
1351 }
1352 format!("{}\n", pretty_ordered(&Value::Object(out), 0))
1353}
1354
1355pub fn encode_access_file(profile: &Profile) -> String {
1357 json_to_yaml(&encode_access(&profile.access))
1358}
1359
1360const EXECUTIONS_DDL: &str = "CREATE TABLE IF NOT EXISTS executions (
1361 id TEXT PRIMARY KEY, job_id TEXT NOT NULL, source TEXT NOT NULL, process_id TEXT NOT NULL, pid INTEGER NOT NULL,
1362 process_started_at INTEGER, status TEXT NOT NULL CHECK(status IN ('claimed','running','completed','failed','unknown')),
1363 claimed_at TEXT NOT NULL, started_at TEXT, finished_at TEXT, error TEXT);
1364CREATE INDEX IF NOT EXISTS idx_executions_job_claimed ON executions(job_id, claimed_at DESC, id DESC);
1365CREATE INDEX IF NOT EXISTS idx_executions_status_claimed ON executions(status, claimed_at DESC, id DESC);";
1366const FOLDER_EXECUTIONS_DDL: &str = "CREATE TABLE IF NOT EXISTS executions (
1371 id TEXT PRIMARY KEY, job_id TEXT NOT NULL, source TEXT NOT NULL, process_id TEXT NOT NULL, pid INTEGER NOT NULL,
1372 process_started_at INTEGER, status TEXT NOT NULL CHECK(status IN ('claimed','running','completed','failed','unknown')),
1373 claimed_at TEXT NOT NULL, started_at TEXT, finished_at TEXT, error TEXT, residue_json TEXT,
1374 session_id TEXT, obligation_id TEXT);
1375CREATE INDEX IF NOT EXISTS idx_executions_job_claimed ON executions(job_id, claimed_at DESC, id DESC);
1376CREATE INDEX IF NOT EXISTS idx_executions_status_claimed ON executions(status, claimed_at DESC, id DESC);";
1377
1378const OBLIGATIONS_DDL: &str = "CREATE TABLE IF NOT EXISTS delivery_obligations (
1379 obligation_id TEXT PRIMARY KEY, session_key TEXT NOT NULL, platform TEXT NOT NULL, chat_id TEXT NOT NULL, thread_id TEXT,
1380 content TEXT NOT NULL, state TEXT NOT NULL, attempts INTEGER NOT NULL DEFAULT 0, created_at REAL NOT NULL, updated_at REAL NOT NULL,
1381 owner_pid INTEGER, owner_started_at INTEGER, last_error TEXT, adapter_profile TEXT,
1382 posted_message_id TEXT, source_json TEXT);";
1383
1384const BINDINGS_DDL: &str = "CREATE TABLE IF NOT EXISTS bindings (
1385 slot TEXT PRIMARY KEY,
1386 platform TEXT NOT NULL, chat_type TEXT NOT NULL, chat_id TEXT, thread_id TEXT, participant_id TEXT,
1387 worker_harness TEXT NOT NULL, worker_session_id TEXT, worker_locator TEXT,
1388 started_at TEXT NOT NULL, last_activity_at TEXT NOT NULL, ended_at TEXT, end_reason TEXT,
1389 handoff_to TEXT, handoff_state TEXT, handoff_error TEXT, recurrence_job_id TEXT, residue_json TEXT);";
1390
1391pub fn write_executions(path: &Path, fires: &[crate::orchestration::Fire]) -> Result<()> {
1393 write_executions_shaped(path, fires, false)
1394}
1395
1396pub fn write_executions_shaped(
1399 path: &Path,
1400 fires: &[crate::orchestration::Fire],
1401 ours: bool,
1402) -> Result<()> {
1403 let mut cols: Vec<&str> = EXECUTION_COLUMNS.to_vec();
1404 if ours {
1405 cols.extend(["residue_json", "session_id", "obligation_id"]);
1408 }
1409 let insert = format!(
1410 "insert into executions ({}) values ({})",
1411 cols.join(", "),
1412 cols.iter().map(|_| "?").collect::<Vec<_>>().join(",")
1413 );
1414 let rows: Vec<Vec<Param>> = fires
1415 .iter()
1416 .map(|f| {
1417 let mut row: Vec<Param> = encode_fire_row(f).iter().map(Param::from).collect();
1418 if ours {
1419 let rest: serde_json::Map<String, Value> = f
1421 .residue
1422 .0
1423 .iter()
1424 .filter(|(k, _)| {
1425 !["source", "process_id", "pid", "process_started_at"].contains(&k.as_str())
1426 })
1427 .map(|(k, v)| (k.clone(), v.clone()))
1428 .collect();
1429 row.push(if rest.is_empty() {
1430 Param::Null
1431 } else {
1432 Param::Text(serde_json::to_string(&rest).unwrap())
1433 });
1434 let opt = |v: &Option<String>| v.clone().map(Param::Text).unwrap_or(Param::Null);
1435 row.push(opt(&f.session_id));
1436 row.push(opt(&f.obligation_id));
1437 }
1438 row
1439 })
1440 .collect();
1441 write_table(
1442 path,
1443 if ours {
1444 FOLDER_EXECUTIONS_DDL
1445 } else {
1446 EXECUTIONS_DDL
1447 },
1448 &insert,
1449 &rows,
1450 )
1451}
1452
1453fn write_state(path: &Path, profile: &Profile) -> Result<()> {
1454 let cols = [
1455 "slot",
1456 "platform",
1457 "chat_type",
1458 "chat_id",
1459 "thread_id",
1460 "participant_id",
1461 "worker_harness",
1462 "worker_session_id",
1463 "worker_locator",
1464 "started_at",
1465 "last_activity_at",
1466 "ended_at",
1467 "end_reason",
1468 "handoff_to",
1469 "handoff_state",
1470 "handoff_error",
1471 "recurrence_job_id",
1472 "residue_json",
1473 ];
1474 let insert = format!(
1475 "insert into bindings ({}) values ({})",
1476 cols.join(", "),
1477 cols.iter().map(|_| "?").collect::<Vec<_>>().join(",")
1478 );
1479 let s = |v: &Option<String>| v.clone().map(Param::Text).unwrap_or(Param::Null);
1480 let rows: Vec<Vec<Param>> = profile
1481 .bindings
1482 .iter()
1483 .map(|(slot, b)| {
1484 vec![
1485 Param::Text(slot.clone()),
1486 Param::Text(b.key.platform.clone().unwrap_or_default()),
1487 Param::Text(b.key.kind.clone().unwrap_or_default()),
1488 Param::Text(b.key.chat_id.clone().unwrap_or_default()),
1489 Param::Text(b.key.thread_id.clone().unwrap_or_default()),
1490 Param::Text(b.key.participant_id.clone().unwrap_or_default()),
1491 Param::Text(b.worker.harness.as_str().into()),
1492 s(&b.worker.session_id.clone().filter(|v| !v.is_empty())),
1493 s(&b.worker.locator),
1494 Param::Text(b.started_at.clone().unwrap_or_default()),
1495 Param::Text(b.last_activity_at.clone().unwrap_or_default()),
1496 s(&b.ended_at),
1497 b.end_reason
1498 .map(|r| Param::Text(r.as_str().into()))
1499 .unwrap_or(Param::Null),
1500 s(&b.handoff.as_ref().and_then(|h| h.to.clone())),
1501 b.handoff
1502 .as_ref()
1503 .map(|h| Param::Text(h.state.clone()))
1504 .unwrap_or(Param::Null),
1505 s(&b.handoff.as_ref().and_then(|h| h.error.clone())),
1506 s(&b.recurrence.as_ref().map(|r| r.job_id.clone())),
1507 if b.residue.is_empty() {
1508 Param::Null
1509 } else {
1510 Param::Text(serde_json::to_string(&b.residue).unwrap())
1511 },
1512 ]
1513 })
1514 .collect();
1515 write_table(
1516 path,
1517 &format!("{BINDINGS_DDL}\n{OBLIGATIONS_DDL}"),
1518 &insert,
1519 &rows,
1520 )?;
1521 let cols: Vec<&str> = OBLIGATION_COLUMNS
1523 .iter()
1524 .chain(FOLDER_OBLIGATION_EXTRA_COLUMNS.iter())
1525 .copied()
1526 .collect();
1527 let insert = format!(
1528 "insert into delivery_obligations ({}) values ({})",
1529 cols.join(", "),
1530 cols.iter().map(|_| "?").collect::<Vec<_>>().join(",")
1531 );
1532 let rows: Vec<Vec<Param>> = profile
1533 .obligations
1534 .iter()
1535 .map(|o| {
1536 encode_obligation_row(o)
1537 .iter()
1538 .chain(encode_obligation_folder_extras(o).iter())
1539 .map(Param::from)
1540 .collect()
1541 })
1542 .collect();
1543 write_table(path, "", &insert, &rows)
1544}
1545
1546fn write_if_changed(
1547 meta: &mut ProfileIo,
1548 dir: &Path,
1549 rel: &str,
1550 record: &Value,
1551 render: impl FnOnce() -> Option<String>,
1552) -> Result<bool> {
1553 let snap = canonical_json(record);
1554 let path = dir.join(rel);
1555 let reuse = meta.flavor == Flavor::Orchestrator && meta.snapshot.get(rel) == Some(&snap);
1558 if reuse && path.exists() {
1559 return Ok(false);
1560 }
1561 if reuse {
1562 if let Some(raw) = meta.raw.get(rel).cloned() {
1563 write_atomic(&path, &raw)?;
1564 return Ok(true);
1565 }
1566 }
1567 let Some(text) = render() else {
1568 return Ok(false);
1569 };
1570 write_atomic(&path, &text)?;
1571 meta.raw.insert(rel.into(), text);
1572 meta.snapshot.insert(rel.into(), snap);
1573 meta.flavor = Flavor::Orchestrator; Ok(true)
1575}
1576
1577pub fn save_home(loaded: &mut LoadedHome, root: Option<&Path>) -> Result<()> {
1579 let root = root
1580 .map(Path::to_path_buf)
1581 .unwrap_or_else(|| loaded.orchestration.root.clone());
1582 fs::create_dir_all(&root)?;
1583 let names: Vec<String> = loaded.orchestration.profiles.keys().cloned().collect();
1584 for name in names {
1585 let dir = if name == "default" {
1586 root.clone()
1587 } else {
1588 root.join("profiles").join(&name)
1589 };
1590 fs::create_dir_all(dir.join("cron"))?;
1591 let profile = loaded.orchestration.profiles[&name].clone();
1592 let meta = loaded
1593 .io
1594 .entry(name.clone())
1595 .or_insert_with(|| ProfileIo::new(Flavor::Orchestrator));
1596 save_profile_dir(&profile, meta, &dir, &loaded.vault)?;
1597 }
1598 Ok(())
1599}
1600
1601fn save_profile_dir(
1602 profile: &Profile,
1603 meta: &mut ProfileIo,
1604 dir: &Path,
1605 vault: &BTreeMap<String, String>,
1606) -> Result<()> {
1607 let cfg_record = config_record(profile);
1608 write_if_changed(meta, dir, "config.yaml", &cfg_record, || {
1609 Some(encode_config(
1610 profile,
1611 None,
1612 Some(vault),
1613 Flavor::Orchestrator,
1614 ))
1615 })?;
1616 if let Some(persona) = &profile.persona {
1617 let text = persona.text.clone().unwrap_or_default();
1618 write_if_changed(
1619 meta,
1620 dir,
1621 "AGENTS.md",
1622 &serde_json::to_value(&profile.persona).unwrap(),
1623 || Some(text),
1624 )?;
1625 if !dir.join("CLAUDE.md").exists() {
1626 write_atomic(&dir.join("CLAUDE.md"), "@AGENTS.md\n")?;
1627 }
1628 }
1629 let jobs_record: Vec<Value> = profile
1630 .jobs
1631 .values()
1632 .map(|j| serde_json::to_value(j).unwrap())
1633 .collect();
1634 let had_jobs = meta.raw.contains_key("cron/jobs.json");
1635 let form = meta.jobs_form.clone();
1636 write_if_changed(
1637 meta,
1638 dir,
1639 "cron/jobs.json",
1640 &Value::Array(jobs_record),
1641 || {
1642 if profile.jobs.is_empty() && !had_jobs {
1643 None
1644 } else {
1645 let stub = ProfileIo {
1646 jobs_form: form.clone(),
1647 ..ProfileIo::new(Flavor::Orchestrator)
1648 };
1649 Some(encode_jobs_file(profile, Some(&stub)))
1650 }
1651 },
1652 )?;
1653 let subs_record: Vec<Value> = profile
1654 .subscriptions
1655 .values()
1656 .map(|s| serde_json::to_value(s).unwrap())
1657 .collect();
1658 let had_subs = meta.raw.contains_key("webhook_subscriptions.json");
1659 write_if_changed(
1660 meta,
1661 dir,
1662 "webhook_subscriptions.json",
1663 &Value::Array(subs_record),
1664 || {
1665 if profile.subscriptions.is_empty() && !had_subs {
1666 None
1667 } else {
1668 Some(encode_subscriptions_file(profile, None))
1669 }
1670 },
1671 )?;
1672 let a = &profile.access;
1673 let access_empty = a.allowlist.is_empty()
1674 && a.admins.is_empty()
1675 && a.pending_pairings.is_empty()
1676 && a.policy.is_empty()
1677 && a.pairing_ttl_minutes.is_none();
1678 let had_access = meta.raw.contains_key("access.yaml");
1679 write_if_changed(
1680 meta,
1681 dir,
1682 "access.yaml",
1683 &serde_json::to_value(a).unwrap(),
1684 || {
1685 if access_empty && !had_access {
1686 None
1687 } else {
1688 Some(encode_access_file(profile))
1689 }
1690 },
1691 )?;
1692
1693 let fires_snap = canonical_json(&serde_json::to_value(&profile.fires).unwrap());
1695 let exec_path = dir.join("cron/executions.db");
1696 if (meta.snapshot.get("cron/executions.db") != Some(&fires_snap) || !exec_path.exists())
1697 && (!profile.fires.is_empty() || exec_path.exists())
1698 {
1699 let tmp = exec_path.with_file_name(format!("executions.db.tmp-{}", std::process::id()));
1700 let _ = fs::remove_file(&tmp);
1701 write_executions_shaped(&tmp, &profile.fires, true)?;
1702 fs::rename(&tmp, &exec_path)?;
1703 meta.snapshot
1704 .insert("cron/executions.db".into(), fires_snap);
1705 }
1706 let state_snap = canonical_json(&state_record(profile));
1707 let state_path = dir.join("state.db");
1708 if (meta.snapshot.get("state.db") != Some(&state_snap) || !state_path.exists())
1709 && (!profile.bindings.is_empty() || !profile.obligations.is_empty() || state_path.exists())
1710 {
1711 let tmp = state_path.with_file_name(format!("state.db.tmp-{}", std::process::id()));
1712 let _ = fs::remove_file(&tmp);
1713 write_state(&tmp, profile)?;
1714 fs::rename(&tmp, &state_path)?;
1715 meta.snapshot.insert("state.db".into(), state_snap);
1716 }
1717
1718 let mut refs: Vec<String> = Vec::new();
1720 for ch in profile.channels.values() {
1721 for r in ch.credentials.values() {
1722 if let crate::ontology::SecretRef::Dotenv(n) = r {
1723 refs.push(n.clone());
1724 }
1725 }
1726 }
1727 for s in profile.subscriptions.values() {
1728 if let Some(crate::ontology::SecretRef::Dotenv(n)) = &s.secret {
1729 refs.push(n.clone());
1730 }
1731 }
1732 if let Some(w) = &profile.worker {
1733 for v in w.env.values() {
1734 if let crate::orchestration::EnvValue::Secret(crate::ontology::SecretRef::Dotenv(n)) = v
1735 {
1736 refs.push(n.clone());
1737 }
1738 }
1739 }
1740 let mut entries: BTreeMap<String, String> = BTreeMap::new();
1741 for r in refs {
1742 if let Some(v) = vault.get(&r) {
1743 entries.insert(r, v.clone());
1744 }
1745 }
1746 if !entries.is_empty() {
1747 let existing = meta.raw.get(".env").cloned();
1748 let mut merged = existing.as_deref().map(parse_dotenv).unwrap_or_default();
1749 for (k, v) in entries {
1750 merged.insert(k, v);
1751 }
1752 let text = render_dotenv(&merged);
1753 if existing.as_deref() != Some(text.as_str()) {
1754 write_atomic(&dir.join(".env"), &text)?;
1755 meta.raw.insert(".env".into(), text);
1756 }
1757 }
1758 Ok(())
1759}
1760
1761pub fn copy_unmodeled(profile: &Profile, meta: &ProfileIo, dest: &Path) -> Result<()> {
1763 let Some(src) = &meta.source_dir else {
1764 return Ok(());
1765 };
1766 carry_unmodeled(&profile.residue.files, src, dest)?;
1767 Ok(())
1768}
1769
1770pub fn carry_unmodeled(files: &[String], src: &Path, dest: &Path) -> Result<Vec<String>> {
1774 let mut carried = Vec::new();
1775 for rel in files {
1776 let from = src.join(rel);
1777 if !from.is_file() {
1778 continue;
1779 }
1780 let to = dest.join(rel);
1781 if let Some(parent) = to.parent() {
1782 fs::create_dir_all(parent)?;
1783 }
1784 fs::copy(&from, &to)?;
1785 carried.push(rel.clone());
1786 }
1787 Ok(carried)
1788}