bamboo_server/tools/
notify_dispatcher.rs

1//! Server adapter implementing `bamboo_server_tools::NotificationDispatcher`
2//! for the agent-invocable `notify` tool.
3//!
4//! Bridges the tool's fire-and-forget port to real delivery:
5//! - mints via [`bamboo_notification::NotificationService::mint_custom`] so
6//!   the master enabled switch and the shared dedup window apply exactly
7//!   like every other notification (WP1's passthrough classification);
8//! - re-broadcasts the resulting `AgentEvent::Notification` on the session's
9//!   event channel so connected SSE/WS clients see it, same as the always-on
10//!   relay (`app_state::session_events::ensure_notification_relay`); and
11//! - fans it out to configured delivery sinks (desktop/ntfy/bark) via
12//!   [`crate::notify_sinks::dispatch_to_sinks`] (WP3), reading the CURRENT
13//!   config so a hot-reloaded topic/token/toggle takes effect immediately.
14//!
15//! Mirrors [`super::child_session_adapter::ChildSessionAdapter`]'s role: a
16//! server-side struct that satisfies a port declared in the
17//! framework-agnostic `bamboo-server-tools` crate, so that crate never needs
18//! to depend on `AppState` (see its crate doc).
19
20use std::collections::HashMap;
21use std::sync::Arc;
22
23use bamboo_agent_core::AgentEvent;
24use bamboo_notification::NotificationPriority;
25use tokio::sync::{broadcast, RwLock};
26
27use bamboo_server_tools::NotificationDispatcher;
28
29use crate::app_state::watchers::SessionWatchers;
30
31/// Server-side adapter satisfying `NotificationDispatcher` for `NotifyTool`.
32pub struct ServerNotificationDispatcher {
33    pub notification_service: Arc<bamboo_notification::NotificationService>,
34    pub session_event_senders: Arc<RwLock<HashMap<String, broadcast::Sender<AgentEvent>>>>,
35    pub session_watchers: Arc<SessionWatchers>,
36    pub config: Arc<RwLock<bamboo_llm::Config>>,
37}
38
39impl ServerNotificationDispatcher {
40    pub fn new(
41        notification_service: Arc<bamboo_notification::NotificationService>,
42        session_event_senders: Arc<RwLock<HashMap<String, broadcast::Sender<AgentEvent>>>>,
43        session_watchers: Arc<SessionWatchers>,
44        config: Arc<RwLock<bamboo_llm::Config>>,
45    ) -> Self {
46        Self {
47            notification_service,
48            session_event_senders,
49            session_watchers,
50            config,
51        }
52    }
53}
54
55/// Maps the tool's validated `"low" | "normal" | "high"` string to the
56/// policy enum. Any other value (there shouldn't be one — `NotifyTool`
57/// validates before calling `dispatch`) falls back to `Normal` rather than
58/// panicking; a delivery path must never crash a run over a cosmetic field.
59fn parse_priority(priority: &str) -> NotificationPriority {
60    match priority {
61        "high" => NotificationPriority::High,
62        "low" => NotificationPriority::Low,
63        _ => NotificationPriority::Normal,
64    }
65}
66
67impl NotificationDispatcher for ServerNotificationDispatcher {
68    fn dispatch(
69        &self,
70        session_id: &str,
71        title: &str,
72        body: &str,
73        priority: &str,
74        url: Option<&str>,
75    ) {
76        let notification_service = self.notification_service.clone();
77        let session_event_senders = self.session_event_senders.clone();
78        let session_watchers = self.session_watchers.clone();
79        let config = self.config.clone();
80        let session_id = session_id.to_string();
81        let title = title.to_string();
82        let body = body.to_string();
83        let priority = parse_priority(priority);
84        let url = url.map(str::to_string);
85
86        // Non-blocking: the tool call already returned a confirmation to the
87        // model. The real classify/broadcast/sink-dispatch work happens here,
88        // off that call, so a slow/unreachable sink can never stall the loop
89        // (same posture as `NotificationSink::deliver` and the always-on
90        // relay in `app_state::session_events`).
91        tokio::spawn(async move {
92            let Some(event) = notification_service.mint_custom(&session_id, title, body, priority)
93            else {
94                // Disabled (master switch off) or deduped within the window —
95                // not an error, just nothing to deliver.
96                return;
97            };
98
99            // Build the sink payload before `event` is moved into the
100            // broadcast send below (mirrors `ensure_notification_relay`),
101            // attaching the tool-supplied click-through URL — the only
102            // producer that currently populates `SinkNotification::click_url`.
103            let sink_notification =
104                crate::notify_sinks::SinkNotification::from_event(&event).map(|mut n| {
105                    n.click_url = url;
106                    n
107                });
108
109            let tx = session_event_senders.read().await.get(&session_id).cloned();
110            if let Some(tx) = tx {
111                let _ = tx.send(event);
112            }
113
114            if let Some(sink_notification) = sink_notification {
115                let has_watcher = session_watchers.has_watcher(&session_id);
116                let config_snapshot = config.read().await.clone();
117                crate::notify_sinks::dispatch_to_sinks(
118                    &config_snapshot,
119                    has_watcher,
120                    &sink_notification,
121                );
122            }
123        });
124    }
125}
126
127#[cfg(test)]
128mod tests {
129    use super::*;
130    use std::time::Duration;
131
132    fn dispatcher() -> (ServerNotificationDispatcher, tempfile::TempDir) {
133        let dir = tempfile::tempdir().unwrap();
134        let notification_service = Arc::new(bamboo_notification::NotificationService::new(
135            dir.path().join("prefs.json"),
136        ));
137        let dispatcher = ServerNotificationDispatcher::new(
138            notification_service,
139            Arc::new(RwLock::new(HashMap::new())),
140            SessionWatchers::new(),
141            Arc::new(RwLock::new(bamboo_llm::Config::default())),
142        );
143        (dispatcher, dir)
144    }
145
146    #[test]
147    fn parse_priority_maps_known_values_and_falls_back_to_normal() {
148        assert_eq!(parse_priority("high"), NotificationPriority::High);
149        assert_eq!(parse_priority("low"), NotificationPriority::Low);
150        assert_eq!(parse_priority("normal"), NotificationPriority::Normal);
151        assert_eq!(parse_priority("unexpected"), NotificationPriority::Normal);
152    }
153
154    #[tokio::test]
155    async fn dispatch_broadcasts_a_notification_event_on_the_session_channel() {
156        let (dispatcher, _dir) = dispatcher();
157        let (tx, mut rx) = broadcast::channel(16);
158        dispatcher
159            .session_event_senders
160            .write()
161            .await
162            .insert("sess-1".to_string(), tx);
163
164        dispatcher.dispatch("sess-1", "Reminder", "Stand up", "high", None);
165
166        let event = tokio::time::timeout(Duration::from_secs(2), rx.recv())
167            .await
168            .expect("dispatch should broadcast within the timeout")
169            .unwrap();
170        match event {
171            AgentEvent::Notification {
172                session_id,
173                category,
174                priority,
175                title,
176                body,
177                ..
178            } => {
179                assert_eq!(session_id, "sess-1");
180                assert_eq!(category, "custom");
181                assert_eq!(priority, "high");
182                assert_eq!(title, "Reminder");
183                assert_eq!(body, "Stand up");
184            }
185            other => panic!("expected Notification, got {other:?}"),
186        }
187    }
188
189    #[tokio::test]
190    async fn dispatch_is_a_no_op_when_notifications_are_disabled() {
191        let (dispatcher, _dir) = dispatcher();
192        dispatcher
193            .notification_service
194            .set_preferences(bamboo_notification::NotificationPreferences {
195                enabled: false,
196                ..bamboo_notification::NotificationPreferences::default()
197            })
198            .unwrap();
199        let (tx, mut rx) = broadcast::channel(16);
200        dispatcher
201            .session_event_senders
202            .write()
203            .await
204            .insert("sess-1".to_string(), tx);
205
206        dispatcher.dispatch("sess-1", "Reminder", "Stand up", "normal", None);
207
208        // Give the spawned task a chance to run; it must not send anything.
209        let outcome = tokio::time::timeout(Duration::from_millis(200), rx.recv()).await;
210        assert!(
211            outcome.is_err(),
212            "disabled notifications must not broadcast an event"
213        );
214    }
215
216    #[tokio::test]
217    async fn dispatch_with_no_subscriber_does_not_panic() {
218        // No entry in `session_event_senders` for this session — dispatch
219        // must degrade gracefully (still mints/dedups/sinks, just nothing to
220        // broadcast onto), not panic.
221        let (dispatcher, _dir) = dispatcher();
222        dispatcher.dispatch("no-such-session", "Title", "Body", "low", None);
223        tokio::time::sleep(Duration::from_millis(50)).await;
224    }
225}
bamboo_server/tools/notify_dispatcher.rs

bamboo_server/tools/
notify_dispatcher.rs
