hatchery_server/
m4a_gateway.rs

1//! Core's m4a gateway: core talks to sessions only through m4a (decisions 4 and 8).
2//!
3//! - **wake**: `POST /v1/sessions/{id}/wake` sends an m4a letter to the session's nick;
4//!   the node's letter hook wakes the session with a short line, and the session reads
5//!   the letter (and any task text in it) from its own m4a inbox.
6//! - **reader**: core polls the DM rooms of sessions that hold an active run and acts on
7//!   letters of the form `run.close <run_id> <review|done|failed|cancelled> [branch] [commit]`
8//!   from the run's own session nick (door `task.close_run` stays as a fallback).
9//!
10//! Core uses its own m4a identity through the m4a CLI and a running m4a client; it never
11//! holds keys itself. Off unless `HATCHERY_CORE_M4A_CLI` and `HATCHERY_CORE_M4A_AS` are set.
12
13use serde_json::Value;
14use std::path::PathBuf;
15use std::process::{Command, Stdio};
16use std::time::Duration;
17
18pub const CLI_ENV: &str = "HATCHERY_CORE_M4A_CLI";
19pub const AS_ENV: &str = "HATCHERY_CORE_M4A_AS";
20
21#[derive(Clone, Debug, PartialEq, Eq)]
22pub struct Letter {
23    pub event_id: String,
24    pub from: String,
25    pub body: String,
26}
27
28pub trait M4a: Send + Sync {
29    fn send(&self, to: &str, text: &str) -> Result<(), String>;
30    /// Newest `n` letters of the DM room with `nick`, oldest first.
31    fn read_dm(&self, nick: &str, n: u32) -> Result<Vec<Letter>, String>;
32    fn me(&self) -> &str;
33}
34
35pub struct CliM4a {
36    bin: PathBuf,
37    as_nick: String,
38}
39
40impl CliM4a {
41    pub fn from_env() -> Option<Self> {
42        let bin = std::env::var(CLI_ENV).ok().filter(|v| !v.is_empty())?;
43        let as_nick = std::env::var(AS_ENV).ok().filter(|v| !v.is_empty())?;
44        Some(Self { bin: bin.into(), as_nick })
45    }
46
47    fn run(&self, args: &[&str], stdin: Option<&str>) -> Result<String, String> {
48        use std::io::Write;
49        let mut child = Command::new(&self.bin)
50            .args(["--as", &self.as_nick])
51            .args(args)
52            .stdin(if stdin.is_some() { Stdio::piped() } else { Stdio::null() })
53            .stdout(Stdio::piped())
54            .stderr(Stdio::piped())
55            .spawn()
56            .map_err(|e| format!("m4a: {e}"))?;
57        if let (Some(text), Some(mut pipe)) = (stdin, child.stdin.take()) {
58            pipe.write_all(text.as_bytes()).map_err(|e| format!("m4a: {e}"))?;
59        }
60        let deadline = std::time::Instant::now() + Duration::from_secs(20);
61        loop {
62            if let Some(status) = child.try_wait().map_err(|e| e.to_string())? {
63                let out = child.wait_with_output().map_err(|e| e.to_string())?;
64                return if status.success() {
65                    Ok(String::from_utf8_lossy(&out.stdout).into_owned())
66                } else {
67                    // stderr may name rooms; it never carries keys.
68                    Err(format!("m4a exited {}: {}", status.code().unwrap_or(-1), String::from_utf8_lossy(&out.stderr).chars().take(200).collect::<String>()))
69                };
70            }
71            if std::time::Instant::now() > deadline {
72                let _ = child.kill();
73                return Err("m4a timed out".into());
74            }
75            std::thread::sleep(Duration::from_millis(50));
76        }
77    }
78}
79
80pub fn parse_rows(out: &str) -> Vec<Letter> {
81    let v: Value = serde_json::from_str(out.trim()).unwrap_or(Value::Null);
82    let rows = v.as_array().cloned().or_else(|| v["messages"].as_array().cloned()).or_else(|| v["rows"].as_array().cloned()).unwrap_or_default();
83    let mut letters: Vec<(i64, Letter)> = rows
84        .iter()
85        .filter(|r| r["kind"].as_str() != Some("undecryptable"))
86        .filter_map(|r| {
87            Some((
88                r["ts"].as_i64().unwrap_or(0),
89                Letter { event_id: r["event_id"].as_str()?.to_owned(), from: r["from"].as_str()?.to_owned(), body: r["body"].as_str().unwrap_or("").to_owned() },
90            ))
91        })
92        .collect();
93    letters.sort_by_key(|(ts, _)| *ts);
94    letters.into_iter().map(|(_, l)| l).collect()
95}
96
97impl M4a for CliM4a {
98    fn send(&self, to: &str, text: &str) -> Result<(), String> {
99        self.run(&["send", to, "-"], Some(text)).map(|_| ())
100    }
101    fn read_dm(&self, nick: &str, n: u32) -> Result<Vec<Letter>, String> {
102        self.run(&["--json", "read", nick, "-n", &n.to_string()], None).map(|o| parse_rows(&o))
103    }
104    fn me(&self) -> &str {
105        &self.as_nick
106    }
107}
108
109/// `run.close <run_id> <status> [branch] [commit]`, first line only.
110#[derive(Debug, PartialEq, Eq)]
111pub struct CloseCmd {
112    pub run_id: i64,
113    pub status: String,
114    pub branch: Option<String>,
115    pub commit: Option<String>,
116}
117
118pub fn parse_close(body: &str) -> Option<CloseCmd> {
119    let line = body.lines().next()?.trim();
120    let mut w = line.split_whitespace();
121    if w.next()? != "run.close" {
122        return None;
123    }
124    let run_id = w.next()?.trim_start_matches('#').parse().ok()?;
125    let status = w.next()?.to_owned();
126    let clean = |s: &str| Some(s.to_owned()).filter(|s| s.len() <= 200 && !s.chars().any(char::is_control));
127    Some(CloseCmd { run_id, status, branch: w.next().and_then(clean), commit: w.next().and_then(clean) })
128}
129
130/// Strip a matrix-style `@nick:server` to `nick` for comparisons.
131pub fn bare_nick(n: &str) -> &str {
132    let n = n.trim_start_matches('@');
133    n.split(':').next().unwrap_or(n)
134}
135
136#[cfg(test)]
137mod tests {
138    use super::*;
139
140    #[test]
141    fn parses_close_letters() {
142        assert_eq!(parse_close("run.close 7 done feat/x abc123\nthanks"), Some(CloseCmd { run_id: 7, status: "done".into(), branch: Some("feat/x".into()), commit: Some("abc123".into()) }));
143        assert_eq!(parse_close("run.close #3 review").unwrap().run_id, 3);
144        assert!(parse_close("hello run.close 1 done").is_none());
145        assert!(parse_close("run.close x done").is_none());
146    }
147
148    #[test]
149    fn parses_rows_oldest_first_and_skips_undecryptable() {
150        let out = r#"[{"event_id":"$b","from":"@s1:h","body":"two","ts":2},{"event_id":"$a","from":"@s1:h","body":"one","ts":1},{"event_id":"$c","from":"x","kind":"undecryptable","ts":3}]"#;
151        let l = parse_rows(out);
152        assert_eq!(l.iter().map(|x| x.body.as_str()).collect::<Vec<_>>(), ["one", "two"]);
153        assert_eq!(bare_nick("@s1:h"), "s1");
154        assert!(parse_rows("garbage").is_empty());
155    }
156}
hatchery_server/m4a_gateway.rs

hatchery_server/
m4a_gateway.rs
