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::paths::state_dir()
172        .ok()
173        .map(|d| d.join("events.jsonl"))
174}
175
176fn is_test_environment() -> bool {
177    use std::sync::OnceLock;
178    static CACHED: OnceLock<bool> = OnceLock::new();
179    *CACHED.get_or_init(|| {
180        if cfg!(test) {
181            return true;
182        }
183        if std::env::var_os("__LEAN_CTX_SKIP_EVENTS").is_some() {
184            return true;
185        }
186        std::env::current_exe().is_ok_and(|p| {
187            let s = p.to_string_lossy();
188            s.contains("/deps/") || s.contains("\\deps\\")
189        })
190    })
191}
192
193fn append_jsonl(event: &LeanCtxEvent) {
194    if is_test_environment() {
195        return;
196    }
197    let Some(path) = jsonl_path() else { return };
198
199    if let Some(parent) = path.parent() {
200        let _ = std::fs::create_dir_all(parent);
201    }
202
203    if let Ok(content) = std::fs::read_to_string(&path) {
204        let lines = content.lines().count();
205        if lines >= JSONL_MAX_LINES {
206            let old = path.with_extension("jsonl.old");
207            let _ = std::fs::remove_file(&old);
208            let _ = std::fs::rename(&path, &old);
209        }
210    }
211
212    if let Ok(json) = serde_json::to_string(event) {
213        use std::io::Write;
214        if let Ok(mut f) = std::fs::OpenOptions::new()
215            .create(true)
216            .append(true)
217            .open(&path)
218        {
219            let _ = writeln!(f, "{json}");
220        }
221    }
222}
223
224// --- Public API ---
225
226pub fn emit(kind: EventKind) -> u64 {
227    bus().emit(kind)
228}
229
230pub fn events_since(after_id: u64) -> Vec<LeanCtxEvent> {
231    bus().events_since(after_id)
232}
233
234pub fn latest_events(n: usize) -> Vec<LeanCtxEvent> {
235    bus().latest_events(n)
236}
237
238#[derive(Default)]
239struct FileEventCache {
240    path: Option<std::path::PathBuf>,
241    mtime: Option<std::time::SystemTime>,
242    len: u64,
243    events: Vec<LeanCtxEvent>,
244}
245
246/// File-backed event load with a process-local cache keyed on (path, mtime, len).
247/// The dashboard polls this every 3 s; without the cache each poll re-read
248/// and re-parsed the entire JSONL (up to 10k lines) even when nothing changed.
249pub fn load_events_from_file(n: usize) -> Vec<LeanCtxEvent> {
250    static CACHE: OnceLock<Mutex<FileEventCache>> = OnceLock::new();
251    let Some(path) = jsonl_path() else {
252        return Vec::new();
253    };
254    let (mtime, len) = match std::fs::metadata(&path) {
255        Ok(m) => (m.modified().ok(), m.len()),
256        Err(_) => return Vec::new(),
257    };
258
259    let cache = CACHE.get_or_init(|| Mutex::new(FileEventCache::default()));
260    let mut guard = match cache.lock() {
261        Ok(g) => g,
262        Err(poisoned) => poisoned.into_inner(),
263    };
264
265    let fresh =
266        guard.path.as_deref() == Some(path.as_path()) && guard.mtime == mtime && guard.len == len;
267    if !fresh {
268        let Ok(content) = std::fs::read_to_string(&path) else {
269            return Vec::new();
270        };
271        guard.events = content
272            .lines()
273            .filter(|l| !l.trim().is_empty())
274            .filter_map(|l| serde_json::from_str(l).ok())
275            .collect();
276        guard.path = Some(path);
277        guard.mtime = mtime;
278        guard.len = len;
279    }
280
281    let start = guard.events.len().saturating_sub(n);
282    guard.events[start..].to_vec()
283}
284
285pub fn emit_tool_call(
286    tool: &str,
287    tokens_original: u64,
288    tokens_saved: u64,
289    mode: Option<String>,
290    duration_ms: u64,
291    path: Option<String>,
292) {
293    emit(EventKind::ToolCall {
294        tool: tool.to_string(),
295        tokens_original,
296        tokens_saved,
297        mode,
298        duration_ms,
299        path,
300    });
301}
302
303pub fn emit_cache_hit(path: &str, saved_tokens: u64) {
304    emit(EventKind::CacheHit {
305        path: path.to_string(),
306        saved_tokens,
307    });
308}
309
310pub fn emit_agent_action(agent_id: &str, action: &str, tool: Option<&str>) {
311    emit(EventKind::AgentAction {
312        agent_id: agent_id.to_string(),
313        action: action.to_string(),
314        tool: tool.map(std::string::ToString::to_string),
315    });
316}
317
318pub fn emit_budget_warning(role: &str, dimension: &str, used: &str, limit: &str, percent: u8) {
319    emit(EventKind::BudgetWarning {
320        role: role.to_string(),
321        dimension: dimension.to_string(),
322        used: used.to_string(),
323        limit: limit.to_string(),
324        percent,
325    });
326}
327
328pub fn emit_budget_exhausted(role: &str, dimension: &str, used: &str, limit: &str) {
329    emit(EventKind::BudgetExhausted {
330        role: role.to_string(),
331        dimension: dimension.to_string(),
332        used: used.to_string(),
333        limit: limit.to_string(),
334    });
335}
336
337pub fn emit_policy_violation(role: &str, tool: &str, reason: &str) {
338    emit(EventKind::PolicyViolation {
339        role: role.to_string(),
340        tool: tool.to_string(),
341        reason: reason.to_string(),
342    });
343}
344
345pub fn emit_role_changed(from: &str, to: &str) {
346    emit(EventKind::RoleChanged {
347        from: from.to_string(),
348        to: to.to_string(),
349    });
350}
351
352pub fn emit_profile_changed(from: &str, to: &str) {
353    emit(EventKind::ProfileChanged {
354        from: from.to_string(),
355        to: to.to_string(),
356    });
357}
358
359pub fn emit_slo_violation(slo_name: &str, metric: &str, threshold: f64, actual: f64, action: &str) {
360    emit(EventKind::SloViolation {
361        slo_name: slo_name.to_string(),
362        metric: metric.to_string(),
363        threshold,
364        actual,
365        action: action.to_string(),
366    });
367}
368
369pub fn emit_anomaly(metric: &str, expected: f64, actual: f64, deviation_factor: f64) {
370    emit(EventKind::Anomaly {
371        metric: metric.to_string(),
372        expected,
373        actual,
374        deviation_factor,
375    });
376}
377
378pub fn emit_verification_warning(warning_kind: &str, detail: &str, severity: &str) {
379    emit(EventKind::VerificationWarning {
380        warning_kind: warning_kind.to_string(),
381        detail: detail.to_string(),
382        severity: severity.to_string(),
383    });
384}
385
386pub fn emit_threshold_adapted(language: &str, arm: &str, old_threshold: f64, new_threshold: f64) {
387    emit(EventKind::ThresholdAdapted {
388        language: language.to_string(),
389        arm: arm.to_string(),
390        old_threshold,
391        new_threshold,
392    });
393}
394
395#[cfg(test)]
396mod tests {
397    use super::*;
398
399    #[test]
400    fn emit_returns_positive_id() {
401        let id = emit(EventKind::ToolCall {
402            tool: "ctx_read".to_string(),
403            tokens_original: 1000,
404            tokens_saved: 800,
405            mode: Some("map".to_string()),
406            duration_ms: 5,
407            path: Some("src/main.rs".to_string()),
408        });
409        assert!(id > 0);
410        let events = latest_events(100);
411        assert!(events.iter().any(|e| e.id == id));
412    }
413
414    #[test]
415    fn events_since_filters_correctly() {
416        let id1 = emit(EventKind::CacheHit {
417            path: "filter_test_a.rs".to_string(),
418            saved_tokens: 100,
419        });
420        let id2 = emit(EventKind::CacheHit {
421            path: "filter_test_b.rs".to_string(),
422            saved_tokens: 200,
423        });
424
425        let after = events_since(id1);
426        assert!(after.iter().any(|e| e.id == id2));
427        assert!(after.iter().all(|e| e.id > id1));
428    }
429
430    /// The (path, mtime, len) cache must never serve stale events: appending a
431    /// line changes the file length, which has nanosecond-independent
432    /// granularity (unlike mtime), so new events show up on the next poll.
433    #[test]
434    fn load_events_from_file_sees_appended_events() {
435        let path = jsonl_path().expect("test sandbox data dir");
436        if let Some(parent) = path.parent() {
437            std::fs::create_dir_all(parent).expect("create data dir");
438        }
439
440        let line_a = r#"{"id":900001,"timestamp":"2026-06-12T08:00:00.000","kind":{"type":"CacheHit","path":"cached_a.rs","saved_tokens":42}}"#;
441        std::fs::write(&path, format!("{line_a}\n")).expect("write events.jsonl");
442
443        let first = load_events_from_file(50);
444        assert!(
445            first.iter().any(|e| e.id == 900_001),
446            "initial load should parse the seeded event"
447        );
448
449        // Second call with unchanged file exercises the cached branch.
450        let cached = load_events_from_file(50);
451        assert_eq!(cached.len(), first.len());
452
453        let line_b = r#"{"id":900002,"timestamp":"2026-06-12T08:00:01.000","kind":{"type":"CacheHit","path":"cached_b.rs","saved_tokens":7}}"#;
454        {
455            use std::io::Write;
456            let mut f = std::fs::OpenOptions::new()
457                .append(true)
458                .open(&path)
459                .expect("append events.jsonl");
460            writeln!(f, "{line_b}").expect("append line");
461        }
462
463        let second = load_events_from_file(50);
464        assert!(
465            second.iter().any(|e| e.id == 900_002),
466            "append must invalidate the cache and surface the new event"
467        );
468    }
469}
lean_ctx/core/events.rs

lean_ctx/core/
events.rs
