1use std::collections::BTreeMap;
6
7use serde_json::{Map, Value};
8
9use crate::ontology::Recurrence;
10use crate::ontology::{
11 Binding, EndReason, Handoff, HarnessId, Residue, SecretRef, SurfaceKey, Worker,
12};
13use crate::orchestration::{
14 Access, AccessPolicy, ChannelConfig, EnvValue, ExpiryPolicy, ExpiryScope, Fire, FireStatus,
15 Job, JobOrigin, Obligation, ObligationSource, ObligationState, OutboundContent, PendingPairing,
16 PermissionDefault, PermissionPolicy, Posted, Repeat, Route, RouteMatch, Schedule, Target,
17 WebhookSubscription, WorkerSpec,
18};
19use crate::Result;
20
21#[derive(Debug, Clone, PartialEq, Eq)]
23pub struct LoadError {
24 pub file: String,
26 pub key: Option<String>,
28 pub message: String,
30}
31
32impl std::fmt::Display for LoadError {
33 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
34 match &self.key {
35 Some(key) => write!(f, "{} [{key}]: {}", self.file, self.message),
36 None => write!(f, "{}: {}", self.file, self.message),
37 }
38 }
39}
40
41impl From<LoadError> for crate::Error {
42 fn from(error: LoadError) -> Self {
43 crate::Error::Other(error.to_string())
44 }
45}
46
47pub(crate) fn load_error(file: &str, key: &str, message: impl Into<String>) -> crate::Error {
48 LoadError {
49 file: file.to_string(),
50 key: if key.is_empty() {
51 None
52 } else {
53 Some(key.to_string())
54 },
55 message: message.into(),
56 }
57 .into()
58}
59
60fn expect_keys(file: &str, key: &str, obj: &Value, allowed: &[&str]) -> Result<()> {
61 let Some(map) = obj.as_object() else {
62 return Err(load_error(file, key, "expected an object"));
63 };
64 for k in map.keys() {
65 if !allowed.contains(&k.as_str()) {
66 let path = if key.is_empty() {
67 k.clone()
68 } else {
69 format!("{key}.{k}")
70 };
71 return Err(load_error(file, &path, "unknown key"));
72 }
73 }
74 Ok(())
75}
76
77fn req_str(file: &str, key: &str, v: Option<&Value>) -> Result<String> {
78 match v {
79 Some(Value::String(s)) if !s.is_empty() => Ok(s.clone()),
80 _ => Err(load_error(file, key, "expected a non-empty string")),
81 }
82}
83
84fn opt_str(file: &str, key: &str, v: Option<&Value>) -> Result<Option<String>> {
85 match v {
86 None | Some(Value::Null) => Ok(None),
87 Some(Value::String(s)) => Ok(Some(s.clone())),
88 _ => Err(load_error(file, key, "expected a string")),
89 }
90}
91
92fn opt_bool(
93 file: &str,
94 key: &str,
95 v: Option<&Value>,
96 default: Option<bool>,
97) -> Result<Option<bool>> {
98 match v {
99 None | Some(Value::Null) => Ok(default),
100 Some(Value::Bool(b)) => Ok(Some(*b)),
101 _ => Err(load_error(file, key, "expected a boolean")),
102 }
103}
104
105fn residue_of(obj: &Map<String, Value>, mapped: &[&str]) -> Residue {
106 let mut out = Residue::default();
107 for (k, v) in obj {
108 if !mapped.contains(&k.as_str()) {
109 out.keep(k.clone(), v.clone());
110 }
111 }
112 out
113}
114
115fn row_residue_of(row: &Map<String, Value>, mapped: &[&str]) -> Residue {
117 let mut out = Residue::default();
118 for (k, v) in row {
119 if !mapped.contains(&k.as_str()) && !v.is_null() {
120 out.keep(k.clone(), v.clone());
121 }
122 }
123 out
124}
125
126fn text_of(v: &Value) -> Option<String> {
128 match v {
129 Value::String(s) => Some(s.clone()),
130 Value::Number(n) => Some(match n.as_f64() {
131 Some(f) if n.is_f64() && f.fract() == 0.0 && f.abs() < 1e21 => format!("{}", f as i64),
132 _ => n.to_string(),
133 }),
134 _ => None,
135 }
136}
137
138fn number_of(v: &Value) -> Option<f64> {
139 match v {
140 Value::Number(n) => n.as_f64(),
141 Value::String(s) => s.parse().ok(),
142 _ => None,
143 }
144}
145
146pub const CHAT_TYPES: &[&str] = &["dm", "group", "channel", "thread"];
150
151pub fn surface_key_string(key: &SurfaceKey) -> String {
153 format!(
154 "{}|{}|{}|{}|{}",
155 key.platform.as_deref().unwrap_or(""),
156 key.kind.as_deref().unwrap_or(""),
157 key.chat_id.as_deref().unwrap_or(""),
158 key.thread_id.as_deref().unwrap_or(""),
159 key.participant_id.as_deref().unwrap_or("")
160 )
161}
162
163pub fn decode_surface_key(file: &str, key: &str, raw: &Value) -> Result<SurfaceKey> {
165 let map = raw
166 .as_object()
167 .ok_or_else(|| load_error(file, key, "expected an object"))?;
168 for k in map.keys() {
169 if ![
170 "platform",
171 "chat_type",
172 "kind",
173 "chat_id",
174 "thread_id",
175 "participant_id",
176 "key",
177 ]
178 .contains(&k.as_str())
179 {
180 return Err(load_error(file, &format!("{key}.{k}"), "unknown key"));
181 }
182 }
183 let chat_type = map
184 .get("chat_type")
185 .or_else(|| map.get("kind"))
186 .and_then(Value::as_str)
187 .map(str::to_string);
188 if !chat_type
189 .as_deref()
190 .is_some_and(|c| CHAT_TYPES.contains(&c))
191 {
192 return Err(load_error(
193 file,
194 key,
195 "expected platform, chat_type, chat_id",
196 ));
197 }
198 Ok(SurfaceKey {
199 key: None,
200 platform: Some(req_str(
201 file,
202 &format!("{key}.platform"),
203 map.get("platform"),
204 )?),
205 kind: chat_type,
206 chat_id: map.get("chat_id").and_then(text_of),
207 thread_id: map
208 .get("thread_id")
209 .and_then(text_of)
210 .filter(|s| !s.is_empty()),
211 participant_id: map
212 .get("participant_id")
213 .and_then(text_of)
214 .filter(|s| !s.is_empty()),
215 })
216}
217
218pub fn encode_surface_key(key: &SurfaceKey) -> Value {
220 let mut out = Map::new();
221 out.insert(
222 "platform".into(),
223 Value::String(key.platform.clone().unwrap_or_default()),
224 );
225 out.insert(
226 "kind".into(),
227 Value::String(key.kind.clone().unwrap_or_default()),
228 );
229 if let Some(c) = &key.chat_id {
230 out.insert("chat_id".into(), Value::String(c.clone()));
231 }
232 if let Some(t) = &key.thread_id {
233 out.insert("thread_id".into(), Value::String(t.clone()));
234 }
235 if let Some(p) = &key.participant_id {
236 out.insert("participant_id".into(), Value::String(p.clone()));
237 }
238 Value::Object(out)
239}
240
241pub fn parse_target(
245 file: &str,
246 key: &str,
247 word: Option<&Value>,
248 extra: Option<&Value>,
249) -> Result<Option<Target>> {
250 let word = match word {
251 None | Some(Value::Null) => return Ok(None),
252 Some(Value::String(s)) if !s.is_empty() => s.as_str(),
253 _ => return Err(load_error(file, key, "expected a delivery word")),
254 };
255 Ok(Some(match word {
256 "origin" => Target::Origin,
257 "local" => Target::Local,
258 "home" => Target::Home,
259 _ => {
260 let parts: Vec<&str> = word.split(':').collect();
261 let extra_get = |name: &str| extra.and_then(|e| e.get(name)).and_then(text_of);
262 let chat_id = parts
263 .get(1)
264 .map(|s| s.to_string())
265 .or_else(|| extra_get("chat_id"))
266 .filter(|s| !s.is_empty());
267 let thread_id = parts
268 .get(2)
269 .map(|s| s.to_string())
270 .or_else(|| extra_get("thread_id"))
271 .filter(|s| !s.is_empty());
272 Target::Explicit {
273 platform: parts[0].to_string(),
274 chat_id,
275 thread_id,
276 }
277 }
278 }))
279}
280
281pub fn decode_expiry(file: &str, raw: Option<&Value>) -> Result<ExpiryPolicy> {
285 let Some(raw) = raw.filter(|v| !v.is_null()) else {
286 return Ok(ExpiryPolicy::default());
287 };
288 expect_keys(
289 file,
290 "expiry",
291 raw,
292 &["idle_minutes", "daily_reset_hour", "scope"],
293 )?;
294 let idle = match raw.get("idle_minutes") {
295 None => 1440,
296 Some(Value::Number(n)) if n.as_u64().is_some_and(|v| v >= 1) => n.as_u64().unwrap() as u32,
297 _ => {
298 return Err(load_error(
299 file,
300 "expiry.idle_minutes",
301 "expected an integer >= 1",
302 ))
303 }
304 };
305 let hour = match raw.get("daily_reset_hour") {
306 None | Some(Value::Null) => None,
307 Some(Value::Number(n)) if n.as_u64().is_some_and(|v| v <= 23) => {
308 Some(n.as_u64().unwrap() as u8)
309 }
310 _ => {
311 return Err(load_error(
312 file,
313 "expiry.daily_reset_hour",
314 "expected null or an integer 0..23",
315 ))
316 }
317 };
318 let scope = match raw.get("scope").and_then(Value::as_str) {
319 None => ExpiryScope::PerChat,
320 Some("per_chat") => ExpiryScope::PerChat,
321 Some("per_user_in_group") => ExpiryScope::PerUserInGroup,
322 Some("per_thread") => ExpiryScope::PerThread,
323 _ => {
324 return Err(load_error(
325 file,
326 "expiry.scope",
327 "expected one of per_chat|per_user_in_group|per_thread",
328 ))
329 }
330 };
331 Ok(ExpiryPolicy {
332 idle_minutes: idle,
333 daily_reset_hour: hour,
334 scope,
335 })
336}
337
338pub fn decode_env_value(file: &str, key: &str, v: &Value) -> Result<EnvValue> {
340 match v {
341 Value::String(s) => Ok(EnvValue::Literal(s.clone())),
342 Value::Object(_) => {
343 expect_keys(file, key, v, &["env", "dotenv"])?;
344 if let Some(Value::String(n)) = v.get("env") {
345 return Ok(EnvValue::Secret(SecretRef::Env(n.clone())));
346 }
347 if let Some(Value::String(n)) = v.get("dotenv") {
348 return Ok(EnvValue::Secret(SecretRef::Dotenv(n.clone())));
349 }
350 Err(load_error(
351 file,
352 key,
353 "expected a string or {env}|{dotenv} ref",
354 ))
355 }
356 _ => Err(load_error(
357 file,
358 key,
359 "expected a string or {env}|{dotenv} ref",
360 )),
361 }
362}
363
364pub fn decode_worker(file: &str, raw: Option<&Value>) -> Result<Option<WorkerSpec>> {
366 let Some(raw) = raw.filter(|v| !v.is_null()) else {
367 return Ok(None);
368 };
369 expect_keys(
370 file,
371 "worker",
372 raw,
373 &["harness", "model", "preset", "cwd", "env", "permission"],
374 )?;
375 let cwd = match raw.get("cwd") {
376 None => ".".to_string(),
377 v => req_str(file, "worker.cwd", v)?,
378 };
379 if cwd.starts_with('/') || cwd.split('/').any(|p| p == "..") {
380 return Err(load_error(
381 file,
382 "worker.cwd",
383 "must be a relative path inside the profile",
384 ));
385 }
386 let mut env = BTreeMap::new();
387 if let Some(e) = raw.get("env") {
388 let map = e
389 .as_object()
390 .ok_or_else(|| load_error(file, "worker.env", "expected a map"))?;
391 for (k, v) in map {
392 env.insert(
393 k.clone(),
394 decode_env_value(file, &format!("worker.env.{k}"), v)?,
395 );
396 }
397 }
398 let mut permission = PermissionPolicy::default();
399 if let Some(p) = raw.get("permission") {
400 expect_keys(
401 file,
402 "worker.permission",
403 p,
404 &["timeout_seconds", "default"],
405 )?;
406 if let Some(t) = p.get("timeout_seconds") {
407 permission.timeout_seconds = t.as_u64().filter(|v| *v >= 1).ok_or_else(|| {
408 load_error(
409 file,
410 "worker.permission.timeout_seconds",
411 "expected an integer >= 1",
412 )
413 })? as u32;
414 }
415 if let Some(d) = p.get("default") {
416 permission.default = match d.as_str() {
417 Some("deny") => PermissionDefault::Deny,
418 Some("allow") => PermissionDefault::Allow,
419 _ => {
420 return Err(load_error(
421 file,
422 "worker.permission.default",
423 "expected deny|allow",
424 ))
425 }
426 };
427 }
428 }
429 Ok(Some(WorkerSpec {
430 harness: HarnessId::new(req_str(file, "worker.harness", raw.get("harness"))?),
431 model: opt_str(file, "worker.model", raw.get("model"))?,
432 preset: opt_str(file, "worker.preset", raw.get("preset"))?,
433 cwd,
434 env,
435 permission,
436 }))
437}
438
439pub fn decode_home(file: &str, raw: Option<&Value>) -> Result<Option<SurfaceKey>> {
441 let Some(raw) = raw.filter(|v| !v.is_null()) else {
442 return Ok(None);
443 };
444 let map = raw
445 .as_object()
446 .ok_or_else(|| load_error(file, "home", "expected an object"))?;
447 for k in map.keys() {
448 if !["platform", "kind", "chat_type", "chat_id", "thread_id"].contains(&k.as_str()) {
449 return Err(load_error(file, &format!("home.{k}"), "unknown key"));
450 }
451 }
452 let platform = map.get("platform").and_then(Value::as_str);
453 let chat_type = map
454 .get("kind")
455 .or_else(|| map.get("chat_type"))
456 .and_then(Value::as_str);
457 let chat_id = map.get("chat_id").and_then(Value::as_str);
458 let (Some(platform), Some(chat_type), Some(chat_id)) = (platform, chat_type, chat_id) else {
459 return Err(load_error(
460 file,
461 "home",
462 "expected platform, chat_type, chat_id",
463 ));
464 };
465 if !CHAT_TYPES.contains(&chat_type) {
466 return Err(load_error(
467 file,
468 "home",
469 "expected platform, chat_type, chat_id",
470 ));
471 }
472 Ok(Some(SurfaceKey {
473 key: None,
474 platform: Some(platform.to_string()),
475 kind: Some(chat_type.to_string()),
476 chat_id: Some(chat_id.to_string()),
477 thread_id: map
478 .get("thread_id")
479 .and_then(text_of)
480 .filter(|s| !s.is_empty()),
481 participant_id: None,
482 }))
483}
484
485pub fn decode_access(file: &str, raw: Option<&Value>) -> Result<Access> {
487 let mut access = Access::default();
488 let Some(raw) = raw.filter(|v| !v.is_null()) else {
489 return Ok(access);
490 };
491 expect_keys(
492 file,
493 "",
494 raw,
495 &[
496 "allowlist",
497 "admins",
498 "pending_pairings",
499 "policy",
500 "pairing_ttl_minutes",
501 ],
502 )?;
503 if let Some(ttl) = raw.get("pairing_ttl_minutes").filter(|v| !v.is_null()) {
504 access.pairing_ttl_minutes =
505 Some(ttl.as_u64().filter(|v| *v >= 1).ok_or_else(|| {
506 load_error(file, "pairing_ttl_minutes", "expected an integer >= 1")
507 })? as u32);
508 }
509 let set_map = |name: &str, into: &mut BTreeMap<String, Vec<String>>| -> Result<()> {
510 let Some(m) = raw.get(name) else {
511 return Ok(());
512 };
513 let map = m
514 .as_object()
515 .ok_or_else(|| load_error(file, name, "expected a map platform -> list"))?;
516 for (platform, users) in map {
517 let list = users
518 .as_array()
519 .filter(|a| a.iter().all(Value::is_string))
520 .ok_or_else(|| {
521 load_error(
522 file,
523 &format!("{name}.{platform}"),
524 "expected a list of user ids",
525 )
526 })?;
527 let mut ids: Vec<String> = list
528 .iter()
529 .filter_map(|v| v.as_str().map(str::to_string))
530 .collect();
531 ids.sort();
532 ids.dedup();
533 into.insert(platform.clone(), ids);
534 }
535 Ok(())
536 };
537 set_map("allowlist", &mut access.allowlist)?;
538 set_map("admins", &mut access.admins)?;
539 if let Some(p) = raw.get("pending_pairings") {
540 let map = p
541 .as_object()
542 .ok_or_else(|| load_error(file, "pending_pairings", "expected a map code -> record"))?;
543 for (code, rec) in map {
544 let k = format!("pending_pairings.{code}");
545 expect_keys(file, &k, rec, &["platform", "user_id", "issued_at"])?;
546 access.pending_pairings.insert(
547 code.clone(),
548 PendingPairing {
549 platform: req_str(file, &format!("{k}.platform"), rec.get("platform"))?,
550 user_id: req_str(file, &format!("{k}.user_id"), rec.get("user_id"))?,
551 issued_at: req_str(file, &format!("{k}.issued_at"), rec.get("issued_at"))?,
552 },
553 );
554 }
555 }
556 if let Some(p) = raw.get("policy") {
557 let map = p.as_object().ok_or_else(|| {
558 load_error(file, "policy", "expected a map platform -> allowlist|open")
559 })?;
560 for (platform, word) in map {
561 let policy = match word.as_str() {
562 Some("allowlist") => AccessPolicy::Allowlist,
563 Some("open") => AccessPolicy::Open,
564 _ => {
565 return Err(load_error(
566 file,
567 &format!("policy.{platform}"),
568 "expected allowlist|open",
569 ))
570 }
571 };
572 access.policy.insert(platform.clone(), policy);
573 }
574 }
575 Ok(access)
576}
577
578pub fn encode_access(access: &Access) -> Value {
580 let mut out = serde_json::to_value(access).unwrap();
581 if access.pairing_ttl_minutes.is_none() {
582 if let Some(map) = out.as_object_mut() {
583 map.remove("pairing_ttl_minutes");
584 }
585 }
586 out
587}
588
589pub fn decode_binding_row(file: &str, row: &Map<String, Value>) -> Result<Binding> {
591 let text = |k: &str| row.get(k).and_then(text_of).filter(|s| !s.is_empty());
592 let chat_type = req_str(file, "chat_type", row.get("chat_type"))?;
593 if !CHAT_TYPES.contains(&chat_type.as_str()) {
594 return Err(load_error(
595 file,
596 "chat_type",
597 format!("expected one of {}", CHAT_TYPES.join("|")),
598 ));
599 }
600 let key = SurfaceKey {
601 key: None,
602 platform: Some(req_str(file, "platform", row.get("platform"))?),
603 kind: Some(chat_type),
604 chat_id: text("chat_id"),
605 thread_id: text("thread_id"),
606 participant_id: text("participant_id"),
607 };
608 let end_reason = match text("end_reason") {
609 None => None,
610 Some(word) => Some(
611 EndReason::parse(&word)
612 .ok_or_else(|| load_error(file, "end_reason", "unknown end reason"))?,
613 ),
614 };
615 let handoff = if text("handoff_to").is_some() || text("handoff_state").is_some() {
616 Some(Handoff {
617 to: text("handoff_to"),
618 state: text("handoff_state").unwrap_or_default(),
619 error: text("handoff_error"),
620 })
621 } else {
622 None
623 };
624 let recurrence = text("recurrence_job_id").map(|job_id| Recurrence {
625 job_id,
626 kind: "cron".into(),
627 });
628 let residue = match row.get("residue_json").and_then(Value::as_str) {
629 Some(json) if !json.is_empty() => serde_json::from_str::<Value>(json)
630 .ok()
631 .and_then(|v| v.as_object().cloned())
632 .map(|m| Residue(m.into_iter().collect()))
633 .unwrap_or_default(),
634 _ => Residue::default(),
635 };
636 Ok(Binding {
637 trigger: if recurrence.is_some() {
638 crate::ontology::Trigger::Cron
639 } else if key.platform.as_deref() == Some("webhook") {
640 crate::ontology::Trigger::Webhook
641 } else {
642 crate::ontology::Trigger::Channel
643 },
644 key,
645 profile: None,
646 worker: Worker {
647 harness: HarnessId::new(req_str(file, "worker_harness", row.get("worker_harness"))?),
648 session_id: opt_str(file, "worker_session_id", row.get("worker_session_id"))?
651 .filter(|s| !s.is_empty()),
652 locator: opt_str(file, "worker_locator", row.get("worker_locator"))?,
653 },
654 recurrence,
655 handoff,
656 started_at: Some(req_str(file, "started_at", row.get("started_at"))?),
657 last_activity_at: Some(req_str(
658 file,
659 "last_activity_at",
660 row.get("last_activity_at"),
661 )?),
662 ended_at: opt_str(file, "ended_at", row.get("ended_at"))?,
663 end_reason,
664 residue,
665 })
666}
667
668const JOB_MAPPED: &[&str] = &[
671 "id",
672 "schedule",
673 "prompt",
674 "workdir",
675 "model",
676 "skills",
677 "context_from",
678 "deliver",
679 "failure_deliver",
680 "origin",
681 "attach_to_session",
682 "repeat",
683 "enabled",
684 "next_run_at",
685 "last_run_at",
686 "last_status",
687 "created_at",
688];
689
690pub const HERMES_JOB_ORDER: &[&str] = &[
692 "id",
693 "schedule",
694 "prompt",
695 "skills",
696 "script",
697 "no_agent",
698 "model",
699 "provider",
700 "workdir",
701 "enabled_toolsets",
702 "context_from",
703 "deliver",
704 "failure_deliver",
705 "attach_to_session",
706 "origin",
707 "repeat",
708 "enabled",
709 "next_run_at",
710 "last_run_at",
711 "last_status",
712 "created_at",
713 "fire_claim",
714];
715
716pub fn decode_context_from(
718 file: &str,
719 key: &str,
720 v: Option<&Value>,
721) -> Result<Option<Vec<String>>> {
722 let items: Vec<String> = match v {
723 None | Some(Value::Null) => return Ok(None),
724 Some(Value::String(s)) => vec![s.clone()],
725 Some(Value::Array(a)) => a
726 .iter()
727 .map(|x| match x {
728 Value::String(s) => s.clone(),
729 other => other.to_string(),
730 })
731 .collect(),
732 _ => {
733 return Err(load_error(
734 file,
735 key,
736 "expected a job id, a list of job ids, or \"self\"",
737 ))
738 }
739 };
740 let refs: Vec<String> = items
741 .into_iter()
742 .map(|s| s.trim().to_string())
743 .filter(|s| !s.is_empty())
744 .collect();
745 Ok(if refs.is_empty() { None } else { Some(refs) })
746}
747
748pub fn decode_repeat(file: &str, key: &str, raw: Option<&Value>) -> Result<Option<Repeat>> {
750 Ok(match raw {
751 None | Some(Value::Null) => None,
752 Some(Value::Bool(true)) => Some(Repeat {
753 times: None,
754 completed: 0,
755 }),
756 Some(Value::Bool(false)) => Some(Repeat {
757 times: Some(1),
758 completed: 0,
759 }),
760 Some(Value::Number(n)) => Some(Repeat {
761 times: n.as_f64().filter(|f| *f > 0.0).map(|f| f.floor() as u32),
762 completed: 0,
763 }),
764 Some(Value::Object(map)) => {
765 let times = match map.get("times") {
766 None | Some(Value::Null) => None,
767 Some(Value::Number(n)) if n.as_u64().is_some_and(|v| v >= 1) => {
768 Some(n.as_u64().unwrap() as u32)
769 }
770 _ => {
771 return Err(load_error(
772 file,
773 &format!("{key}.times"),
774 "expected null or an integer >= 1",
775 ))
776 }
777 };
778 let completed = match map.get("completed") {
779 None => 0,
780 Some(Value::Number(n)) if n.as_u64().is_some() => n.as_u64().unwrap() as u32,
781 _ => {
782 return Err(load_error(
783 file,
784 &format!("{key}.completed"),
785 "expected an integer >= 0",
786 ))
787 }
788 };
789 Some(Repeat { times, completed })
790 }
791 _ => {
792 return Err(load_error(
793 file,
794 key,
795 "expected null, {times, completed}, a number or a boolean",
796 ))
797 }
798 })
799}
800
801pub fn decode_job(file: &str, raw: &Value) -> Result<Job> {
803 let map = raw
804 .as_object()
805 .ok_or_else(|| load_error(file, "", "expected a job object"))?;
806 let id = req_str(file, "id", map.get("id"))?;
807 let k = |s: &str| format!("{id}.{s}");
808 let sched = map
809 .get("schedule")
810 .and_then(Value::as_object)
811 .ok_or_else(|| load_error(file, &k("schedule"), "expected an object"))?;
812 let schedule = match sched.get("kind").and_then(Value::as_str) {
813 Some("once") => Schedule::Once {
814 run_at: req_str(file, &k("schedule.run_at"), sched.get("run_at"))?,
815 },
816 Some("interval") => {
817 let minutes = sched
818 .get("minutes")
819 .and_then(Value::as_f64)
820 .filter(|m| *m > 0.0)
821 .ok_or_else(|| {
822 load_error(file, &k("schedule.minutes"), "expected a positive number")
823 })?;
824 Schedule::Interval { minutes }
825 }
826 Some("cron") => Schedule::Cron {
827 expr: req_str(file, &k("schedule.expr"), sched.get("expr"))?,
828 tz: opt_str(file, &k("schedule.tz"), sched.get("tz"))?.unwrap_or_else(|| "UTC".into()),
829 },
830 _ => {
831 return Err(load_error(
832 file,
833 &k("schedule.kind"),
834 "expected once|interval|cron",
835 ))
836 }
837 };
838 let schedule_residue = residue_of(sched, &["kind", "run_at", "minutes", "expr", "tz"]);
839 let mut residue = residue_of(map, JOB_MAPPED);
840 if !schedule_residue.is_empty() {
841 residue.keep(
842 "__schedule",
843 Value::Object(schedule_residue.0.into_iter().collect()),
844 );
845 }
846 let origin = match map.get("origin") {
847 Some(Value::Object(o)) => {
848 let r = residue_of(o, &["platform", "chat_id", "thread_id"]);
849 if !r.is_empty() {
850 residue.keep("__origin", Value::Object(r.0.into_iter().collect()));
851 }
852 Some(JobOrigin {
853 platform: req_str(file, &k("origin.platform"), o.get("platform"))?,
854 chat_type: None,
855 chat_id: opt_str(file, &k("origin.chat_id"), o.get("chat_id"))?,
856 thread_id: opt_str(file, &k("origin.thread_id"), o.get("thread_id"))?,
857 })
858 }
859 _ => None,
860 };
861 Ok(Job {
862 schedule,
863 prompt: opt_str(file, &k("prompt"), map.get("prompt"))?,
864 workdir: opt_str(file, &k("workdir"), map.get("workdir"))?,
865 model: opt_str(file, &k("model"), map.get("model"))?,
866 skills: map
867 .get("skills")
868 .and_then(Value::as_array)
869 .map(|a| {
870 a.iter()
871 .map(|v| {
872 v.as_str()
873 .map(str::to_string)
874 .unwrap_or_else(|| v.to_string())
875 })
876 .collect()
877 })
878 .unwrap_or_default(),
879 context_from: decode_context_from(file, &k("context_from"), map.get("context_from"))?,
880 deliver: parse_target(file, &k("deliver"), map.get("deliver"), None)?
881 .unwrap_or(Target::Local),
882 failure_deliver: parse_target(
883 file,
884 &k("failure_deliver"),
885 map.get("failure_deliver"),
886 None,
887 )?,
888 origin,
889 attach_to_session: opt_bool(
890 file,
891 &k("attach_to_session"),
892 map.get("attach_to_session"),
893 None,
894 )?,
895 repeat: decode_repeat(file, &k("repeat"), map.get("repeat"))?,
896 enabled: opt_bool(file, &k("enabled"), map.get("enabled"), Some(true))?.unwrap_or(true),
897 next_run_at: opt_str(file, &k("next_run_at"), map.get("next_run_at"))?,
898 last_run_at: opt_str(file, &k("last_run_at"), map.get("last_run_at"))?,
899 last_status: opt_str(file, &k("last_status"), map.get("last_status"))?,
900 created_at: opt_str(file, &k("created_at"), map.get("created_at"))?,
901 residue,
902 id,
903 })
904}
905
906pub fn encode_job(job: &Job) -> Vec<(String, Value)> {
909 let mut sched = Map::new();
910 sched.insert("kind".into(), Value::String(job.schedule.kind().into()));
911 match &job.schedule {
912 Schedule::Once { run_at } => {
913 sched.insert("run_at".into(), Value::String(run_at.clone()));
914 }
915 Schedule::Interval { minutes } => {
916 sched.insert("minutes".into(), serde_json::json!(*minutes));
917 }
918 Schedule::Cron { expr, tz } => {
919 sched.insert("expr".into(), Value::String(expr.clone()));
920 if tz != "UTC" {
921 sched.insert("tz".into(), Value::String(tz.clone()));
922 }
923 }
924 }
925 if let Some(Value::Object(extra)) = job.residue.0.get("__schedule") {
926 for (k, v) in extra {
927 sched.insert(k.clone(), v.clone());
928 }
929 }
930 let origin = job
931 .origin
932 .as_ref()
933 .map(|o| {
934 let mut m = Map::new();
935 m.insert("platform".into(), Value::String(o.platform.clone()));
936 m.insert(
937 "chat_id".into(),
938 o.chat_id.clone().map(Value::String).unwrap_or(Value::Null),
939 );
940 m.insert(
941 "thread_id".into(),
942 o.thread_id
943 .clone()
944 .map(Value::String)
945 .unwrap_or(Value::Null),
946 );
947 if let Some(Value::Object(extra)) = job.residue.0.get("__origin") {
948 for (k, v) in extra {
949 m.insert(k.clone(), v.clone());
950 }
951 }
952 Value::Object(m)
953 })
954 .unwrap_or(Value::Null);
955 let opt = |s: &Option<String>| s.clone().map(Value::String).unwrap_or(Value::Null);
956 let mut mapped: Vec<(String, Value)> = vec![
957 ("id".into(), Value::String(job.id.clone())),
958 ("schedule".into(), Value::Object(sched)),
959 ("prompt".into(), opt(&job.prompt)),
960 (
961 "skills".into(),
962 Value::Array(
963 job.skills
964 .iter()
965 .map(|s| Value::String(s.clone()))
966 .collect(),
967 ),
968 ),
969 ("model".into(), opt(&job.model)),
970 ("workdir".into(), opt(&job.workdir)),
971 (
972 "context_from".into(),
973 job.context_from
974 .as_ref()
975 .map(|l| Value::Array(l.iter().map(|s| Value::String(s.clone())).collect()))
976 .unwrap_or(Value::Null),
977 ),
978 ("deliver".into(), Value::String(job.deliver.render())),
979 (
980 "failure_deliver".into(),
981 job.failure_deliver
982 .as_ref()
983 .map(|t| Value::String(t.render()))
984 .unwrap_or(Value::Null),
985 ),
986 (
987 "attach_to_session".into(),
988 job.attach_to_session
989 .map(Value::Bool)
990 .unwrap_or(Value::Null),
991 ),
992 ("origin".into(), origin),
993 (
994 "repeat".into(),
995 job.repeat
996 .as_ref()
997 .map(|r| serde_json::json!({"times": r.times, "completed": r.completed}))
998 .unwrap_or(Value::Null),
999 ),
1000 ("enabled".into(), Value::Bool(job.enabled)),
1001 ("next_run_at".into(), opt(&job.next_run_at)),
1002 ("last_run_at".into(), opt(&job.last_run_at)),
1003 ("last_status".into(), opt(&job.last_status)),
1004 ("created_at".into(), opt(&job.created_at)),
1005 ];
1006 let mut residue: Vec<(String, Value)> = job
1007 .residue
1008 .0
1009 .iter()
1010 .filter(|(k, _)| k.as_str() != "__schedule" && k.as_str() != "__origin")
1016 .map(|(k, v)| (k.clone(), v.clone()))
1017 .collect();
1018 let mut out: Vec<(String, Value)> = Vec::new();
1019 for key in HERMES_JOB_ORDER {
1020 if let Some(pos) = mapped.iter().position(|(k, _)| k == key) {
1021 out.push(mapped.remove(pos));
1022 } else if let Some(pos) = residue.iter().position(|(k, _)| k == key) {
1023 out.push(residue.remove(pos));
1024 }
1025 }
1026 out.extend(mapped);
1027 out.extend(residue);
1028 out
1029}
1030
1031pub fn decode_fire_row(file: &str, row: &Map<String, Value>) -> Result<Fire> {
1033 let id = req_str(file, "id", row.get("id"))?;
1034 let status_word = row.get("status").and_then(Value::as_str).unwrap_or("");
1035 let status = FireStatus::from_hermes_word(status_word).ok_or_else(|| {
1036 load_error(
1037 file,
1038 &format!("{id}.status"),
1039 format!(
1040 "unknown fire status {}",
1041 serde_json::to_string(status_word).unwrap()
1042 ),
1043 )
1044 })?;
1045 Ok(Fire {
1046 job_id: req_str(file, &format!("{id}.job_id"), row.get("job_id"))?,
1047 session_id: row
1048 .get("session_id")
1049 .and_then(text_of)
1050 .filter(|s| !s.is_empty()),
1051 status,
1052 claimed_at: req_str(file, &format!("{id}.claimed_at"), row.get("claimed_at"))?,
1053 started_at: opt_str(file, &format!("{id}.started_at"), row.get("started_at"))?,
1054 finished_at: opt_str(file, &format!("{id}.finished_at"), row.get("finished_at"))?,
1055 error: opt_str(file, &format!("{id}.error"), row.get("error"))?,
1056 obligation_id: row
1057 .get("obligation_id")
1058 .and_then(text_of)
1059 .filter(|s| !s.is_empty()),
1060 residue: {
1061 let mut residue = row_residue_of(
1062 row,
1063 &[
1064 "id",
1065 "job_id",
1066 "status",
1067 "claimed_at",
1068 "started_at",
1069 "finished_at",
1070 "error",
1071 "residue_json",
1072 "session_id",
1073 "obligation_id",
1074 ],
1075 );
1076 if let Some(extra) = row
1079 .get("residue_json")
1080 .and_then(text_of)
1081 .and_then(|t| serde_json::from_str::<Map<String, Value>>(&t).ok())
1082 {
1083 for (k, v) in extra {
1084 residue.keep(k, v);
1085 }
1086 }
1087 residue
1088 },
1089 id,
1090 })
1091}
1092
1093pub fn encode_fire_row(fire: &Fire) -> Vec<Value> {
1095 let r = &fire.residue.0;
1096 let opt = |s: &Option<String>| s.clone().map(Value::String).unwrap_or(Value::Null);
1097 vec![
1098 Value::String(fire.id.clone()),
1099 Value::String(fire.job_id.clone()),
1100 r.get("source")
1101 .cloned()
1102 .unwrap_or(Value::String("scheduler".into())),
1103 r.get("process_id")
1104 .cloned()
1105 .unwrap_or(Value::String(String::new())),
1106 r.get("pid").cloned().unwrap_or(Value::from(0)),
1107 r.get("process_started_at").cloned().unwrap_or(Value::Null),
1108 Value::String(fire.status.hermes_word().into()),
1109 Value::String(fire.claimed_at.clone()),
1110 opt(&fire.started_at),
1111 opt(&fire.finished_at),
1112 opt(&fire.error),
1113 ]
1114}
1115
1116pub const EXECUTION_COLUMNS: &[&str] = &[
1118 "id",
1119 "job_id",
1120 "source",
1121 "process_id",
1122 "pid",
1123 "process_started_at",
1124 "status",
1125 "claimed_at",
1126 "started_at",
1127 "finished_at",
1128 "error",
1129];
1130
1131pub fn decode_obligation_row(file: &str, row: &Map<String, Value>) -> Result<Obligation> {
1133 let id = req_str(file, "obligation_id", row.get("obligation_id"))?;
1134 let state_word = row.get("state").and_then(Value::as_str).unwrap_or("");
1135 let state = ObligationState::from_hermes_word(state_word).ok_or_else(|| {
1136 load_error(
1137 file,
1138 &format!("{id}.state"),
1139 format!(
1140 "unknown obligation state {}",
1141 serde_json::to_string(state_word).unwrap()
1142 ),
1143 )
1144 })?;
1145 let session_key = row
1146 .get("session_key")
1147 .and_then(Value::as_str)
1148 .filter(|s| !s.is_empty())
1149 .map(str::to_string);
1150 let parsed = session_key
1151 .as_deref()
1152 .and_then(crate::ontology::parse_hermes_session_key)
1153 .map(|(k, _)| k);
1154 let target = SurfaceKey {
1155 key: None,
1156 platform: Some(req_str(
1157 file,
1158 &format!("{id}.platform"),
1159 row.get("platform"),
1160 )?),
1161 kind: parsed.as_ref().and_then(|k| k.kind.clone()),
1162 chat_id: Some(row.get("chat_id").and_then(text_of).unwrap_or_default()),
1163 thread_id: row
1164 .get("thread_id")
1165 .and_then(text_of)
1166 .filter(|s| !s.is_empty()),
1167 participant_id: None,
1168 };
1169 let text = |k: &str| row.get(k).and_then(text_of);
1170 let created_at = text("created_at").unwrap_or_else(|| "undefined".into());
1171 let updated_at = text("updated_at").unwrap_or_else(|| created_at.clone());
1172 Ok(Obligation {
1173 target,
1174 session_key,
1175 content: OutboundContent {
1176 text: row.get("content").and_then(text_of).unwrap_or_default(),
1177 attachments: None,
1178 reply_to: None,
1179 format: None,
1180 },
1181 state,
1182 attempts: row.get("attempts").and_then(number_of).unwrap_or(0.0) as u64,
1183 last_error: row.get("last_error").and_then(text_of),
1184 delivered_at: if state == ObligationState::Sent {
1185 Some(updated_at.clone())
1186 } else {
1187 None
1188 },
1189 created_at,
1190 updated_at,
1191 posted: row
1195 .get("posted_message_id")
1196 .and_then(text_of)
1197 .filter(|s| !s.is_empty())
1198 .map(|message_id| Posted { message_id }),
1199 source: row
1200 .get("source_json")
1201 .and_then(text_of)
1202 .and_then(|text| serde_json::from_str::<ObligationSource>(&text).ok())
1203 .unwrap_or(ObligationSource::Turn { key: None }),
1204 residue: row_residue_of(
1205 row,
1206 &[
1207 "obligation_id",
1208 "session_key",
1209 "platform",
1210 "chat_id",
1211 "thread_id",
1212 "content",
1213 "state",
1214 "attempts",
1215 "created_at",
1216 "updated_at",
1217 "last_error",
1218 "posted_message_id",
1219 "source_json",
1220 ],
1221 ),
1222 id,
1223 })
1224}
1225
1226pub const FOLDER_OBLIGATION_EXTRA_COLUMNS: &[&str] = &["posted_message_id", "source_json"];
1228
1229pub fn encode_obligation_folder_extras(o: &Obligation) -> Vec<Value> {
1231 vec![
1232 o.posted
1233 .as_ref()
1234 .map(|p| Value::String(p.message_id.clone()))
1235 .unwrap_or(Value::Null),
1236 Value::String(serde_json::to_string(&o.source).unwrap()),
1237 ]
1238}
1239
1240pub fn encode_obligation_row(o: &Obligation) -> Vec<Value> {
1242 let r = &o.residue.0;
1243 let num = |s: &str| {
1244 s.parse::<f64>()
1245 .map(|f| serde_json::json!(f))
1246 .unwrap_or(Value::from(0))
1247 };
1248 vec![
1249 Value::String(o.id.clone()),
1250 Value::String(o.session_key.clone().unwrap_or_default()),
1251 Value::String(o.target.platform.clone().unwrap_or_default()),
1252 Value::String(o.target.chat_id.clone().unwrap_or_default()),
1253 o.target
1254 .thread_id
1255 .clone()
1256 .map(Value::String)
1257 .unwrap_or(Value::Null),
1258 Value::String(o.content.text.clone()),
1259 Value::String(o.state.hermes_word().into()),
1260 Value::from(o.attempts),
1261 num(&o.created_at),
1262 num(o
1263 .delivered_at
1264 .as_deref()
1265 .map(|_| o.updated_at.as_str())
1266 .unwrap_or(&o.updated_at)),
1267 r.get("owner_pid").cloned().unwrap_or(Value::Null),
1268 r.get("owner_started_at").cloned().unwrap_or(Value::Null),
1269 o.last_error
1270 .clone()
1271 .map(Value::String)
1272 .unwrap_or(Value::Null),
1273 r.get("adapter_profile").cloned().unwrap_or(Value::Null),
1274 ]
1275}
1276
1277pub const OBLIGATION_COLUMNS: &[&str] = &[
1279 "obligation_id",
1280 "session_key",
1281 "platform",
1282 "chat_id",
1283 "thread_id",
1284 "content",
1285 "state",
1286 "attempts",
1287 "created_at",
1288 "updated_at",
1289 "owner_pid",
1290 "owner_started_at",
1291 "last_error",
1292 "adapter_profile",
1293];
1294
1295const ROUTE_MATCH: &[&str] = &["platform", "guild_id", "chat_id", "thread_id"];
1296
1297pub fn decode_route(file: &str, index: usize, raw: &Value) -> Result<Route> {
1299 let map = raw.as_object().ok_or_else(|| {
1300 load_error(
1301 file,
1302 &format!("profile_routes[{index}]"),
1303 "expected an object",
1304 )
1305 })?;
1306 let text = |k: &str| map.get(k).filter(|v| !v.is_null()).and_then(text_of);
1307 Ok(Route {
1308 name: opt_str(
1309 file,
1310 &format!("profile_routes[{index}].name"),
1311 map.get("name"),
1312 )?,
1313 matches: RouteMatch {
1314 platform: req_str(
1315 file,
1316 &format!("profile_routes[{index}].platform"),
1317 map.get("platform"),
1318 )?,
1319 guild_id: text("guild_id"),
1320 chat_id: text("chat_id"),
1321 thread_id: text("thread_id"),
1322 },
1323 profile: req_str(
1324 file,
1325 &format!("profile_routes[{index}].profile"),
1326 map.get("profile"),
1327 )?,
1328 residue: residue_of(
1329 map,
1330 &[
1331 "name",
1332 "profile",
1333 "platform",
1334 "guild_id",
1335 "chat_id",
1336 "thread_id",
1337 ],
1338 ),
1339 })
1340}
1341
1342pub fn encode_route(r: &Route) -> Value {
1344 let mut out = Map::new();
1345 if let Some(n) = &r.name {
1346 out.insert("name".into(), Value::String(n.clone()));
1347 }
1348 let m = &r.matches;
1349 for (k, v) in [
1350 ("platform", Some(&m.platform)),
1351 ("guild_id", m.guild_id.as_ref()),
1352 ("chat_id", m.chat_id.as_ref()),
1353 ("thread_id", m.thread_id.as_ref()),
1354 ] {
1355 if let Some(v) = v {
1356 out.insert(k.into(), Value::String(v.clone()));
1357 }
1358 }
1359 let _ = ROUTE_MATCH;
1360 out.insert("profile".into(), Value::String(r.profile.clone()));
1361 for (k, v) in &r.residue.0 {
1362 out.insert(k.clone(), v.clone());
1363 }
1364 Value::Object(out)
1365}
1366
1367pub fn is_credential_key(key: &str) -> bool {
1369 let k = key.to_ascii_lowercase();
1370 [
1371 "token",
1372 "secret",
1373 "key",
1374 "password",
1375 "passwd",
1376 "api_key",
1377 "app_secret",
1378 "signing",
1379 "webhook_url",
1380 "private",
1381 ]
1382 .iter()
1383 .any(|w| k.contains(w))
1384}
1385
1386pub fn placeholder_ref(text: &str) -> Option<&str> {
1393 let inner = text.strip_prefix("${")?.strip_suffix('}')?;
1394 inner
1395 .strip_prefix("dotenv:")
1396 .or_else(|| inner.strip_prefix("env:"))
1397 .filter(|name| !name.is_empty())
1398}
1399
1400pub fn credential_ref_name(platform: &str, key: &str) -> String {
1402 format!("{platform}_{key}")
1403 .chars()
1404 .map(|c| {
1405 if c.is_ascii_alphanumeric() {
1406 c.to_ascii_uppercase()
1407 } else {
1408 '_'
1409 }
1410 })
1411 .collect()
1412}
1413
1414pub fn decode_channel(
1416 file: &str,
1417 platform: &str,
1418 raw: &Value,
1419 vault: &mut BTreeMap<String, String>,
1420) -> Result<ChannelConfig> {
1421 let map = raw
1422 .as_object()
1423 .ok_or_else(|| load_error(file, &format!("platforms.{platform}"), "expected an object"))?;
1424 let mut ch = ChannelConfig {
1425 platform: platform.to_string(),
1426 enabled: opt_bool(
1427 file,
1428 &format!("platforms.{platform}.enabled"),
1429 map.get("enabled"),
1430 Some(true),
1431 )?
1432 .unwrap_or(true),
1433 credentials: BTreeMap::new(),
1434 extra: BTreeMap::new(),
1435 };
1436 fn walk(
1437 map: &Map<String, Value>,
1438 prefix: &str,
1439 platform: &str,
1440 ch: &mut ChannelConfig,
1441 vault: &mut BTreeMap<String, String>,
1442 ) {
1443 for (k, v) in map {
1444 if k == "enabled" && prefix.is_empty() {
1445 continue;
1446 }
1447 let name = format!("{prefix}{k}");
1448 if let Some(obj) = v.as_object() {
1449 if obj.len() == 1 {
1450 if let Some(Value::String(n)) = obj.get("dotenv") {
1451 ch.credentials.insert(name, SecretRef::Dotenv(n.clone()));
1452 continue;
1453 }
1454 if let Some(Value::String(n)) = obj.get("env") {
1455 ch.credentials.insert(name, SecretRef::Env(n.clone()));
1456 continue;
1457 }
1458 }
1459 if k == "extra" && prefix.is_empty() {
1460 walk(obj, "extra.", platform, ch, vault);
1461 continue;
1462 }
1463 }
1464 if let Value::String(s) = v {
1465 if is_credential_key(k) {
1466 let r = credential_ref_name(platform, &name);
1467 vault.insert(r.clone(), s.clone());
1468 ch.credentials.insert(name, SecretRef::Dotenv(r));
1469 continue;
1470 }
1471 }
1472 ch.extra.insert(name, v.clone());
1473 }
1474 }
1475 walk(map, "", platform, &mut ch, vault);
1476 Ok(ch)
1477}
1478
1479pub fn encode_channel(ch: &ChannelConfig, vault: Option<&BTreeMap<String, String>>) -> Value {
1481 let mut out = Map::new();
1482 out.insert("enabled".into(), Value::Bool(ch.enabled));
1483 let mut extra = Map::new();
1484 for (k, v) in &ch.extra {
1485 match k.strip_prefix("extra.") {
1486 Some(inner) => {
1487 extra.insert(inner.to_string(), v.clone());
1488 }
1489 None => {
1490 out.insert(k.clone(), v.clone());
1491 }
1492 }
1493 }
1494 for (k, r) in &ch.credentials {
1495 let rendered = match vault.and_then(|v| v.get(r.name())) {
1496 Some(value) => Value::String(value.clone()),
1497 None => serde_json::to_value(r).unwrap(),
1498 };
1499 match k.strip_prefix("extra.") {
1500 Some(inner) => {
1501 extra.insert(inner.to_string(), rendered);
1502 }
1503 None => {
1504 out.insert(k.clone(), rendered);
1505 }
1506 }
1507 }
1508 if !extra.is_empty() {
1509 out.insert("extra".into(), Value::Object(extra));
1510 }
1511 Value::Object(out)
1512}
1513
1514const SUB_MAPPED: &[&str] = &[
1515 "events",
1516 "prompt",
1517 "skills",
1518 "deliver",
1519 "deliver_extra",
1520 "secret",
1521 "description",
1522 "created_at",
1523];
1524
1525pub fn decode_subscription(
1527 file: &str,
1528 name: &str,
1529 raw: &Value,
1530 vault: &mut BTreeMap<String, String>,
1531) -> Result<WebhookSubscription> {
1532 let map = raw
1533 .as_object()
1534 .ok_or_else(|| load_error(file, name, "expected an object"))?;
1535 let mut sub = WebhookSubscription {
1536 name: name.to_string(),
1537 secret: None,
1538 events: map.get("events").and_then(Value::as_array).map(|a| {
1539 a.iter()
1540 .map(|v| {
1541 v.as_str()
1542 .map(str::to_string)
1543 .unwrap_or_else(|| v.to_string())
1544 })
1545 .collect()
1546 }),
1547 prompt_template: opt_str(file, &format!("{name}.prompt"), map.get("prompt"))?
1548 .unwrap_or_default(),
1549 deliver: parse_target(
1550 file,
1551 &format!("{name}.deliver"),
1552 map.get("deliver"),
1553 map.get("deliver_extra"),
1554 )?,
1555 skills: map
1556 .get("skills")
1557 .and_then(Value::as_array)
1558 .map(|a| {
1559 a.iter()
1560 .map(|v| {
1561 v.as_str()
1562 .map(str::to_string)
1563 .unwrap_or_else(|| v.to_string())
1564 })
1565 .collect()
1566 })
1567 .unwrap_or_default(),
1568 description: opt_str(file, &format!("{name}.description"), map.get("description"))?,
1569 created_at: opt_str(file, &format!("{name}.created_at"), map.get("created_at"))?,
1570 residue: residue_of(map, SUB_MAPPED),
1571 };
1572 match map.get("secret") {
1573 Some(Value::String(secret)) => {
1577 let r = match placeholder_ref(secret) {
1578 Some(name) => name.to_string(),
1579 None => {
1580 let r = credential_ref_name("webhook", &format!("{name}_secret"));
1581 vault.insert(r.clone(), secret.clone());
1582 r
1583 }
1584 };
1585 sub.secret = Some(SecretRef::Dotenv(r));
1586 }
1587 Some(Value::Object(obj)) => {
1588 if let Some(Value::String(n)) = obj.get("dotenv") {
1589 sub.secret = Some(SecretRef::Dotenv(n.clone()));
1590 } else if let Some(Value::String(n)) = obj.get("env") {
1591 sub.secret = Some(SecretRef::Env(n.clone()));
1592 } else {
1593 return Err(load_error(
1594 file,
1595 &format!("{name}.secret"),
1596 "a secret ref is {dotenv: NAME} or {env: NAME}",
1597 ));
1598 }
1599 }
1600 Some(Value::Null) | None => {}
1601 Some(_) => {
1602 return Err(load_error(
1603 file,
1604 &format!("{name}.secret"),
1605 "expected a string or a {dotenv|env: NAME} ref",
1606 ))
1607 }
1608 }
1609 Ok(sub)
1610}
1611
1612pub fn encode_subscription(
1614 sub: &WebhookSubscription,
1615 vault: Option<&BTreeMap<String, String>>,
1616) -> Value {
1617 let mut out = Map::new();
1618 if let Some(d) = &sub.description {
1619 out.insert("description".into(), Value::String(d.clone()));
1620 }
1621 if let Some(e) = &sub.events {
1622 out.insert(
1623 "events".into(),
1624 Value::Array(e.iter().map(|s| Value::String(s.clone())).collect()),
1625 );
1626 }
1627 out.insert("prompt".into(), Value::String(sub.prompt_template.clone()));
1628 out.insert(
1629 "skills".into(),
1630 Value::Array(
1631 sub.skills
1632 .iter()
1633 .map(|s| Value::String(s.clone()))
1634 .collect(),
1635 ),
1636 );
1637 if let Some(t) = &sub.deliver {
1638 match t {
1639 Target::Explicit {
1640 platform,
1641 chat_id,
1642 thread_id,
1643 } => {
1644 out.insert("deliver".into(), Value::String(platform.clone()));
1645 let mut extra = Map::new();
1646 if let Some(c) = chat_id {
1647 extra.insert("chat_id".into(), Value::String(c.clone()));
1648 }
1649 if let Some(th) = thread_id {
1650 extra.insert("thread_id".into(), Value::String(th.clone()));
1651 }
1652 if !extra.is_empty() {
1653 out.insert("deliver_extra".into(), Value::Object(extra));
1654 }
1655 }
1656 other => {
1657 out.insert("deliver".into(), Value::String(other.render()));
1658 }
1659 }
1660 }
1661 if let Some(s) = &sub.secret {
1662 let value = vault
1663 .and_then(|v| v.get(s.name()).cloned())
1664 .unwrap_or_else(|| format!("${{dotenv:{}}}", s.name()));
1665 out.insert("secret".into(), Value::String(value));
1666 }
1667 if let Some(c) = &sub.created_at {
1668 out.insert("created_at".into(), Value::String(c.clone()));
1669 }
1670 for (k, v) in &sub.residue.0 {
1671 out.insert(k.clone(), v.clone());
1672 }
1673 Value::Object(out)
1674}
1675
1676#[cfg(test)]
1677mod tests {
1678 use super::*;
1679
1680 #[test]
1681 fn job_round_trip_keeps_hermes_order_and_residue() {
1682 let raw = serde_json::json!({
1683 "id": "j1", "schedule": {"kind": "cron", "expr": "0 8 * * *", "tz": "UTC", "jitter": 3},
1684 "prompt": "p", "deliver": "slack:C1:t1", "context_from": "self", "repeat": true, "attach_to_session": true,
1685 "origin": {"platform": "telegram", "chat_id": "1", "thread_id": null, "chat_name": "x"},
1686 "script": "echo hi", "fire_claim": {"pid": 1}, "enabled": false,
1687 });
1688 let job = decode_job("jobs.json", &raw).unwrap();
1689 assert_eq!(job.context_from.as_deref(), Some(&["self".to_string()][..]));
1690 assert_eq!(
1691 job.repeat,
1692 Some(Repeat {
1693 times: None,
1694 completed: 0
1695 })
1696 );
1697 assert_eq!(job.attach_to_session, Some(true));
1698 assert_eq!(
1699 job.deliver,
1700 Target::Explicit {
1701 platform: "slack".into(),
1702 chat_id: Some("C1".into()),
1703 thread_id: Some("t1".into())
1704 }
1705 );
1706 assert_eq!(
1707 job.residue.0.get("script").and_then(Value::as_str),
1708 Some("echo hi")
1709 );
1710 assert_eq!(job.residue.0["__schedule"]["jitter"], serde_json::json!(3));
1711 assert_eq!(
1712 job.residue.0["__origin"]["chat_name"],
1713 serde_json::json!("x")
1714 );
1715 let ordered = encode_job(&job);
1716 let keys: Vec<&str> = ordered.iter().map(|(k, _)| k.as_str()).collect();
1717 assert_eq!(
1718 keys,
1719 vec![
1720 "id",
1721 "schedule",
1722 "prompt",
1723 "skills",
1724 "script",
1725 "model",
1726 "workdir",
1727 "context_from",
1728 "deliver",
1729 "failure_deliver",
1730 "attach_to_session",
1731 "origin",
1732 "repeat",
1733 "enabled",
1734 "next_run_at",
1735 "last_run_at",
1736 "last_status",
1737 "created_at",
1738 "fire_claim"
1739 ]
1740 );
1741 let encoded: Map<String, Value> = encode_job(&job).into_iter().collect();
1742 assert_eq!(encoded["schedule"]["jitter"], serde_json::json!(3));
1743 assert_eq!(
1744 encoded["repeat"],
1745 serde_json::json!({"times": null, "completed": 0})
1746 );
1747 assert_eq!(encoded["origin"]["chat_name"], serde_json::json!("x"));
1748 assert!(decode_job(
1749 "jobs.json",
1750 &serde_json::json!({"id": "x", "schedule": {"kind": "weekly"}})
1751 )
1752 .is_err());
1753 assert!(decode_job("jobs.json", &serde_json::json!({"id": "x", "schedule": {"kind": "once", "run_at": "2026-01-01T00:00:00Z"}, "context_from": 7})).is_err());
1754 }
1755
1756 #[test]
1757 fn strict_o_records() {
1758 let e = decode_expiry(
1759 "config.yaml",
1760 Some(&serde_json::json!({"idle_minutes": 5, "bogus": 1})),
1761 )
1762 .unwrap_err();
1763 assert!(e.to_string().contains("[expiry.bogus]: unknown key"), "{e}");
1764 let w = decode_worker(
1765 "config.yaml",
1766 Some(&serde_json::json!({"harness": "codex", "cwd": "../x"})),
1767 )
1768 .unwrap_err();
1769 assert!(w.to_string().contains("worker.cwd"), "{w}");
1770 let w = decode_worker("config.yaml", Some(&serde_json::json!({"harness": "codex", "env": {"A": {"dotenv": "A_KEY"}, "B": "lit"}, "permission": {"default": "allow"}}))).unwrap().unwrap();
1771 assert_eq!(
1772 w.env["A"],
1773 EnvValue::Secret(SecretRef::Dotenv("A_KEY".into()))
1774 );
1775 assert_eq!(w.permission.default, PermissionDefault::Allow);
1776 assert_eq!(w.cwd, ".");
1777 let a = decode_access("access.yaml", Some(&serde_json::json!({"allowlist": {"telegram": ["b", "a", "a"]}, "policy": {"slack": "open"}, "pairing_ttl_minutes": 30}))).unwrap();
1778 assert_eq!(a.allowlist["telegram"], vec!["a", "b"]);
1779 assert_eq!(a.policy["slack"], AccessPolicy::Open);
1780 assert_eq!(a.pairing_ttl_minutes, Some(30));
1781 assert!(decode_access(
1782 "access.yaml",
1783 Some(&serde_json::json!({"policy": {"slack": "maybe"}}))
1784 )
1785 .is_err());
1786 let h = decode_home(
1787 "config.yaml",
1788 Some(&serde_json::json!({"platform": "telegram", "chat_type": "dm", "chat_id": "1"})),
1789 )
1790 .unwrap()
1791 .unwrap();
1792 assert_eq!(surface_key_string(&h), "telegram|dm|1||");
1793 let h = decode_home(
1794 "config.yaml",
1795 Some(&serde_json::json!({"platform": "telegram", "kind": "dm", "chat_id": "1"})),
1796 )
1797 .unwrap()
1798 .unwrap();
1799 assert_eq!(encode_surface_key(&h)["kind"], serde_json::json!("dm"));
1800 }
1801
1802 #[test]
1803 fn channels_redact_credentials_into_the_vault() {
1804 let mut vault = BTreeMap::new();
1805 let ch = decode_channel("config.yaml", "telegram", &serde_json::json!({"enabled": true, "token": "T", "extra": {"key": "K", "host": "h"}, "mode": "polling"}), &mut vault).unwrap();
1806 assert_eq!(vault.get("TELEGRAM_TOKEN").map(String::as_str), Some("T"));
1807 assert_eq!(
1808 vault.get("TELEGRAM_EXTRA_KEY").map(String::as_str),
1809 Some("K")
1810 );
1811 assert_eq!(
1812 ch.credentials["token"],
1813 SecretRef::Dotenv("TELEGRAM_TOKEN".into())
1814 );
1815 assert_eq!(
1816 ch.credentials["extra.key"],
1817 SecretRef::Dotenv("TELEGRAM_EXTRA_KEY".into())
1818 );
1819 assert_eq!(ch.extra["extra.host"], serde_json::json!("h"));
1820 assert_eq!(ch.extra["mode"], serde_json::json!("polling"));
1821 let ours = encode_channel(&ch, None);
1822 assert_eq!(
1823 ours["token"],
1824 serde_json::json!({"dotenv": "TELEGRAM_TOKEN"})
1825 );
1826 let hermes = encode_channel(&ch, Some(&vault));
1827 assert_eq!(hermes["token"], serde_json::json!("T"));
1828 assert_eq!(hermes["extra"]["key"], serde_json::json!("K"));
1829 }
1830
1831 #[test]
1832 fn fires_obligations_subscriptions() {
1833 let row: Map<String, Value> = serde_json::from_value(serde_json::json!({"id": "f1", "job_id": "j", "status": "completed", "claimed_at": "t0", "started_at": null, "finished_at": "t1", "error": null, "source": "scheduler", "pid": 4})).unwrap();
1834 let fire = decode_fire_row("executions.db", &row).unwrap();
1835 assert_eq!(fire.status, FireStatus::Succeeded);
1836 assert_eq!(fire.residue.0.get("pid"), Some(&serde_json::json!(4)));
1837 assert_eq!(encode_fire_row(&fire)[6], serde_json::json!("completed"));
1838 let row: Map<String, Value> = serde_json::from_value(serde_json::json!({"obligation_id": "o1", "session_key": "agent:coder:telegram:group:-1:55", "platform": "telegram", "chat_id": "-1", "thread_id": "55", "content": "hi", "state": "delivered", "attempts": 1, "created_at": 1.5, "updated_at": 2.5, "adapter_profile": "coder"})).unwrap();
1839 let o = decode_obligation_row("state.db", &row).unwrap();
1840 assert_eq!(o.state, ObligationState::Sent);
1841 assert_eq!(o.target.kind.as_deref(), Some("group"));
1842 assert_eq!(o.delivered_at.as_deref(), Some("2.5"));
1843 assert_eq!(encode_obligation_row(&o)[6], serde_json::json!("delivered"));
1844 let mut vault = BTreeMap::new();
1845 let sub = decode_subscription("webhook_subscriptions.json", "deploys", &serde_json::json!({"events": ["push"], "prompt": "P", "deliver": "telegram", "deliver_extra": {"chat_id": "1"}, "secret": "S", "note": 1}), &mut vault).unwrap();
1846 assert_eq!(
1847 sub.deliver,
1848 Some(Target::Explicit {
1849 platform: "telegram".into(),
1850 chat_id: Some("1".into()),
1851 thread_id: None
1852 })
1853 );
1854 assert_eq!(
1855 vault.get("WEBHOOK_DEPLOYS_SECRET").map(String::as_str),
1856 Some("S")
1857 );
1858 let back = encode_subscription(&sub, Some(&vault));
1859 assert_eq!(back["secret"], serde_json::json!("S"));
1860 assert_eq!(back["deliver_extra"]["chat_id"], serde_json::json!("1"));
1861 assert_eq!(back["note"], serde_json::json!(1));
1862 assert_eq!(
1863 encode_subscription(&sub, None)["secret"],
1864 serde_json::json!("${dotenv:WEBHOOK_DEPLOYS_SECRET}")
1865 );
1866 let mut ours = BTreeMap::new();
1870 let reread = decode_subscription(
1871 "webhook_subscriptions.json",
1872 "deploys",
1873 &encode_subscription(&sub, None),
1874 &mut ours,
1875 )
1876 .unwrap();
1877 assert_eq!(reread.secret, sub.secret);
1878 assert!(ours.is_empty(), "a placeholder is never a value: {ours:?}");
1879 }
1880}