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 parse_openclaw_session_key(
417 key: &str,
418) -> Option<(Option<String>, SurfaceKey, Trigger, Option<Recurrence>)> {
419 let parts: Vec<&str> = key.split(':').collect();
420 match parts.first().copied() {
421 Some("agent") if parts.len() >= 3 => {
422 let agent = Some(parts[1].to_string());
423 if parts[2] == "main" {
424 let surface = SurfaceKey {
425 key: Some(key.to_string()),
426 kind: Some("main".to_string()),
427 ..SurfaceKey::default()
428 };
429 return Some((agent, surface, Trigger::Unknown, None));
430 }
431 if parts.len() < 5 {
432 return None;
433 }
434 let thread_id = match (parts.get(5), parts.get(6)) {
435 (Some(&"thread"), Some(t)) | (Some(&"topic"), Some(t)) => Some(t.to_string()),
436 _ => None,
437 };
438 let surface = SurfaceKey {
439 key: Some(key.to_string()),
440 platform: Some(parts[2].to_string()),
441 kind: Some(parts[3].to_string()),
442 chat_id: Some(parts[4].to_string()),
443 thread_id,
444 participant_id: None,
445 };
446 Some((agent, surface, Trigger::Channel, None))
447 }
448 Some("cron") if parts.len() >= 2 => Some((
449 None,
450 SurfaceKey {
451 key: Some(key.to_string()),
452 ..SurfaceKey::default()
453 },
454 Trigger::Cron,
455 Some(Recurrence {
456 job_id: parts[1..].join(":"),
457 kind: "cron".into(),
458 }),
459 )),
460 Some("hook") if parts.len() >= 2 => Some((
461 None,
462 SurfaceKey {
463 key: Some(key.to_string()),
464 ..SurfaceKey::default()
465 },
466 Trigger::Webhook,
467 None,
468 )),
469 Some("acp-bridge") => Some((
470 None,
471 SurfaceKey {
472 key: Some(key.to_string()),
473 platform: Some("acp".into()),
474 ..SurfaceKey::default()
475 },
476 Trigger::Api,
477 None,
478 )),
479 _ => None,
480 }
481}
482
483impl Binding {
484 pub fn from_openclaw_key(
488 key: &str,
489 agent_from_path: Option<&str>,
490 session_id: Option<&str>,
491 locator: Option<&str>,
492 ) -> Option<Self> {
493 let (agent, surface, trigger, recurrence) = parse_openclaw_session_key(key)?;
494 Some(Self {
495 key: surface,
496 profile: agent.or_else(|| agent_from_path.map(str::to_string)),
497 worker: Worker {
498 harness: HarnessId::new(HarnessId::OPENCLAW),
499 session_id: session_id.map(str::to_string),
500 locator: locator.map(str::to_string),
501 },
502 trigger,
503 recurrence,
504 ..Self::default()
505 })
506 }
507}
508
509#[derive(Debug, Clone, Default)]
513pub struct OrchestratorBindingRow {
514 pub platform: String,
516 pub chat_type: String,
518 pub chat_id: Option<String>,
520 pub thread_id: Option<String>,
522 pub participant_id: Option<String>,
524 pub worker_harness: String,
526 pub worker_session_id: Option<String>,
528 pub worker_locator: Option<String>,
530 pub started_at: Option<String>,
532 pub last_activity_at: Option<String>,
534 pub ended_at: Option<String>,
536 pub end_reason: Option<String>,
538 pub handoff_to: Option<String>,
540 pub handoff_state: Option<String>,
542 pub handoff_error: Option<String>,
544 pub recurrence_job_id: Option<String>,
546}
547
548impl Binding {
549 pub fn from_orchestrator_row(profile: &str, row: &OrchestratorBindingRow) -> Self {
553 let mut key = SurfaceKey {
554 key: None,
555 platform: Some(row.platform.clone()),
556 kind: Some(row.chat_type.clone()),
557 chat_id: row.chat_id.clone(),
558 thread_id: row.thread_id.clone(),
559 participant_id: row.participant_id.clone(),
560 };
561 key.key = Some(render_hermes_session_key(profile, &key));
562 let trigger = if row.recurrence_job_id.is_some() {
563 Trigger::Cron
564 } else if row.platform == "webhook" {
565 Trigger::Webhook
566 } else {
567 Trigger::Channel
568 };
569 let mut residue = Residue::default();
570 let end_reason = match row.end_reason.as_deref() {
571 Some(word) => match EndReason::parse(word) {
572 Some(r) => Some(r),
573 None => {
574 residue.keep("end_reason", serde_json::Value::String(word.to_string()));
575 None
576 }
577 },
578 None => None,
579 };
580 Self {
581 key,
582 profile: Some(profile.to_string()),
583 worker: Worker {
584 harness: HarnessId::new(&row.worker_harness),
585 session_id: row.worker_session_id.clone().filter(|s| !s.is_empty()),
586 locator: row.worker_locator.clone(),
587 },
588 trigger,
589 recurrence: row.recurrence_job_id.clone().map(|job_id| Recurrence {
590 job_id,
591 kind: "cron".into(),
592 }),
593 handoff: row.handoff_state.clone().map(|state| Handoff {
594 to: row.handoff_to.clone(),
595 state,
596 error: row.handoff_error.clone(),
597 }),
598 started_at: row.started_at.clone(),
599 last_activity_at: row.last_activity_at.clone(),
600 ended_at: row.ended_at.clone(),
601 end_reason,
602 residue,
603 }
604 }
605}
606
607#[cfg(test)]
608mod tests {
609 use super::*;
610
611 #[test]
612 fn hermes_row_columns_win_over_the_key_and_api_server_keeps_its_key() {
613 let row = HermesSessionRow {
614 id: "s1".into(),
615 source: Some("telegram".into()),
616 session_key: Some("agent:coder:telegram:group:-100777:55".into()),
617 chat_id: Some("-100999".into()),
618 profile_name: Some("coder".into()),
619 started_at: Some(1_788_000_000.5),
620 ..Default::default()
621 };
622 let b = Binding::from_hermes_row(&row, Some("state.db"));
623 assert_eq!(b.trigger, Trigger::Channel);
624 assert_eq!(b.key.chat_id.as_deref(), Some("-100999"));
625 assert_eq!(b.key.thread_id.as_deref(), Some("55"));
626 assert_eq!(b.profile.as_deref(), Some("coder"));
627 assert_eq!(b.started_at.as_deref(), Some("2026-08-29T10:40:00.500Z"));
628 let n = b.nouns();
629 assert_eq!(
630 n.surface
631 .as_ref()
632 .and_then(|s| s.platform.clone())
633 .as_deref(),
634 Some("telegram")
635 );
636
637 let api = HermesSessionRow {
638 id: "s2".into(),
639 source: Some("api_server".into()),
640 session_key: Some("agent:main:chat:dm:ada-dm".into()),
641 ..Default::default()
642 };
643 let b = Binding::from_hermes_row(&api, None);
644 assert_eq!(b.trigger, Trigger::Api);
645 assert_eq!(b.key.chat_id.as_deref(), Some("ada-dm"));
646 assert_eq!(b.profile, None);
647 }
648
649 #[test]
650 fn hermes_cron_and_delegate_and_terminal_rows() {
651 let cron = HermesSessionRow {
652 id: "cron_job42_20260902_120000".into(),
653 source: Some("cron".into()),
654 ..Default::default()
655 };
656 let b = Binding::from_hermes_row(&cron, None);
657 assert_eq!(b.trigger, Trigger::Cron);
658 assert_eq!(
659 b.recurrence.as_ref().map(|r| r.job_id.as_str()),
660 Some("job42")
661 );
662 let child = HermesSessionRow {
663 id: "c".into(),
664 source: Some("cli".into()),
665 lineage_kind: Some("delegate".into()),
666 ..Default::default()
667 };
668 assert_eq!(
669 Binding::from_hermes_row(&child, None).trigger,
670 Trigger::Parent
671 );
672 let terminal = HermesSessionRow {
673 id: "t".into(),
674 source: Some("cli".into()),
675 end_reason: Some("weird".into()),
676 ..Default::default()
677 };
678 let b = Binding::from_hermes_row(&terminal, None);
679 assert_eq!(b.surface(), None, "a terminal session has a degenerate key");
680 assert_eq!(b.nouns().surface, None);
681 assert_eq!(b.end_reason, None);
682 assert_eq!(
683 b.residue.0.get("end_reason").and_then(|v| v.as_str()),
684 Some("weird")
685 );
686 }
687
688 #[test]
689 fn openclaw_keys_and_orchestrator_rows() {
690 let b = Binding::from_openclaw_key(
691 "agent:ops:telegram:group:-1:thread:7",
692 None,
693 Some("u1"),
694 None,
695 )
696 .unwrap();
697 assert_eq!(b.profile.as_deref(), Some("ops"));
698 assert_eq!(b.key.thread_id.as_deref(), Some("7"));
699 assert_eq!(b.trigger, Trigger::Channel);
700 let c = Binding::from_openclaw_key("cron:abc:def", Some("ops"), None, None).unwrap();
701 assert_eq!(
702 c.recurrence.as_ref().map(|r| r.job_id.as_str()),
703 Some("abc:def")
704 );
705 assert_eq!(c.profile.as_deref(), Some("ops"));
706 assert!(Binding::from_openclaw_key("nonsense", None, None, None).is_none());
707
708 let row = OrchestratorBindingRow {
709 platform: "telegram".into(),
710 chat_type: "dm".into(),
711 chat_id: Some("123456".into()),
712 worker_harness: "codex".into(),
713 worker_session_id: Some("sess-1".into()),
714 end_reason: Some("idle".into()),
715 ended_at: Some("2026-09-04T10:00:00.000Z".into()),
716 ..Default::default()
717 };
718 let b = Binding::from_orchestrator_row("default", &row);
719 assert_eq!(
720 b.key.key.as_deref(),
721 Some("agent:default:telegram:dm:123456")
722 );
723 assert_eq!(b.trigger, Trigger::Channel);
724 assert_eq!(b.end_reason, Some(EndReason::Idle));
725 let fire = OrchestratorBindingRow {
726 platform: "cron".into(),
727 chat_type: "dm".into(),
728 chat_id: Some("job42".into()),
729 recurrence_job_id: Some("job42".into()),
730 worker_harness: "hermes".into(),
731 worker_session_id: Some("f".into()),
732 ..Default::default()
733 };
734 assert_eq!(
735 Binding::from_orchestrator_row("default", &fire)
736 .nouns()
737 .trigger,
738 Some(Trigger::Cron)
739 );
740 let hook = OrchestratorBindingRow {
741 platform: "webhook".into(),
742 chat_type: "dm".into(),
743 worker_harness: "hermes".into(),
744 worker_session_id: Some("w".into()),
745 ..Default::default()
746 };
747 assert_eq!(
748 Binding::from_orchestrator_row("default", &hook).trigger,
749 Trigger::Webhook
750 );
751 let (parsed, profile) = parse_hermes_session_key(b.key.key.as_deref().unwrap()).unwrap();
753 assert_eq!(parsed.chat_id, b.key.chat_id);
754 assert_eq!(profile, None);
755 }
756}