1use schemars::JsonSchema;
8use serde::{Deserialize, Serialize};
9
10use super::residue::Residue;
11use super::surface::{CrossSurface, Recurrence, SurfaceKey, Trigger};
12use super::HarnessId;
13use crate::session::OrchestrationNouns;
14
15#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
17#[serde(rename_all = "snake_case")]
18pub enum EndReason {
19 Idle,
21 Daily,
23 Reset,
25 New,
27 Handoff,
29 Error,
31}
32
33impl EndReason {
34 pub fn parse(word: &str) -> Option<Self> {
36 Some(match word {
37 "idle" => Self::Idle,
38 "daily" => Self::Daily,
39 "reset" => Self::Reset,
40 "new" => Self::New,
41 "handoff" => Self::Handoff,
42 "error" => Self::Error,
43 _ => return None,
44 })
45 }
46
47 pub fn as_str(self) -> &'static str {
49 match self {
50 Self::Idle => "idle",
51 Self::Daily => "daily",
52 Self::Reset => "reset",
53 Self::New => "new",
54 Self::Handoff => "handoff",
55 Self::Error => "error",
56 }
57 }
58}
59
60#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
62pub struct Worker {
63 pub harness: HarnessId,
65 #[serde(default, skip_serializing_if = "Option::is_none")]
67 pub session_id: Option<String>,
68 #[serde(default, skip_serializing_if = "Option::is_none")]
70 pub locator: Option<String>,
71}
72
73#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
75pub struct Handoff {
76 #[serde(default, skip_serializing_if = "Option::is_none")]
78 pub to: Option<String>,
79 pub state: String,
81 #[serde(default, skip_serializing_if = "Option::is_none")]
82 pub error: Option<String>,
84}
85
86impl Default for Worker {
87 fn default() -> Self {
88 Self {
89 harness: HarnessId::new(""),
90 session_id: None,
91 locator: None,
92 }
93 }
94}
95
96#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
98pub struct Binding {
99 pub key: SurfaceKey,
101 #[serde(default)]
103 pub profile: Option<String>,
104 pub worker: Worker,
106 #[serde(default)]
110 pub trigger: Trigger,
111 #[serde(default)]
112 pub recurrence: Option<Recurrence>,
114 #[serde(default)]
115 pub handoff: Option<Handoff>,
117 #[serde(default)]
119 pub started_at: Option<String>,
120 #[serde(default)]
121 pub last_activity_at: Option<String>,
123 #[serde(default)]
124 pub ended_at: Option<String>,
126 #[serde(default)]
127 pub end_reason: Option<EndReason>,
129 #[serde(default)]
131 pub residue: Residue,
132}
133
134impl Default for Binding {
135 fn default() -> Self {
136 Self {
137 key: SurfaceKey::default(),
138 profile: None,
139 worker: Worker::default(),
140 trigger: Trigger::Unknown,
141 recurrence: None,
142 handoff: None,
143 started_at: None,
144 last_activity_at: None,
145 ended_at: None,
146 end_reason: None,
147 residue: Residue::default(),
148 }
149 }
150}
151
152impl Binding {
153 pub fn surface(&self) -> Option<SurfaceKey> {
156 let k = &self.key;
157 if k.key.is_some() || k.platform.is_some() || k.chat_id.is_some() {
158 Some(k.clone())
159 } else {
160 None
161 }
162 }
163
164 pub fn nouns(&self) -> OrchestrationNouns {
168 OrchestrationNouns {
169 trigger: Some(self.trigger),
170 surface: self.surface(),
171 profile: self.profile.clone(),
172 recurrence: self.recurrence.clone(),
173 cross_surface: self.handoff.as_ref().map(|h| CrossSurface {
174 state: h.state.clone(),
175 platform: h.to.clone(),
176 error: h.error.clone(),
177 }),
178 workspace: None,
179 }
180 }
181}
182
183#[derive(Debug, Clone, Default)]
188pub struct HermesSessionRow {
189 pub id: String,
191 pub source: Option<String>,
193 pub lineage_kind: Option<String>,
195 pub session_key: Option<String>,
197 pub chat_id: Option<String>,
199 pub chat_type: Option<String>,
201 pub thread_id: Option<String>,
203 pub user_id: Option<String>,
205 pub profile_name: Option<String>,
207 pub handoff_state: Option<String>,
209 pub handoff_platform: Option<String>,
211 pub handoff_error: Option<String>,
213 pub started_at: Option<f64>,
215 pub ended_at: Option<f64>,
217 pub end_reason: Option<String>,
219}
220
221pub fn hermes_trigger_for_source(source: &str) -> Trigger {
225 match source {
226 "" => Trigger::Unknown,
227 "cron" => Trigger::Cron,
228 "webhook" => Trigger::Webhook,
229 "cli" | "tui" | "acp" | "console" => Trigger::Human,
230 "api_server" | "api" => Trigger::Api,
231 _ => Trigger::Channel,
232 }
233}
234
235pub fn hermes_cron_job_id(session_id: &str) -> Option<String> {
238 let rest = session_id.strip_prefix("cron_")?;
239 let (job, stamp) = rest.rsplit_once('_')?;
240 let (job, date) = job.rsplit_once('_')?;
241 let ok = date.len() == 8
242 && stamp.len() == 6
243 && date.chars().all(|c| c.is_ascii_digit())
244 && stamp.chars().all(|c| c.is_ascii_digit());
245 if ok && !job.is_empty() {
246 Some(job.to_string())
247 } else {
248 None
249 }
250}
251
252pub fn parse_hermes_session_key(key: &str) -> Option<(SurfaceKey, Option<String>)> {
256 let parts: Vec<&str> = key.split(':').collect();
257 if parts.len() < 4 || parts[0] != "agent" {
258 return None;
259 }
260 let profile = match parts[1] {
261 "" | "main" | "default" => None,
262 p => Some(p.to_string()),
263 };
264 let surface = SurfaceKey {
265 key: Some(key.to_string()),
266 platform: Some(parts[2].to_string()),
267 kind: Some(parts[3].to_string()),
268 chat_id: parts.get(4).map(|s| s.to_string()),
269 thread_id: parts.get(5).map(|s| s.to_string()),
270 participant_id: parts.get(6).map(|s| s.to_string()),
271 };
272 Some((surface, profile))
273}
274
275pub fn render_hermes_session_key(profile: &str, key: &SurfaceKey) -> String {
278 let mut parts = vec![
279 "agent".to_string(),
280 if profile.is_empty() {
281 "main".to_string()
282 } else {
283 profile.to_string()
284 },
285 key.platform.clone().unwrap_or_default(),
286 key.kind.clone().unwrap_or_default(),
287 ];
288 parts.extend(
289 [
290 key.chat_id.clone(),
291 key.thread_id.clone(),
292 key.participant_id.clone(),
293 ]
294 .into_iter()
295 .flatten(),
296 );
297 parts.join(":")
298}
299
300fn epoch_to_rfc3339(seconds: f64) -> String {
301 let millis = (seconds * 1000.0).round() as i64;
302 let secs = millis.div_euclid(1000);
303 let sub = millis.rem_euclid(1000) as u32;
304 let days = secs.div_euclid(86_400);
306 let sod = secs.rem_euclid(86_400);
307 let z = days + 719_468;
308 let era = z.div_euclid(146_097);
309 let doe = z - era * 146_097;
310 let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365;
311 let y = yoe + era * 400;
312 let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
313 let mp = (5 * doy + 2) / 153;
314 let d = doy - (153 * mp + 2) / 5 + 1;
315 let m = if mp < 10 { mp + 3 } else { mp - 9 };
316 let y = if m <= 2 { y + 1 } else { y };
317 format!(
318 "{y:04}-{m:02}-{d:02}T{:02}:{:02}:{:02}.{sub:03}Z",
319 sod / 3600,
320 (sod % 3600) / 60,
321 sod % 60
322 )
323}
324
325impl Binding {
326 pub fn from_hermes_row(row: &HermesSessionRow, locator: Option<&str>) -> Self {
331 let nonempty = |v: &Option<String>| v.clone().filter(|s| !s.is_empty());
332 let source = nonempty(&row.source).unwrap_or_default();
333 let mut trigger = hermes_trigger_for_source(&source);
334 if row.lineage_kind.as_deref() == Some("delegate") {
335 trigger = Trigger::Parent;
336 }
337 let mut recurrence = None;
338 if let Some(job_id) = hermes_cron_job_id(&row.id) {
339 recurrence = Some(Recurrence {
340 job_id,
341 kind: "cron".into(),
342 });
343 trigger = Trigger::Cron;
344 }
345 let mut profile = None;
346 let mut key = nonempty(&row.session_key)
347 .and_then(|k| parse_hermes_session_key(&k))
348 .map(|(surface, key_profile)| {
349 profile = key_profile;
350 surface
351 })
352 .unwrap_or_default();
353 if key.key.is_none() {
354 key.key = nonempty(&row.session_key);
355 }
356 if let Some(v) = nonempty(&row.chat_id) {
357 key.chat_id = Some(v);
358 }
359 if let Some(v) = nonempty(&row.chat_type) {
360 key.kind = Some(v);
361 }
362 if let Some(v) = nonempty(&row.thread_id) {
363 key.thread_id = Some(v);
364 }
365 if let Some(v) = nonempty(&row.user_id) {
366 key.participant_id = Some(v);
367 }
368 if key.platform.is_none() && trigger == Trigger::Channel {
369 key.platform = Some(source.clone());
370 }
371 if let Some(p) = nonempty(&row.profile_name) {
372 profile = Some(p);
373 }
374 let handoff = nonempty(&row.handoff_state).map(|state| Handoff {
375 to: nonempty(&row.handoff_platform),
376 state,
377 error: nonempty(&row.handoff_error),
378 });
379 let mut residue = Residue::default();
380 let end_reason = match nonempty(&row.end_reason) {
381 Some(word) => match EndReason::parse(&word) {
382 Some(r) => Some(r),
383 None => {
384 residue.keep("end_reason", serde_json::Value::String(word));
385 None
386 }
387 },
388 None => None,
389 };
390 Self {
391 key,
392 profile,
393 worker: Worker {
394 harness: HarnessId::new(HarnessId::HERMES),
395 session_id: Some(row.id.clone()),
396 locator: locator.map(str::to_string),
397 },
398 trigger,
399 recurrence,
400 handoff,
401 started_at: row.started_at.map(epoch_to_rfc3339),
402 last_activity_at: row.ended_at.or(row.started_at).map(epoch_to_rfc3339),
403 ended_at: row.ended_at.map(epoch_to_rfc3339),
404 end_reason,
405 residue,
406 }
407 }
408}
409
410pub fn hermes_source_for_binding(binding: &Binding) -> String {
414 match binding.trigger {
415 Trigger::Cron => "cron".into(),
416 Trigger::Webhook => "webhook".into(),
417 Trigger::Api => "api_server".into(),
418 Trigger::Human => "cli".into(),
419 Trigger::Parent => "delegate".into(),
420 Trigger::Heartbeat => "heartbeat".into(),
421 Trigger::Channel | Trigger::Unknown => {
422 binding.key.platform.clone().unwrap_or_else(|| "cli".into())
423 }
424 }
425}
426
427pub fn render_openclaw_session_key(agent: &str, binding: &Binding) -> String {
433 let key = &binding.key;
434 if let Some(k) = &key.key {
435 return k.clone();
436 }
437 if let Some(r) = &binding.recurrence {
438 return format!("cron:{}", r.job_id);
439 }
440 if key.kind.as_deref() == Some("main") || key.platform.is_none() {
441 return format!("agent:{agent}:main");
442 }
443 let mut out = format!(
444 "agent:{agent}:{}:{}:{}",
445 key.platform.clone().unwrap_or_default(),
446 key.kind.clone().unwrap_or_else(|| "dm".into()),
447 key.chat_id.clone().unwrap_or_default()
448 );
449 if let Some(t) = &key.thread_id {
450 out.push_str(&format!(":thread:{t}"));
451 }
452 out
453}
454
455pub fn parse_openclaw_session_key(
460 key: &str,
461) -> Option<(Option<String>, SurfaceKey, Trigger, Option<Recurrence>)> {
462 let parts: Vec<&str> = key.split(':').collect();
463 match parts.first().copied() {
464 Some("agent") if parts.len() >= 3 => {
465 let agent = Some(parts[1].to_string());
466 if parts[2] == "main" {
467 let surface = SurfaceKey {
468 key: Some(key.to_string()),
469 kind: Some("main".to_string()),
470 ..SurfaceKey::default()
471 };
472 return Some((agent, surface, Trigger::Unknown, None));
473 }
474 if parts.len() < 5 {
475 return None;
476 }
477 let thread_id = match (parts.get(5), parts.get(6)) {
478 (Some(&"thread"), Some(t)) | (Some(&"topic"), Some(t)) => Some(t.to_string()),
479 _ => None,
480 };
481 let surface = SurfaceKey {
482 key: Some(key.to_string()),
483 platform: Some(parts[2].to_string()),
484 kind: Some(parts[3].to_string()),
485 chat_id: Some(parts[4].to_string()),
486 thread_id,
487 participant_id: None,
488 };
489 Some((agent, surface, Trigger::Channel, None))
490 }
491 Some("cron") if parts.len() >= 2 => Some((
492 None,
493 SurfaceKey {
494 key: Some(key.to_string()),
495 ..SurfaceKey::default()
496 },
497 Trigger::Cron,
498 Some(Recurrence {
499 job_id: parts[1..].join(":"),
500 kind: "cron".into(),
501 }),
502 )),
503 Some("hook") if parts.len() >= 2 => Some((
504 None,
505 SurfaceKey {
506 key: Some(key.to_string()),
507 ..SurfaceKey::default()
508 },
509 Trigger::Webhook,
510 None,
511 )),
512 Some("acp-bridge") => Some((
513 None,
514 SurfaceKey {
515 key: Some(key.to_string()),
516 platform: Some("acp".into()),
517 ..SurfaceKey::default()
518 },
519 Trigger::Api,
520 None,
521 )),
522 _ => None,
523 }
524}
525
526impl Binding {
527 pub fn from_openclaw_key(
531 key: &str,
532 agent_from_path: Option<&str>,
533 session_id: Option<&str>,
534 locator: Option<&str>,
535 ) -> Option<Self> {
536 let (agent, surface, trigger, recurrence) = parse_openclaw_session_key(key)?;
537 Some(Self {
538 key: surface,
539 profile: agent.or_else(|| agent_from_path.map(str::to_string)),
540 worker: Worker {
541 harness: HarnessId::new(HarnessId::OPENCLAW),
542 session_id: session_id.map(str::to_string),
543 locator: locator.map(str::to_string),
544 },
545 trigger,
546 recurrence,
547 ..Self::default()
548 })
549 }
550}
551
552#[derive(Debug, Clone, Default)]
556pub struct OrchestratorBindingRow {
557 pub platform: String,
559 pub chat_type: String,
561 pub chat_id: Option<String>,
563 pub thread_id: Option<String>,
565 pub participant_id: Option<String>,
567 pub worker_harness: String,
569 pub worker_session_id: Option<String>,
571 pub worker_locator: Option<String>,
573 pub started_at: Option<String>,
575 pub last_activity_at: Option<String>,
577 pub ended_at: Option<String>,
579 pub end_reason: Option<String>,
581 pub handoff_to: Option<String>,
583 pub handoff_state: Option<String>,
585 pub handoff_error: Option<String>,
587 pub recurrence_job_id: Option<String>,
589}
590
591impl Binding {
592 pub fn from_orchestrator_row(profile: &str, row: &OrchestratorBindingRow) -> Self {
596 let mut key = SurfaceKey {
597 key: None,
598 platform: Some(row.platform.clone()),
599 kind: Some(row.chat_type.clone()),
600 chat_id: row.chat_id.clone(),
601 thread_id: row.thread_id.clone(),
602 participant_id: row.participant_id.clone(),
603 };
604 key.key = Some(render_hermes_session_key(profile, &key));
605 let trigger = if row.recurrence_job_id.is_some() {
606 Trigger::Cron
607 } else if row.platform == "webhook" {
608 Trigger::Webhook
609 } else {
610 Trigger::Channel
611 };
612 let mut residue = Residue::default();
613 let end_reason = match row.end_reason.as_deref() {
614 Some(word) => match EndReason::parse(word) {
615 Some(r) => Some(r),
616 None => {
617 residue.keep("end_reason", serde_json::Value::String(word.to_string()));
618 None
619 }
620 },
621 None => None,
622 };
623 Self {
624 key,
625 profile: Some(profile.to_string()),
626 worker: Worker {
627 harness: HarnessId::new(&row.worker_harness),
628 session_id: row.worker_session_id.clone().filter(|s| !s.is_empty()),
629 locator: row.worker_locator.clone(),
630 },
631 trigger,
632 recurrence: row.recurrence_job_id.clone().map(|job_id| Recurrence {
633 job_id,
634 kind: "cron".into(),
635 }),
636 handoff: row.handoff_state.clone().map(|state| Handoff {
637 to: row.handoff_to.clone(),
638 state,
639 error: row.handoff_error.clone(),
640 }),
641 started_at: row.started_at.clone(),
642 last_activity_at: row.last_activity_at.clone(),
643 ended_at: row.ended_at.clone(),
644 end_reason,
645 residue,
646 }
647 }
648}
649
650#[cfg(test)]
651mod tests {
652 use super::*;
653
654 #[test]
655 fn hermes_row_columns_win_over_the_key_and_api_server_keeps_its_key() {
656 let row = HermesSessionRow {
657 id: "s1".into(),
658 source: Some("telegram".into()),
659 session_key: Some("agent:coder:telegram:group:-100777:55".into()),
660 chat_id: Some("-100999".into()),
661 profile_name: Some("coder".into()),
662 started_at: Some(1_788_000_000.5),
663 ..Default::default()
664 };
665 let b = Binding::from_hermes_row(&row, Some("state.db"));
666 assert_eq!(b.trigger, Trigger::Channel);
667 assert_eq!(b.key.chat_id.as_deref(), Some("-100999"));
668 assert_eq!(b.key.thread_id.as_deref(), Some("55"));
669 assert_eq!(b.profile.as_deref(), Some("coder"));
670 assert_eq!(b.started_at.as_deref(), Some("2026-08-29T10:40:00.500Z"));
671 let n = b.nouns();
672 assert_eq!(
673 n.surface
674 .as_ref()
675 .and_then(|s| s.platform.clone())
676 .as_deref(),
677 Some("telegram")
678 );
679
680 let api = HermesSessionRow {
681 id: "s2".into(),
682 source: Some("api_server".into()),
683 session_key: Some("agent:main:chat:dm:ada-dm".into()),
684 ..Default::default()
685 };
686 let b = Binding::from_hermes_row(&api, None);
687 assert_eq!(b.trigger, Trigger::Api);
688 assert_eq!(b.key.chat_id.as_deref(), Some("ada-dm"));
689 assert_eq!(b.profile, None);
690 }
691
692 #[test]
693 fn hermes_cron_and_delegate_and_terminal_rows() {
694 let cron = HermesSessionRow {
695 id: "cron_job42_20260902_120000".into(),
696 source: Some("cron".into()),
697 ..Default::default()
698 };
699 let b = Binding::from_hermes_row(&cron, None);
700 assert_eq!(b.trigger, Trigger::Cron);
701 assert_eq!(
702 b.recurrence.as_ref().map(|r| r.job_id.as_str()),
703 Some("job42")
704 );
705 let child = HermesSessionRow {
706 id: "c".into(),
707 source: Some("cli".into()),
708 lineage_kind: Some("delegate".into()),
709 ..Default::default()
710 };
711 assert_eq!(
712 Binding::from_hermes_row(&child, None).trigger,
713 Trigger::Parent
714 );
715 let terminal = HermesSessionRow {
716 id: "t".into(),
717 source: Some("cli".into()),
718 end_reason: Some("weird".into()),
719 ..Default::default()
720 };
721 let b = Binding::from_hermes_row(&terminal, None);
722 assert_eq!(b.surface(), None, "a terminal session has a degenerate key");
723 assert_eq!(b.nouns().surface, None);
724 assert_eq!(b.end_reason, None);
725 assert_eq!(
726 b.residue.0.get("end_reason").and_then(|v| v.as_str()),
727 Some("weird")
728 );
729 }
730
731 #[test]
732 fn openclaw_keys_and_orchestrator_rows() {
733 let b = Binding::from_openclaw_key(
734 "agent:ops:telegram:group:-1:thread:7",
735 None,
736 Some("u1"),
737 None,
738 )
739 .unwrap();
740 assert_eq!(b.profile.as_deref(), Some("ops"));
741 assert_eq!(b.key.thread_id.as_deref(), Some("7"));
742 assert_eq!(b.trigger, Trigger::Channel);
743 let c = Binding::from_openclaw_key("cron:abc:def", Some("ops"), None, None).unwrap();
744 assert_eq!(
745 c.recurrence.as_ref().map(|r| r.job_id.as_str()),
746 Some("abc:def")
747 );
748 assert_eq!(c.profile.as_deref(), Some("ops"));
749 assert!(Binding::from_openclaw_key("nonsense", None, None, None).is_none());
750
751 let row = OrchestratorBindingRow {
752 platform: "telegram".into(),
753 chat_type: "dm".into(),
754 chat_id: Some("123456".into()),
755 worker_harness: "codex".into(),
756 worker_session_id: Some("sess-1".into()),
757 end_reason: Some("idle".into()),
758 ended_at: Some("2026-09-04T10:00:00.000Z".into()),
759 ..Default::default()
760 };
761 let b = Binding::from_orchestrator_row("default", &row);
762 assert_eq!(
763 b.key.key.as_deref(),
764 Some("agent:default:telegram:dm:123456")
765 );
766 assert_eq!(b.trigger, Trigger::Channel);
767 assert_eq!(b.end_reason, Some(EndReason::Idle));
768 let fire = OrchestratorBindingRow {
769 platform: "cron".into(),
770 chat_type: "dm".into(),
771 chat_id: Some("job42".into()),
772 recurrence_job_id: Some("job42".into()),
773 worker_harness: "hermes".into(),
774 worker_session_id: Some("f".into()),
775 ..Default::default()
776 };
777 assert_eq!(
778 Binding::from_orchestrator_row("default", &fire)
779 .nouns()
780 .trigger,
781 Some(Trigger::Cron)
782 );
783 let hook = OrchestratorBindingRow {
784 platform: "webhook".into(),
785 chat_type: "dm".into(),
786 worker_harness: "hermes".into(),
787 worker_session_id: Some("w".into()),
788 ..Default::default()
789 };
790 assert_eq!(
791 Binding::from_orchestrator_row("default", &hook).trigger,
792 Trigger::Webhook
793 );
794 let (parsed, profile) = parse_hermes_session_key(b.key.key.as_deref().unwrap()).unwrap();
796 assert_eq!(parsed.chat_id, b.key.chat_id);
797 assert_eq!(profile, None);
798 }
799}