ljos_cli/
wake.rs

1//! Wake a seat when mail reaches it.
2//!
3//! Mail waits in the pack until the recipient's next prompt reads it, and
4//! an idle agent has no next prompt. A seat names how to wake it in
5//! `wake.toml` beside `harnesses.toml`: a command this machine runs, or a
6//! webhook it posts to. `ljos send` and `ljos reply` start that wake in its
7//! own session after the message is written, so the sender neither waits
8//! on it nor fails with it, and the sender needs to know nothing about the
9//! recipient's runner.
10//!
11//! A command runs as `sh -c` with the recipient's seat in `LJOS_SEAT`, so
12//! `ljos inbox` in it reads the recipient's mail. It is judged by the same
13//! guard a tool call is, when it is set and again each time it runs, and a
14//! `deny` or an `ask` keeps it from running: nobody is there to ask.
15//! A webhook gets only the envelope (seat, sender, message id, issue,
16//! interrupt), never the body, and its key is read from a file or an
17//! environment variable, never from `wake.toml`.
18
19use std::collections::BTreeMap;
20use std::io::Write;
21use std::path::PathBuf;
22use std::time::{Duration, Instant};
23
24use anyhow::{bail, Context, Result};
25use serde::{Deserialize, Serialize};
26
27/// How long a wake command may run before it is stopped, in seconds.
28pub const WAKE_TIMEOUT_SECS: u64 = 30;
29
30/// How long a webhook post may take, in seconds.
31const WEBHOOK_TIMEOUT_SECS: u64 = 10;
32
33/// One seat's wake: a command or a webhook.
34#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
35pub struct Wake {
36    /// A shell line this machine runs.
37    #[serde(default, skip_serializing_if = "Option::is_none")]
38    pub command: Option<String>,
39    /// A URL this machine posts the envelope to, such as a Grok Bot
40    /// routine's webhook.
41    #[serde(default, skip_serializing_if = "Option::is_none")]
42    pub webhook: Option<String>,
43    /// A file holding the webhook's bearer key.
44    #[serde(default, skip_serializing_if = "Option::is_none")]
45    pub key_file: Option<String>,
46    /// An environment variable holding the webhook's bearer key.
47    #[serde(default, skip_serializing_if = "Option::is_none")]
48    pub key_env: Option<String>,
49    /// Seconds before a command is stopped. Absent: [`WAKE_TIMEOUT_SECS`].
50    #[serde(default, skip_serializing_if = "Option::is_none")]
51    pub timeout: Option<u64>,
52}
53
54/// The message a wake is about.
55#[derive(Debug, Clone, Default, PartialEq, Eq)]
56pub struct Event {
57    pub seat: String,
58    pub from: String,
59    pub msg: String,
60    pub issue: Option<String>,
61    pub interrupt: bool,
62}
63
64/// Where the wakes are kept: `wake.toml` beside `harnesses.toml`.
65#[must_use]
66pub fn path() -> PathBuf {
67    super::harnesses_path().with_file_name("wake.toml")
68}
69
70/// Every seat's wake. An absent or unreadable file is no wakes.
71#[must_use]
72pub fn load() -> BTreeMap<String, Wake> {
73    std::fs::read_to_string(path())
74        .ok()
75        .and_then(|text| toml::from_str(&text).ok())
76        .unwrap_or_default()
77}
78
79fn save(table: &BTreeMap<String, Wake>) -> Result<()> {
80    let path = path();
81    if let Some(dir) = path.parent() {
82        std::fs::create_dir_all(dir)?;
83    }
84    let text = format!(
85        "# Wakes `ljos send` starts when mail reaches a seat. `ljos wake --help` sets them.\n{}",
86        toml::to_string(table)?
87    );
88    std::fs::write(&path, text).with_context(|| format!("wake: {}", path.display()))
89}
90
91/// The guard's verdict on a wake command: the seat's own guard, the
92/// trusted computing base, then the pack's rules and the ask pack.
93/// `None` lets it run.
94#[must_use]
95pub fn judge(command: &str) -> Option<super::Rule> {
96    super::seat_guard(command)
97        .or_else(|| super::tcb_verdict(command))
98        .or_else(|| super::ruled_verdict(&super::rules_for_gate(), command))
99}
100
101fn refusal(command: &str, rule: &super::Rule) -> String {
102    format!(
103        "the guard says {} on `{command}`: {} (rule `{}`). A wake runs with nobody to ask, so it does not run",
104        rule.verdict, rule.reason, rule.pattern
105    )
106}
107
108fn check_webhook(url: &str) -> Result<()> {
109    let loopback = ["http://127.0.0.1", "http://localhost", "http://[::1]"]
110        .iter()
111        .any(|p| url.starts_with(p));
112    if !url.starts_with("https://") && !loopback {
113        bail!("wake: {url} is not https, nor http on this machine");
114    }
115    Ok(())
116}
117
118/// Set a seat's wake command, after the guard passes it.
119///
120/// # Errors
121///
122/// The guard refuses the command, or `wake.toml` cannot be written.
123pub fn set_command(seat: &str, command: &str, timeout: Option<u64>) -> Result<String> {
124    let command = command.trim();
125    if command.is_empty() {
126        bail!("wake: empty command");
127    }
128    if let Some(rule) = judge(command) {
129        bail!("wake: {}", refusal(command, &rule));
130    }
131    let mut table = load();
132    table.insert(
133        seat.to_string(),
134        Wake {
135            command: Some(command.to_string()),
136            timeout,
137            ..Wake::default()
138        },
139    );
140    save(&table)?;
141    Ok(format!(
142        "wake {seat}: runs `{command}` when mail reaches it\n"
143    ))
144}
145
146/// Set a seat's webhook. The key stays in its file or variable.
147///
148/// # Errors
149///
150/// The URL is not https or loopback, or `wake.toml` cannot be written.
151pub fn set_webhook(
152    seat: &str,
153    url: &str,
154    key_file: Option<&str>,
155    key_env: Option<&str>,
156) -> Result<String> {
157    let url = url.trim();
158    check_webhook(url)?;
159    let mut table = load();
160    table.insert(
161        seat.to_string(),
162        Wake {
163            webhook: Some(url.to_string()),
164            key_file: key_file.map(str::to_string),
165            key_env: key_env.map(str::to_string),
166            ..Wake::default()
167        },
168    );
169    save(&table)?;
170    Ok(format!(
171        "wake {seat}: posts to {url} when mail reaches it\n"
172    ))
173}
174
175/// Remove a seat's wake.
176///
177/// # Errors
178///
179/// `wake.toml` cannot be written.
180pub fn clear(seat: &str) -> Result<String> {
181    let mut table = load();
182    if table.remove(seat).is_none() {
183        return Ok(format!("wake {seat}: none set\n"));
184    }
185    save(&table)?;
186    Ok(format!("wake {seat}: removed\n"))
187}
188
189/// What a seat's wake is, in one line.
190#[must_use]
191pub fn show(seat: &str) -> String {
192    match load().get(seat) {
193        Some(Wake {
194            command: Some(c), ..
195        }) => format!("wake {seat}: runs `{c}`\n"),
196        Some(Wake {
197            webhook: Some(u), ..
198        }) => format!("wake {seat}: posts to {u}\n"),
199        _ => format!(
200            "wake {seat}: none set, so mail waits for its next prompt. `ljos wake --command CMD` sets one\n"
201        ),
202    }
203}
204
205fn log_path() -> PathBuf {
206    std::env::var_os("XDG_STATE_HOME")
207        .map(PathBuf::from)
208        .or_else(|| std::env::var_os("HOME").map(|h| PathBuf::from(h).join(".local/state")))
209        .unwrap_or_else(|| PathBuf::from(".local/state"))
210        .join("ljos")
211        .join("mail")
212        .join("wake.log")
213}
214
215fn log(line: &str) {
216    let path = log_path();
217    if let Some(dir) = path.parent() {
218        let _ = std::fs::create_dir_all(dir);
219    }
220    if let Ok(mut f) = std::fs::OpenOptions::new()
221        .create(true)
222        .append(true)
223        .open(&path)
224    {
225        let _ = writeln!(f, "{} {line}", super::now_utc());
226    }
227}
228
229/// The `ljos` a detached wake runs. `LJOS_WAKE_BIN` is the test stand-in.
230fn wake_bin() -> Option<String> {
231    if let Some(bin) = std::env::var_os("LJOS_WAKE_BIN").filter(|b| !b.is_empty()) {
232        return Some(bin.to_string_lossy().into_owned());
233    }
234    let me = std::env::current_exe().ok();
235    let beside = me.as_ref().map(|m| m.with_file_name("ljos"));
236    me.filter(|m| m.file_stem().is_some_and(|s| s == "ljos"))
237        .or(beside.filter(|p| p.is_file()))
238        .or_else(|| which::which("ljos").ok())
239        .map(|p| p.display().to_string())
240}
241
242/// Start the wake of each recipient that has one, each in its own session,
243/// and name the seats woken. Nothing here fails the send: a wake that
244/// cannot start is a line in the wake log.
245#[must_use]
246pub fn spawn(events: &[Event]) -> Vec<String> {
247    let table = load();
248    let mut woken = Vec::new();
249    for event in events.iter().filter(|e| table.contains_key(&e.seat)) {
250        let Some(bin) = wake_bin() else {
251            log(&format!("{}: no ljos to run the wake", event.seat));
252            continue;
253        };
254        let mut args = vec![
255            "wake".to_string(),
256            event.seat.clone(),
257            "--fire".into(),
258            "--from".into(),
259            event.from.clone(),
260            "--msg".into(),
261            event.msg.clone(),
262        ];
263        if let Some(issue) = &event.issue {
264            args.push("--issue".into());
265            args.push(issue.clone());
266        }
267        if event.interrupt {
268            args.push("--interrupt".into());
269        }
270        // The detached run writes its outcome to the wake log itself, and
271        // what it prints would only repeat it.
272        match super::detach(&bin, &args, std::path::Path::new("/dev/null")) {
273            Ok(()) => woken.push(event.seat.clone()),
274            Err(e) => log(&format!("{}: the wake did not start: {e:#}", event.seat)),
275        }
276    }
277    woken
278}
279
280/// The envelope a webhook gets. The body stays in the pack.
281#[must_use]
282pub fn envelope(event: &Event) -> serde_json::Value {
283    serde_json::json!({
284        "event": "ljos-mail",
285        "seat": event.seat,
286        "from": event.from,
287        "msg": event.msg,
288        "issue": event.issue,
289        "interrupt": event.interrupt,
290        "read": format!("LJOS_SEAT={} ljos inbox", event.seat),
291    })
292}
293
294fn webhook_key(wake: &Wake) -> Option<String> {
295    let from_file = wake.key_file.as_deref().and_then(|p| {
296        let path = p
297            .strip_prefix("~/")
298            .and_then(|rest| std::env::var_os("HOME").map(|h| PathBuf::from(h).join(rest)))
299            .unwrap_or_else(|| PathBuf::from(p));
300        std::fs::read_to_string(path).ok()
301    });
302    from_file
303        .or_else(|| wake.key_env.as_deref().and_then(|v| std::env::var(v).ok()))
304        .map(|k| k.trim().to_string())
305        .filter(|k| !k.is_empty())
306}
307
308fn run_command(command: &str, event: &Event, timeout: Duration) -> Result<String> {
309    if let Some(rule) = judge(command) {
310        bail!("{}", refusal(command, &rule));
311    }
312    let mut cmd = std::process::Command::new("sh");
313    cmd.arg("-c")
314        .arg(command)
315        .env("LJOS_SEAT", &event.seat)
316        .env("LJOS_WAKE_SEAT", &event.seat)
317        .env("LJOS_WAKE_FROM", &event.from)
318        .env("LJOS_WAKE_MSG", &event.msg)
319        .env("LJOS_WAKE_ISSUE", event.issue.as_deref().unwrap_or(""))
320        .env(
321            "LJOS_WAKE_INTERRUPT",
322            if event.interrupt { "1" } else { "0" },
323        )
324        .stdin(std::process::Stdio::null());
325    // What the command prints goes to the wake log, above the line that
326    // says how it ended.
327    let path = log_path();
328    if let Some(dir) = path.parent() {
329        let _ = std::fs::create_dir_all(dir);
330    }
331    if let Ok(out) = std::fs::OpenOptions::new()
332        .create(true)
333        .append(true)
334        .open(&path)
335    {
336        if let Ok(err) = out.try_clone() {
337            cmd.stdout(out).stderr(err);
338        }
339    }
340    let mut child = cmd.spawn().context("start the wake command")?;
341    let started = Instant::now();
342    loop {
343        if let Some(status) = child.try_wait()? {
344            if status.success() {
345                return Ok(format!("ran in {} ms", started.elapsed().as_millis()));
346            }
347            bail!("exited {status}");
348        }
349        if started.elapsed() >= timeout {
350            // The child's own pid, which this process started.
351            let _ = child.kill();
352            let _ = child.wait();
353            bail!("stopped after {} ms", timeout.as_millis());
354        }
355        std::thread::sleep(Duration::from_millis(20));
356    }
357}
358
359fn post(url: &str, wake: &Wake, event: &Event) -> Result<String> {
360    check_webhook(url)?;
361    let agent = ureq::AgentBuilder::new()
362        .timeout(Duration::from_secs(WEBHOOK_TIMEOUT_SECS))
363        .build();
364    let mut req = agent.post(url).set("Content-Type", "application/json");
365    if let Some(key) = webhook_key(wake) {
366        req = req.set("Authorization", &format!("Bearer {key}"));
367    }
368    let started = Instant::now();
369    let resp = req.send_string(&envelope(event).to_string()).map_err(|e| {
370        anyhow::anyhow!("POST {url}: {}", e.to_string().replace(url, "the webhook"))
371    })?;
372    Ok(format!(
373        "posted, {} in {} ms",
374        resp.status(),
375        started.elapsed().as_millis()
376    ))
377}
378
379/// Run one seat's wake now and say what happened. The outcome is also a
380/// line in the wake log, since a detached wake has nobody to tell.
381///
382/// # Errors
383///
384/// The seat has no wake, the guard refuses its command, or the command or
385/// the post fails.
386pub fn fire(event: &Event) -> Result<String> {
387    let table = load();
388    let Some(wake) = table.get(&event.seat) else {
389        bail!("wake: {} has no wake set", event.seat);
390    };
391    let outcome = if let Some(command) = &wake.command {
392        let timeout = Duration::from_secs(wake.timeout.unwrap_or(WAKE_TIMEOUT_SECS).max(1));
393        run_command(command, event, timeout)
394    } else if let Some(url) = &wake.webhook {
395        post(url, wake, event)
396    } else {
397        Err(anyhow::anyhow!("neither a command nor a webhook"))
398    };
399    match outcome {
400        Ok(said) => {
401            log(&format!("{} {}: {said}", event.seat, event.msg));
402            Ok(format!("wake {}: {said}\n", event.seat))
403        }
404        Err(e) => {
405            log(&format!("{} {}: {e:#}", event.seat, event.msg));
406            Err(e).context(format!("wake {}", event.seat))
407        }
408    }
409}
410
411#[cfg(test)]
412mod tests {
413    use super::*;
414
415    #[test]
416    fn a_wake_reads_back_as_written_and_keeps_no_key() {
417        let mut table = BTreeMap::new();
418        table.insert(
419            "grok".to_string(),
420            Wake {
421                webhook: Some("https://api2.cursor.sh/automations/webhook/abc".into()),
422                key_file: Some("~/.config/ljos/grok.key".into()),
423                ..Wake::default()
424            },
425        );
426        table.insert(
427            "claude-review".to_string(),
428            Wake {
429                command: Some("claude -p --resume \"$SESSION\" 'ljos inbox'".into()),
430                timeout: Some(120),
431                ..Wake::default()
432            },
433        );
434        let text = toml::to_string(&table).unwrap();
435        assert!(!text.contains("key ="), "{text}");
436        let back: BTreeMap<String, Wake> = toml::from_str(&text).unwrap();
437        assert_eq!(back, table);
438    }
439
440    #[test]
441    fn a_webhook_gets_the_envelope_and_not_the_body() {
442        let event = Event {
443            seat: "grok".into(),
444            from: "saddle".into(),
445            msg: "m1".into(),
446            issue: Some("Software-d9n6".into()),
447            interrupt: true,
448        };
449        let v = envelope(&event);
450        assert_eq!(v["seat"], "grok");
451        assert_eq!(v["msg"], "m1");
452        assert_eq!(v["interrupt"], true);
453        assert!(v.get("body").is_none() && v.get("text").is_none(), "{v}");
454        assert!(check_webhook("http://example.com/hook").is_err());
455        assert!(check_webhook("https://api2.cursor.sh/automations/webhook/x").is_ok());
456        assert!(check_webhook("http://127.0.0.1:9/hook").is_ok());
457    }
458
459    #[test]
460    fn a_wake_that_runs_past_its_timeout_is_stopped() {
461        let event = Event {
462            seat: "slow".into(),
463            ..Event::default()
464        };
465        let started = Instant::now();
466        let err = run_command("sleep 5", &event, Duration::from_millis(300)).unwrap_err();
467        assert!(err.to_string().contains("stopped after"), "{err}");
468        assert!(started.elapsed() < Duration::from_secs(3));
469        let ok = run_command("test \"$LJOS_SEAT\" = slow", &event, Duration::from_secs(5));
470        assert!(ok.is_ok(), "{ok:?}");
471    }
472}
ljos_cli/wake.rs

ljos_cli/
wake.rs
