lean_ctx/core/
events.rs

1use serde::{Deserialize, Serialize};
2use std::collections::VecDeque;
3use std::sync::atomic::{AtomicU64, Ordering};
4use std::sync::{Mutex, OnceLock};
5
6const RING_CAPACITY: usize = 1000;
7const JSONL_MAX_LINES: usize = 10_000;
8
9#[derive(Clone, Debug, Serialize, Deserialize)]
10pub struct LeanCtxEvent {
11    pub id: u64,
12    pub timestamp: String,
13    pub kind: EventKind,
14}
15
16#[derive(Clone, Debug, Serialize, Deserialize)]
17#[serde(tag = "type")]
18pub enum EventKind {
19    ToolCall {
20        tool: String,
21        tokens_original: u64,
22        tokens_saved: u64,
23        mode: Option<String>,
24        duration_ms: u64,
25        path: Option<String>,
26    },
27    CacheHit {
28        path: String,
29        saved_tokens: u64,
30    },
31    Compression {
32        path: String,
33        before_lines: u32,
34        after_lines: u32,
35        strategy: String,
36        kept_line_count: u32,
37        removed_line_count: u32,
38    },
39    AgentAction {
40        agent_id: String,
41        action: String,
42        tool: Option<String>,
43    },
44    KnowledgeUpdate {
45        category: String,
46        key: String,
47        action: String,
48    },
49    ThresholdShift {
50        language: String,
51        old_entropy: f64,
52        new_entropy: f64,
53        old_jaccard: f64,
54        new_jaccard: f64,
55    },
56    BudgetWarning {
57        role: String,
58        dimension: String,
59        used: String,
60        limit: String,
61        percent: u8,
62    },
63    BudgetExhausted {
64        role: String,
65        dimension: String,
66        used: String,
67        limit: String,
68    },
69    PolicyViolation {
70        role: String,
71        tool: String,
72        reason: String,
73    },
74    RoleChanged {
75        from: String,
76        to: String,
77    },
78    ProfileChanged {
79        from: String,
80        to: String,
81    },
82    SloViolation {
83        slo_name: String,
84        metric: String,
85        threshold: f64,
86        actual: f64,
87        action: String,
88    },
89    Anomaly {
90        metric: String,
91        expected: f64,
92        actual: f64,
93        deviation_factor: f64,
94    },
95    VerificationWarning {
96        warning_kind: String,
97        detail: String,
98        severity: String,
99    },
100    ThresholdAdapted {
101        language: String,
102        arm: String,
103        old_threshold: f64,
104        new_threshold: f64,
105    },
106}
107
108struct EventBus {
109    seq: AtomicU64,
110    ring: Mutex<VecDeque<LeanCtxEvent>>,
111}
112
113impl EventBus {
114    fn new() -> Self {
115        Self {
116            seq: AtomicU64::new(0),
117            ring: Mutex::new(VecDeque::with_capacity(RING_CAPACITY)),
118        }
119    }
120
121    fn emit(&self, kind: EventKind) -> u64 {
122        let id = self.seq.fetch_add(1, Ordering::Relaxed) + 1;
123        let event = LeanCtxEvent {
124            id,
125            timestamp: chrono::Local::now()
126                .format("%Y-%m-%dT%H:%M:%S%.3f")
127                .to_string(),
128            kind,
129        };
130
131        {
132            let mut ring = self
133                .ring
134                .lock()
135                .unwrap_or_else(std::sync::PoisonError::into_inner);
136            if ring.len() >= RING_CAPACITY {
137                ring.pop_front();
138            }
139            ring.push_back(event.clone());
140        }
141
142        append_jsonl(&event);
143        id
144    }
145
146    fn events_since(&self, after_id: u64) -> Vec<LeanCtxEvent> {
147        let ring = self
148            .ring
149            .lock()
150            .unwrap_or_else(std::sync::PoisonError::into_inner);
151        ring.iter().filter(|e| e.id > after_id).cloned().collect()
152    }
153
154    fn latest_events(&self, n: usize) -> Vec<LeanCtxEvent> {
155        let ring = self
156            .ring
157            .lock()
158            .unwrap_or_else(std::sync::PoisonError::into_inner);
159        let len = ring.len();
160        let start = len.saturating_sub(n);
161        ring.iter().skip(start).cloned().collect()
162    }
163}
164
165fn bus() -> &'static EventBus {
166    static INSTANCE: OnceLock<EventBus> = OnceLock::new();
167    INSTANCE.get_or_init(EventBus::new)
168}
169
170fn jsonl_path() -> Option<std::path::PathBuf> {
171    crate::core::data_dir::lean_ctx_data_dir()
172        .ok()
173        .map(|d| d.join("events.jsonl"))
174}
175
176fn append_jsonl(event: &LeanCtxEvent) {
177    let Some(path) = jsonl_path() else { return };
178
179    if let Some(parent) = path.parent() {
180        let _ = std::fs::create_dir_all(parent);
181    }
182
183    if let Ok(content) = std::fs::read_to_string(&path) {
184        let lines = content.lines().count();
185        if lines >= JSONL_MAX_LINES {
186            let old = path.with_extension("jsonl.old");
187            let _ = std::fs::remove_file(&old);
188            let _ = std::fs::rename(&path, &old);
189        }
190    }
191
192    if let Ok(json) = serde_json::to_string(event) {
193        use std::io::Write;
194        if let Ok(mut f) = std::fs::OpenOptions::new()
195            .create(true)
196            .append(true)
197            .open(&path)
198        {
199            let _ = writeln!(f, "{json}");
200        }
201    }
202}
203
204// --- Public API ---
205
206pub fn emit(kind: EventKind) -> u64 {
207    bus().emit(kind)
208}
209
210pub fn events_since(after_id: u64) -> Vec<LeanCtxEvent> {
211    bus().events_since(after_id)
212}
213
214pub fn latest_events(n: usize) -> Vec<LeanCtxEvent> {
215    bus().latest_events(n)
216}
217
218pub fn load_events_from_file(n: usize) -> Vec<LeanCtxEvent> {
219    let Some(path) = jsonl_path() else {
220        return Vec::new();
221    };
222    let Ok(content) = std::fs::read_to_string(&path) else {
223        return Vec::new();
224    };
225    let all: Vec<LeanCtxEvent> = content
226        .lines()
227        .filter(|l| !l.trim().is_empty())
228        .filter_map(|l| serde_json::from_str(l).ok())
229        .collect();
230    let start = all.len().saturating_sub(n);
231    all[start..].to_vec()
232}
233
234pub fn emit_tool_call(
235    tool: &str,
236    tokens_original: u64,
237    tokens_saved: u64,
238    mode: Option<String>,
239    duration_ms: u64,
240    path: Option<String>,
241) {
242    emit(EventKind::ToolCall {
243        tool: tool.to_string(),
244        tokens_original,
245        tokens_saved,
246        mode,
247        duration_ms,
248        path,
249    });
250}
251
252pub fn emit_cache_hit(path: &str, saved_tokens: u64) {
253    emit(EventKind::CacheHit {
254        path: path.to_string(),
255        saved_tokens,
256    });
257}
258
259pub fn emit_agent_action(agent_id: &str, action: &str, tool: Option<&str>) {
260    emit(EventKind::AgentAction {
261        agent_id: agent_id.to_string(),
262        action: action.to_string(),
263        tool: tool.map(std::string::ToString::to_string),
264    });
265}
266
267pub fn emit_budget_warning(role: &str, dimension: &str, used: &str, limit: &str, percent: u8) {
268    emit(EventKind::BudgetWarning {
269        role: role.to_string(),
270        dimension: dimension.to_string(),
271        used: used.to_string(),
272        limit: limit.to_string(),
273        percent,
274    });
275}
276
277pub fn emit_budget_exhausted(role: &str, dimension: &str, used: &str, limit: &str) {
278    emit(EventKind::BudgetExhausted {
279        role: role.to_string(),
280        dimension: dimension.to_string(),
281        used: used.to_string(),
282        limit: limit.to_string(),
283    });
284}
285
286pub fn emit_policy_violation(role: &str, tool: &str, reason: &str) {
287    emit(EventKind::PolicyViolation {
288        role: role.to_string(),
289        tool: tool.to_string(),
290        reason: reason.to_string(),
291    });
292}
293
294pub fn emit_role_changed(from: &str, to: &str) {
295    emit(EventKind::RoleChanged {
296        from: from.to_string(),
297        to: to.to_string(),
298    });
299}
300
301pub fn emit_profile_changed(from: &str, to: &str) {
302    emit(EventKind::ProfileChanged {
303        from: from.to_string(),
304        to: to.to_string(),
305    });
306}
307
308pub fn emit_slo_violation(slo_name: &str, metric: &str, threshold: f64, actual: f64, action: &str) {
309    emit(EventKind::SloViolation {
310        slo_name: slo_name.to_string(),
311        metric: metric.to_string(),
312        threshold,
313        actual,
314        action: action.to_string(),
315    });
316}
317
318pub fn emit_anomaly(metric: &str, expected: f64, actual: f64, deviation_factor: f64) {
319    emit(EventKind::Anomaly {
320        metric: metric.to_string(),
321        expected,
322        actual,
323        deviation_factor,
324    });
325}
326
327pub fn emit_verification_warning(warning_kind: &str, detail: &str, severity: &str) {
328    emit(EventKind::VerificationWarning {
329        warning_kind: warning_kind.to_string(),
330        detail: detail.to_string(),
331        severity: severity.to_string(),
332    });
333}
334
335pub fn emit_threshold_adapted(language: &str, arm: &str, old_threshold: f64, new_threshold: f64) {
336    emit(EventKind::ThresholdAdapted {
337        language: language.to_string(),
338        arm: arm.to_string(),
339        old_threshold,
340        new_threshold,
341    });
342}
343
344#[cfg(test)]
345mod tests {
346    use super::*;
347
348    #[test]
349    fn emit_returns_positive_id() {
350        let id = emit(EventKind::ToolCall {
351            tool: "ctx_read".to_string(),
352            tokens_original: 1000,
353            tokens_saved: 800,
354            mode: Some("map".to_string()),
355            duration_ms: 5,
356            path: Some("src/main.rs".to_string()),
357        });
358        assert!(id > 0);
359        let events = latest_events(100);
360        assert!(events.iter().any(|e| e.id == id));
361    }
362
363    #[test]
364    fn events_since_filters_correctly() {
365        let id1 = emit(EventKind::CacheHit {
366            path: "filter_test_a.rs".to_string(),
367            saved_tokens: 100,
368        });
369        let id2 = emit(EventKind::CacheHit {
370            path: "filter_test_b.rs".to_string(),
371            saved_tokens: 200,
372        });
373
374        let after = events_since(id1);
375        assert!(after.iter().any(|e| e.id == id2));
376        assert!(after.iter().all(|e| e.id > id1));
377    }
378}
lean_ctx/core/events.rs

lean_ctx/core/
events.rs
