deepstrike_core/signals/
router.rs

1use std::collections::HashSet;
2
3use compact_str::CompactString;
4
5use super::attention::UrgencyBasedPolicy;
6use super::queue::SignalQueue;
7use crate::types::policy::SignalDisposition;
8use crate::types::signal::RuntimeSignal;
9
10/// Signal router: dedup set + urgency-based attention + bounded priority queue.
11pub struct SignalRouter {
12    seen: HashSet<CompactString>,
13    queue: SignalQueue,
14    attention: UrgencyBasedPolicy,
15}
16
17impl SignalRouter {
18    pub fn new(max_queue_size: usize) -> Self {
19        Self {
20            seen: HashSet::new(),
21            queue: SignalQueue::new(max_queue_size),
22            attention: UrgencyBasedPolicy,
23        }
24    }
25
26    /// Ingest a signal. Returns the disposition after dedup + attention evaluation.
27    /// `Queue` dispositions are buffered; if the queue is full, returns `Dropped`
28    /// so the SDK can apply backpressure or surface the loss to telemetry.
29    /// All other dispositions are returned directly to the caller.
30    pub fn ingest(&mut self, signal: RuntimeSignal, is_running: bool) -> SignalDisposition {
31        if let Some(ref key) = signal.dedupe_key {
32            if !self.seen.insert(key.clone()) {
33                return SignalDisposition::Ignore;
34            }
35        }
36
37        let disposition = self.attention.evaluate(&signal, is_running);
38
39        if disposition == SignalDisposition::Queue {
40            if !self.queue.push(signal) {
41                return SignalDisposition::Dropped;
42            }
43        }
44
45        disposition
46    }
47
48    /// Pull next queued signal.
49    pub fn next(&mut self) -> Option<RuntimeSignal> {
50        self.queue.pop()
51    }
52
53    /// Number of queued signals.
54    pub fn depth(&self) -> usize {
55        self.queue.len()
56    }
57
58    /// Clear the dedup set (call at session boundaries to prevent unbounded growth).
59    pub fn clear_dedup(&mut self) {
60        self.seen.clear();
61    }
62}
63
64#[cfg(test)]
65mod tests {
66    use super::*;
67    use crate::types::signal::{SignalSource, SignalType, Urgency};
68
69    #[test]
70    fn deduplicates_signals() {
71        let mut router = SignalRouter::new(100);
72        let sig = RuntimeSignal::new(
73            SignalSource::Cron,
74            SignalType::Event,
75            Urgency::Normal,
76            "tick",
77        )
78        .with_dedupe("cron-tick-1");
79
80        let d1 = router.ingest(sig.clone(), false);
81        assert_ne!(d1, SignalDisposition::Ignore);
82
83        let d2 = router.ingest(sig, false);
84        assert_eq!(d2, SignalDisposition::Ignore);
85    }
86
87    #[test]
88    fn normal_signal_queued() {
89        let mut router = SignalRouter::new(100);
90        let sig = RuntimeSignal::new(
91            SignalSource::Cron,
92            SignalType::Event,
93            Urgency::Normal,
94            "job",
95        );
96
97        let d = router.ingest(sig, false);
98        assert_eq!(d, SignalDisposition::Queue);
99        assert_eq!(router.depth(), 1);
100        assert!(router.next().is_some());
101    }
102
103    #[test]
104    fn interrupt_signals_not_queued() {
105        let mut router = SignalRouter::new(100);
106        let sig = RuntimeSignal::new(
107            SignalSource::Gateway,
108            SignalType::Alert,
109            Urgency::Critical,
110            "fire",
111        );
112
113        let d = router.ingest(sig, true);
114        assert_eq!(d, SignalDisposition::InterruptNow);
115        assert_eq!(router.depth(), 0);
116    }
117
118    #[test]
119    fn full_queue_drops_signal() {
120        let mut router = SignalRouter::new(1);
121        let s1 = RuntimeSignal::new(
122            SignalSource::Cron,
123            SignalType::Event,
124            Urgency::Normal,
125            "first",
126        );
127        let s2 = RuntimeSignal::new(
128            SignalSource::Cron,
129            SignalType::Event,
130            Urgency::Normal,
131            "second",
132        );
133
134        assert_eq!(router.ingest(s1, false), SignalDisposition::Queue);
135        assert_eq!(router.ingest(s2, false), SignalDisposition::Dropped);
136    }
137
138    #[test]
139    fn clear_dedup_allows_reingest() {
140        let mut router = SignalRouter::new(100);
141        let sig = RuntimeSignal::new(
142            SignalSource::Cron,
143            SignalType::Event,
144            Urgency::Normal,
145            "tick",
146        )
147        .with_dedupe("key-1");
148
149        router.ingest(sig.clone(), false);
150        assert_eq!(router.ingest(sig.clone(), false), SignalDisposition::Ignore);
151
152        router.clear_dedup();
153        assert_ne!(router.ingest(sig, false), SignalDisposition::Ignore);
154    }
155}
deepstrike_core/signals/router.rs

deepstrike_core/signals/
router.rs
