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

ljos_cli/
wake.rs
