agentmux_launcher/notify/
linux.rs

1// Copyright 2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! Linux notification backend — `SPEC_OS_NOTIFICATIONS_SYSTEM_2026_09_24.md` §6.3.
5//!
6//! freedesktop Desktop Notifications (`org.freedesktop.Notifications`) spoken
7//! directly over async zbus on the launcher's Tokio runtime — no libnotify,
8//! no GTK, no blocking D-Bus calls. One task owns the session-bus connection:
9//!
10//! - `Notify` with `replaces_id` = the server id last used for this tag, so a
11//!   same-group update replaces in place (spec §5.3);
12//! - `CloseNotification` for retract / clear-all;
13//! - `ActionInvoked("default")` → click, `NotificationClosed(reason 2)` →
14//!   user dismissal (1 = expired, 3 = closed by us — neither is feedback).
15//!
16//! Clicks only reach a process still on the bus — fine, the launcher lives for
17//! the whole session (and clears its notifications on exit).
18//!
19//! Content safety (§9.1): if the server advertises `body-markup`, the body is
20//! escaped so agent-influenced text can't inject markup. The summary is never
21//! markup per the spec.
22
23use std::collections::HashMap;
24use std::sync::mpsc as std_mpsc;
25
26use futures_util::StreamExt;
27use tokio::sync::mpsc;
28use zbus::zvariant::Value;
29
30use super::{Notification, Presenter, UserAction};
31
32const DEST: &str = "org.freedesktop.Notifications";
33const PATH: &str = "/org/freedesktop/Notifications";
34const IFACE: &str = "org.freedesktop.Notifications";
35
36enum Cmd {
37    Show(Notification),
38    Retract(String),
39    ClearAll(std_mpsc::Sender<()>),
40}
41
42pub struct LinuxPresenter {
43    tx: mpsc::UnboundedSender<Cmd>,
44}
45
46impl LinuxPresenter {
47    pub fn spawn(actions: mpsc::UnboundedSender<UserAction>) -> Self {
48        let (tx, rx) = mpsc::unbounded_channel();
49        tokio::spawn(run(rx, actions));
50        Self { tx }
51    }
52}
53
54impl Presenter for LinuxPresenter {
55    fn show(&self, n: &Notification) {
56        let _ = self.tx.send(Cmd::Show(n.clone()));
57    }
58    fn retract(&self, tag: &str) {
59        let _ = self.tx.send(Cmd::Retract(tag.to_string()));
60    }
61    fn clear_all(&self) {
62        let (done_tx, done_rx) = std_mpsc::channel();
63        if self.tx.send(Cmd::ClearAll(done_tx)).is_ok() {
64            let _ = done_rx.recv_timeout(std::time::Duration::from_secs(2));
65        }
66    }
67}
68
69/// Escape for servers that parse the body as (a subset of) markup.
70pub fn escape_markup(s: &str) -> String {
71    s.replace('&', "&amp;").replace('<', "&lt;").replace('>', "&gt;")
72}
73
74/// `(summary, body)` exactly as sent — pure, so the escaping rule is testable.
75/// (The freedesktop "summary" is our title.) The agent's session summary has
76/// no slot of its own on Linux, so it is the body's last line (rich-content
77/// spec §1.3), escaped like the rest of the body.
78pub fn texts(n: &Notification, body_markup: bool) -> (String, String) {
79    let body = [n.body.as_deref(), n.summary.as_deref()].into_iter().flatten().collect::<Vec<_>>().join("\n");
80    (n.title.clone(), if body_markup { escape_markup(&body) } else { body })
81}
82
83/// `(urgency, expire_timeout, resident)` for a notification.
84///
85/// Attention (needs input / crashed / needs review) must not silently expire
86/// before the user sees it (ReAgent P1 on #3668) — so it never expires
87/// (`expire_timeout = 0`) and is `resident` (stays in the list after an
88/// action). It deliberately stays at urgency **1**, not 2: "critical" breaks
89/// through Do Not Disturb on GNOME, which the spec forbids (§5.1, §6.0) —
90/// Windows' High priority doesn't bypass Focus either.
91pub fn delivery(n: &Notification) -> (u8, i32, bool) {
92    if n.is_attention() {
93        (1, 0, true)
94    } else {
95        (1, -1, false)
96    }
97}
98
99/// Map a `NotificationClosed` reason to user feedback: only 2 ("dismissed by
100/// the user") counts.
101pub fn closed_is_user_dismissal(reason: u32) -> bool {
102    reason == 2
103}
104
105async fn run(mut rx: mpsc::UnboundedReceiver<Cmd>, actions: mpsc::UnboundedSender<UserAction>) {
106    let conn = match zbus::Connection::session().await {
107        Ok(c) => c,
108        Err(e) => {
109            crate::logging::log(&format!("notify: no D-Bus session bus ({e}); notifications disabled"));
110            // Keep draining so callers (clear_all's bounded wait) never stall.
111            while let Some(cmd) = rx.recv().await {
112                if let Cmd::ClearAll(done) = cmd {
113                    let _ = done.send(());
114                }
115            }
116            return;
117        }
118    };
119
120    let body_markup = match conn.call_method(Some(DEST), PATH, Some(IFACE), "GetCapabilities", &()).await {
121        Ok(reply) => reply.body().deserialize::<Vec<String>>().map(|c| c.iter().any(|x| x == "body-markup")).unwrap_or(false),
122        Err(e) => {
123            crate::logging::log(&format!("notify: no notification server ({e})"));
124            false
125        }
126    };
127
128    // Only the notification server's own signals: without `sender`, any
129    // session-bus client could emit a forged ActionInvoked for one of our
130    // server ids (ReAgent P2 on #3668). The bus resolves the well-known name
131    // to its current owner when matching.
132    let rule = zbus::MatchRule::builder()
133        .msg_type(zbus::message::Type::Signal)
134        .sender(DEST)
135        .and_then(|b| b.path(PATH))
136        .and_then(|b| b.interface(IFACE))
137        .map(|b| b.build());
138    let mut signals = match rule {
139        Ok(rule) => zbus::MessageStream::for_match_rule(rule, &conn, None).await.ok(),
140        Err(_) => None,
141    };
142
143    // tag → (server id, our notification id); server id → our id.
144    let mut by_tag: HashMap<String, (u32, String)> = HashMap::new();
145    let mut by_server: HashMap<u32, String> = HashMap::new();
146
147    loop {
148        tokio::select! {
149            cmd = rx.recv() => {
150                let Some(cmd) = cmd else { return };
151                match cmd {
152                    Cmd::Show(n) => {
153                        let replaces = by_tag.get(&n.tag).map(|(sid, _)| *sid).unwrap_or(0);
154                        let (summary, body) = texts(&n, body_markup);
155                        let mut hints: HashMap<&str, Value<'_>> = HashMap::new();
156                        hints.insert("desktop-entry", Value::from("agentmux"));
157                        let (urgency, timeout, resident) = delivery(&n);
158                        hints.insert("urgency", Value::from(urgency));
159                        if resident {
160                            hints.insert("resident", Value::from(true));
161                        }
162                        let args = (
163                            "AgentMux",
164                            replaces,
165                            "agentmux",
166                            summary.as_str(),
167                            body.as_str(),
168                            vec!["default", "Open"],
169                            hints,
170                            timeout,
171                        );
172                        match conn.call_method(Some(DEST), PATH, Some(IFACE), "Notify", &args).await {
173                            Ok(reply) => {
174                                if let Ok(sid) = reply.body().deserialize::<u32>() {
175                                    if let Some((old, _)) = by_tag.insert(n.tag.clone(), (sid, n.id.clone())) {
176                                        by_server.remove(&old);
177                                    }
178                                    by_server.insert(sid, n.id.clone());
179                                }
180                            }
181                            Err(e) => crate::logging::log(&format!("notify: Notify failed: {e}")),
182                        }
183                    }
184                    Cmd::Retract(tag) => {
185                        if let Some((sid, _)) = by_tag.remove(&tag) {
186                            by_server.remove(&sid);
187                            let _ = conn.call_method(Some(DEST), PATH, Some(IFACE), "CloseNotification", &(sid,)).await;
188                        }
189                    }
190                    Cmd::ClearAll(done) => {
191                        for (_, (sid, _)) in by_tag.drain() {
192                            let _ = conn.call_method(Some(DEST), PATH, Some(IFACE), "CloseNotification", &(sid,)).await;
193                        }
194                        by_server.clear();
195                        let _ = done.send(());
196                    }
197                }
198            }
199            msg = async {
200                match signals.as_mut() {
201                    Some(s) => s.next().await,
202                    None => std::future::pending().await,
203                }
204            } => {
205                let Some(Ok(msg)) = msg else { continue };
206                let header = msg.header();
207                let member = header.member().map(|m| m.as_str().to_string()).unwrap_or_default();
208                match member.as_str() {
209                    "ActionInvoked" => {
210                        if let Ok((sid, action)) = msg.body().deserialize::<(u32, String)>() {
211                            if action == "default" {
212                                if let Some(id) = by_server.get(&sid) {
213                                    let _ = actions.send(UserAction::Clicked(id.clone()));
214                                }
215                            }
216                        }
217                    }
218                    "NotificationClosed" => {
219                        if let Ok((sid, reason)) = msg.body().deserialize::<(u32, u32)>() {
220                            if let Some(id) = by_server.remove(&sid) {
221                                by_tag.retain(|_, (s, _)| *s != sid);
222                                if closed_is_user_dismissal(reason) {
223                                    let _ = actions.send(UserAction::Dismissed(id));
224                                }
225                            }
226                        }
227                    }
228                    _ => {}
229                }
230            }
231        }
232    }
233}
234
235#[cfg(test)]
236mod tests {
237    use super::*;
238
239    fn n(body: Option<&str>) -> Notification {
240        Notification {
241            id: "n1".into(),
242            kind: "input_waiting".into(),
243            priority: "attention".into(),
244            title: "lark has a question".into(),
245            body: body.map(Into::into),
246            summary: None,
247            tag: "t1".into(),
248        }
249    }
250
251    #[test]
252    fn body_is_escaped_only_for_markup_servers() {
253        let evil = "<a href='x'>click</a> & more";
254        assert_eq!(texts(&n(Some(evil)), true).1, "&lt;a href='x'&gt;click&lt;/a&gt; &amp; more");
255        assert_eq!(texts(&n(Some(evil)), false).1, evil);
256        assert_eq!(texts(&n(None), true), ("lark has a question".to_string(), String::new()));
257    }
258
259    #[test]
260    fn summary_is_the_last_body_line_and_escaped_with_it() {
261        let with = |body: Option<&str>, summary: &str| Notification { summary: Some(summary.into()), ..n(body) };
262        assert_eq!(texts(&with(Some("Which branch?"), "Fix repaint"), false).1, "Which branch?\nFix repaint");
263        assert_eq!(texts(&with(None, "Fix repaint"), false).1, "Fix repaint");
264        assert_eq!(texts(&with(Some("a"), "<b>x</b>"), true).1, "a\n&lt;b&gt;x&lt;/b&gt;");
265    }
266
267    #[test]
268    fn attention_never_expires_but_never_goes_critical() {
269        let mut a = n(None);
270        a.priority = "attention".into();
271        assert_eq!(delivery(&a), (1, 0, true));
272        let mut normal = n(None);
273        normal.priority = "normal".into();
274        assert_eq!(delivery(&normal), (1, -1, false));
275    }
276
277    #[test]
278    fn only_user_dismissal_is_feedback() {
279        assert!(closed_is_user_dismissal(2));
280        assert!(!closed_is_user_dismissal(1)); // expired
281        assert!(!closed_is_user_dismissal(3)); // closed by us
282        assert!(!closed_is_user_dismissal(4));
283    }
284}
agentmux_launcher/notify/linux.rs

agentmux_launcher/notify/
linux.rs
