agentmux_launcher/notify/
mod.rs

1// Copyright 2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! OS notification **Presenter** — `docs/specs/SPEC_OS_NOTIFICATIONS_SYSTEM_2026_09_24.md` §3.2, §6.
5//!
6//! The srv Router decides *whether* to notify and publishes `notification` /
7//! `notification:retract` on its event bus. This module is the launcher's
8//! subscriber: it keeps a WebSocket to srv (the launcher's first), renders
9//! each notification with the platform backend, and reports the user's
10//! click/dismiss back as `notify.ack`.
11//!
12//! Why the launcher (spec §3.2): it outlives host restarts, stays up with no
13//! window in background mode, owns the tray, and — on macOS — is the bundle's
14//! main executable. Click callbacks need a live process; this is the
15//! longest-lived one.
16//!
17//! ## Lifecycle
18//!
19//! `start` once srv is up; `update_endpoint` whenever srv is respawned (new
20//! port + auth key); `shutdown` before the launcher exits, which clears every
21//! toast this app put in the notification center — a click on a toast from a
22//! dead instance could do nothing useful (Phase 1 has no cold-start
23//! activation, spec §6.1).
24
25use agentmux_common::AUTH_KEY_HEADER;
26use std::sync::{Arc, OnceLock};
27
28use futures_util::{SinkExt, StreamExt};
29use serde::Deserialize;
30use tokio::sync::{mpsc, watch};
31use tokio_tungstenite::tungstenite::{ClientRequestBuilder, Message};
32
33#[cfg(target_os = "windows")]
34mod windows;
35#[cfg(target_os = "linux")]
36mod linux;
37
38/// Mirrors srv's `OsNotification` (`crates/srv/src/backend/notify/policy.rs`).
39/// Unknown fields are ignored so srv can grow the payload first.
40#[derive(Clone, Debug, Deserialize, PartialEq, Eq)]
41pub struct Notification {
42    pub id: String,
43    pub kind: String,
44    pub priority: String,
45    pub title: String,
46    #[serde(default)]
47    pub body: Option<String>,
48    /// What the agent is working on — a small line under the body
49    /// (`SPEC_OS_NOTIFICATIONS_RICH_CONTENT_AND_CLICK_TO_PANE_2026_09_25.md`
50    /// §1.3). Absent from an older srv, and when the agent has none.
51    #[serde(default)]
52    pub summary: Option<String>,
53    pub tag: String,
54}
55
56impl Notification {
57    pub fn is_attention(&self) -> bool {
58        self.priority == "attention"
59    }
60}
61
62/// What the user did — on a toast (reported by a backend) or in the tray's
63/// notification menu (`tray_request`).
64#[derive(Clone, Debug, PartialEq, Eq)]
65pub enum UserAction {
66    Clicked(String),
67    Dismissed(String),
68    /// Set `notify:pause:until` (epoch ms; 0 = resume).
69    SetPause(i64),
70    /// Set `app:startatlogin` (the tray's check item; `start_at_login`).
71    SetStartAtLogin(bool),
72}
73
74/// A platform notification surface.
75pub trait Presenter: Send + Sync {
76    fn show(&self, n: &Notification);
77    fn retract(&self, tag: &str);
78    fn clear_all(&self);
79}
80
81/// No platform backend (macOS today — needs a signed bundle and a Mac to
82/// verify, spec §6.2): log and drop.
83struct NullPresenter;
84impl Presenter for NullPresenter {
85    fn show(&self, n: &Notification) {
86        crate::logging::log(&format!("notify: no OS backend on this platform; dropping '{}'", n.kind));
87    }
88    fn retract(&self, _tag: &str) {}
89    fn clear_all(&self) {}
90}
91
92#[derive(Clone, Debug, PartialEq, Eq)]
93struct Endpoint {
94    ws: String,
95    auth_key: String,
96}
97
98struct Handle {
99    endpoint: watch::Sender<Option<Endpoint>>,
100    presenter: Arc<dyn Presenter>,
101}
102
103/// The launcher's reducer state — where a clicked toast's window is looked up
104/// (backend window id → label → HWND).
105pub type LauncherState = Arc<tokio::sync::Mutex<crate::state::State>>;
106
107static HANDLE: OnceLock<Handle> = OnceLock::new();
108
109/// The user-action channel exists from first use, not from `start`: the tray
110/// (and its Pause/Resume/Open menu) comes up before srv does, and a pick made
111/// in that window must queue until the presenter connects rather than vanish
112/// (ReAgent P2 on #3654). `start` takes the receiver.
113type ActionChannel = (mpsc::UnboundedSender<UserAction>, std::sync::Mutex<Option<mpsc::UnboundedReceiver<UserAction>>>);
114static ACTIONS: OnceLock<ActionChannel> = OnceLock::new();
115
116fn actions() -> &'static ActionChannel {
117    ACTIONS.get_or_init(|| {
118        let (tx, rx) = mpsc::unbounded_channel();
119        (tx, std::sync::Mutex::new(Some(rx)))
120    })
121}
122
123fn make_presenter(actions: mpsc::UnboundedSender<UserAction>, data_dir: &std::path::Path) -> Arc<dyn Presenter> {
124    #[cfg(target_os = "windows")]
125    {
126        match windows::WindowsPresenter::spawn(actions, data_dir) {
127            Ok(p) => return Arc::new(p),
128            Err(e) => crate::logging::log(&format!("notify: Windows toast backend unavailable: {e}")),
129        }
130    }
131    #[cfg(target_os = "linux")]
132    {
133        let _ = data_dir;
134        return Arc::new(linux::LinuxPresenter::spawn(actions));
135    }
136    #[cfg(not(any(target_os = "windows", target_os = "linux")))]
137    let _ = (actions, data_dir);
138    #[allow(unreachable_code)]
139    Arc::new(NullPresenter)
140}
141
142/// Start the presenter. Idempotent: a second call only updates the endpoint.
143pub fn start(
144    ws_endpoint: &str,
145    auth_key: &str,
146    data_dir: std::path::PathBuf,
147    dir_hash: String,
148    state: LauncherState,
149) {
150    let ep = Endpoint { ws: ws_endpoint.to_string(), auth_key: auth_key.to_string() };
151    if let Some(h) = HANDLE.get() {
152        let _ = h.endpoint.send(Some(ep));
153        return;
154    }
155    let act_tx = actions().0.clone();
156    let Some(act_rx) = actions().1.lock().unwrap_or_else(|e| e.into_inner()).take() else {
157        return;
158    };
159    let presenter = make_presenter(act_tx, &data_dir);
160    // A fresh launcher owns nothing on screen yet — anything left over from a
161    // previous (crashed) run points at a dead instance.
162    presenter.clear_all();
163    let (ep_tx, ep_rx) = watch::channel(Some(ep));
164    if HANDLE.set(Handle { endpoint: ep_tx, presenter: presenter.clone() }).is_err() {
165        return;
166    }
167    tokio::spawn(run(ep_rx, act_rx, presenter, data_dir, dir_hash, state));
168}
169
170/// srv was respawned: reconnect to its new endpoint. The toasts on screen
171/// name ids the new srv never issued — clear them rather than leave clicks
172/// that can't find their pane (rich-content spec §3.3 F).
173#[cfg(target_os = "windows")]
174pub fn update_endpoint(ws_endpoint: &str, auth_key: &str) {
175    if let Some(h) = HANDLE.get() {
176        h.presenter.clear_all();
177        let _ = h.endpoint.send(Some(Endpoint { ws: ws_endpoint.to_string(), auth_key: auth_key.to_string() }));
178    }
179}
180
181/// The tray's notification menu (spec §4.2): open an attention item exactly
182/// as its toast click would, or pause/resume. Non-blocking; callable from the
183/// tray thread.
184pub fn tray_request(action: crate::tray::notify_menu::NotifyMenuAction) {
185    use crate::tray::notify_menu::{resolve_pause, NotifyMenuAction};
186    let ua = match action {
187        NotifyMenuAction::Open(id) => {
188            // The menu pick made us foreground-eligible; hand that to the host
189            // so it can raise its window.
190            #[cfg(target_os = "windows")]
191            unsafe {
192                let _ = ::windows::Win32::UI::WindowsAndMessaging::AllowSetForegroundWindow(
193                    ::windows::Win32::UI::WindowsAndMessaging::ASFW_ANY,
194                );
195            }
196            UserAction::Clicked(id)
197        }
198        NotifyMenuAction::Pause(choice) => UserAction::SetPause(resolve_pause(choice, chrono::Local::now())),
199        NotifyMenuAction::Resume => UserAction::SetPause(0),
200    };
201    // Queues even before `start` — delivered once the presenter connects.
202    let _ = actions().0.send(ua);
203}
204
205/// Write `app:startatlogin` through srv, like any settings change, so the
206/// Settings page and the OS login entry follow it (`start_at_login`). Queues
207/// until the session is up.
208pub fn request_start_at_login(on: bool) {
209    let _ = actions().0.send(UserAction::SetStartAtLogin(on));
210}
211
212/// Launcher is exiting: remove our toasts from the notification center.
213pub fn shutdown() {
214    if let Some(h) = HANDLE.get() {
215        h.presenter.clear_all();
216    }
217}
218
219/// `host:port` (srv's ESTART form) or a full URL → the `/ws` URL.
220pub fn ws_url(endpoint: &str) -> String {
221    let base = endpoint.trim_end_matches('/');
222    let base = if base.starts_with("ws://") || base.starts_with("wss://") {
223        base.to_string()
224    } else {
225        format!("ws://{base}")
226    };
227    format!("{base}/ws")
228}
229
230/// What an incoming srv frame means to the presenter.
231#[derive(Debug, PartialEq, Eq)]
232pub enum Frame {
233    Show(Notification),
234    Retract(String),
235    State(crate::tray::notify_menu::NotifyTrayState),
236    /// A full-config broadcast: the value of `app:startatlogin` (`None` = unset).
237    StartAtLogin(Option<bool>),
238    Response { reqid: String, ack: AckResponse },
239    Other,
240}
241
242/// srv's `notify.ack` answer (`NotifyAckResult`).
243#[derive(Clone, Debug, Default, PartialEq, Eq, Deserialize)]
244pub struct AckResponse {
245    #[serde(default)]
246    pub has_window: Option<bool>,
247    #[serde(default)]
248    pub window_id: Option<String>,
249    #[serde(default)]
250    pub workspace_id: Option<String>,
251}
252
253/// What a click makes the launcher do (rich-content spec §3.3 C, D, F).
254#[derive(Debug, PartialEq, Eq)]
255pub enum ClickPlan {
256    /// Raise this open window (backend window id); its frontend selects the
257    /// pane from `notification:activate`.
258    Raise(String),
259    /// The pane's workspace is open in no window: open one on it.
260    OpenWorkspace(String),
261    /// No window at all (background mode, or the pane is gone): open one.
262    OpenWindow,
263    Nothing,
264}
265
266/// Pure: srv decided the target; this only maps its answer to an action.
267pub fn click_plan(ack: &AckResponse) -> ClickPlan {
268    if let Some(w) = ack.window_id.as_ref().filter(|w| !w.is_empty()) {
269        return ClickPlan::Raise(w.clone());
270    }
271    if let Some(ws) = ack.workspace_id.as_ref().filter(|w| !w.is_empty()) {
272        return ClickPlan::OpenWorkspace(ws.clone());
273    }
274    if ack.has_window == Some(false) {
275        return ClickPlan::OpenWindow;
276    }
277    ClickPlan::Nothing
278}
279
280/// The host window showing backend window `window_id`: `(label, hwnd, iconic)`.
281pub fn window_for(state: &crate::state::State, window_id: &str) -> Option<(String, Option<u64>, bool)> {
282    let label = state.backend_window_ids.iter().find(|(_, w)| w.as_str() == window_id).map(|(l, _)| l.clone())?;
283    let mirror = state.windows.get(&label);
284    Some((label, mirror.and_then(|m| m.hwnd), mirror.is_some_and(|m| m.iconic)))
285}
286
287/// Parse one srv WS text frame. Events arrive as
288/// `{"eventtype":"rpc","data":{"command":"eventrecv","data":<MuxEvent>}}`;
289/// RPC responses as `{"eventtype":"rpc","data":{"resid":..,"data":..}}`.
290pub fn parse_frame(text: &str) -> Frame {
291    let Ok(v) = serde_json::from_str::<serde_json::Value>(text) else {
292        return Frame::Other;
293    };
294    let Some(data) = v.get("data") else { return Frame::Other };
295    if let Some(resid) = data.get("resid").and_then(|r| r.as_str()) {
296        return Frame::Response {
297            reqid: resid.to_string(),
298            ack: data.get("data").and_then(|d| serde_json::from_value(d.clone()).ok()).unwrap_or_default(),
299        };
300    }
301    if data.get("command").and_then(|c| c.as_str()) != Some("eventrecv") {
302        return Frame::Other;
303    }
304    let Some(ev) = data.get("data") else { return Frame::Other };
305    match ev.get("event").and_then(|e| e.as_str()) {
306        Some("notification") => ev
307            .get("data")
308            .and_then(|d| serde_json::from_value::<Notification>(d.clone()).ok())
309            .map(Frame::Show)
310            .unwrap_or(Frame::Other),
311        Some("notification:state") => ev
312            .get("data")
313            .and_then(|d| serde_json::from_value(d.clone()).ok())
314            .map(Frame::State)
315            .unwrap_or(Frame::Other),
316        Some("config") => ev
317            .get("data")
318            .and_then(|d| d.get("fullconfig"))
319            .and_then(|c| c.get("settings"))
320            .map(|s| Frame::StartAtLogin(s.get(crate::start_at_login::SETTINGS_KEY).and_then(|v| v.as_bool())))
321            .unwrap_or(Frame::Other),
322        Some("notification:retract") => ev
323            .get("data")
324            .and_then(|d| d.get("tag"))
325            .and_then(|t| t.as_str())
326            .map(|t| Frame::Retract(t.to_string()))
327            .unwrap_or(Frame::Other),
328        _ => Frame::Other,
329    }
330}
331
332fn rpc(command: &str, reqid: Option<&str>, data: serde_json::Value) -> String {
333    let mut msg = serde_json::json!({ "command": command, "data": data });
334    if let Some(r) = reqid {
335        msg["reqid"] = serde_json::Value::String(r.to_string());
336    }
337    serde_json::json!({ "wscommand": "rpc", "message": msg }).to_string()
338}
339
340async fn run(
341    mut ep_rx: watch::Receiver<Option<Endpoint>>,
342    mut act_rx: mpsc::UnboundedReceiver<UserAction>,
343    presenter: Arc<dyn Presenter>,
344    data_dir: std::path::PathBuf,
345    dir_hash: String,
346    state: LauncherState,
347) {
348    let mut backoff_ms = 500u64;
349    loop {
350        let ep = ep_rx.borrow_and_update().clone();
351        let Some(ep) = ep else {
352            if ep_rx.changed().await.is_err() {
353                return;
354            }
355            continue;
356        };
357        match session(&ep, &mut ep_rx, &mut act_rx, &presenter, &data_dir, &dir_hash, &state).await {
358            SessionEnd::EndpointChanged => {
359                backoff_ms = 500;
360                continue;
361            }
362            SessionEnd::Closed(reason) => {
363                crate::logging::log(&format!("notify: srv connection ended ({reason}); retrying in {backoff_ms}ms"));
364            }
365        }
366        tokio::select! {
367            _ = tokio::time::sleep(std::time::Duration::from_millis(backoff_ms)) => {}
368            r = ep_rx.changed() => if r.is_err() { return; },
369        }
370        backoff_ms = (backoff_ms * 2).min(30_000);
371    }
372}
373
374enum SessionEnd {
375    EndpointChanged,
376    Closed(String),
377}
378
379async fn session(
380    ep: &Endpoint,
381    ep_rx: &mut watch::Receiver<Option<Endpoint>>,
382    act_rx: &mut mpsc::UnboundedReceiver<UserAction>,
383    presenter: &Arc<dyn Presenter>,
384    data_dir: &std::path::Path,
385    dir_hash: &str,
386    state: &LauncherState,
387) -> SessionEnd {
388    let uri: tokio_tungstenite::tungstenite::http::Uri = match ws_url(&ep.ws).parse() {
389        Ok(u) => u,
390        Err(e) => return SessionEnd::Closed(format!("bad endpoint: {e}")),
391    };
392    // Header auth (`X-AuthKey`), as srv's auth_middleware accepts for /ws.
393    // ClientRequestBuilder generates the handshake headers itself — see
394    // cloud_subscriber.rs for why a hand-built Request fails the handshake.
395    let request = ClientRequestBuilder::new(uri).with_header(AUTH_KEY_HEADER, ep.auth_key.clone());
396    let (ws, _) = match tokio_tungstenite::connect_async(request).await {
397        Ok(s) => s,
398        Err(e) => return SessionEnd::Closed(format!("connect: {e}")),
399    };
400    let (mut write, mut read) = ws.split();
401    for event in ["notification", "notification:retract", "notification:state"] {
402        let sub = rpc("eventsub", None, serde_json::json!({ "event": event, "scopes": [], "allscopes": true }));
403        if let Err(e) = write.send(Message::Text(sub.into())).await {
404            return SessionEnd::Closed(format!("subscribe: {e}"));
405        }
406    }
407    crate::logging::log("notify: subscribed to srv notifications");
408
409    // reqid → was it a click (needs has_window handling)?
410    let mut pending_clicks: std::collections::HashSet<String> = Default::default();
411
412    loop {
413        tokio::select! {
414            changed = ep_rx.changed() => {
415                if changed.is_err() || ep_rx.borrow().as_ref() != Some(ep) {
416                    return SessionEnd::EndpointChanged;
417                }
418            }
419            action = act_rx.recv() => {
420                let Some(action) = action else { return SessionEnd::Closed("presenter gone".into()) };
421                let (id, clicked) = match &action {
422                    UserAction::Clicked(id) => (id.clone(), true),
423                    UserAction::Dismissed(id) => (id.clone(), false),
424                    UserAction::SetPause(until) => {
425                        let msg = rpc("setconfig", Some(&uuid::Uuid::new_v4().to_string()), serde_json::json!({ "notify:pause:until": until }));
426                        if let Err(e) = write.send(Message::Text(msg.into())).await {
427                            return SessionEnd::Closed(format!("send setconfig: {e}"));
428                        }
429                        continue;
430                    }
431                    UserAction::SetStartAtLogin(on) => {
432                        let msg = rpc("setconfig", Some(&uuid::Uuid::new_v4().to_string()), serde_json::json!({ crate::start_at_login::SETTINGS_KEY: on }));
433                        if let Err(e) = write.send(Message::Text(msg.into())).await {
434                            return SessionEnd::Closed(format!("send setconfig: {e}"));
435                        }
436                        continue;
437                    }
438                };
439                let reqid = uuid::Uuid::new_v4().to_string();
440                if clicked {
441                    pending_clicks.insert(reqid.clone());
442                }
443                let msg = rpc("notify.ack", Some(&reqid), serde_json::json!({ "id": id, "clicked": clicked }));
444                if let Err(e) = write.send(Message::Text(msg.into())).await {
445                    return SessionEnd::Closed(format!("send ack: {e}"));
446                }
447            }
448            msg = read.next() => {
449                let text = match msg {
450                    Some(Ok(Message::Text(t))) => t.to_string(),
451                    Some(Ok(Message::Close(_))) | None => return SessionEnd::Closed("closed by srv".into()),
452                    Some(Ok(_)) => continue,
453                    Some(Err(e)) => return SessionEnd::Closed(format!("read: {e}")),
454                };
455                match parse_frame(&text) {
456                    Frame::Show(n) => presenter.show(&n),
457                    Frame::Retract(tag) => presenter.retract(&tag),
458                    Frame::State(state) => crate::tray::notify_menu::set(state),
459                    Frame::StartAtLogin(v) => crate::start_at_login::observe(v),
460                    Frame::Response { reqid, ack } => {
461                        if pending_clicks.remove(&reqid) {
462                            let plan = click_plan(&ack);
463                            crate::logging::log(&format!("notify: click → {plan:?}"));
464                            let window = match &plan {
465                                ClickPlan::Raise(w) => window_for(&*state.lock().await, w),
466                                _ => None,
467                            };
468                            let (dd, dh) = (data_dir.to_path_buf(), dir_hash.to_string());
469                            tokio::task::spawn_blocking(move || carry_out_click(plan, window, &dd, &dh));
470                        }
471                    }
472                    Frame::Other => {}
473                }
474            }
475        }
476    }
477}
478
479/// Raise first, in this process: the toast click made the launcher the
480/// foreground-eligible process, and that right is what a raise needs —
481/// waiting for the pane selection to go srv → renderer → host first can
482/// lose it (rich-content spec §3.3 C, F4). Every branch logs.
483fn carry_out_click(
484    plan: ClickPlan,
485    window: Option<(String, Option<u64>, bool)>,
486    data_dir: &std::path::Path,
487    dir_hash: &str,
488) {
489    use crate::second_instance::{forward_host_cmd_with_args, forward_open_new_window, ForwardError};
490    let log_err = |what: &str, r: Result<(), ForwardError>| {
491        if let Err(ForwardError::Transient(e) | ForwardError::Fatal(e)) = r {
492            crate::logging::log(&format!("notify: {what} for click failed: {e}"));
493        }
494    };
495    match plan {
496        ClickPlan::Raise(window_id) => {
497            let Some((label, hwnd, iconic)) = window else {
498                // The window reported itself to srv but the launcher hasn't
499                // mapped it (yet): let the host raise it by its frontend's own
500                // request instead — the frontend calls focus_window itself.
501                crate::logging::log(&format!("notify: click window {window_id} not in the launcher's mirror"));
502                return;
503            };
504            if hwnd.is_some_and(|h| raise_hwnd(h, iconic)) {
505                return;
506            }
507            crate::logging::log(&format!("notify: raising {label} directly failed; asking the host"));
508            log_err("focus_window", forward_host_cmd_with_args(data_dir, dir_hash, "focus_window", serde_json::json!({ "label": label })));
509            if let Some(h) = hwnd {
510                flash_hwnd(h);
511            }
512        }
513        ClickPlan::OpenWorkspace(workspace_id) => log_err(
514            "open_new_window {workspace_id}",
515            forward_host_cmd_with_args(data_dir, dir_hash, "open_new_window", serde_json::json!({ "workspace_id": workspace_id })),
516        ),
517        ClickPlan::OpenWindow => log_err("open_new_window", forward_open_new_window(data_dir, dir_hash)),
518        ClickPlan::Nothing => {}
519    }
520}
521
522/// Restore if minimized, then bring to the foreground. False when Windows
523/// refused the foreground change.
524#[cfg(target_os = "windows")]
525fn raise_hwnd(hwnd: u64, iconic: bool) -> bool {
526    use windows_sys::Win32::UI::WindowsAndMessaging::{IsIconic, SetForegroundWindow, ShowWindow, SW_RESTORE};
527    let h = hwnd as usize as windows_sys::Win32::Foundation::HWND;
528    unsafe {
529        if iconic || IsIconic(h) != 0 {
530            ShowWindow(h, SW_RESTORE);
531        }
532        SetForegroundWindow(h) != 0
533    }
534}
535
536#[cfg(not(target_os = "windows"))]
537fn raise_hwnd(_hwnd: u64, _iconic: bool) -> bool {
538    false
539}
540
541/// Windows' own fallback when the foreground change is refused: flash the
542/// taskbar button so the user at least sees where to go.
543#[cfg(target_os = "windows")]
544fn flash_hwnd(hwnd: u64) {
545    use windows_sys::Win32::UI::WindowsAndMessaging::{FlashWindowEx, FLASHWINFO, FLASHW_ALL, FLASHW_TIMERNOFG};
546    let info = FLASHWINFO {
547        cbSize: std::mem::size_of::<FLASHWINFO>() as u32,
548        hwnd: hwnd as usize as windows_sys::Win32::Foundation::HWND,
549        dwFlags: FLASHW_ALL | FLASHW_TIMERNOFG,
550        uCount: 3,
551        dwTimeout: 0,
552    };
553    unsafe {
554        FlashWindowEx(&info);
555    }
556}
557
558#[cfg(not(target_os = "windows"))]
559fn flash_hwnd(_hwnd: u64) {}
560
561#[cfg(test)]
562mod tests {
563    use super::*;
564
565    #[test]
566    fn click_plan_follows_srvs_answer() {
567        let ack = |window: Option<&str>, ws: Option<&str>, has: Option<bool>| AckResponse {
568            has_window: has,
569            window_id: window.map(Into::into),
570            workspace_id: ws.map(Into::into),
571        };
572        assert_eq!(click_plan(&ack(Some("w1"), None, Some(true))), ClickPlan::Raise("w1".into()));
573        assert_eq!(click_plan(&ack(None, Some("ws1"), Some(false))), ClickPlan::OpenWorkspace("ws1".into()));
574        assert_eq!(click_plan(&ack(None, None, Some(false))), ClickPlan::OpenWindow);
575        // An older srv: only has_window.
576        assert_eq!(click_plan(&ack(None, None, Some(true))), ClickPlan::Nothing);
577        assert_eq!(click_plan(&ack(Some(""), Some(""), None)), ClickPlan::Nothing);
578    }
579
580    #[test]
581    fn window_for_maps_backend_id_to_label_hwnd_and_minimized() {
582        let mut s = crate::state::State::default();
583        s.backend_window_ids.insert("main".into(), "wid-1".into());
584        assert_eq!(window_for(&s, "wid-1"), Some(("main".into(), None, false)), "label known before its HWND");
585        assert_eq!(window_for(&s, "wid-2"), None);
586    }
587
588    #[test]
589    fn ws_url_forms() {
590        assert_eq!(ws_url("127.0.0.1:5000"), "ws://127.0.0.1:5000/ws");
591        assert_eq!(ws_url("ws://127.0.0.1:5000/"), "ws://127.0.0.1:5000/ws");
592    }
593
594    #[test]
595    fn parses_notification_event() {
596        let f = r#"{"eventtype":"rpc","data":{"command":"eventrecv","data":{"event":"notification","data":{
597            "id":"n1-1","kind":"input_waiting","priority":"attention","block_id":"b1","agent_name":"lark",
598            "title":"lark needs your input","body":"Which branch?","tag":"abcdef0123456789","created_at_ms":1}}}}"#;
599        match parse_frame(f) {
600            Frame::Show(n) => {
601                assert_eq!(n.id, "n1-1");
602                assert!(n.is_attention());
603                assert_eq!(n.body.as_deref(), Some("Which branch?"));
604                assert_eq!(n.tag, "abcdef0123456789");
605            }
606            other => panic!("unexpected {other:?}"),
607        }
608    }
609
610    #[test]
611    fn parses_retract_and_response_and_ignores_others() {
612        let r = r#"{"eventtype":"rpc","data":{"command":"eventrecv","data":{"event":"notification:retract","data":{"id":"n1","tag":"t1"}}}}"#;
613        assert_eq!(parse_frame(r), Frame::Retract("t1".into()));
614        let resp = r#"{"eventtype":"rpc","data":{"resid":"q1","data":{"has_window":false}}}"#;
615        assert_eq!(
616            parse_frame(resp),
617            Frame::Response { reqid: "q1".into(), ack: AckResponse { has_window: Some(false), ..Default::default() } }
618        );
619        let resp = r#"{"eventtype":"rpc","data":{"resid":"q2","data":{"has_window":true,"known":true,"window_id":"w1"}}}"#;
620        assert_eq!(
621            parse_frame(resp),
622            Frame::Response {
623                reqid: "q2".into(),
624                ack: AckResponse { has_window: Some(true), window_id: Some("w1".into()), workspace_id: None }
625            }
626        );
627        let st = r#"{"eventtype":"rpc","data":{"command":"eventrecv","data":{"event":"notification:state","data":{"attention":[{"id":"n1","block_id":"b1","title":"lark needs your input"}],"paused_until_ms":5}}}}"#;
628        match parse_frame(st) {
629            Frame::State(s) => {
630                assert_eq!(s.attention.len(), 1);
631                assert_eq!(s.paused_until_ms, 5);
632            }
633            other => panic!("unexpected {other:?}"),
634        }
635        let cfg = r#"{"eventtype":"rpc","data":{"command":"eventrecv","data":{"event":"config","data":{}}}}"#;
636        assert_eq!(parse_frame(cfg), Frame::Other);
637        let with = |settings: &str| {
638            format!(r#"{{"eventtype":"rpc","data":{{"command":"eventrecv","data":{{"event":"config","data":{{"fullconfig":{{"settings":{settings}}}}}}}}}}}"#)
639        };
640        assert_eq!(parse_frame(&with(r#"{"app:startatlogin":true}"#)), Frame::StartAtLogin(Some(true)));
641        assert_eq!(parse_frame(&with(r#"{"app:startatlogin":false}"#)), Frame::StartAtLogin(Some(false)));
642        assert_eq!(parse_frame(&with(r#"{"term:fontsize":12}"#)), Frame::StartAtLogin(None));
643        assert_eq!(parse_frame(r#"{"type":"ping","stime":1}"#), Frame::Other);
644        assert_eq!(parse_frame("garbage"), Frame::Other);
645    }
646
647    /// End-to-end over a real socket: a fake srv checks the auth header and
648    /// subscriptions, pushes a notification + retract, and receives the ack
649    /// the presenter's click produces.
650    #[tokio::test]
651    async fn session_round_trip_against_fake_srv() {
652        use std::sync::Mutex;
653        use tokio::net::TcpListener;
654        use tokio_tungstenite::tungstenite::handshake::server::{Request, Response};
655
656        #[derive(Default)]
657        struct Fake {
658            calls: Mutex<Vec<String>>,
659        }
660        impl Presenter for Fake {
661            fn show(&self, n: &Notification) {
662                self.calls.lock().unwrap().push(format!("show:{}", n.id));
663            }
664            fn retract(&self, tag: &str) {
665                self.calls.lock().unwrap().push(format!("retract:{tag}"));
666            }
667            fn clear_all(&self) {}
668        }
669
670        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
671        let addr = listener.local_addr().unwrap();
672        let (got_tx, mut got_rx) = mpsc::unbounded_channel::<serde_json::Value>();
673        tokio::spawn(async move {
674            let (sock, _) = listener.accept().await.unwrap();
675            let mut ws = tokio_tungstenite::accept_hdr_async(sock, |req: &Request, resp: Response| {
676                assert_eq!(req.uri().path(), "/ws");
677                assert_eq!(req.headers().get("X-AuthKey").unwrap(), "k1");
678                Ok(resp)
679            })
680            .await
681            .unwrap();
682            // Three eventsubs first.
683            for _ in 0..3 {
684                let m = ws.next().await.unwrap().unwrap();
685                got_tx.send(serde_json::from_str(m.to_text().unwrap()).unwrap()).unwrap();
686            }
687            let show = r#"{"eventtype":"rpc","data":{"command":"eventrecv","data":{"event":"notification","data":{"id":"n1","kind":"turn_completed","priority":"normal","title":"lark finished","body":null,"tag":"t1"}}}}"#;
688            ws.send(Message::Text(show.into())).await.unwrap();
689            let retract = r#"{"eventtype":"rpc","data":{"command":"eventrecv","data":{"event":"notification:retract","data":{"id":"n1","tag":"t1"}}}}"#;
690            ws.send(Message::Text(retract.into())).await.unwrap();
691            // The ack produced by the simulated click.
692            let m = ws.next().await.unwrap().unwrap();
693            got_tx.send(serde_json::from_str(m.to_text().unwrap()).unwrap()).unwrap();
694        });
695
696        let fake = Arc::new(Fake::default());
697        let presenter: Arc<dyn Presenter> = fake.clone();
698        let (_ep_tx, ep_rx) = watch::channel(Some(Endpoint { ws: addr.to_string(), auth_key: "k1".into() }));
699        let (act_tx, act_rx) = mpsc::unbounded_channel();
700        let dir = std::env::temp_dir();
701        let state: LauncherState = Default::default();
702        tokio::spawn(run(ep_rx, act_rx, presenter, dir, "h".into(), state));
703
704        let sub1 = got_rx.recv().await.unwrap();
705        let sub2 = got_rx.recv().await.unwrap();
706        let sub3 = got_rx.recv().await.unwrap();
707        assert_eq!(sub3["message"]["data"]["event"], "notification:state");
708        assert_eq!(sub1["message"]["command"], "eventsub");
709        assert_eq!(sub1["message"]["data"]["event"], "notification");
710        assert_eq!(sub2["message"]["data"]["event"], "notification:retract");
711
712        // Wait until show + retract reached the presenter, then click.
713        for _ in 0..100 {
714            if fake.calls.lock().unwrap().len() >= 2 {
715                break;
716            }
717            tokio::time::sleep(std::time::Duration::from_millis(10)).await;
718        }
719        assert_eq!(*fake.calls.lock().unwrap(), vec!["show:n1".to_string(), "retract:t1".to_string()]);
720        act_tx.send(UserAction::Clicked("n1".into())).unwrap();
721        let ack = got_rx.recv().await.unwrap();
722        assert_eq!(ack["message"]["command"], "notify.ack");
723        assert_eq!(ack["message"]["data"]["id"], "n1");
724        assert_eq!(ack["message"]["data"]["clicked"], true);
725        assert!(ack["message"]["reqid"].is_string());
726    }
727
728    #[test]
729    fn rpc_envelope_shape() {
730        let v: serde_json::Value = serde_json::from_str(&rpc("notify.ack", Some("r1"), serde_json::json!({"id":"n1","clicked":true}))).unwrap();
731        assert_eq!(v["wscommand"], "rpc");
732        assert_eq!(v["message"]["command"], "notify.ack");
733        assert_eq!(v["message"]["reqid"], "r1");
734        assert_eq!(v["message"]["data"]["clicked"], true);
735    }
736}
agentmux_launcher/notify/mod.rs

agentmux_launcher/notify/
mod.rs
