bamboo_server/app_state/
session_events.rs

1//! Session event sender management.
2//!
3//! Re-exports the shared `get_or_create_event_sender` function from the runtime crate.
4//! The `impl AppState` method delegates to the re-exported function.
5
6pub use bamboo_engine::execution::session_events::get_or_create_event_sender;
7
8impl super::AppState {
9    /// Get (or create) a long-lived session event sender for a session id.
10    ///
11    /// This stream is intended for UI consumption and background activity; it should remain
12    /// available even when no agent execution is running.
13    pub async fn get_session_event_sender(
14        &self,
15        session_id: &str,
16    ) -> tokio::sync::broadcast::Sender<bamboo_agent_core::AgentEvent> {
17        get_or_create_event_sender(&self.session_event_senders, session_id).await
18    }
19
20    /// Ensure a notification relay task is running for `session_id`.
21    ///
22    /// The relay subscribes to the session's event broadcast, runs the backend
23    /// notification policy on each event, and re-broadcasts any resulting
24    /// `AgentEvent::Notification` onto the same channel so connected SSE clients
25    /// receive it. Idempotent: at most one relay per session (tracked by the
26    /// notification service). The relay does not hold a `Sender` clone, so the
27    /// channel still closes naturally when the session is torn down.
28    pub fn ensure_notification_relay(
29        &self,
30        session_id: &str,
31        sender: tokio::sync::broadcast::Sender<bamboo_agent_core::AgentEvent>,
32    ) {
33        if !self.notification_service.try_begin_relay(session_id) {
34            return;
35        }
36        let service = self.notification_service.clone();
37        let senders = self.session_event_senders.clone();
38        let sid = session_id.to_string();
39        let mut rx = sender.subscribe();
40        drop(sender);
41        tokio::spawn(async move {
42            use tokio::sync::broadcast::error::RecvError;
43            loop {
44                match rx.recv().await {
45                    Ok(event) => {
46                        if let Some(notification) = service.notify(&sid, &event) {
47                            let tx = senders.read().await.get(&sid).cloned();
48                            if let Some(tx) = tx {
49                                let _ = tx.send(notification);
50                            }
51                        }
52                    }
53                    Err(RecvError::Lagged(_)) => continue,
54                    Err(RecvError::Closed) => break,
55                }
56            }
57            service.end_relay(&sid);
58        });
59    }
60}
bamboo_server/app_state/session_events.rs

bamboo_server/app_state/
session_events.rs
