agent_engine/events/
queue.rs

1use super::types::{Event, Severity};
2use std::collections::VecDeque;
3use std::sync::Mutex;
4
5pub struct EventQueue {
6    inner: Mutex<VecDeque<Event>>,
7    capacity: usize,
8    notify: tokio::sync::Notify,
9}
10
11impl EventQueue {
12    pub fn new(capacity: usize) -> Self {
13        Self {
14            inner: Mutex::new(VecDeque::with_capacity(capacity)),
15            capacity,
16            notify: tokio::sync::Notify::new(),
17        }
18    }
19
20    /// Push an event. Critical goes to front, High goes after existing Criticals,
21    /// everything else goes to the back. Returns Err if queue is full.
22    pub fn push(&self, event: Event) -> Result<(), String> {
23        let mut q = self.inner.lock().unwrap();
24        if q.len() >= self.capacity {
25            return Err(format!("event queue full (capacity {})", self.capacity));
26        }
27        let sev = event.content.severity.clone();
28        match sev {
29            Some(Severity::Critical) => q.push_front(event),
30            Some(Severity::High) => {
31                // insert after trailing Critical events at the front
32                let mut idx = 0;
33                while idx < q.len() && matches!(q[idx].content.severity, Some(Severity::Critical)) {
34                    idx += 1;
35                }
36                q.insert(idx, event);
37            }
38            _ => q.push_back(event),
39        }
40        drop(q);
41        self.notify.notify_one();
42        Ok(())
43    }
44
45    /// Force push to front regardless of severity. Evicts the oldest (back) if full.
46    pub fn push_priority(&self, event: Event) {
47        let mut q = self.inner.lock().unwrap();
48        if q.len() >= self.capacity {
49            if let Some(evicted) = q.back() {
50                tracing::warn!("event queue full — evicting event id={}", evicted.id);
51            }
52            q.pop_back();
53        }
54        q.push_front(event);
55        drop(q);
56        self.notify.notify_one();
57    }
58
59    pub fn pop(&self) -> Option<Event> {
60        self.inner.lock().unwrap().pop_front()
61    }
62
63    pub fn peek(&self) -> Option<Event> {
64        self.inner.lock().unwrap().front().cloned()
65    }
66
67    pub fn len(&self) -> usize {
68        self.inner.lock().unwrap().len()
69    }
70
71    /// Wait until an event is pushed. Use in tokio::select! for instant wake.
72    pub fn notified(&self) -> impl std::future::Future<Output = ()> + '_ {
73        self.notify.notified()
74    }
75
76    pub fn is_empty(&self) -> bool {
77        self.inner.lock().unwrap().is_empty()
78    }
79
80    pub fn drain(&self) -> Vec<Event> {
81        let mut q = self.inner.lock().unwrap();
82        q.drain(..).collect()
83    }
84}
85
86#[cfg(test)]
87mod tests {
88    use super::*;
89
90    fn ev(text: &str, sev: Option<Severity>) -> Event {
91        Event::simple("test", text, sev)
92    }
93
94    #[test]
95    fn push_pop_fifo_for_medium() {
96        let q = EventQueue::new(10);
97        q.push(ev("a", Some(Severity::Medium))).unwrap();
98        q.push(ev("b", Some(Severity::Low))).unwrap();
99        q.push(ev("c", None)).unwrap();
100        assert_eq!(q.pop().unwrap().content.text, "a");
101        assert_eq!(q.pop().unwrap().content.text, "b");
102        assert_eq!(q.pop().unwrap().content.text, "c");
103        assert!(q.is_empty());
104    }
105
106    #[test]
107    fn critical_jumps_to_front() {
108        let q = EventQueue::new(10);
109        q.push(ev("a", Some(Severity::Medium))).unwrap();
110        q.push(ev("b", Some(Severity::Medium))).unwrap();
111        q.push(ev("CRIT", Some(Severity::Critical))).unwrap();
112        assert_eq!(q.pop().unwrap().content.text, "CRIT");
113        assert_eq!(q.pop().unwrap().content.text, "a");
114    }
115
116    #[test]
117    fn high_sits_after_critical() {
118        let q = EventQueue::new(10);
119        q.push(ev("med", Some(Severity::Medium))).unwrap();
120        q.push(ev("c1", Some(Severity::Critical))).unwrap();
121        q.push(ev("c2", Some(Severity::Critical))).unwrap();
122        q.push(ev("high", Some(Severity::High))).unwrap();
123        // Order should be: c2, c1, high, med
124        assert_eq!(q.pop().unwrap().content.text, "c2");
125        assert_eq!(q.pop().unwrap().content.text, "c1");
126        assert_eq!(q.pop().unwrap().content.text, "high");
127        assert_eq!(q.pop().unwrap().content.text, "med");
128    }
129
130    #[test]
131    fn capacity_limit() {
132        let q = EventQueue::new(2);
133        q.push(ev("a", None)).unwrap();
134        q.push(ev("b", None)).unwrap();
135        assert!(q.push(ev("c", None)).is_err());
136    }
137
138    #[test]
139    fn drain_takes_all() {
140        let q = EventQueue::new(10);
141        q.push(ev("a", None)).unwrap();
142        q.push(ev("b", None)).unwrap();
143        let all = q.drain();
144        assert_eq!(all.len(), 2);
145        assert!(q.is_empty());
146    }
147
148    #[test]
149    fn peek_does_not_remove() {
150        let q = EventQueue::new(10);
151        q.push(ev("a", None)).unwrap();
152        assert_eq!(q.peek().unwrap().content.text, "a");
153        assert_eq!(q.len(), 1);
154    }
155}
agent_engine/events/queue.rs

agent_engine/events/
queue.rs
