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, PermissionUnattended, Posted, Repeat, Route, RouteMatch,
17 Schedule, Target, 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", "unattended"],
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 if let Some(u) = p.get("unattended") {
429 permission.unattended = match u.as_str() {
430 Some("deny") => PermissionUnattended::Deny,
431 Some("approve") => PermissionUnattended::Approve,
432 _ => {
433 return Err(load_error(
434 file,
435 "worker.permission.unattended",
436 "expected deny|approve",
437 ))
438 }
439 };
440 }
441 }
442 Ok(Some(WorkerSpec {
443 harness: HarnessId::new(req_str(file, "worker.harness", raw.get("harness"))?),
444 model: opt_str(file, "worker.model", raw.get("model"))?,
445 preset: opt_str(file, "worker.preset", raw.get("preset"))?,
446 cwd,
447 env,
448 permission,
449 }))
450}
451
452pub fn decode_home(file: &str, raw: Option<&Value>) -> Result<Option<SurfaceKey>> {
454 let Some(raw) = raw.filter(|v| !v.is_null()) else {
455 return Ok(None);
456 };
457 let map = raw
458 .as_object()
459 .ok_or_else(|| load_error(file, "home", "expected an object"))?;
460 for k in map.keys() {
461 if !["platform", "kind", "chat_type", "chat_id", "thread_id"].contains(&k.as_str()) {
462 return Err(load_error(file, &format!("home.{k}"), "unknown key"));
463 }
464 }
465 let platform = map.get("platform").and_then(Value::as_str);
466 let chat_type = map
467 .get("kind")
468 .or_else(|| map.get("chat_type"))
469 .and_then(Value::as_str);
470 let chat_id = map.get("chat_id").and_then(Value::as_str);
471 let (Some(platform), Some(chat_type), Some(chat_id)) = (platform, chat_type, chat_id) else {
472 return Err(load_error(
473 file,
474 "home",
475 "expected platform, chat_type, chat_id",
476 ));
477 };
478 if !CHAT_TYPES.contains(&chat_type) {
479 return Err(load_error(
480 file,
481 "home",
482 "expected platform, chat_type, chat_id",
483 ));
484 }
485 Ok(Some(SurfaceKey {
486 key: None,
487 platform: Some(platform.to_string()),
488 kind: Some(chat_type.to_string()),
489 chat_id: Some(chat_id.to_string()),
490 thread_id: map
491 .get("thread_id")
492 .and_then(text_of)
493 .filter(|s| !s.is_empty()),
494 participant_id: None,
495 }))
496}
497
498pub fn decode_access(file: &str, raw: Option<&Value>) -> Result<Access> {
500 let mut access = Access::default();
501 let Some(raw) = raw.filter(|v| !v.is_null()) else {
502 return Ok(access);
503 };
504 expect_keys(
505 file,
506 "",
507 raw,
508 &[
509 "allowlist",
510 "admins",
511 "pending_pairings",
512 "policy",
513 "pairing_ttl_minutes",
514 ],
515 )?;
516 if let Some(ttl) = raw.get("pairing_ttl_minutes").filter(|v| !v.is_null()) {
517 access.pairing_ttl_minutes =
518 Some(ttl.as_u64().filter(|v| *v >= 1).ok_or_else(|| {
519 load_error(file, "pairing_ttl_minutes", "expected an integer >= 1")
520 })? as u32);
521 }
522 let set_map = |name: &str, into: &mut BTreeMap<String, Vec<String>>| -> Result<()> {
523 let Some(m) = raw.get(name) else {
524 return Ok(());
525 };
526 let map = m
527 .as_object()
528 .ok_or_else(|| load_error(file, name, "expected a map platform -> list"))?;
529 for (platform, users) in map {
530 let list = users
531 .as_array()
532 .filter(|a| a.iter().all(Value::is_string))
533 .ok_or_else(|| {
534 load_error(
535 file,
536 &format!("{name}.{platform}"),
537 "expected a list of user ids",
538 )
539 })?;
540 let mut ids: Vec<String> = list
541 .iter()
542 .filter_map(|v| v.as_str().map(str::to_string))
543 .collect();
544 ids.sort();
545 ids.dedup();
546 into.insert(platform.clone(), ids);
547 }
548 Ok(())
549 };
550 set_map("allowlist", &mut access.allowlist)?;
551 set_map("admins", &mut access.admins)?;
552 if let Some(p) = raw.get("pending_pairings") {
553 let map = p
554 .as_object()
555 .ok_or_else(|| load_error(file, "pending_pairings", "expected a map code -> record"))?;
556 for (code, rec) in map {
557 let k = format!("pending_pairings.{code}");
558 expect_keys(file, &k, rec, &["platform", "user_id", "issued_at"])?;
559 access.pending_pairings.insert(
560 code.clone(),
561 PendingPairing {
562 platform: req_str(file, &format!("{k}.platform"), rec.get("platform"))?,
563 user_id: req_str(file, &format!("{k}.user_id"), rec.get("user_id"))?,
564 issued_at: req_str(file, &format!("{k}.issued_at"), rec.get("issued_at"))?,
565 },
566 );
567 }
568 }
569 if let Some(p) = raw.get("policy") {
570 let map = p.as_object().ok_or_else(|| {
571 load_error(file, "policy", "expected a map platform -> allowlist|open")
572 })?;
573 for (platform, word) in map {
574 let policy = match word.as_str() {
575 Some("allowlist") => AccessPolicy::Allowlist,
576 Some("open") => AccessPolicy::Open,
577 _ => {
578 return Err(load_error(
579 file,
580 &format!("policy.{platform}"),
581 "expected allowlist|open",
582 ))
583 }
584 };
585 access.policy.insert(platform.clone(), policy);
586 }
587 }
588 Ok(access)
589}
590
591pub fn encode_access(access: &Access) -> Value {
593 let mut out = serde_json::to_value(access).unwrap();
594 if access.pairing_ttl_minutes.is_none() {
595 if let Some(map) = out.as_object_mut() {
596 map.remove("pairing_ttl_minutes");
597 }
598 }
599 out
600}
601
602pub fn decode_binding_row(file: &str, row: &Map<String, Value>) -> Result<Binding> {
604 let text = |k: &str| row.get(k).and_then(text_of).filter(|s| !s.is_empty());
605 let chat_type = req_str(file, "chat_type", row.get("chat_type"))?;
606 if !CHAT_TYPES.contains(&chat_type.as_str()) {
607 return Err(load_error(
608 file,
609 "chat_type",
610 format!("expected one of {}", CHAT_TYPES.join("|")),
611 ));
612 }
613 let key = SurfaceKey {
614 key: None,
615 platform: Some(req_str(file, "platform", row.get("platform"))?),
616 kind: Some(chat_type),
617 chat_id: text("chat_id"),
618 thread_id: text("thread_id"),
619 participant_id: text("participant_id"),
620 };
621 let end_reason = match text("end_reason") {
622 None => None,
623 Some(word) => Some(
624 EndReason::parse(&word)
625 .ok_or_else(|| load_error(file, "end_reason", "unknown end reason"))?,
626 ),
627 };
628 let handoff = if text("handoff_to").is_some() || text("handoff_state").is_some() {
629 Some(Handoff {
630 to: text("handoff_to"),
631 state: text("handoff_state").unwrap_or_default(),
632 error: text("handoff_error"),
633 })
634 } else {
635 None
636 };
637 let recurrence = text("recurrence_job_id").map(|job_id| Recurrence {
638 job_id,
639 kind: "cron".into(),
640 });
641 let residue = match row.get("residue_json").and_then(Value::as_str) {
642 Some(json) if !json.is_empty() => serde_json::from_str::<Value>(json)
643 .ok()
644 .and_then(|v| v.as_object().cloned())
645 .map(|m| Residue(m.into_iter().collect()))
646 .unwrap_or_default(),
647 _ => Residue::default(),
648 };
649 Ok(Binding {
650 trigger: if recurrence.is_some() {
651 crate::ontology::Trigger::Cron
652 } else if key.platform.as_deref() == Some("webhook") {
653 crate::ontology::Trigger::Webhook
654 } else {
655 crate::ontology::Trigger::Channel
656 },
657 key,
658 profile: None,
659 worker: Worker {
660 harness: HarnessId::new(req_str(file, "worker_harness", row.get("worker_harness"))?),
661 session_id: opt_str(file, "worker_session_id", row.get("worker_session_id"))?
664 .filter(|s| !s.is_empty()),
665 locator: opt_str(file, "worker_locator", row.get("worker_locator"))?,
666 },
667 recurrence,
668 handoff,
669 started_at: Some(req_str(file, "started_at", row.get("started_at"))?),
670 last_activity_at: Some(req_str(
671 file,
672 "last_activity_at",
673 row.get("last_activity_at"),
674 )?),
675 ended_at: opt_str(file, "ended_at", row.get("ended_at"))?,
676 end_reason,
677 residue,
678 })
679}
680
681const JOB_MAPPED: &[&str] = &[
684 "id",
685 "schedule",
686 "prompt",
687 "workdir",
688 "model",
689 "skills",
690 "context_from",
691 "deliver",
692 "failure_deliver",
693 "origin",
694 "attach_to_session",
695 "repeat",
696 "enabled",
697 "next_run_at",
698 "last_run_at",
699 "last_status",
700 "created_at",
701];
702
703pub const HERMES_JOB_ORDER: &[&str] = &[
705 "id",
706 "schedule",
707 "prompt",
708 "skills",
709 "script",
710 "no_agent",
711 "model",
712 "provider",
713 "workdir",
714 "enabled_toolsets",
715 "context_from",
716 "deliver",
717 "failure_deliver",
718 "attach_to_session",
719 "origin",
720 "repeat",
721 "enabled",
722 "next_run_at",
723 "last_run_at",
724 "last_status",
725 "created_at",
726 "fire_claim",
727];
728
729pub fn decode_context_from(
731 file: &str,
732 key: &str,
733 v: Option<&Value>,
734) -> Result<Option<Vec<String>>> {
735 let items: Vec<String> = match v {
736 None | Some(Value::Null) => return Ok(None),
737 Some(Value::String(s)) => vec![s.clone()],
738 Some(Value::Array(a)) => a
739 .iter()
740 .map(|x| match x {
741 Value::String(s) => s.clone(),
742 other => other.to_string(),
743 })
744 .collect(),
745 _ => {
746 return Err(load_error(
747 file,
748 key,
749 "expected a job id, a list of job ids, or \"self\"",
750 ))
751 }
752 };
753 let refs: Vec<String> = items
754 .into_iter()
755 .map(|s| s.trim().to_string())
756 .filter(|s| !s.is_empty())
757 .collect();
758 Ok(if refs.is_empty() { None } else { Some(refs) })
759}
760
761pub fn decode_repeat(file: &str, key: &str, raw: Option<&Value>) -> Result<Option<Repeat>> {
763 Ok(match raw {
764 None | Some(Value::Null) => None,
765 Some(Value::Bool(true)) => Some(Repeat {
766 times: None,
767 completed: 0,
768 }),
769 Some(Value::Bool(false)) => Some(Repeat {
770 times: Some(1),
771 completed: 0,
772 }),
773 Some(Value::Number(n)) => Some(Repeat {
774 times: n.as_f64().filter(|f| *f > 0.0).map(|f| f.floor() as u32),
775 completed: 0,
776 }),
777 Some(Value::Object(map)) => {
778 let times = match map.get("times") {
779 None | Some(Value::Null) => None,
780 Some(Value::Number(n)) if n.as_u64().is_some_and(|v| v >= 1) => {
781 Some(n.as_u64().unwrap() as u32)
782 }
783 _ => {
784 return Err(load_error(
785 file,
786 &format!("{key}.times"),
787 "expected null or an integer >= 1",
788 ))
789 }
790 };
791 let completed = match map.get("completed") {
792 None => 0,
793 Some(Value::Number(n)) if n.as_u64().is_some() => n.as_u64().unwrap() as u32,
794 _ => {
795 return Err(load_error(
796 file,
797 &format!("{key}.completed"),
798 "expected an integer >= 0",
799 ))
800 }
801 };
802 Some(Repeat { times, completed })
803 }
804 _ => {
805 return Err(load_error(
806 file,
807 key,
808 "expected null, {times, completed}, a number or a boolean",
809 ))
810 }
811 })
812}
813
814pub fn decode_job(file: &str, raw: &Value) -> Result<Job> {
816 let map = raw
817 .as_object()
818 .ok_or_else(|| load_error(file, "", "expected a job object"))?;
819 let id = req_str(file, "id", map.get("id"))?;
820 let k = |s: &str| format!("{id}.{s}");
821 let sched = map
822 .get("schedule")
823 .and_then(Value::as_object)
824 .ok_or_else(|| load_error(file, &k("schedule"), "expected an object"))?;
825 let schedule = match sched.get("kind").and_then(Value::as_str) {
826 Some("once") => Schedule::Once {
827 run_at: req_str(file, &k("schedule.run_at"), sched.get("run_at"))?,
828 },
829 Some("interval") => {
830 let minutes = sched
831 .get("minutes")
832 .and_then(Value::as_f64)
833 .filter(|m| *m > 0.0)
834 .ok_or_else(|| {
835 load_error(file, &k("schedule.minutes"), "expected a positive number")
836 })?;
837 Schedule::Interval { minutes }
838 }
839 Some("cron") => Schedule::Cron {
840 expr: req_str(file, &k("schedule.expr"), sched.get("expr"))?,
841 tz: opt_str(file, &k("schedule.tz"), sched.get("tz"))?.unwrap_or_else(|| "UTC".into()),
842 },
843 _ => {
844 return Err(load_error(
845 file,
846 &k("schedule.kind"),
847 "expected once|interval|cron",
848 ))
849 }
850 };
851 let schedule_residue = residue_of(sched, &["kind", "run_at", "minutes", "expr", "tz"]);
852 let mut residue = residue_of(map, JOB_MAPPED);
853 if !schedule_residue.is_empty() {
854 residue.keep(
855 "__schedule",
856 Value::Object(schedule_residue.0.into_iter().collect()),
857 );
858 }
859 let origin = match map.get("origin") {
860 Some(Value::Object(o)) => {
861 let r = residue_of(o, &["platform", "chat_id", "thread_id"]);
862 if !r.is_empty() {
863 residue.keep("__origin", Value::Object(r.0.into_iter().collect()));
864 }
865 Some(JobOrigin {
866 platform: req_str(file, &k("origin.platform"), o.get("platform"))?,
867 chat_type: None,
868 chat_id: opt_str(file, &k("origin.chat_id"), o.get("chat_id"))?,
869 thread_id: opt_str(file, &k("origin.thread_id"), o.get("thread_id"))?,
870 })
871 }
872 _ => None,
873 };
874 Ok(Job {
875 schedule,
876 prompt: opt_str(file, &k("prompt"), map.get("prompt"))?,
877 workdir: opt_str(file, &k("workdir"), map.get("workdir"))?,
878 model: opt_str(file, &k("model"), map.get("model"))?,
879 skills: map
880 .get("skills")
881 .and_then(Value::as_array)
882 .map(|a| {
883 a.iter()
884 .map(|v| {
885 v.as_str()
886 .map(str::to_string)
887 .unwrap_or_else(|| v.to_string())
888 })
889 .collect()
890 })
891 .unwrap_or_default(),
892 context_from: decode_context_from(file, &k("context_from"), map.get("context_from"))?,
893 deliver: parse_target(file, &k("deliver"), map.get("deliver"), None)?
894 .unwrap_or(Target::Local),
895 failure_deliver: parse_target(
896 file,
897 &k("failure_deliver"),
898 map.get("failure_deliver"),
899 None,
900 )?,
901 origin,
902 attach_to_session: opt_bool(
903 file,
904 &k("attach_to_session"),
905 map.get("attach_to_session"),
906 None,
907 )?,
908 repeat: decode_repeat(file, &k("repeat"), map.get("repeat"))?,
909 enabled: opt_bool(file, &k("enabled"), map.get("enabled"), Some(true))?.unwrap_or(true),
910 next_run_at: opt_str(file, &k("next_run_at"), map.get("next_run_at"))?,
911 last_run_at: opt_str(file, &k("last_run_at"), map.get("last_run_at"))?,
912 last_status: opt_str(file, &k("last_status"), map.get("last_status"))?,
913 created_at: opt_str(file, &k("created_at"), map.get("created_at"))?,
914 residue,
915 id,
916 })
917}
918
919pub fn encode_job(job: &Job) -> Vec<(String, Value)> {
922 let mut sched = Map::new();
923 sched.insert("kind".into(), Value::String(job.schedule.kind().into()));
924 match &job.schedule {
925 Schedule::Once { run_at } => {
926 sched.insert("run_at".into(), Value::String(run_at.clone()));
927 }
928 Schedule::Interval { minutes } => {
929 sched.insert("minutes".into(), serde_json::json!(*minutes));
930 }
931 Schedule::Cron { expr, tz } => {
932 sched.insert("expr".into(), Value::String(expr.clone()));
933 if tz != "UTC" {
934 sched.insert("tz".into(), Value::String(tz.clone()));
935 }
936 }
937 }
938 if let Some(Value::Object(extra)) = job.residue.0.get("__schedule") {
939 for (k, v) in extra {
940 sched.insert(k.clone(), v.clone());
941 }
942 }
943 let origin = job
944 .origin
945 .as_ref()
946 .map(|o| {
947 let mut m = Map::new();
948 m.insert("platform".into(), Value::String(o.platform.clone()));
949 m.insert(
950 "chat_id".into(),
951 o.chat_id.clone().map(Value::String).unwrap_or(Value::Null),
952 );
953 m.insert(
954 "thread_id".into(),
955 o.thread_id
956 .clone()
957 .map(Value::String)
958 .unwrap_or(Value::Null),
959 );
960 if let Some(Value::Object(extra)) = job.residue.0.get("__origin") {
961 for (k, v) in extra {
962 m.insert(k.clone(), v.clone());
963 }
964 }
965 Value::Object(m)
966 })
967 .unwrap_or(Value::Null);
968 let opt = |s: &Option<String>| s.clone().map(Value::String).unwrap_or(Value::Null);
969 let mut mapped: Vec<(String, Value)> = vec![
970 ("id".into(), Value::String(job.id.clone())),
971 ("schedule".into(), Value::Object(sched)),
972 ("prompt".into(), opt(&job.prompt)),
973 (
974 "skills".into(),
975 Value::Array(
976 job.skills
977 .iter()
978 .map(|s| Value::String(s.clone()))
979 .collect(),
980 ),
981 ),
982 ("model".into(), opt(&job.model)),
983 ("workdir".into(), opt(&job.workdir)),
984 (
985 "context_from".into(),
986 job.context_from
987 .as_ref()
988 .map(|l| Value::Array(l.iter().map(|s| Value::String(s.clone())).collect()))
989 .unwrap_or(Value::Null),
990 ),
991 ("deliver".into(), Value::String(job.deliver.render())),
992 (
993 "failure_deliver".into(),
994 job.failure_deliver
995 .as_ref()
996 .map(|t| Value::String(t.render()))
997 .unwrap_or(Value::Null),
998 ),
999 (
1000 "attach_to_session".into(),
1001 job.attach_to_session
1002 .map(Value::Bool)
1003 .unwrap_or(Value::Null),
1004 ),
1005 ("origin".into(), origin),
1006 (
1007 "repeat".into(),
1008 job.repeat
1009 .as_ref()
1010 .map(|r| serde_json::json!({"times": r.times, "completed": r.completed}))
1011 .unwrap_or(Value::Null),
1012 ),
1013 ("enabled".into(), Value::Bool(job.enabled)),
1014 ("next_run_at".into(), opt(&job.next_run_at)),
1015 ("last_run_at".into(), opt(&job.last_run_at)),
1016 ("last_status".into(), opt(&job.last_status)),
1017 ("created_at".into(), opt(&job.created_at)),
1018 ];
1019 let mut residue: Vec<(String, Value)> = job
1020 .residue
1021 .0
1022 .iter()
1023 .filter(|(k, _)| k.as_str() != "__schedule" && k.as_str() != "__origin")
1029 .map(|(k, v)| (k.clone(), v.clone()))
1030 .collect();
1031 let mut out: Vec<(String, Value)> = Vec::new();
1032 for key in HERMES_JOB_ORDER {
1033 if let Some(pos) = mapped.iter().position(|(k, _)| k == key) {
1034 out.push(mapped.remove(pos));
1035 } else if let Some(pos) = residue.iter().position(|(k, _)| k == key) {
1036 out.push(residue.remove(pos));
1037 }
1038 }
1039 out.extend(mapped);
1040 out.extend(residue);
1041 out
1042}
1043
1044pub fn decode_fire_row(file: &str, row: &Map<String, Value>) -> Result<Fire> {
1046 let id = req_str(file, "id", row.get("id"))?;
1047 let status_word = row.get("status").and_then(Value::as_str).unwrap_or("");
1048 let status = FireStatus::from_hermes_word(status_word).ok_or_else(|| {
1049 load_error(
1050 file,
1051 &format!("{id}.status"),
1052 format!(
1053 "unknown fire status {}",
1054 serde_json::to_string(status_word).unwrap()
1055 ),
1056 )
1057 })?;
1058 Ok(Fire {
1059 job_id: req_str(file, &format!("{id}.job_id"), row.get("job_id"))?,
1060 session_id: row
1061 .get("session_id")
1062 .and_then(text_of)
1063 .filter(|s| !s.is_empty()),
1064 status,
1065 claimed_at: req_str(file, &format!("{id}.claimed_at"), row.get("claimed_at"))?,
1066 started_at: opt_str(file, &format!("{id}.started_at"), row.get("started_at"))?,
1067 finished_at: opt_str(file, &format!("{id}.finished_at"), row.get("finished_at"))?,
1068 error: opt_str(file, &format!("{id}.error"), row.get("error"))?,
1069 obligation_id: row
1070 .get("obligation_id")
1071 .and_then(text_of)
1072 .filter(|s| !s.is_empty()),
1073 residue: {
1074 let mut residue = row_residue_of(
1075 row,
1076 &[
1077 "id",
1078 "job_id",
1079 "status",
1080 "claimed_at",
1081 "started_at",
1082 "finished_at",
1083 "error",
1084 "residue_json",
1085 "session_id",
1086 "obligation_id",
1087 ],
1088 );
1089 if let Some(extra) = row
1092 .get("residue_json")
1093 .and_then(text_of)
1094 .and_then(|t| serde_json::from_str::<Map<String, Value>>(&t).ok())
1095 {
1096 for (k, v) in extra {
1097 residue.keep(k, v);
1098 }
1099 }
1100 residue
1101 },
1102 id,
1103 })
1104}
1105
1106pub fn encode_fire_row(fire: &Fire) -> Vec<Value> {
1108 let r = &fire.residue.0;
1109 let opt = |s: &Option<String>| s.clone().map(Value::String).unwrap_or(Value::Null);
1110 vec![
1111 Value::String(fire.id.clone()),
1112 Value::String(fire.job_id.clone()),
1113 r.get("source")
1114 .cloned()
1115 .unwrap_or(Value::String("scheduler".into())),
1116 r.get("process_id")
1117 .cloned()
1118 .unwrap_or(Value::String(String::new())),
1119 r.get("pid").cloned().unwrap_or(Value::from(0)),
1120 r.get("process_started_at").cloned().unwrap_or(Value::Null),
1121 Value::String(fire.status.hermes_word().into()),
1122 Value::String(fire.claimed_at.clone()),
1123 opt(&fire.started_at),
1124 opt(&fire.finished_at),
1125 opt(&fire.error),
1126 ]
1127}
1128
1129pub const EXECUTION_COLUMNS: &[&str] = &[
1131 "id",
1132 "job_id",
1133 "source",
1134 "process_id",
1135 "pid",
1136 "process_started_at",
1137 "status",
1138 "claimed_at",
1139 "started_at",
1140 "finished_at",
1141 "error",
1142];
1143
1144pub fn decode_obligation_row(file: &str, row: &Map<String, Value>) -> Result<Obligation> {
1146 let id = req_str(file, "obligation_id", row.get("obligation_id"))?;
1147 let state_word = row.get("state").and_then(Value::as_str).unwrap_or("");
1148 let state = ObligationState::from_hermes_word(state_word).ok_or_else(|| {
1149 load_error(
1150 file,
1151 &format!("{id}.state"),
1152 format!(
1153 "unknown obligation state {}",
1154 serde_json::to_string(state_word).unwrap()
1155 ),
1156 )
1157 })?;
1158 let session_key = row
1159 .get("session_key")
1160 .and_then(Value::as_str)
1161 .filter(|s| !s.is_empty())
1162 .map(str::to_string);
1163 let parsed = session_key
1164 .as_deref()
1165 .and_then(crate::ontology::parse_hermes_session_key)
1166 .map(|(k, _)| k);
1167 let target = SurfaceKey {
1168 key: None,
1169 platform: Some(req_str(
1170 file,
1171 &format!("{id}.platform"),
1172 row.get("platform"),
1173 )?),
1174 kind: parsed.as_ref().and_then(|k| k.kind.clone()),
1175 chat_id: Some(row.get("chat_id").and_then(text_of).unwrap_or_default()),
1176 thread_id: row
1177 .get("thread_id")
1178 .and_then(text_of)
1179 .filter(|s| !s.is_empty()),
1180 participant_id: None,
1181 };
1182 let text = |k: &str| row.get(k).and_then(text_of);
1183 let created_at = text("created_at").unwrap_or_else(|| "undefined".into());
1184 let updated_at = text("updated_at").unwrap_or_else(|| created_at.clone());
1185 Ok(Obligation {
1186 target,
1187 session_key,
1188 content: OutboundContent {
1189 text: row.get("content").and_then(text_of).unwrap_or_default(),
1190 attachments: None,
1191 reply_to: None,
1192 format: None,
1193 },
1194 state,
1195 attempts: row.get("attempts").and_then(number_of).unwrap_or(0.0) as u64,
1196 last_error: row.get("last_error").and_then(text_of),
1197 delivered_at: if state == ObligationState::Sent {
1198 Some(updated_at.clone())
1199 } else {
1200 None
1201 },
1202 created_at,
1203 updated_at,
1204 posted: row
1208 .get("posted_message_id")
1209 .and_then(text_of)
1210 .filter(|s| !s.is_empty())
1211 .map(|message_id| Posted { message_id }),
1212 source: row
1213 .get("source_json")
1214 .and_then(text_of)
1215 .and_then(|text| serde_json::from_str::<ObligationSource>(&text).ok())
1216 .unwrap_or(ObligationSource::Turn { key: None }),
1217 residue: row_residue_of(
1218 row,
1219 &[
1220 "obligation_id",
1221 "session_key",
1222 "platform",
1223 "chat_id",
1224 "thread_id",
1225 "content",
1226 "state",
1227 "attempts",
1228 "created_at",
1229 "updated_at",
1230 "last_error",
1231 "posted_message_id",
1232 "source_json",
1233 ],
1234 ),
1235 id,
1236 })
1237}
1238
1239pub const FOLDER_OBLIGATION_EXTRA_COLUMNS: &[&str] = &["posted_message_id", "source_json"];
1241
1242pub fn encode_obligation_folder_extras(o: &Obligation) -> Vec<Value> {
1244 vec![
1245 o.posted
1246 .as_ref()
1247 .map(|p| Value::String(p.message_id.clone()))
1248 .unwrap_or(Value::Null),
1249 Value::String(serde_json::to_string(&o.source).unwrap()),
1250 ]
1251}
1252
1253pub fn encode_obligation_row(o: &Obligation) -> Vec<Value> {
1255 let r = &o.residue.0;
1256 let num = |s: &str| {
1257 s.parse::<f64>()
1258 .map(|f| serde_json::json!(f))
1259 .unwrap_or(Value::from(0))
1260 };
1261 vec![
1262 Value::String(o.id.clone()),
1263 Value::String(o.session_key.clone().unwrap_or_default()),
1264 Value::String(o.target.platform.clone().unwrap_or_default()),
1265 Value::String(o.target.chat_id.clone().unwrap_or_default()),
1266 o.target
1267 .thread_id
1268 .clone()
1269 .map(Value::String)
1270 .unwrap_or(Value::Null),
1271 Value::String(o.content.text.clone()),
1272 Value::String(o.state.hermes_word().into()),
1273 Value::from(o.attempts),
1274 num(&o.created_at),
1275 num(o
1276 .delivered_at
1277 .as_deref()
1278 .map(|_| o.updated_at.as_str())
1279 .unwrap_or(&o.updated_at)),
1280 r.get("owner_pid").cloned().unwrap_or(Value::Null),
1281 r.get("owner_started_at").cloned().unwrap_or(Value::Null),
1282 o.last_error
1283 .clone()
1284 .map(Value::String)
1285 .unwrap_or(Value::Null),
1286 r.get("adapter_profile").cloned().unwrap_or(Value::Null),
1287 ]
1288}
1289
1290pub const OBLIGATION_COLUMNS: &[&str] = &[
1292 "obligation_id",
1293 "session_key",
1294 "platform",
1295 "chat_id",
1296 "thread_id",
1297 "content",
1298 "state",
1299 "attempts",
1300 "created_at",
1301 "updated_at",
1302 "owner_pid",
1303 "owner_started_at",
1304 "last_error",
1305 "adapter_profile",
1306];
1307
1308const ROUTE_MATCH: &[&str] = &["platform", "guild_id", "chat_id", "thread_id"];
1309
1310pub fn decode_route(file: &str, index: usize, raw: &Value) -> Result<Route> {
1312 let map = raw.as_object().ok_or_else(|| {
1313 load_error(
1314 file,
1315 &format!("profile_routes[{index}]"),
1316 "expected an object",
1317 )
1318 })?;
1319 let text = |k: &str| map.get(k).filter(|v| !v.is_null()).and_then(text_of);
1320 Ok(Route {
1321 name: opt_str(
1322 file,
1323 &format!("profile_routes[{index}].name"),
1324 map.get("name"),
1325 )?,
1326 matches: RouteMatch {
1327 platform: req_str(
1328 file,
1329 &format!("profile_routes[{index}].platform"),
1330 map.get("platform"),
1331 )?,
1332 guild_id: text("guild_id"),
1333 chat_id: text("chat_id"),
1334 thread_id: text("thread_id"),
1335 },
1336 profile: req_str(
1337 file,
1338 &format!("profile_routes[{index}].profile"),
1339 map.get("profile"),
1340 )?,
1341 residue: residue_of(
1342 map,
1343 &[
1344 "name",
1345 "profile",
1346 "platform",
1347 "guild_id",
1348 "chat_id",
1349 "thread_id",
1350 ],
1351 ),
1352 })
1353}
1354
1355pub fn encode_route(r: &Route) -> Value {
1357 let mut out = Map::new();
1358 if let Some(n) = &r.name {
1359 out.insert("name".into(), Value::String(n.clone()));
1360 }
1361 let m = &r.matches;
1362 for (k, v) in [
1363 ("platform", Some(&m.platform)),
1364 ("guild_id", m.guild_id.as_ref()),
1365 ("chat_id", m.chat_id.as_ref()),
1366 ("thread_id", m.thread_id.as_ref()),
1367 ] {
1368 if let Some(v) = v {
1369 out.insert(k.into(), Value::String(v.clone()));
1370 }
1371 }
1372 let _ = ROUTE_MATCH;
1373 out.insert("profile".into(), Value::String(r.profile.clone()));
1374 for (k, v) in &r.residue.0 {
1375 out.insert(k.clone(), v.clone());
1376 }
1377 Value::Object(out)
1378}
1379
1380pub fn is_credential_key(key: &str) -> bool {
1382 let k = key.to_ascii_lowercase();
1383 [
1384 "token",
1385 "secret",
1386 "key",
1387 "password",
1388 "passwd",
1389 "api_key",
1390 "app_secret",
1391 "signing",
1392 "webhook_url",
1393 "private",
1394 ]
1395 .iter()
1396 .any(|w| k.contains(w))
1397}
1398
1399pub fn placeholder_ref(text: &str) -> Option<&str> {
1406 let inner = text.strip_prefix("${")?.strip_suffix('}')?;
1407 inner
1408 .strip_prefix("dotenv:")
1409 .or_else(|| inner.strip_prefix("env:"))
1410 .filter(|name| !name.is_empty())
1411}
1412
1413pub fn credential_ref_name(platform: &str, key: &str) -> String {
1415 format!("{platform}_{key}")
1416 .chars()
1417 .map(|c| {
1418 if c.is_ascii_alphanumeric() {
1419 c.to_ascii_uppercase()
1420 } else {
1421 '_'
1422 }
1423 })
1424 .collect()
1425}
1426
1427pub fn decode_channel(
1429 file: &str,
1430 platform: &str,
1431 raw: &Value,
1432 vault: &mut BTreeMap<String, String>,
1433) -> Result<ChannelConfig> {
1434 let map = raw
1435 .as_object()
1436 .ok_or_else(|| load_error(file, &format!("platforms.{platform}"), "expected an object"))?;
1437 let mut ch = ChannelConfig {
1438 platform: platform.to_string(),
1439 enabled: opt_bool(
1440 file,
1441 &format!("platforms.{platform}.enabled"),
1442 map.get("enabled"),
1443 Some(true),
1444 )?
1445 .unwrap_or(true),
1446 credentials: BTreeMap::new(),
1447 extra: BTreeMap::new(),
1448 };
1449 fn walk(
1450 map: &Map<String, Value>,
1451 prefix: &str,
1452 platform: &str,
1453 ch: &mut ChannelConfig,
1454 vault: &mut BTreeMap<String, String>,
1455 ) {
1456 for (k, v) in map {
1457 if k == "enabled" && prefix.is_empty() {
1458 continue;
1459 }
1460 let name = format!("{prefix}{k}");
1461 if let Some(obj) = v.as_object() {
1462 if obj.len() == 1 {
1463 if let Some(Value::String(n)) = obj.get("dotenv") {
1464 ch.credentials.insert(name, SecretRef::Dotenv(n.clone()));
1465 continue;
1466 }
1467 if let Some(Value::String(n)) = obj.get("env") {
1468 ch.credentials.insert(name, SecretRef::Env(n.clone()));
1469 continue;
1470 }
1471 }
1472 if k == "extra" && prefix.is_empty() {
1473 walk(obj, "extra.", platform, ch, vault);
1474 continue;
1475 }
1476 }
1477 if let Value::String(s) = v {
1478 if is_credential_key(k) {
1479 let r = credential_ref_name(platform, &name);
1480 vault.insert(r.clone(), s.clone());
1481 ch.credentials.insert(name, SecretRef::Dotenv(r));
1482 continue;
1483 }
1484 }
1485 ch.extra.insert(name, v.clone());
1486 }
1487 }
1488 walk(map, "", platform, &mut ch, vault);
1489 Ok(ch)
1490}
1491
1492pub fn encode_channel(ch: &ChannelConfig, vault: Option<&BTreeMap<String, String>>) -> Value {
1494 let mut out = Map::new();
1495 out.insert("enabled".into(), Value::Bool(ch.enabled));
1496 let mut extra = Map::new();
1497 for (k, v) in &ch.extra {
1498 match k.strip_prefix("extra.") {
1499 Some(inner) => {
1500 extra.insert(inner.to_string(), v.clone());
1501 }
1502 None => {
1503 out.insert(k.clone(), v.clone());
1504 }
1505 }
1506 }
1507 for (k, r) in &ch.credentials {
1508 let rendered = match vault.and_then(|v| v.get(r.name())) {
1509 Some(value) => Value::String(value.clone()),
1510 None => serde_json::to_value(r).unwrap(),
1511 };
1512 match k.strip_prefix("extra.") {
1513 Some(inner) => {
1514 extra.insert(inner.to_string(), rendered);
1515 }
1516 None => {
1517 out.insert(k.clone(), rendered);
1518 }
1519 }
1520 }
1521 if !extra.is_empty() {
1522 out.insert("extra".into(), Value::Object(extra));
1523 }
1524 Value::Object(out)
1525}
1526
1527const SUB_MAPPED: &[&str] = &[
1528 "events",
1529 "prompt",
1530 "skills",
1531 "deliver",
1532 "deliver_extra",
1533 "secret",
1534 "description",
1535 "created_at",
1536];
1537
1538pub fn decode_subscription(
1540 file: &str,
1541 name: &str,
1542 raw: &Value,
1543 vault: &mut BTreeMap<String, String>,
1544) -> Result<WebhookSubscription> {
1545 let map = raw
1546 .as_object()
1547 .ok_or_else(|| load_error(file, name, "expected an object"))?;
1548 let mut sub = WebhookSubscription {
1549 name: name.to_string(),
1550 secret: None,
1551 events: map.get("events").and_then(Value::as_array).map(|a| {
1552 a.iter()
1553 .map(|v| {
1554 v.as_str()
1555 .map(str::to_string)
1556 .unwrap_or_else(|| v.to_string())
1557 })
1558 .collect()
1559 }),
1560 prompt_template: opt_str(file, &format!("{name}.prompt"), map.get("prompt"))?
1561 .unwrap_or_default(),
1562 deliver: parse_target(
1563 file,
1564 &format!("{name}.deliver"),
1565 map.get("deliver"),
1566 map.get("deliver_extra"),
1567 )?,
1568 skills: map
1569 .get("skills")
1570 .and_then(Value::as_array)
1571 .map(|a| {
1572 a.iter()
1573 .map(|v| {
1574 v.as_str()
1575 .map(str::to_string)
1576 .unwrap_or_else(|| v.to_string())
1577 })
1578 .collect()
1579 })
1580 .unwrap_or_default(),
1581 description: opt_str(file, &format!("{name}.description"), map.get("description"))?,
1582 created_at: opt_str(file, &format!("{name}.created_at"), map.get("created_at"))?,
1583 residue: residue_of(map, SUB_MAPPED),
1584 };
1585 match map.get("secret") {
1586 Some(Value::String(secret)) => {
1590 let r = match placeholder_ref(secret) {
1591 Some(name) => name.to_string(),
1592 None => {
1593 let r = credential_ref_name("webhook", &format!("{name}_secret"));
1594 vault.insert(r.clone(), secret.clone());
1595 r
1596 }
1597 };
1598 sub.secret = Some(SecretRef::Dotenv(r));
1599 }
1600 Some(Value::Object(obj)) => {
1601 if let Some(Value::String(n)) = obj.get("dotenv") {
1602 sub.secret = Some(SecretRef::Dotenv(n.clone()));
1603 } else if let Some(Value::String(n)) = obj.get("env") {
1604 sub.secret = Some(SecretRef::Env(n.clone()));
1605 } else {
1606 return Err(load_error(
1607 file,
1608 &format!("{name}.secret"),
1609 "a secret ref is {dotenv: NAME} or {env: NAME}",
1610 ));
1611 }
1612 }
1613 Some(Value::Null) | None => {}
1614 Some(_) => {
1615 return Err(load_error(
1616 file,
1617 &format!("{name}.secret"),
1618 "expected a string or a {dotenv|env: NAME} ref",
1619 ))
1620 }
1621 }
1622 Ok(sub)
1623}
1624
1625pub fn encode_subscription(
1627 sub: &WebhookSubscription,
1628 vault: Option<&BTreeMap<String, String>>,
1629) -> Value {
1630 let mut out = Map::new();
1631 if let Some(d) = &sub.description {
1632 out.insert("description".into(), Value::String(d.clone()));
1633 }
1634 if let Some(e) = &sub.events {
1635 out.insert(
1636 "events".into(),
1637 Value::Array(e.iter().map(|s| Value::String(s.clone())).collect()),
1638 );
1639 }
1640 out.insert("prompt".into(), Value::String(sub.prompt_template.clone()));
1641 out.insert(
1642 "skills".into(),
1643 Value::Array(
1644 sub.skills
1645 .iter()
1646 .map(|s| Value::String(s.clone()))
1647 .collect(),
1648 ),
1649 );
1650 if let Some(t) = &sub.deliver {
1651 match t {
1652 Target::Explicit {
1653 platform,
1654 chat_id,
1655 thread_id,
1656 } => {
1657 out.insert("deliver".into(), Value::String(platform.clone()));
1658 let mut extra = Map::new();
1659 if let Some(c) = chat_id {
1660 extra.insert("chat_id".into(), Value::String(c.clone()));
1661 }
1662 if let Some(th) = thread_id {
1663 extra.insert("thread_id".into(), Value::String(th.clone()));
1664 }
1665 if !extra.is_empty() {
1666 out.insert("deliver_extra".into(), Value::Object(extra));
1667 }
1668 }
1669 other => {
1670 out.insert("deliver".into(), Value::String(other.render()));
1671 }
1672 }
1673 }
1674 if let Some(s) = &sub.secret {
1675 let value = vault
1676 .and_then(|v| v.get(s.name()).cloned())
1677 .unwrap_or_else(|| format!("${{dotenv:{}}}", s.name()));
1678 out.insert("secret".into(), Value::String(value));
1679 }
1680 if let Some(c) = &sub.created_at {
1681 out.insert("created_at".into(), Value::String(c.clone()));
1682 }
1683 for (k, v) in &sub.residue.0 {
1684 out.insert(k.clone(), v.clone());
1685 }
1686 Value::Object(out)
1687}
1688
1689#[cfg(test)]
1690mod tests {
1691 use super::*;
1692
1693 #[test]
1694 fn job_round_trip_keeps_hermes_order_and_residue() {
1695 let raw = serde_json::json!({
1696 "id": "j1", "schedule": {"kind": "cron", "expr": "0 8 * * *", "tz": "UTC", "jitter": 3},
1697 "prompt": "p", "deliver": "slack:C1:t1", "context_from": "self", "repeat": true, "attach_to_session": true,
1698 "origin": {"platform": "telegram", "chat_id": "1", "thread_id": null, "chat_name": "x"},
1699 "script": "echo hi", "fire_claim": {"pid": 1}, "enabled": false,
1700 });
1701 let job = decode_job("jobs.json", &raw).unwrap();
1702 assert_eq!(job.context_from.as_deref(), Some(&["self".to_string()][..]));
1703 assert_eq!(
1704 job.repeat,
1705 Some(Repeat {
1706 times: None,
1707 completed: 0
1708 })
1709 );
1710 assert_eq!(job.attach_to_session, Some(true));
1711 assert_eq!(
1712 job.deliver,
1713 Target::Explicit {
1714 platform: "slack".into(),
1715 chat_id: Some("C1".into()),
1716 thread_id: Some("t1".into())
1717 }
1718 );
1719 assert_eq!(
1720 job.residue.0.get("script").and_then(Value::as_str),
1721 Some("echo hi")
1722 );
1723 assert_eq!(job.residue.0["__schedule"]["jitter"], serde_json::json!(3));
1724 assert_eq!(
1725 job.residue.0["__origin"]["chat_name"],
1726 serde_json::json!("x")
1727 );
1728 let ordered = encode_job(&job);
1729 let keys: Vec<&str> = ordered.iter().map(|(k, _)| k.as_str()).collect();
1730 assert_eq!(
1731 keys,
1732 vec![
1733 "id",
1734 "schedule",
1735 "prompt",
1736 "skills",
1737 "script",
1738 "model",
1739 "workdir",
1740 "context_from",
1741 "deliver",
1742 "failure_deliver",
1743 "attach_to_session",
1744 "origin",
1745 "repeat",
1746 "enabled",
1747 "next_run_at",
1748 "last_run_at",
1749 "last_status",
1750 "created_at",
1751 "fire_claim"
1752 ]
1753 );
1754 let encoded: Map<String, Value> = encode_job(&job).into_iter().collect();
1755 assert_eq!(encoded["schedule"]["jitter"], serde_json::json!(3));
1756 assert_eq!(
1757 encoded["repeat"],
1758 serde_json::json!({"times": null, "completed": 0})
1759 );
1760 assert_eq!(encoded["origin"]["chat_name"], serde_json::json!("x"));
1761 assert!(decode_job(
1762 "jobs.json",
1763 &serde_json::json!({"id": "x", "schedule": {"kind": "weekly"}})
1764 )
1765 .is_err());
1766 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());
1767 }
1768
1769 #[test]
1770 fn strict_o_records() {
1771 let e = decode_expiry(
1772 "config.yaml",
1773 Some(&serde_json::json!({"idle_minutes": 5, "bogus": 1})),
1774 )
1775 .unwrap_err();
1776 assert!(e.to_string().contains("[expiry.bogus]: unknown key"), "{e}");
1777 let w = decode_worker(
1778 "config.yaml",
1779 Some(&serde_json::json!({"harness": "codex", "cwd": "../x"})),
1780 )
1781 .unwrap_err();
1782 assert!(w.to_string().contains("worker.cwd"), "{w}");
1783 let w = decode_worker("config.yaml", Some(&serde_json::json!({"harness": "codex", "env": {"A": {"dotenv": "A_KEY"}, "B": "lit"}, "permission": {"default": "allow"}}))).unwrap().unwrap();
1784 assert_eq!(
1785 w.env["A"],
1786 EnvValue::Secret(SecretRef::Dotenv("A_KEY".into()))
1787 );
1788 assert_eq!(w.permission.default, PermissionDefault::Allow);
1789 assert_eq!(w.cwd, ".");
1790 let a = decode_access("access.yaml", Some(&serde_json::json!({"allowlist": {"telegram": ["b", "a", "a"]}, "policy": {"slack": "open"}, "pairing_ttl_minutes": 30}))).unwrap();
1791 assert_eq!(a.allowlist["telegram"], vec!["a", "b"]);
1792 assert_eq!(a.policy["slack"], AccessPolicy::Open);
1793 assert_eq!(a.pairing_ttl_minutes, Some(30));
1794 assert!(decode_access(
1795 "access.yaml",
1796 Some(&serde_json::json!({"policy": {"slack": "maybe"}}))
1797 )
1798 .is_err());
1799 let h = decode_home(
1800 "config.yaml",
1801 Some(&serde_json::json!({"platform": "telegram", "chat_type": "dm", "chat_id": "1"})),
1802 )
1803 .unwrap()
1804 .unwrap();
1805 assert_eq!(surface_key_string(&h), "telegram|dm|1||");
1806 let h = decode_home(
1807 "config.yaml",
1808 Some(&serde_json::json!({"platform": "telegram", "kind": "dm", "chat_id": "1"})),
1809 )
1810 .unwrap()
1811 .unwrap();
1812 assert_eq!(encode_surface_key(&h)["kind"], serde_json::json!("dm"));
1813 }
1814
1815 #[test]
1816 fn channels_redact_credentials_into_the_vault() {
1817 let mut vault = BTreeMap::new();
1818 let ch = decode_channel("config.yaml", "telegram", &serde_json::json!({"enabled": true, "token": "T", "extra": {"key": "K", "host": "h"}, "mode": "polling"}), &mut vault).unwrap();
1819 assert_eq!(vault.get("TELEGRAM_TOKEN").map(String::as_str), Some("T"));
1820 assert_eq!(
1821 vault.get("TELEGRAM_EXTRA_KEY").map(String::as_str),
1822 Some("K")
1823 );
1824 assert_eq!(
1825 ch.credentials["token"],
1826 SecretRef::Dotenv("TELEGRAM_TOKEN".into())
1827 );
1828 assert_eq!(
1829 ch.credentials["extra.key"],
1830 SecretRef::Dotenv("TELEGRAM_EXTRA_KEY".into())
1831 );
1832 assert_eq!(ch.extra["extra.host"], serde_json::json!("h"));
1833 assert_eq!(ch.extra["mode"], serde_json::json!("polling"));
1834 let ours = encode_channel(&ch, None);
1835 assert_eq!(
1836 ours["token"],
1837 serde_json::json!({"dotenv": "TELEGRAM_TOKEN"})
1838 );
1839 let hermes = encode_channel(&ch, Some(&vault));
1840 assert_eq!(hermes["token"], serde_json::json!("T"));
1841 assert_eq!(hermes["extra"]["key"], serde_json::json!("K"));
1842 }
1843
1844 #[test]
1845 fn fires_obligations_subscriptions() {
1846 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();
1847 let fire = decode_fire_row("executions.db", &row).unwrap();
1848 assert_eq!(fire.status, FireStatus::Succeeded);
1849 assert_eq!(fire.residue.0.get("pid"), Some(&serde_json::json!(4)));
1850 assert_eq!(encode_fire_row(&fire)[6], serde_json::json!("completed"));
1851 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();
1852 let o = decode_obligation_row("state.db", &row).unwrap();
1853 assert_eq!(o.state, ObligationState::Sent);
1854 assert_eq!(o.target.kind.as_deref(), Some("group"));
1855 assert_eq!(o.delivered_at.as_deref(), Some("2.5"));
1856 assert_eq!(encode_obligation_row(&o)[6], serde_json::json!("delivered"));
1857 let mut vault = BTreeMap::new();
1858 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();
1859 assert_eq!(
1860 sub.deliver,
1861 Some(Target::Explicit {
1862 platform: "telegram".into(),
1863 chat_id: Some("1".into()),
1864 thread_id: None
1865 })
1866 );
1867 assert_eq!(
1868 vault.get("WEBHOOK_DEPLOYS_SECRET").map(String::as_str),
1869 Some("S")
1870 );
1871 let back = encode_subscription(&sub, Some(&vault));
1872 assert_eq!(back["secret"], serde_json::json!("S"));
1873 assert_eq!(back["deliver_extra"]["chat_id"], serde_json::json!("1"));
1874 assert_eq!(back["note"], serde_json::json!(1));
1875 assert_eq!(
1876 encode_subscription(&sub, None)["secret"],
1877 serde_json::json!("${dotenv:WEBHOOK_DEPLOYS_SECRET}")
1878 );
1879 let mut ours = BTreeMap::new();
1883 let reread = decode_subscription(
1884 "webhook_subscriptions.json",
1885 "deploys",
1886 &encode_subscription(&sub, None),
1887 &mut ours,
1888 )
1889 .unwrap();
1890 assert_eq!(reread.secret, sub.secret);
1891 assert!(ours.is_empty(), "a placeholder is never a value: {ours:?}");
1892 }
1893}