lnmp_net/
routing.rs

1//! Routing logic for LNMP-Net messages
2
3use lnmp_sfe::{ContextScorer, ContextScorerConfig};
4
5use crate::error::Result;
6use crate::message::NetMessage;
7
8/// Routing decision for a message
9#[derive(Debug, Clone, Copy, PartialEq, Eq)]
10pub enum RoutingDecision {
11    /// Send message to LLM for processing
12    SendToLLM,
13    /// Process message locally at edge/service level
14    ProcessLocally,
15    /// Drop message (expired, low priority, etc.)
16    Drop,
17}
18
19/// Policy for routing messages to LLM vs local processing
20///
21/// Implements the ECO (Energy/Token Optimization) profile logic:
22/// - Alerts with high priority always routed to LLM
23/// - Expired messages dropped
24/// - Event/State messages scored using SFE and routed based on threshold
25/// - Commands/Queries typically processed locally
26///
27/// # Examples
28///
29/// ```
30/// use lnmp_core::LnmpRecord;
31/// use lnmp_envelope::EnvelopeBuilder;
32/// use lnmp_net::{MessageKind, NetMessage, RoutingPolicy, RoutingDecision};
33///
34/// let policy = RoutingPolicy::default();
35///
36/// // Alert message -> always to LLM
37/// let envelope = EnvelopeBuilder::new(LnmpRecord::new())
38///     .timestamp(1000)
39///     .build();
40/// let alert = NetMessage::new(envelope, MessageKind::Alert);
41/// assert_eq!(policy.decide(&alert, 2000).unwrap(), RoutingDecision::SendToLLM);
42/// ```
43#[derive(Debug, Clone)]
44pub struct RoutingPolicy {
45    /// Minimum importance score (0.0-1.0) to route to LLM
46    pub llm_threshold: f64,
47
48    /// Always route Alert messages to LLM regardless of score
49    pub always_route_alerts: bool,
50
51    /// Automatically drop expired messages
52    pub drop_expired: bool,
53
54    /// SFE scorer for computing importance/freshness
55    scorer_config: ContextScorerConfig,
56}
57
58impl RoutingPolicy {
59    /// Creates a new routing policy with custom threshold
60    pub fn new(llm_threshold: f64) -> Self {
61        Self {
62            llm_threshold,
63            always_route_alerts: true,
64            drop_expired: true,
65            scorer_config: ContextScorerConfig::default(),
66        }
67    }
68
69    /// Sets whether to always route alerts to LLM
70    pub fn with_always_route_alerts(mut self, enabled: bool) -> Self {
71        self.always_route_alerts = enabled;
72        self
73    }
74
75    /// Sets whether to drop expired messages
76    pub fn with_drop_expired(mut self, enabled: bool) -> Self {
77        self.drop_expired = enabled;
78        self
79    }
80
81    /// Sets custom SFE scorer configuration
82    pub fn with_scorer_config(mut self, config: ContextScorerConfig) -> Self {
83        self.scorer_config = config;
84        self
85    }
86
87    /// Decides how to route a message
88    ///
89    /// Decision flow:
90    /// 1. Check expiry (if enabled) -> Drop
91    /// 2. Check if Alert + high priority -> SendToLLM
92    /// 3. For Event/State: compute importance score -> threshold check
93    /// 4. Commands/Queries -> ProcessLocally
94    ///
95    /// # Arguments
96    ///
97    /// * `msg` - The message to route
98    /// * `now_ms` - Current time in epoch milliseconds
99    pub fn decide(&self, msg: &NetMessage, now_ms: u64) -> Result<RoutingDecision> {
100        // 1. Check expiry
101        if self.drop_expired && msg.is_expired(now_ms)? {
102            return Ok(RoutingDecision::Drop);
103        }
104
105        // 2. Always route high-priority alerts
106        if self.always_route_alerts && msg.kind.is_alert() && msg.priority > 200 {
107            return Ok(RoutingDecision::SendToLLM);
108        }
109
110        // 3. For Event/State: compute importance and check threshold
111        if msg.kind.is_event() || msg.kind.is_state() {
112            let importance = self.base_importance(msg, now_ms)?;
113            return if importance >= self.llm_threshold {
114                Ok(RoutingDecision::SendToLLM)
115            } else {
116                Ok(RoutingDecision::ProcessLocally)
117            };
118        }
119
120        // 4. Commands and Queries: local processing by default
121        // (Future: complex queries could be routed to LLM)
122        Ok(RoutingDecision::ProcessLocally)
123    }
124
125    /// Computes base importance score for a message (0.0-1.0)
126    ///
127    /// Combines:
128    /// - Priority (0-255 normalized to 0.0-1.0): 50% weight
129    /// - SFE score (freshness + importance from lnmp-sfe): 50% weight
130    ///
131    /// # Formula
132    ///
133    /// ```text
134    /// importance = (priority / 255.0) * 0.5 + sfe_score * 0.5
135    /// ```
136    pub fn base_importance(&self, msg: &NetMessage, now_ms: u64) -> Result<f64> {
137        // Normalize priority to 0.0-1.0
138        let priority_score = msg.priority as f64 / 255.0;
139
140        // Compute SFE score using ContextScorer
141        let sfe_score = if msg.timestamp().is_some() {
142            let scorer = ContextScorer::with_config(self.scorer_config.clone());
143            let profile = scorer.score_envelope(&msg.envelope, now_ms);
144
145            // Use composite_score which combines freshness + importance + confidence
146            profile.composite_score()
147        } else {
148            // No timestamp, use half score
149            0.5
150        };
151
152        // Combine priority and SFE
153        Ok(priority_score * 0.5 + sfe_score * 0.5)
154    }
155}
156
157impl Default for RoutingPolicy {
158    fn default() -> Self {
159        Self::new(0.7)
160    }
161}
162
163#[cfg(test)]
164mod tests {
165    use super::*;
166    use crate::kind::MessageKind;
167    use lnmp_core::{LnmpField, LnmpRecord, LnmpValue};
168    use lnmp_envelope::EnvelopeBuilder;
169
170    fn sample_record() -> LnmpRecord {
171        let mut record = LnmpRecord::new();
172        record.add_field(LnmpField {
173            fid: 42,
174            value: LnmpValue::Int(100),
175        });
176        record
177    }
178
179    #[test]
180    fn test_alert_always_routed_to_llm() {
181        let policy = RoutingPolicy::default();
182
183        let envelope = EnvelopeBuilder::new(sample_record())
184            .timestamp(1000)
185            .build();
186        let msg = NetMessage::new(envelope, MessageKind::Alert);
187
188        assert_eq!(
189            policy.decide(&msg, 2000).unwrap(),
190            RoutingDecision::SendToLLM
191        );
192    }
193
194    #[test]
195    fn test_expired_message_dropped() {
196        let policy = RoutingPolicy::default();
197
198        let envelope = EnvelopeBuilder::new(sample_record())
199            .timestamp(1000)
200            .build();
201
202        // Event with TTL=5000ms, age=6000ms -> expired
203        let msg = NetMessage::with_qos(envelope, MessageKind::Event, 100, 5000);
204
205        assert_eq!(policy.decide(&msg, 7000).unwrap(), RoutingDecision::Drop);
206    }
207
208    #[test]
209    fn test_low_priority_event_processed_locally() {
210        let policy = RoutingPolicy::default(); // threshold = 0.7
211
212        let envelope = EnvelopeBuilder::new(sample_record())
213            .timestamp(1000)
214            .build();
215
216        // Low priority event (30/255 = 0.12) -> below threshold
217        let msg = NetMessage::with_qos(envelope, MessageKind::Event, 30, 10000);
218
219        assert_eq!(
220            policy.decide(&msg, 2000).unwrap(),
221            RoutingDecision::ProcessLocally
222        );
223    }
224
225    #[test]
226    fn test_high_priority_state_sent_to_llm() {
227        let policy = RoutingPolicy::default();
228
229        let envelope = EnvelopeBuilder::new(sample_record())
230            .timestamp(1000)
231            .build();
232
233        // High priority state (220/255 = 0.86) -> above threshold
234        let msg = NetMessage::with_qos(envelope, MessageKind::State, 220, 10000);
235
236        assert_eq!(
237            policy.decide(&msg, 2000).unwrap(),
238            RoutingDecision::SendToLLM
239        );
240    }
241
242    #[test]
243    fn test_command_processed_locally() {
244        let policy = RoutingPolicy::default();
245
246        let envelope = EnvelopeBuilder::new(sample_record())
247            .timestamp(1000)
248            .build();
249
250        let msg = NetMessage::new(envelope, MessageKind::Command);
251
252        assert_eq!(
253            policy.decide(&msg, 2000).unwrap(),
254            RoutingDecision::ProcessLocally
255        );
256    }
257
258    #[test]
259    fn test_query_processed_locally() {
260        let policy = RoutingPolicy::default();
261
262        let envelope = EnvelopeBuilder::new(sample_record())
263            .timestamp(1000)
264            .build();
265
266        let msg = NetMessage::new(envelope, MessageKind::Query);
267
268        assert_eq!(
269            policy.decide(&msg, 2000).unwrap(),
270            RoutingDecision::ProcessLocally
271        );
272    }
273
274    #[test]
275    fn test_base_importance_high_priority() {
276        let policy = RoutingPolicy::default();
277
278        let envelope = EnvelopeBuilder::new(sample_record())
279            .timestamp(1000)
280            .build();
281
282        let msg = NetMessage::with_qos(envelope, MessageKind::Event, 255, 10000);
283
284        let importance = policy.base_importance(&msg, 2000).unwrap();
285
286        // High priority (1.0) with fresh timestamp should yield high score
287        assert!(
288            importance > 0.5,
289            "Expected high importance, got {}",
290            importance
291        );
292    }
293
294    #[test]
295    fn test_base_importance_low_priority() {
296        let policy = RoutingPolicy::default();
297
298        let envelope = EnvelopeBuilder::new(sample_record())
299            .timestamp(1000)
300            .build();
301
302        let msg = NetMessage::with_qos(envelope, MessageKind::Event, 10, 10000);
303
304        let importance = policy.base_importance(&msg, 2000).unwrap();
305
306        // Low priority should yield lower score
307        assert!(
308            importance < 0.7,
309            "Expected low importance, got {}",
310            importance
311        );
312    }
313
314    #[test]
315    fn test_custom_threshold() {
316        let policy = RoutingPolicy::new(0.9); // Very high threshold
317
318        let envelope = EnvelopeBuilder::new(sample_record())
319            .timestamp(1000)
320            .build();
321
322        // Medium priority (150/255 = 0.59)
323        let msg = NetMessage::with_qos(envelope, MessageKind::Event, 150, 10000);
324
325        // Should be processed locally due to high threshold
326        assert_eq!(
327            policy.decide(&msg, 2000).unwrap(),
328            RoutingDecision::ProcessLocally
329        );
330    }
331
332    #[test]
333    fn test_disable_alert_routing() {
334        let policy = RoutingPolicy::default().with_always_route_alerts(false);
335
336        let envelope = EnvelopeBuilder::new(sample_record())
337            .timestamp(1000)
338            .build();
339
340        // Low priority alert (if alerts didn't auto-route, this would be local)
341        let msg = NetMessage::with_qos(envelope, MessageKind::Alert, 50, 1000);
342
343        // With alerts disabled from auto-route, should check importance
344        let decision = policy.decide(&msg, 2000).unwrap();
345
346        // Low priority alert should not auto-route
347        assert_ne!(decision, RoutingDecision::SendToLLM);
348    }
349
350    #[test]
351    fn test_disable_drop_expired() {
352        let policy = RoutingPolicy::default().with_drop_expired(false);
353
354        let envelope = EnvelopeBuilder::new(sample_record())
355            .timestamp(1000)
356            .build();
357
358        // Expired event
359        let msg = NetMessage::with_qos(envelope, MessageKind::Event, 30, 2000);
360
361        // Should not drop even though expired
362        assert_ne!(policy.decide(&msg, 10000).unwrap(), RoutingDecision::Drop);
363    }
364}
lnmp_net/routing.rs

lnmp_net/
routing.rs
