lean_ctx/proxy/
routing_feedback.rs

1//! Adapter between proxy routing events and OCLA quality tracking.
2
3use std::collections::{HashMap, VecDeque};
4use std::sync::{Arc, Mutex};
5
6use crate::core::ocla::routing_quality::{RoutingDecision, RoutingOutcome, RoutingQualityTracker};
7
8const MAX_PENDING_DECISIONS: usize = 1_000;
9type PendingDecisions = HashMap<(String, String), VecDeque<RoutingDecision>>;
10
11/// Collects proxy routing decisions and their measured outcomes.
12#[derive(Clone, Debug)]
13pub struct RoutingFeedback {
14    tracker: Arc<Mutex<RoutingQualityTracker>>,
15    pending_decisions: Arc<Mutex<PendingDecisions>>,
16}
17
18impl RoutingFeedback {
19    /// Creates an empty routing feedback collector.
20    pub fn new() -> Self {
21        Self {
22            tracker: Arc::new(Mutex::new(RoutingQualityTracker::new())),
23            pending_decisions: Arc::new(Mutex::new(HashMap::new())),
24        }
25    }
26
27    /// Records a route selection until its measured outcome arrives.
28    pub fn record_decision(&self, original: &str, routed: &str, reason: &str) {
29        let decision = RoutingDecision {
30            original_model: original.to_string(),
31            routed_model: routed.to_string(),
32            reason: reason.to_string(),
33            timestamp: chrono::Utc::now().to_rfc3339(),
34        };
35        let key = (original.to_string(), routed.to_string());
36        let mut pending = self
37            .pending_decisions
38            .lock()
39            .expect("routing feedback pending decision mutex poisoned");
40        let queue = pending.entry(key).or_default();
41        queue.push_back(decision);
42        while pending.values().map(VecDeque::len).sum::<usize>() > MAX_PENDING_DECISIONS {
43            let Some(evicted_key) = pending
44                .iter()
45                .find(|(_, decisions)| !decisions.is_empty())
46                .map(|(key, _)| key.clone())
47            else {
48                break;
49            };
50            if let Some(decisions) = pending.get_mut(&evicted_key) {
51                decisions.pop_front();
52                if decisions.is_empty() {
53                    pending.remove(&evicted_key);
54                }
55            }
56        }
57    }
58
59    /// Records measured quality for a route and forwards it to the tracker.
60    pub fn record_outcome(
61        &self,
62        original: &str,
63        routed: &str,
64        quality: f64,
65        tokens_saved: u64,
66        latency_delta_ms: i64,
67    ) {
68        let key = (original.to_string(), routed.to_string());
69        let decision = {
70            let mut pending = self
71                .pending_decisions
72                .lock()
73                .expect("routing feedback pending decision mutex poisoned");
74            let decision = pending.get_mut(&key).and_then(VecDeque::pop_front);
75            if pending.get(&key).is_some_and(VecDeque::is_empty) {
76                pending.remove(&key);
77            }
78            decision
79        };
80        let decision = decision.unwrap_or_else(|| RoutingDecision {
81            original_model: original.to_string(),
82            routed_model: routed.to_string(),
83            reason: "outcome without recorded decision".to_string(),
84            timestamp: chrono::Utc::now().to_rfc3339(),
85        });
86
87        self.tracker
88            .lock()
89            .expect("routing feedback tracker mutex poisoned")
90            .record(RoutingOutcome {
91                decision,
92                quality_score: Some(quality),
93                tokens_saved,
94                latency_delta_ms,
95            });
96    }
97
98    /// Returns whether tracked route quality warrants fallback.
99    pub fn should_use_fallback(&self) -> bool {
100        self.tracker
101            .lock()
102            .expect("routing feedback tracker mutex poisoned")
103            .should_fallback()
104    }
105
106    /// Returns tracked success rate and average token savings.
107    pub fn stats(&self) -> (f64, f64) {
108        let tracker = self
109            .tracker
110            .lock()
111            .expect("routing feedback tracker mutex poisoned");
112        (tracker.success_rate(), tracker.average_savings())
113    }
114}
115
116impl Default for RoutingFeedback {
117    fn default() -> Self {
118        Self::new()
119    }
120}
121
122#[cfg(test)]
123mod tests {
124    use super::*;
125
126    #[test]
127    fn records_decision_until_matching_outcome() {
128        let feedback = RoutingFeedback::new();
129
130        feedback.record_decision("expensive", "fast", "token budget");
131        assert_eq!(feedback.stats(), (0.0, 0.0));
132        assert!(!feedback.should_use_fallback());
133
134        let pending = feedback
135            .pending_decisions
136            .lock()
137            .expect("test pending decision mutex poisoned");
138        let key = (String::from("expensive"), String::from("fast"));
139        let decision = &pending[&key][0];
140        assert_eq!(decision.reason, "token budget");
141    }
142
143    #[test]
144    fn records_successful_outcome_and_statistics() {
145        let feedback = RoutingFeedback::new();
146
147        feedback.record_decision("expensive", "fast", "token budget");
148        feedback.record_outcome("expensive", "fast", 0.95, 120, -10);
149
150        assert_eq!(feedback.stats(), (1.0, 120.0));
151        assert!(!feedback.should_use_fallback());
152    }
153
154    #[test]
155    fn poor_outcome_triggers_fallback() {
156        let feedback = RoutingFeedback::new();
157
158        feedback.record_outcome("expensive", "fast", 0.4, 20, 15);
159
160        assert_eq!(feedback.stats(), (0.0, 20.0));
161        assert!(feedback.should_use_fallback());
162    }
163}
lean_ctx/proxy/routing_feedback.rs

lean_ctx/proxy/
routing_feedback.rs
