roder_core/
bus.rs

1use std::sync::{
2    Arc,
3    atomic::{AtomicU64, Ordering},
4};
5
6use roder_api::events::{EventEnvelope, EventSource, RoderEvent, ThreadId, TurnId};
7use time::OffsetDateTime;
8use tokio::sync::broadcast;
9
10#[derive(Debug, Clone, Default)]
11pub struct EventFilter {
12    pub thread_id: Option<ThreadId>,
13    pub turn_id: Option<TurnId>,
14    pub kinds: Vec<String>,
15    pub sources: Vec<EventSource>,
16}
17
18impl EventFilter {
19    pub fn matches(&self, envelope: &EventEnvelope) -> bool {
20        if let Some(thread_id) = &self.thread_id
21            && envelope.thread_id.as_ref() != Some(thread_id)
22        {
23            return false;
24        }
25        if let Some(turn_id) = &self.turn_id
26            && envelope.turn_id.as_ref() != Some(turn_id)
27        {
28            return false;
29        }
30        if !self.kinds.is_empty() && !self.kinds.iter().any(|kind| kind == &envelope.kind) {
31            return false;
32        }
33        if !self.sources.is_empty() && !self.sources.iter().any(|source| source == &envelope.source)
34        {
35            return false;
36        }
37        true
38    }
39}
40
41#[derive(Clone)]
42pub struct EventBus {
43    sender: broadcast::Sender<EventEnvelope>,
44    next_seq: Arc<AtomicU64>,
45}
46
47impl EventBus {
48    pub fn new(capacity: usize) -> Self {
49        let (sender, _) = broadcast::channel(capacity);
50        Self {
51            sender,
52            next_seq: Arc::new(AtomicU64::new(1)),
53        }
54    }
55
56    pub fn subscribe(&self) -> broadcast::Receiver<EventEnvelope> {
57        self.sender.subscribe()
58    }
59
60    pub fn emit(&self, event: RoderEvent) -> EventEnvelope {
61        let envelope = EventEnvelope {
62            event_id: uuid::Uuid::new_v4().to_string(),
63            seq: self.next_seq.fetch_add(1, Ordering::SeqCst),
64            timestamp: OffsetDateTime::now_utc(),
65            source: event.source(),
66            kind: event.kind().to_string(),
67            thread_id: event.thread_id().cloned(),
68            turn_id: event.turn_id().cloned(),
69            event,
70        };
71        let _ = self.sender.send(envelope.clone());
72        envelope
73    }
74}
75
76#[cfg(test)]
77mod tests {
78    use super::*;
79    use roder_api::events::{RoderEvent, TurnStarted};
80    use roder_api::inference::RuntimeProfile;
81
82    fn sample_event() -> RoderEvent {
83        RoderEvent::TurnStarted(TurnStarted {
84            thread_id: "thread".to_string(),
85            turn_id: "turn".to_string(),
86            runtime_profile: RuntimeProfile::Interactive,
87            timestamp: OffsetDateTime::now_utc(),
88        })
89    }
90
91    #[tokio::test]
92    async fn retains_burst_up_to_capacity_without_lagging() {
93        // A slow consumer (the TUI render loop only drains every ~166ms during
94        // an active turn) must be able to buffer a large burst of streaming
95        // events without the broadcast ring overflowing. This guards the
96        // capacity headroom that keeps tool/thinking rows from being dropped.
97        let capacity = 16_384;
98        let bus = EventBus::new(capacity);
99        let mut rx = bus.subscribe();
100
101        for _ in 0..capacity {
102            bus.emit(sample_event());
103        }
104
105        // Every buffered event is still readable; none were dropped.
106        for _ in 0..capacity {
107            assert!(rx.try_recv().is_ok());
108        }
109        assert!(matches!(
110            rx.try_recv(),
111            Err(tokio::sync::broadcast::error::TryRecvError::Empty)
112        ));
113    }
114
115    #[tokio::test]
116    async fn overflow_beyond_capacity_surfaces_lagged() {
117        // When the buffer truly overflows the consumer must still observe a
118        // `Lagged` signal so the TUI can record the drop and run its stuck-turn
119        // recovery instead of hanging.
120        let capacity = 16usize;
121        let bus = EventBus::new(capacity);
122        let mut rx = bus.subscribe();
123
124        for _ in 0..(capacity + 8) {
125            bus.emit(sample_event());
126        }
127
128        assert!(matches!(
129            rx.try_recv(),
130            Err(tokio::sync::broadcast::error::TryRecvError::Lagged(_))
131        ));
132    }
133}
roder_core/bus.rs

roder_core/
bus.rs
