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, 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: None,
1048 status,
1049 claimed_at: req_str(file, &format!("{id}.claimed_at"), row.get("claimed_at"))?,
1050 started_at: opt_str(file, &format!("{id}.started_at"), row.get("started_at"))?,
1051 finished_at: opt_str(file, &format!("{id}.finished_at"), row.get("finished_at"))?,
1052 error: opt_str(file, &format!("{id}.error"), row.get("error"))?,
1053 obligation_id: None,
1054 residue: row_residue_of(
1055 row,
1056 &[
1057 "id",
1058 "job_id",
1059 "status",
1060 "claimed_at",
1061 "started_at",
1062 "finished_at",
1063 "error",
1064 ],
1065 ),
1066 id,
1067 })
1068}
1069
1070pub fn encode_fire_row(fire: &Fire) -> Vec<Value> {
1072 let r = &fire.residue.0;
1073 let opt = |s: &Option<String>| s.clone().map(Value::String).unwrap_or(Value::Null);
1074 vec![
1075 Value::String(fire.id.clone()),
1076 Value::String(fire.job_id.clone()),
1077 r.get("source")
1078 .cloned()
1079 .unwrap_or(Value::String("scheduler".into())),
1080 r.get("process_id")
1081 .cloned()
1082 .unwrap_or(Value::String(String::new())),
1083 r.get("pid").cloned().unwrap_or(Value::from(0)),
1084 r.get("process_started_at").cloned().unwrap_or(Value::Null),
1085 Value::String(fire.status.hermes_word().into()),
1086 Value::String(fire.claimed_at.clone()),
1087 opt(&fire.started_at),
1088 opt(&fire.finished_at),
1089 opt(&fire.error),
1090 ]
1091}
1092
1093pub const EXECUTION_COLUMNS: &[&str] = &[
1095 "id",
1096 "job_id",
1097 "source",
1098 "process_id",
1099 "pid",
1100 "process_started_at",
1101 "status",
1102 "claimed_at",
1103 "started_at",
1104 "finished_at",
1105 "error",
1106];
1107
1108pub fn decode_obligation_row(file: &str, row: &Map<String, Value>) -> Result<Obligation> {
1110 let id = req_str(file, "obligation_id", row.get("obligation_id"))?;
1111 let state_word = row.get("state").and_then(Value::as_str).unwrap_or("");
1112 let state = ObligationState::from_hermes_word(state_word).ok_or_else(|| {
1113 load_error(
1114 file,
1115 &format!("{id}.state"),
1116 format!(
1117 "unknown obligation state {}",
1118 serde_json::to_string(state_word).unwrap()
1119 ),
1120 )
1121 })?;
1122 let session_key = row
1123 .get("session_key")
1124 .and_then(Value::as_str)
1125 .filter(|s| !s.is_empty())
1126 .map(str::to_string);
1127 let parsed = session_key
1128 .as_deref()
1129 .and_then(crate::ontology::parse_hermes_session_key)
1130 .map(|(k, _)| k);
1131 let target = SurfaceKey {
1132 key: None,
1133 platform: Some(req_str(
1134 file,
1135 &format!("{id}.platform"),
1136 row.get("platform"),
1137 )?),
1138 kind: parsed.as_ref().and_then(|k| k.kind.clone()),
1139 chat_id: Some(row.get("chat_id").and_then(text_of).unwrap_or_default()),
1140 thread_id: row
1141 .get("thread_id")
1142 .and_then(text_of)
1143 .filter(|s| !s.is_empty()),
1144 participant_id: None,
1145 };
1146 let text = |k: &str| row.get(k).and_then(text_of);
1147 let created_at = text("created_at").unwrap_or_else(|| "undefined".into());
1148 let updated_at = text("updated_at").unwrap_or_else(|| created_at.clone());
1149 Ok(Obligation {
1150 target,
1151 session_key,
1152 content: OutboundContent {
1153 text: row.get("content").and_then(text_of).unwrap_or_default(),
1154 attachments: None,
1155 reply_to: None,
1156 format: None,
1157 },
1158 state,
1159 attempts: row.get("attempts").and_then(number_of).unwrap_or(0.0) as u64,
1160 last_error: row.get("last_error").and_then(text_of),
1161 delivered_at: if state == ObligationState::Sent {
1162 Some(updated_at.clone())
1163 } else {
1164 None
1165 },
1166 created_at,
1167 updated_at,
1168 posted: None,
1169 source: ObligationSource::Turn { key: None },
1170 residue: row_residue_of(
1171 row,
1172 &[
1173 "obligation_id",
1174 "session_key",
1175 "platform",
1176 "chat_id",
1177 "thread_id",
1178 "content",
1179 "state",
1180 "attempts",
1181 "created_at",
1182 "updated_at",
1183 "last_error",
1184 ],
1185 ),
1186 id,
1187 })
1188}
1189
1190pub fn encode_obligation_row(o: &Obligation) -> Vec<Value> {
1192 let r = &o.residue.0;
1193 let num = |s: &str| {
1194 s.parse::<f64>()
1195 .map(|f| serde_json::json!(f))
1196 .unwrap_or(Value::from(0))
1197 };
1198 vec![
1199 Value::String(o.id.clone()),
1200 Value::String(o.session_key.clone().unwrap_or_default()),
1201 Value::String(o.target.platform.clone().unwrap_or_default()),
1202 Value::String(o.target.chat_id.clone().unwrap_or_default()),
1203 o.target
1204 .thread_id
1205 .clone()
1206 .map(Value::String)
1207 .unwrap_or(Value::Null),
1208 Value::String(o.content.text.clone()),
1209 Value::String(o.state.hermes_word().into()),
1210 Value::from(o.attempts),
1211 num(&o.created_at),
1212 num(o
1213 .delivered_at
1214 .as_deref()
1215 .map(|_| o.updated_at.as_str())
1216 .unwrap_or(&o.updated_at)),
1217 r.get("owner_pid").cloned().unwrap_or(Value::Null),
1218 r.get("owner_started_at").cloned().unwrap_or(Value::Null),
1219 o.last_error
1220 .clone()
1221 .map(Value::String)
1222 .unwrap_or(Value::Null),
1223 r.get("adapter_profile").cloned().unwrap_or(Value::Null),
1224 ]
1225}
1226
1227pub const OBLIGATION_COLUMNS: &[&str] = &[
1229 "obligation_id",
1230 "session_key",
1231 "platform",
1232 "chat_id",
1233 "thread_id",
1234 "content",
1235 "state",
1236 "attempts",
1237 "created_at",
1238 "updated_at",
1239 "owner_pid",
1240 "owner_started_at",
1241 "last_error",
1242 "adapter_profile",
1243];
1244
1245const ROUTE_MATCH: &[&str] = &["platform", "guild_id", "chat_id", "thread_id"];
1246
1247pub fn decode_route(file: &str, index: usize, raw: &Value) -> Result<Route> {
1249 let map = raw.as_object().ok_or_else(|| {
1250 load_error(
1251 file,
1252 &format!("profile_routes[{index}]"),
1253 "expected an object",
1254 )
1255 })?;
1256 let text = |k: &str| map.get(k).filter(|v| !v.is_null()).and_then(text_of);
1257 Ok(Route {
1258 name: opt_str(
1259 file,
1260 &format!("profile_routes[{index}].name"),
1261 map.get("name"),
1262 )?,
1263 matches: RouteMatch {
1264 platform: req_str(
1265 file,
1266 &format!("profile_routes[{index}].platform"),
1267 map.get("platform"),
1268 )?,
1269 guild_id: text("guild_id"),
1270 chat_id: text("chat_id"),
1271 thread_id: text("thread_id"),
1272 },
1273 profile: req_str(
1274 file,
1275 &format!("profile_routes[{index}].profile"),
1276 map.get("profile"),
1277 )?,
1278 residue: residue_of(
1279 map,
1280 &[
1281 "name",
1282 "profile",
1283 "platform",
1284 "guild_id",
1285 "chat_id",
1286 "thread_id",
1287 ],
1288 ),
1289 })
1290}
1291
1292pub fn encode_route(r: &Route) -> Value {
1294 let mut out = Map::new();
1295 if let Some(n) = &r.name {
1296 out.insert("name".into(), Value::String(n.clone()));
1297 }
1298 let m = &r.matches;
1299 for (k, v) in [
1300 ("platform", Some(&m.platform)),
1301 ("guild_id", m.guild_id.as_ref()),
1302 ("chat_id", m.chat_id.as_ref()),
1303 ("thread_id", m.thread_id.as_ref()),
1304 ] {
1305 if let Some(v) = v {
1306 out.insert(k.into(), Value::String(v.clone()));
1307 }
1308 }
1309 let _ = ROUTE_MATCH;
1310 out.insert("profile".into(), Value::String(r.profile.clone()));
1311 for (k, v) in &r.residue.0 {
1312 out.insert(k.clone(), v.clone());
1313 }
1314 Value::Object(out)
1315}
1316
1317pub fn is_credential_key(key: &str) -> bool {
1319 let k = key.to_ascii_lowercase();
1320 [
1321 "token",
1322 "secret",
1323 "key",
1324 "password",
1325 "passwd",
1326 "api_key",
1327 "app_secret",
1328 "signing",
1329 "webhook_url",
1330 "private",
1331 ]
1332 .iter()
1333 .any(|w| k.contains(w))
1334}
1335
1336pub fn placeholder_ref(text: &str) -> Option<&str> {
1343 let inner = text.strip_prefix("${")?.strip_suffix('}')?;
1344 inner
1345 .strip_prefix("dotenv:")
1346 .or_else(|| inner.strip_prefix("env:"))
1347 .filter(|name| !name.is_empty())
1348}
1349
1350pub fn credential_ref_name(platform: &str, key: &str) -> String {
1352 format!("{platform}_{key}")
1353 .chars()
1354 .map(|c| {
1355 if c.is_ascii_alphanumeric() {
1356 c.to_ascii_uppercase()
1357 } else {
1358 '_'
1359 }
1360 })
1361 .collect()
1362}
1363
1364pub fn decode_channel(
1366 file: &str,
1367 platform: &str,
1368 raw: &Value,
1369 vault: &mut BTreeMap<String, String>,
1370) -> Result<ChannelConfig> {
1371 let map = raw
1372 .as_object()
1373 .ok_or_else(|| load_error(file, &format!("platforms.{platform}"), "expected an object"))?;
1374 let mut ch = ChannelConfig {
1375 platform: platform.to_string(),
1376 enabled: opt_bool(
1377 file,
1378 &format!("platforms.{platform}.enabled"),
1379 map.get("enabled"),
1380 Some(true),
1381 )?
1382 .unwrap_or(true),
1383 credentials: BTreeMap::new(),
1384 extra: BTreeMap::new(),
1385 };
1386 fn walk(
1387 map: &Map<String, Value>,
1388 prefix: &str,
1389 platform: &str,
1390 ch: &mut ChannelConfig,
1391 vault: &mut BTreeMap<String, String>,
1392 ) {
1393 for (k, v) in map {
1394 if k == "enabled" && prefix.is_empty() {
1395 continue;
1396 }
1397 let name = format!("{prefix}{k}");
1398 if let Some(obj) = v.as_object() {
1399 if obj.len() == 1 {
1400 if let Some(Value::String(n)) = obj.get("dotenv") {
1401 ch.credentials.insert(name, SecretRef::Dotenv(n.clone()));
1402 continue;
1403 }
1404 if let Some(Value::String(n)) = obj.get("env") {
1405 ch.credentials.insert(name, SecretRef::Env(n.clone()));
1406 continue;
1407 }
1408 }
1409 if k == "extra" && prefix.is_empty() {
1410 walk(obj, "extra.", platform, ch, vault);
1411 continue;
1412 }
1413 }
1414 if let Value::String(s) = v {
1415 if is_credential_key(k) {
1416 let r = credential_ref_name(platform, &name);
1417 vault.insert(r.clone(), s.clone());
1418 ch.credentials.insert(name, SecretRef::Dotenv(r));
1419 continue;
1420 }
1421 }
1422 ch.extra.insert(name, v.clone());
1423 }
1424 }
1425 walk(map, "", platform, &mut ch, vault);
1426 Ok(ch)
1427}
1428
1429pub fn encode_channel(ch: &ChannelConfig, vault: Option<&BTreeMap<String, String>>) -> Value {
1431 let mut out = Map::new();
1432 out.insert("enabled".into(), Value::Bool(ch.enabled));
1433 let mut extra = Map::new();
1434 for (k, v) in &ch.extra {
1435 match k.strip_prefix("extra.") {
1436 Some(inner) => {
1437 extra.insert(inner.to_string(), v.clone());
1438 }
1439 None => {
1440 out.insert(k.clone(), v.clone());
1441 }
1442 }
1443 }
1444 for (k, r) in &ch.credentials {
1445 let rendered = match vault.and_then(|v| v.get(r.name())) {
1446 Some(value) => Value::String(value.clone()),
1447 None => serde_json::to_value(r).unwrap(),
1448 };
1449 match k.strip_prefix("extra.") {
1450 Some(inner) => {
1451 extra.insert(inner.to_string(), rendered);
1452 }
1453 None => {
1454 out.insert(k.clone(), rendered);
1455 }
1456 }
1457 }
1458 if !extra.is_empty() {
1459 out.insert("extra".into(), Value::Object(extra));
1460 }
1461 Value::Object(out)
1462}
1463
1464const SUB_MAPPED: &[&str] = &[
1465 "events",
1466 "prompt",
1467 "skills",
1468 "deliver",
1469 "deliver_extra",
1470 "secret",
1471 "description",
1472 "created_at",
1473];
1474
1475pub fn decode_subscription(
1477 file: &str,
1478 name: &str,
1479 raw: &Value,
1480 vault: &mut BTreeMap<String, String>,
1481) -> Result<WebhookSubscription> {
1482 let map = raw
1483 .as_object()
1484 .ok_or_else(|| load_error(file, name, "expected an object"))?;
1485 let mut sub = WebhookSubscription {
1486 name: name.to_string(),
1487 secret: None,
1488 events: map.get("events").and_then(Value::as_array).map(|a| {
1489 a.iter()
1490 .map(|v| {
1491 v.as_str()
1492 .map(str::to_string)
1493 .unwrap_or_else(|| v.to_string())
1494 })
1495 .collect()
1496 }),
1497 prompt_template: opt_str(file, &format!("{name}.prompt"), map.get("prompt"))?
1498 .unwrap_or_default(),
1499 deliver: parse_target(
1500 file,
1501 &format!("{name}.deliver"),
1502 map.get("deliver"),
1503 map.get("deliver_extra"),
1504 )?,
1505 skills: map
1506 .get("skills")
1507 .and_then(Value::as_array)
1508 .map(|a| {
1509 a.iter()
1510 .map(|v| {
1511 v.as_str()
1512 .map(str::to_string)
1513 .unwrap_or_else(|| v.to_string())
1514 })
1515 .collect()
1516 })
1517 .unwrap_or_default(),
1518 description: opt_str(file, &format!("{name}.description"), map.get("description"))?,
1519 created_at: opt_str(file, &format!("{name}.created_at"), map.get("created_at"))?,
1520 residue: residue_of(map, SUB_MAPPED),
1521 };
1522 match map.get("secret") {
1523 Some(Value::String(secret)) => {
1527 let r = match placeholder_ref(secret) {
1528 Some(name) => name.to_string(),
1529 None => {
1530 let r = credential_ref_name("webhook", &format!("{name}_secret"));
1531 vault.insert(r.clone(), secret.clone());
1532 r
1533 }
1534 };
1535 sub.secret = Some(SecretRef::Dotenv(r));
1536 }
1537 Some(Value::Object(obj)) => {
1538 if let Some(Value::String(n)) = obj.get("dotenv") {
1539 sub.secret = Some(SecretRef::Dotenv(n.clone()));
1540 } else if let Some(Value::String(n)) = obj.get("env") {
1541 sub.secret = Some(SecretRef::Env(n.clone()));
1542 } else {
1543 return Err(load_error(
1544 file,
1545 &format!("{name}.secret"),
1546 "a secret ref is {dotenv: NAME} or {env: NAME}",
1547 ));
1548 }
1549 }
1550 Some(Value::Null) | None => {}
1551 Some(_) => {
1552 return Err(load_error(
1553 file,
1554 &format!("{name}.secret"),
1555 "expected a string or a {dotenv|env: NAME} ref",
1556 ))
1557 }
1558 }
1559 Ok(sub)
1560}
1561
1562pub fn encode_subscription(
1564 sub: &WebhookSubscription,
1565 vault: Option<&BTreeMap<String, String>>,
1566) -> Value {
1567 let mut out = Map::new();
1568 if let Some(d) = &sub.description {
1569 out.insert("description".into(), Value::String(d.clone()));
1570 }
1571 if let Some(e) = &sub.events {
1572 out.insert(
1573 "events".into(),
1574 Value::Array(e.iter().map(|s| Value::String(s.clone())).collect()),
1575 );
1576 }
1577 out.insert("prompt".into(), Value::String(sub.prompt_template.clone()));
1578 out.insert(
1579 "skills".into(),
1580 Value::Array(
1581 sub.skills
1582 .iter()
1583 .map(|s| Value::String(s.clone()))
1584 .collect(),
1585 ),
1586 );
1587 if let Some(t) = &sub.deliver {
1588 match t {
1589 Target::Explicit {
1590 platform,
1591 chat_id,
1592 thread_id,
1593 } => {
1594 out.insert("deliver".into(), Value::String(platform.clone()));
1595 let mut extra = Map::new();
1596 if let Some(c) = chat_id {
1597 extra.insert("chat_id".into(), Value::String(c.clone()));
1598 }
1599 if let Some(th) = thread_id {
1600 extra.insert("thread_id".into(), Value::String(th.clone()));
1601 }
1602 if !extra.is_empty() {
1603 out.insert("deliver_extra".into(), Value::Object(extra));
1604 }
1605 }
1606 other => {
1607 out.insert("deliver".into(), Value::String(other.render()));
1608 }
1609 }
1610 }
1611 if let Some(s) = &sub.secret {
1612 let value = vault
1613 .and_then(|v| v.get(s.name()).cloned())
1614 .unwrap_or_else(|| format!("${{dotenv:{}}}", s.name()));
1615 out.insert("secret".into(), Value::String(value));
1616 }
1617 if let Some(c) = &sub.created_at {
1618 out.insert("created_at".into(), Value::String(c.clone()));
1619 }
1620 for (k, v) in &sub.residue.0 {
1621 out.insert(k.clone(), v.clone());
1622 }
1623 Value::Object(out)
1624}
1625
1626#[cfg(test)]
1627mod tests {
1628 use super::*;
1629
1630 #[test]
1631 fn job_round_trip_keeps_hermes_order_and_residue() {
1632 let raw = serde_json::json!({
1633 "id": "j1", "schedule": {"kind": "cron", "expr": "0 8 * * *", "tz": "UTC", "jitter": 3},
1634 "prompt": "p", "deliver": "slack:C1:t1", "context_from": "self", "repeat": true, "attach_to_session": true,
1635 "origin": {"platform": "telegram", "chat_id": "1", "thread_id": null, "chat_name": "x"},
1636 "script": "echo hi", "fire_claim": {"pid": 1}, "enabled": false,
1637 });
1638 let job = decode_job("jobs.json", &raw).unwrap();
1639 assert_eq!(job.context_from.as_deref(), Some(&["self".to_string()][..]));
1640 assert_eq!(
1641 job.repeat,
1642 Some(Repeat {
1643 times: None,
1644 completed: 0
1645 })
1646 );
1647 assert_eq!(job.attach_to_session, Some(true));
1648 assert_eq!(
1649 job.deliver,
1650 Target::Explicit {
1651 platform: "slack".into(),
1652 chat_id: Some("C1".into()),
1653 thread_id: Some("t1".into())
1654 }
1655 );
1656 assert_eq!(
1657 job.residue.0.get("script").and_then(Value::as_str),
1658 Some("echo hi")
1659 );
1660 assert_eq!(job.residue.0["__schedule"]["jitter"], serde_json::json!(3));
1661 assert_eq!(
1662 job.residue.0["__origin"]["chat_name"],
1663 serde_json::json!("x")
1664 );
1665 let ordered = encode_job(&job);
1666 let keys: Vec<&str> = ordered.iter().map(|(k, _)| k.as_str()).collect();
1667 assert_eq!(
1668 keys,
1669 vec![
1670 "id",
1671 "schedule",
1672 "prompt",
1673 "skills",
1674 "script",
1675 "model",
1676 "workdir",
1677 "context_from",
1678 "deliver",
1679 "failure_deliver",
1680 "attach_to_session",
1681 "origin",
1682 "repeat",
1683 "enabled",
1684 "next_run_at",
1685 "last_run_at",
1686 "last_status",
1687 "created_at",
1688 "fire_claim"
1689 ]
1690 );
1691 let encoded: Map<String, Value> = encode_job(&job).into_iter().collect();
1692 assert_eq!(encoded["schedule"]["jitter"], serde_json::json!(3));
1693 assert_eq!(
1694 encoded["repeat"],
1695 serde_json::json!({"times": null, "completed": 0})
1696 );
1697 assert_eq!(encoded["origin"]["chat_name"], serde_json::json!("x"));
1698 assert!(decode_job(
1699 "jobs.json",
1700 &serde_json::json!({"id": "x", "schedule": {"kind": "weekly"}})
1701 )
1702 .is_err());
1703 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());
1704 }
1705
1706 #[test]
1707 fn strict_o_records() {
1708 let e = decode_expiry(
1709 "config.yaml",
1710 Some(&serde_json::json!({"idle_minutes": 5, "bogus": 1})),
1711 )
1712 .unwrap_err();
1713 assert!(e.to_string().contains("[expiry.bogus]: unknown key"), "{e}");
1714 let w = decode_worker(
1715 "config.yaml",
1716 Some(&serde_json::json!({"harness": "codex", "cwd": "../x"})),
1717 )
1718 .unwrap_err();
1719 assert!(w.to_string().contains("worker.cwd"), "{w}");
1720 let w = decode_worker("config.yaml", Some(&serde_json::json!({"harness": "codex", "env": {"A": {"dotenv": "A_KEY"}, "B": "lit"}, "permission": {"default": "allow"}}))).unwrap().unwrap();
1721 assert_eq!(
1722 w.env["A"],
1723 EnvValue::Secret(SecretRef::Dotenv("A_KEY".into()))
1724 );
1725 assert_eq!(w.permission.default, PermissionDefault::Allow);
1726 assert_eq!(w.cwd, ".");
1727 let a = decode_access("access.yaml", Some(&serde_json::json!({"allowlist": {"telegram": ["b", "a", "a"]}, "policy": {"slack": "open"}, "pairing_ttl_minutes": 30}))).unwrap();
1728 assert_eq!(a.allowlist["telegram"], vec!["a", "b"]);
1729 assert_eq!(a.policy["slack"], AccessPolicy::Open);
1730 assert_eq!(a.pairing_ttl_minutes, Some(30));
1731 assert!(decode_access(
1732 "access.yaml",
1733 Some(&serde_json::json!({"policy": {"slack": "maybe"}}))
1734 )
1735 .is_err());
1736 let h = decode_home(
1737 "config.yaml",
1738 Some(&serde_json::json!({"platform": "telegram", "chat_type": "dm", "chat_id": "1"})),
1739 )
1740 .unwrap()
1741 .unwrap();
1742 assert_eq!(surface_key_string(&h), "telegram|dm|1||");
1743 let h = decode_home(
1744 "config.yaml",
1745 Some(&serde_json::json!({"platform": "telegram", "kind": "dm", "chat_id": "1"})),
1746 )
1747 .unwrap()
1748 .unwrap();
1749 assert_eq!(encode_surface_key(&h)["kind"], serde_json::json!("dm"));
1750 }
1751
1752 #[test]
1753 fn channels_redact_credentials_into_the_vault() {
1754 let mut vault = BTreeMap::new();
1755 let ch = decode_channel("config.yaml", "telegram", &serde_json::json!({"enabled": true, "token": "T", "extra": {"key": "K", "host": "h"}, "mode": "polling"}), &mut vault).unwrap();
1756 assert_eq!(vault.get("TELEGRAM_TOKEN").map(String::as_str), Some("T"));
1757 assert_eq!(
1758 vault.get("TELEGRAM_EXTRA_KEY").map(String::as_str),
1759 Some("K")
1760 );
1761 assert_eq!(
1762 ch.credentials["token"],
1763 SecretRef::Dotenv("TELEGRAM_TOKEN".into())
1764 );
1765 assert_eq!(
1766 ch.credentials["extra.key"],
1767 SecretRef::Dotenv("TELEGRAM_EXTRA_KEY".into())
1768 );
1769 assert_eq!(ch.extra["extra.host"], serde_json::json!("h"));
1770 assert_eq!(ch.extra["mode"], serde_json::json!("polling"));
1771 let ours = encode_channel(&ch, None);
1772 assert_eq!(
1773 ours["token"],
1774 serde_json::json!({"dotenv": "TELEGRAM_TOKEN"})
1775 );
1776 let hermes = encode_channel(&ch, Some(&vault));
1777 assert_eq!(hermes["token"], serde_json::json!("T"));
1778 assert_eq!(hermes["extra"]["key"], serde_json::json!("K"));
1779 }
1780
1781 #[test]
1782 fn fires_obligations_subscriptions() {
1783 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();
1784 let fire = decode_fire_row("executions.db", &row).unwrap();
1785 assert_eq!(fire.status, FireStatus::Succeeded);
1786 assert_eq!(fire.residue.0.get("pid"), Some(&serde_json::json!(4)));
1787 assert_eq!(encode_fire_row(&fire)[6], serde_json::json!("completed"));
1788 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();
1789 let o = decode_obligation_row("state.db", &row).unwrap();
1790 assert_eq!(o.state, ObligationState::Sent);
1791 assert_eq!(o.target.kind.as_deref(), Some("group"));
1792 assert_eq!(o.delivered_at.as_deref(), Some("2.5"));
1793 assert_eq!(encode_obligation_row(&o)[6], serde_json::json!("delivered"));
1794 let mut vault = BTreeMap::new();
1795 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();
1796 assert_eq!(
1797 sub.deliver,
1798 Some(Target::Explicit {
1799 platform: "telegram".into(),
1800 chat_id: Some("1".into()),
1801 thread_id: None
1802 })
1803 );
1804 assert_eq!(
1805 vault.get("WEBHOOK_DEPLOYS_SECRET").map(String::as_str),
1806 Some("S")
1807 );
1808 let back = encode_subscription(&sub, Some(&vault));
1809 assert_eq!(back["secret"], serde_json::json!("S"));
1810 assert_eq!(back["deliver_extra"]["chat_id"], serde_json::json!("1"));
1811 assert_eq!(back["note"], serde_json::json!(1));
1812 assert_eq!(
1813 encode_subscription(&sub, None)["secret"],
1814 serde_json::json!("${dotenv:WEBHOOK_DEPLOYS_SECRET}")
1815 );
1816 let mut ours = BTreeMap::new();
1820 let reread = decode_subscription(
1821 "webhook_subscriptions.json",
1822 "deploys",
1823 &encode_subscription(&sub, None),
1824 &mut ours,
1825 )
1826 .unwrap();
1827 assert_eq!(reread.secret, sub.secret);
1828 assert!(ours.is_empty(), "a placeholder is never a value: {ours:?}");
1829 }
1830}