agentmux_srv/server/
notify_handlers.rs

1// Copyright 2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! RPC surface of the notification Router —
5//! `docs/specs/SPEC_OS_NOTIFICATIONS_SYSTEM_2026_09_24.md` §3.3.
6//!
7//! - `notify.emit`   renderer → router: a pane event the renderer is the
8//!                   authority for (turn outcome, waiting-for-input).
9//! - `notify.focus`  renderer → router: this window's focus state.
10//! - `notify.ack`    presenter → router: user clicked / dismissed a toast.
11//! - `notify.test`   Settings "Send test notification".
12//! - `notify.takeactivation` new window → router: a click that arrived while
13//!                   no window was open.
14//! - `block.reveal`  any renderer → router: reveal a block that lives in
15//!                   ANOTHER window. Same target resolution as a toast click,
16//!                   so it lives with the Router, which knows which windows
17//!                   are open (`SPEC_REVEAL_BLOCK_ONE_PATH_2026_09_27.md` §4.3).
18//!
19//! `conn_id` is captured per connection so focus reports are keyed by window
20//! and dropped on disconnect (see `websocket.rs`'s disconnect path).
21
22use std::sync::Arc;
23
24use crate::backend::notify::policy::{Family, FocusReport};
25use crate::backend::notify::router;
26use crate::backend::rpc::engine::WshRpcEngine;
27
28use super::AppState;
29
30/// Pane-level events the renderer reports. `question` accompanies
31/// `input_waiting` only; the router redacts it (§9.2).
32#[derive(serde::Deserialize, ts_rs::TS)]
33#[serde(rename_all = "snake_case")]
34#[ts(export, export_to = "../../../frontend/types/rpc/")]
35pub enum NotifyPaneEvent {
36    /// Ignored since srv's controller status became the source of truth for
37    /// turns (the renderer's reducer can report turn-ended mid-turn). Kept so
38    /// an older frontend's calls still deserialize.
39    TurnStarted,
40    /// Ignored — see `TurnStarted`.
41    TurnCompleted,
42    /// Ignored — see `TurnStarted`. Real failures arrive as `agentfailure`.
43    TurnErrored,
44    /// The user stopped/interrupted the turn in the pane.
45    TurnStopped,
46    InputWaiting,
47    InputResolved,
48}
49
50#[derive(serde::Deserialize, ts_rs::TS)]
51#[ts(export, export_to = "../../../frontend/types/rpc/")]
52pub struct NotifyEmitParams {
53    pub block_id: String,
54    pub event: NotifyPaneEvent,
55    #[ts(optional)]
56    pub question: Option<String>,
57    /// How many questions the pending call asks; > 1 adds " (+N more)".
58    #[ts(optional)]
59    pub question_count: Option<u32>,
60}
61
62#[derive(serde::Deserialize, ts_rs::TS)]
63#[ts(export, export_to = "../../../frontend/types/rpc/")]
64pub struct NotifyFocusParams {
65    pub window_focused: bool,
66    #[ts(optional)]
67    pub block_id: Option<String>,
68    /// This window's own `Window` oid, so srv knows which windows are open.
69    #[ts(optional)]
70    pub window_id: Option<String>,
71}
72
73#[derive(serde::Deserialize, ts_rs::TS)]
74#[ts(export, export_to = "../../../frontend/types/rpc/")]
75pub struct NotifyAckParams {
76    pub id: String,
77    pub clicked: bool,
78}
79
80/// `notify.attention`: the calls to action srv holds open now
81/// (`backend/user_attention.rs`), each as its `userattention` event's data.
82#[derive(serde::Serialize, serde::Deserialize, ts_rs::TS)]
83#[ts(export, export_to = "../../../frontend/types/rpc/")]
84pub struct NotifyAttentionResult {
85    pub attention: Vec<crate::backend::user_attention::UserAttentionEvent>,
86}
87
88#[derive(serde::Serialize, serde::Deserialize, ts_rs::TS)]
89#[ts(export, export_to = "../../../frontend/types/rpc/")]
90pub struct NotifyAckResult {
91    /// False when the click's window isn't open — an older launcher opens a
92    /// window; the new one picks the pane up via `notify.takeactivation`.
93    pub has_window: bool,
94    /// The acked id named a live notification.
95    #[serde(default)]
96    pub known: bool,
97    /// Raise this window (a connected frontend's `Window` oid).
98    #[serde(default, skip_serializing_if = "Option::is_none")]
99    #[ts(optional)]
100    pub window_id: Option<String>,
101    /// No window shows the pane: open one on this workspace.
102    #[serde(default, skip_serializing_if = "Option::is_none")]
103    #[ts(optional)]
104    pub workspace_id: Option<String>,
105}
106
107#[derive(serde::Deserialize, ts_rs::TS)]
108#[ts(export, export_to = "../../../frontend/types/rpc/")]
109pub struct NotifyTakeActivationParams {
110    /// The asking window's workspace; only a click for that workspace is
111    /// handed over.
112    #[serde(default)]
113    #[ts(optional)]
114    pub workspace_id: Option<String>,
115}
116
117#[derive(serde::Serialize, serde::Deserialize, ts_rs::TS)]
118#[ts(export, export_to = "../../../frontend/types/rpc/")]
119pub struct NotifyTakeActivationResult {
120    #[ts(optional)]
121    pub block_id: Option<String>,
122    #[ts(optional)]
123    pub tab_id: Option<String>,
124}
125
126#[derive(serde::Deserialize, ts_rs::TS)]
127#[ts(export, export_to = "../../../frontend/types/rpc/")]
128pub struct NotifyNoArgs {}
129
130#[derive(serde::Serialize, serde::Deserialize, ts_rs::TS)]
131#[ts(export, export_to = "../../../frontend/types/rpc/")]
132pub struct NotifyOk {
133    pub ok: bool,
134}
135
136#[derive(serde::Deserialize, ts_rs::TS)]
137#[ts(export, export_to = "../../../frontend/types/rpc/")]
138pub struct BlockRevealParams {
139    pub block_id: String,
140}
141
142#[derive(serde::Serialize, serde::Deserialize, ts_rs::TS)]
143#[ts(export, export_to = "../../../frontend/types/rpc/")]
144pub struct BlockRevealResult {
145    /// The block exists and its workspace is open in a window, which was
146    /// told to reveal it (`block:reveal`).
147    pub found: bool,
148    /// That window (a connected frontend's `Window` oid) — the caller raises it.
149    #[serde(default, skip_serializing_if = "Option::is_none")]
150    #[ts(optional)]
151    pub window_id: Option<String>,
152}
153
154fn router_for(state: &AppState) -> Arc<router::Router> {
155    router::init(state.broker.clone(), state.config_watcher.clone(), state.mstore.clone(), state.reactive_handler)
156}
157
158/// `block.reveal`: resolve the block's tab and open window like a toast
159/// click, make the block its pane's visible tab in the STORED layout (so a
160/// window that loads the tab fresh already shows it), then tell only that
161/// window's renderer to reveal it (`block:reveal`). A workspace open in no
162/// window is out of scope: `found: false`, nothing changed.
163pub(crate) async fn reveal_block(
164    state: &AppState,
165    r: &Arc<router::Router>,
166    block_id: String,
167) -> Result<BlockRevealResult, String> {
168    if block_id.is_empty() {
169        return Err("block.reveal: block_id required".to_string());
170    }
171    let rr = r.clone();
172    let target = tokio::task::spawn_blocking(move || rr.reveal_target(&block_id))
173        .await
174        .map_err(|e| format!("block.reveal: {e}"))?;
175    let Some(target) = target else {
176        return Ok(BlockRevealResult { found: false, window_id: None });
177    };
178    activate_in_stored_layout(state, &target.tab_id, &target.block_id).await;
179    r.publish_block_reveal(&target);
180    Ok(BlockRevealResult { found: true, window_id: Some(target.window_id) })
181}
182
183/// `LayoutStackActivate` through the reducer, then mstore — the same
184/// dispatch → apply → publish shape as `pane.moveTab`. Best-effort: the
185/// renderer switches the pane itself on `block:reveal`, so a reducer that
186/// can't apply it (no tree for the tab yet) only costs the head start.
187async fn activate_in_stored_layout(state: &AppState, tab_id: &str, block_id: &str) {
188    let events = crate::server::service::dispatch_to_reducer(
189        state,
190        agentmux_common::ipc::Command::LayoutStackActivate {
191            tab_id: tab_id.to_string(),
192            block_id: block_id.to_string(),
193            correlation_id: String::new(),
194        },
195    )
196    .await;
197    if let Some(msg) = events.iter().find_map(|e| match e {
198        agentmux_common::ipc::Event::Error { message, .. } => Some(message.clone()),
199        _ => None,
200    }) {
201        tracing::debug!(block_id, "block.reveal: stored layout not updated: {msg}");
202        return;
203    }
204    for ev in &events {
205        if let Err(e) = crate::persist_subscriber::apply_event_to_mstore(ev, &state.mstore) {
206            tracing::warn!(block_id, "block.reveal: mstore apply failed: {e}");
207        }
208    }
209    crate::server::service::publish_events(state, &events);
210}
211
212pub fn register_notify_handlers(engine: &Arc<WshRpcEngine>, state: &AppState, conn_id: String) {
213    let r = router_for(state);
214
215    let rr = r.clone();
216    engine.register_typed("notify.emit", move |p: NotifyEmitParams, _ctx| {
217        let r = rr.clone();
218        async move {
219            if p.block_id.is_empty() {
220                return Err("notify.emit: block_id required".to_string());
221            }
222            // Everything goes through the Router's single ordered queue, so a
223            // resolve can never overtake the emit it cancels — whichever
224            // source (renderer or srv) sent either one (Codex P2 on #3662).
225            // Emits resolve the agent name off the async workers there too.
226            match p.event {
227                // Turn start/finish come from srv's controller status (see
228                // router `TurnStatus`), not from the renderer.
229                NotifyPaneEvent::TurnStarted | NotifyPaneEvent::TurnCompleted | NotifyPaneEvent::TurnErrored => {}
230                NotifyPaneEvent::TurnStopped => r.turn_stopped_nonblocking(&p.block_id),
231                NotifyPaneEvent::InputResolved => r.resolve_nonblocking(&p.block_id, Family::Input),
232                NotifyPaneEvent::InputWaiting => {
233                    r.input_waiting_nonblocking(&p.block_id, p.question, p.question_count.unwrap_or(0) as usize)
234                }
235            }
236            Ok(NotifyOk { ok: true })
237        }
238    });
239
240    let rr = r.clone();
241    let conn = conn_id.clone();
242    engine.register_typed("notify.focus", move |p: NotifyFocusParams, _ctx| {
243        let r = rr.clone();
244        let conn = conn.clone();
245        async move {
246            r.focus(
247                &conn,
248                FocusReport {
249                    window_focused: p.window_focused,
250                    block_id: p.block_id.filter(|b| !b.is_empty()),
251                    window_id: p.window_id.filter(|w| !w.is_empty()),
252                },
253            );
254            Ok(NotifyOk { ok: true })
255        }
256    });
257
258    let rr = r.clone();
259    engine.register_typed("notify.ack", move |p: NotifyAckParams, _ctx| {
260        let r = rr.clone();
261        async move {
262            // Resolving the click's pane reads the store — off the async workers.
263            let o = tokio::task::spawn_blocking(move || r.ack(&p.id, p.clicked))
264                .await
265                .map_err(|e| format!("notify.ack: {e}"))?;
266            Ok(NotifyAckResult { has_window: o.has_window, known: o.known, window_id: o.window_id, workspace_id: o.workspace_id })
267        }
268    });
269
270    let rr = r.clone();
271    // notify.attention: the calls to action srv holds open now
272    // (backend/user_attention.rs), for a window that starts while one is.
273    engine.register_typed("notify.attention", |_p: NotifyNoArgs, _ctx| async move {
274        Ok(NotifyAttentionResult { attention: crate::backend::user_attention::open_requests() })
275    });
276
277    engine.register_typed("notify.test", move |_p: NotifyNoArgs, _ctx| {
278        let r = rr.clone();
279        async move {
280            r.test();
281            Ok(NotifyOk { ok: true })
282        }
283    });
284
285    let rr = r.clone();
286    engine.register_typed("notify.takeactivation", move |p: NotifyTakeActivationParams, _ctx| {
287        let r = rr.clone();
288        async move {
289            let taken = r.take_pending_activation(p.workspace_id.as_deref().filter(|w| !w.is_empty()));
290            Ok(NotifyTakeActivationResult {
291                block_id: taken.as_ref().map(|t| t.block_id.clone()),
292                tab_id: taken.map(|t| t.tab_id),
293            })
294        }
295    });
296
297    let rr = r;
298    let st = state.clone();
299    engine.register_typed("block.reveal", move |p: BlockRevealParams, _ctx| {
300        let r = rr.clone();
301        let st = st.clone();
302        async move { reveal_block(&st, &r, p.block_id).await }
303    });
304}
305
306#[cfg(test)]
307mod block_reveal_tests {
308    use std::sync::{Arc, Mutex};
309
310    use agentmux_common::ipc::{Command, Event};
311
312    use crate::backend::mps::MuxEvent;
313    use crate::backend::notify::policy::FocusReport;
314    use crate::backend::notify::router::{self, BlockRevealEvent, EVENT_BLOCK_REVEAL};
315    use crate::backend::obj::Window;
316    use crate::server::AppState;
317
318    async fn dispatch_apply(state: &AppState, cmd: Command) -> Vec<Event> {
319        let evs = crate::server::service::dispatch_to_reducer(state, cmd).await;
320        for ev in &evs {
321            crate::persist_subscriber::apply_event_to_mstore(ev, &state.mstore).unwrap();
322        }
323        evs
324    }
325
326    struct Fixture {
327        state: AppState,
328        router: Arc<router::Router>,
329        tab_id: String,
330        /// Pane tabs, `a` showing, `b` in the background.
331        a: String,
332        b: String,
333        window_id: String,
334        reveals: Arc<Mutex<Vec<serde_json::Value>>>,
335    }
336
337    /// A workspace with one tab holding a two-tab pane [a, b] (a showing), a
338    /// `Window` object on that workspace, and a recorder for `block:reveal`.
339    /// `open`: a connected frontend reports being that window.
340    async fn fixture(open: bool) -> Fixture {
341        let state = crate::server::tests::test_state();
342        let ws_id = dispatch_apply(&state, Command::CreateWorkspace { name: "w".into() })
343            .await
344            .iter()
345            .find_map(|e| match e {
346                Event::WorkspaceCreated { workspace_id, .. } => Some(workspace_id.clone()),
347                _ => None,
348            })
349            .unwrap();
350        let tab_id = dispatch_apply(&state, Command::CreateTab { workspace_id: ws_id.clone(), name: "t".into() })
351            .await
352            .iter()
353            .find_map(|e| match e {
354                Event::TabCreated { tab_id, .. } => Some(tab_id.clone()),
355                _ => None,
356            })
357            .unwrap();
358        let mut ids = Vec::new();
359        for _ in 0..2 {
360            let id = dispatch_apply(&state, Command::CreateBlock { tab_id: tab_id.clone(), meta: serde_json::Value::Null })
361                .await
362                .iter()
363                .find_map(|e| match e {
364                    Event::BlockCreated { block_id, .. } => Some(block_id.clone()),
365                    _ => None,
366                })
367                .unwrap();
368            ids.push(id);
369        }
370        let (a, b) = (ids[0].clone(), ids[1].clone());
371        dispatch_apply(
372            &state,
373            Command::LayoutSetTree {
374                tab_id: tab_id.clone(),
375                new_tree: Some(agentmux_common::LayoutNode {
376                    id: "pane".into(),
377                    data: Some(agentmux_common::LayoutNodeData {
378                        block_id: a.clone(),
379                        block_stack: vec![a.clone(), b.clone()],
380                        active_block_id: a.clone(),
381                        ..Default::default()
382                    }),
383                    ..Default::default()
384                }),
385                correlation_id: String::new(),
386                slices: None,
387            },
388        )
389        .await;
390        // A window showing that workspace (the seeded one shows another).
391        let mut window = state.mstore.get_all::<Window>().unwrap().into_iter().next().expect("seeded window");
392        window.oid = uuid::Uuid::new_v4().to_string();
393        window.workspaceid = ws_id;
394        state.mstore.insert(&mut window).unwrap();
395
396        let reveals = Arc::new(Mutex::new(Vec::new()));
397        let rec = reveals.clone();
398        state.broker.add_observer(Arc::new(move |ev: &MuxEvent| {
399            if ev.event == EVENT_BLOCK_REVEAL {
400                rec.lock().unwrap().push(ev.data.clone().unwrap_or_default());
401            }
402        }));
403        let r = super::router_for(&state);
404        if open {
405            r.focus("conn-1", FocusReport { window_focused: false, block_id: None, window_id: Some(window.oid.clone()) });
406        }
407        Fixture { state, router: r, tab_id, a, b, window_id: window.oid, reveals }
408    }
409
410    #[tokio::test]
411    async fn reveals_a_background_pane_tab_in_the_window_showing_it() {
412        let f = fixture(true).await;
413        let res = super::reveal_block(&f.state, &f.router, f.b.clone()).await.unwrap();
414        assert!(res.found);
415        assert_eq!(res.window_id.as_deref(), Some(f.window_id.as_str()));
416
417        // The stored tree shows b — in the reducer and in mstore.
418        let tree = f.state.srv_state.lock().await.tabs[&f.tab_id].rootnode.clone().unwrap();
419        let data = tree.data.as_ref().unwrap();
420        assert_eq!((data.block_id.as_str(), data.active_block_id.as_str()), (f.b.as_str(), f.b.as_str()));
421        let tab = f.state.mstore.must_get::<crate::backend::obj::Tab>(&f.tab_id).unwrap();
422        let layout = f.state.mstore.must_get::<crate::backend::obj::LayoutState>(&tab.layoutstate).unwrap();
423        assert_eq!(layout.rootnode.as_ref(), Some(&tree), "db_layout has b active");
424
425        // Only that window is told, with the tab it should switch to.
426        let reveals = f.reveals.lock().unwrap().clone();
427        assert_eq!(reveals.len(), 1, "{reveals:?}");
428        let ev: BlockRevealEvent = serde_json::from_value(reveals[0].clone()).unwrap();
429        assert_eq!(ev, BlockRevealEvent { block_id: f.b.clone(), tab_id: f.tab_id.clone(), window_id: f.window_id.clone() });
430    }
431
432    #[tokio::test]
433    async fn unknown_block_is_not_found_and_publishes_nothing() {
434        let f = fixture(true).await;
435        let res = super::reveal_block(&f.state, &f.router, "no-such-block".into()).await.unwrap();
436        assert!(!res.found);
437        assert_eq!(res.window_id, None);
438        assert!(f.reveals.lock().unwrap().is_empty());
439    }
440
441    #[tokio::test]
442    async fn workspace_open_in_no_window_is_not_found_and_changes_nothing() {
443        let f = fixture(false).await;
444        let res = super::reveal_block(&f.state, &f.router, f.b.clone()).await.unwrap();
445        assert!(!res.found);
446        assert!(f.reveals.lock().unwrap().is_empty());
447        let tree = f.state.srv_state.lock().await.tabs[&f.tab_id].rootnode.clone().unwrap();
448        assert_eq!(tree.data.as_ref().unwrap().active_block_id, f.a, "stored tree untouched");
449    }
450
451    #[tokio::test]
452    async fn empty_block_id_is_an_error() {
453        let f = fixture(true).await;
454        assert!(super::reveal_block(&f.state, &f.router, String::new()).await.is_err());
455    }
456}
457
458#[cfg(test)]
459mod live_tests {
460    use futures_util::{SinkExt, StreamExt};
461    use tokio_tungstenite::tungstenite::{ClientRequestBuilder, Message};
462
463    async fn next_event<S>(rx: &mut S, want: &str) -> serde_json::Value
464    where
465        S: futures_util::Stream<Item = Result<Message, tokio_tungstenite::tungstenite::Error>> + Unpin,
466    {
467        loop {
468            let msg = tokio::time::timeout(std::time::Duration::from_secs(5), rx.next())
469                .await
470                .unwrap_or_else(|_| panic!("timed out waiting for {want}"))
471                .unwrap()
472                .unwrap();
473            let Ok(text) = msg.to_text() else { continue };
474            let Ok(v) = serde_json::from_str::<serde_json::Value>(text) else { continue };
475            if v["data"]["command"] == "eventrecv" && v["data"]["data"]["event"] == want {
476                return v["data"]["data"]["data"].clone();
477            }
478        }
479    }
480
481    /// The launcher presenter's exact wire shape, against the REAL srv router:
482    /// `X-AuthKey` header auth on /ws, `eventsub`, then a Router-published
483    /// `notification` (via notify.test) and its `notification:retract` after
484    /// a dismiss ack.
485    #[tokio::test]
486    async fn header_authed_client_receives_router_events_over_ws() {
487        let state = crate::server::tests::test_state();
488        // test_state() leaves the broker without its fan-out client; wire it
489        // exactly as bootstrap/background.rs does so published events reach
490        // /ws clients.
491        state
492            .broker
493            .set_client(Box::new(crate::backend::eventbus::EventBusBridge::new(state.event_bus.clone())));
494        let auth = state.auth_key.clone();
495        let app = crate::server::build_router(state);
496        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
497        let addr = listener.local_addr().unwrap();
498        tokio::spawn(async move {
499            axum::serve(listener, app.into_make_service_with_connect_info::<std::net::SocketAddr>())
500                .await
501                .unwrap();
502        });
503
504        let uri: tokio_tungstenite::tungstenite::http::Uri = format!("ws://{addr}/ws").parse().unwrap();
505        let req = ClientRequestBuilder::new(uri).with_header("X-AuthKey", auth);
506        let (ws, _) = tokio_tungstenite::connect_async(req).await.expect("header auth accepted on /ws");
507        let (mut tx, mut rx) = ws.split();
508
509        let send = |cmd: &str, reqid: &str, data: serde_json::Value| {
510            serde_json::json!({"wscommand":"rpc","message":{"command":cmd,"reqid":reqid,"data":data}}).to_string()
511        };
512        for (i, ev) in ["notification", "notification:retract"].iter().enumerate() {
513            tx.send(Message::Text(send("eventsub", &format!("s{i}"), serde_json::json!({"event":ev,"scopes":[],"allscopes":true})).into()))
514                .await
515                .unwrap();
516        }
517        tx.send(Message::Text(send("notify.test", "t1", serde_json::json!({})).into())).await.unwrap();
518
519        let n = next_event(&mut rx, "notification").await;
520        assert_eq!(n["kind"], "test");
521        assert_eq!(n["title"], "AgentMux notifications are working");
522        let id = n["id"].as_str().unwrap().to_string();
523        let tag = n["tag"].as_str().unwrap().to_string();
524
525        tx.send(Message::Text(send("notify.ack", "a1", serde_json::json!({"id": id, "clicked": false})).into()))
526            .await
527            .unwrap();
528        let r = next_event(&mut rx, "notification:retract").await;
529        assert_eq!(r["tag"], tag);
530    }
531
532    /// Rich-content spec §3.3 A/B: a click on a pane's toast resolves the
533    /// pane's tab and open window server-side, names that window in
534    /// `notification:activate`, and tells the launcher which window to raise.
535    #[tokio::test]
536    async fn clicking_a_pane_toast_names_its_open_window_and_tab() {
537        use crate::backend::obj::{Block, Window};
538        let state = crate::server::tests::test_state();
539        state
540            .broker
541            .set_client(Box::new(crate::backend::eventbus::EventBusBridge::new(state.event_bus.clone())));
542        let (broker, store) = (state.broker.clone(), state.mstore.clone());
543        crate::backend::wcore::ensure_initial_data(&store).unwrap();
544        let block = store.get_all::<Block>().unwrap().into_iter().next().expect("seeded block");
545        let window = store.get_all::<Window>().unwrap().into_iter().next().expect("seeded window");
546        let auth = state.auth_key.clone();
547        let app = crate::server::build_router(state);
548        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
549        let addr = listener.local_addr().unwrap();
550        tokio::spawn(async move {
551            axum::serve(listener, app.into_make_service_with_connect_info::<std::net::SocketAddr>())
552                .await
553                .unwrap();
554        });
555        let uri: tokio_tungstenite::tungstenite::http::Uri = format!("ws://{addr}/ws").parse().unwrap();
556        let (ws, _) = tokio_tungstenite::connect_async(ClientRequestBuilder::new(uri).with_header("X-AuthKey", auth))
557            .await
558            .unwrap();
559        let (mut tx, mut rx) = ws.split();
560        let send = |cmd: &str, reqid: &str, data: serde_json::Value| {
561            serde_json::json!({"wscommand":"rpc","message":{"command":cmd,"reqid":reqid,"data":data}}).to_string()
562        };
563        for (i, ev) in ["notification", "notification:activate"].iter().enumerate() {
564            tx.send(Message::Text(send("eventsub", &format!("s{i}"), serde_json::json!({"event":ev,"scopes":[],"allscopes":true})).into()))
565                .await
566                .unwrap();
567        }
568        // This connection is the seeded window, open but not focused.
569        tx.send(Message::Text(send("notify.focus", "f1", serde_json::json!({"window_focused": false, "window_id": window.oid})).into()))
570            .await
571            .unwrap();
572        let r = crate::backend::notify::router::get(&broker).expect("router created by the connection");
573        let b = block.oid.clone();
574        tokio::task::spawn_blocking(move || {
575            r.emit_fixed(crate::backend::notify::policy::NotifyKind::MessageNeedsReview, &b, Some("x".into()))
576        })
577        .await
578        .unwrap();
579        let n = next_event(&mut rx, "notification").await;
580        let id = n["id"].as_str().unwrap().to_string();
581
582        tx.send(Message::Text(send("notify.ack", "a1", serde_json::json!({"id": id, "clicked": true})).into()))
583            .await
584            .unwrap();
585        let (mut activate, mut response) = (None, None);
586        while activate.is_none() || response.is_none() {
587            let msg = tokio::time::timeout(std::time::Duration::from_secs(5), rx.next()).await.expect("ack answered").unwrap().unwrap();
588            let Ok(v) = serde_json::from_str::<serde_json::Value>(msg.to_text().unwrap_or("")) else { continue };
589            if v["data"]["resid"] == "a1" {
590                response = Some(v["data"]["data"].clone());
591            } else if v["data"]["command"] == "eventrecv" && v["data"]["data"]["event"] == "notification:activate" {
592                activate = Some(v["data"]["data"]["data"].clone());
593            }
594        }
595        let (a, resp) = (activate.unwrap(), response.unwrap());
596        let tab_id = block.parentoref.strip_prefix("tab:").unwrap();
597        assert_eq!(a["block_id"], block.oid.as_str());
598        assert_eq!(a["tab_id"], tab_id);
599        assert_eq!(a["window_id"], window.oid.as_str());
600        assert_eq!(a["workspace_id"], window.workspaceid.as_str());
601        assert_eq!(resp["window_id"], window.oid.as_str(), "the launcher raises that window: {resp}");
602        assert_eq!(resp["known"], true);
603        assert_eq!(resp["has_window"], true);
604    }
605}
agentmux_srv/server/notify_handlers.rs

agentmux_srv/server/
notify_handlers.rs
