refactor(brain): A2 — split arje-brain en 3 sub-crates
1//! Observador estadístico. Mantiene marginales y co-ocurrencias dentro de una
2//! ventana deslizante. Calcula entropía de Shannon e información mutua para
3//! identificar correlaciones significativas.
4//!
5//! Diseño:
6//! - Counters incrementales: cada `record()` es O(window_size) en el peor
7//! caso (actualiza co-ocurrencias con cada evento del window).
8//! - Sin recomputaciones globales: marginales y joint counts son state.
9//! - El cálculo de H(X), P(B|A), I(A;B) es O(|distinct events|).
10
11use arje_brain_rules::{EventKind, TimedEvent};
12use std::collections::{HashMap, VecDeque};
13use std::time::Instant;
14
15/// Histograma de gaps temporales con buckets exponenciales en segundos.
16/// Cubre 6 órdenes de magnitud: 1ms hasta 1000s.
17#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
18pub struct GapHistogram {
19 /// Buckets cumulativos (Prometheus-style): cada índice cuenta eventos
20 /// con gap ≤ ese límite. Limites: 1ms, 10ms, 100ms, 1s, 10s, 100s, 1000s.
21 pub buckets: [u64; 7],
22 pub count: u64,
23 pub sum_secs: f64,
24 /// Suma de cuadrados — permite calcular varianza/stddev en O(1).
25 pub sum_squares_secs: f64,
26 pub max_secs: f64,
27}
28
29/// Estadísticas resumidas de un GapHistogram, usables en cristales temporales.
30#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
31pub struct GapStats {
32 pub count: u64,
33 pub mean_secs: f64,
34 pub stddev_secs: f64,
35 pub max_secs: f64,
36}
37
38const GAP_BUCKET_LIMITS_SECS: [f64; 7] = [
39 0.001, 0.01, 0.1, 1.0, 10.0, 100.0, 1000.0,
40];
41
42impl GapHistogram {
43 pub fn observe(&mut self, gap_secs: f64) {
44 for (i, &limit) in GAP_BUCKET_LIMITS_SECS.iter().enumerate() {
45 if gap_secs <= limit {
46 self.buckets[i] += 1;
47 }
48 }
49 self.count += 1;
50 self.sum_secs += gap_secs;
51 self.sum_squares_secs += gap_secs * gap_secs;
52 if gap_secs > self.max_secs { self.max_secs = gap_secs; }
53 }
54
55 pub fn mean_secs(&self) -> f64 {
56 if self.count == 0 { 0.0 } else { self.sum_secs / self.count as f64 }
57 }
58
59 /// Desviación estándar muestral. Computada vía `sum_squares - n*mean²`
60 /// para precisión razonable sin almacenar las muestras.
61 pub fn stddev_secs(&self) -> f64 {
62 if self.count < 2 { return 0.0; }
63 let n = self.count as f64;
64 let mean = self.mean_secs();
65 let var = (self.sum_squares_secs - n * mean * mean) / (n - 1.0);
66 // Numerical floor: var puede ser ligeramente negativo por float ε.
67 if var <= 0.0 { 0.0 } else { var.sqrt() }
68 }
69
70 pub fn stats(&self) -> GapStats {
71 GapStats {
72 count: self.count,
73 mean_secs: self.mean_secs(),
74 stddev_secs: self.stddev_secs(),
75 max_secs: self.max_secs,
76 }
77 }
78
79 pub fn bucket_limits() -> &'static [f64; 7] { &GAP_BUCKET_LIMITS_SECS }
80}
81
82pub struct Observer {
83 window: VecDeque<TimedEvent>,
84 window_size: usize,
85 marginal: HashMap<EventKind, u64>,
86 cooccur: HashMap<(EventKind, EventKind), u64>,
87 total: u64,
88 /// Last-seen timestamps para aplicar decay en query time. None = sin
89 /// time-decay (modo tradicional).
90 last_seen_marginal: HashMap<EventKind, Instant>,
91 last_seen_cooccur: HashMap<(EventKind, EventKind), Instant>,
92 /// Half-life del decay exponencial en segundos. None = sin decay
93 /// (las consultas devuelven los counts crudos).
94 half_life_secs: Option<f64>,
95 /// Histograma de gaps temporales por par (a, b). Capturado al `record()`.
96 gap_histograms: HashMap<(EventKind, EventKind), GapHistogram>,
97 /// Sets de "qué cambió desde el último snapshot". Se vacían en
98 /// `snapshot()` y `snapshot_delta()`. Usado para escritura incremental.
99 dirty_marginal: std::collections::HashSet<EventKind>,
100 dirty_cooccur: std::collections::HashSet<(EventKind, EventKind)>,
101}
102
103impl Observer {
104 pub fn new(window_size: usize) -> Self {
105 Self {
106 window: VecDeque::with_capacity(window_size),
107 window_size,
108 marginal: HashMap::new(),
109 cooccur: HashMap::new(),
110 total: 0,
111 last_seen_marginal: HashMap::new(),
112 last_seen_cooccur: HashMap::new(),
113 half_life_secs: None,
114 gap_histograms: HashMap::new(),
115 dirty_marginal: std::collections::HashSet::new(),
116 dirty_cooccur: std::collections::HashSet::new(),
117 }
118 }
119
120 /// Activa decay exponencial con half-life en segundos. λ = ln(2)/half_life.
121 /// Aplicado en query time sobre los counts crudos usando last_seen.
122 pub fn with_half_life(mut self, half_life_secs: f64) -> Self {
123 if half_life_secs > 0.0 {
124 self.half_life_secs = Some(half_life_secs);
125 }
126 self
127 }
128
129 pub fn half_life(&self) -> Option<f64> { self.half_life_secs }
130
131 /// Registra un evento. Actualiza marginales y co-ocurrencias contra todo
132 /// evento aún en la ventana.
133 pub fn record(&mut self, kind: EventKind) {
134 let now = Instant::now();
135 let timed = TimedEvent { kind: kind.clone(), at: now };
136
137 // Co-ocurrencias: este evento con cada uno previo en ventana.
138 // Capturamos también el gap temporal (now - w.at) para histograma.
139 for w in &self.window {
140 let key = (w.kind.clone(), kind.clone());
141 *self.cooccur.entry(key.clone()).or_insert(0) += 1;
142 self.last_seen_cooccur.insert(key.clone(), now);
143 let gap_secs = now.duration_since(w.at).as_secs_f64();
144 self.gap_histograms.entry(key.clone()).or_default().observe(gap_secs);
145 self.dirty_cooccur.insert(key);
146 }
147
148 self.window.push_back(timed);
149 if self.window.len() > self.window_size {
150 self.window.pop_front();
151 }
152
153 *self.marginal.entry(kind.clone()).or_insert(0) += 1;
154 self.last_seen_marginal.insert(kind.clone(), now);
155 self.dirty_marginal.insert(kind);
156 self.total += 1;
157 }
158
159 /// Aplica el decay sobre un count crudo dado el `last_seen` correspondiente.
160 /// Si half_life es None, devuelve el count tal cual (sin decay).
161 fn decay(&self, count: u64, last_seen: Option<Instant>) -> f64 {
162 let raw = count as f64;
163 let (hl, last) = match (self.half_life_secs, last_seen) {
164 (Some(hl), Some(t)) => (hl, t),
165 _ => return raw,
166 };
167 let age_secs = Instant::now().duration_since(last).as_secs_f64();
168 raw * 0.5_f64.powf(age_secs / hl)
169 }
170
171 /// Marginal con decay aplicado.
172 pub fn marginal_decayed(&self, k: &EventKind) -> f64 {
173 let raw = self.marginal.get(k).copied().unwrap_or(0);
174 let last = self.last_seen_marginal.get(k).copied();
175 self.decay(raw, last)
176 }
177
178 /// Cooccurrence con decay aplicado.
179 pub fn cooccur_decayed(&self, a: &EventKind, b: &EventKind) -> f64 {
180 let raw = self.cooccur.get(&(a.clone(), b.clone())).copied().unwrap_or(0);
181 let last = self.last_seen_cooccur.get(&(a.clone(), b.clone())).copied();
182 self.decay(raw, last)
183 }
184
185 /// Entropía de Shannon de la distribución marginal de eventos.
186 /// H(X) = −Σ p(x) log₂ p(x). Unidad: bits.
187 pub fn shannon_entropy(&self) -> f64 {
188 if self.total == 0 { return 0.0; }
189 let total = self.total as f64;
190 self.marginal.values()
191 .map(|&c| {
192 let p = c as f64 / total;
193 if p > 0.0 { -p * p.log2() } else { 0.0 }
194 })
195 .sum()
196 }
197
198 /// P(b | a) = "dado que algo siguió a `a` dentro del window, qué fracción
199 /// fue `b`". Suma 1 sobre todos los b posibles para un a fijo.
200 ///
201 /// Implementación: cooccur_decayed(a, b) / Σ_x cooccur_decayed(a, x).
202 /// Si half_life is None, los decayed values son los counts crudos.
203 pub fn conditional_prob(&self, a: &EventKind, b: &EventKind) -> f64 {
204 let joint = self.cooccur_decayed(a, b);
205 let row_total: f64 = self.cooccur.keys()
206 .filter(|(x, _)| x == a)
207 .map(|(x, y)| self.cooccur_decayed(x, y))
208 .sum();
209 if row_total <= 0.0 { 0.0 } else { joint / row_total }
210 }
211
212 /// Información mutua puntual entre `a` y `b` con decay aplicado:
213 /// PMI(a, b) = log₂( P(a, b) / (P(a) · P(b)) ).
214 /// Positivo → más correlacionados de lo que sugiere independencia.
215 pub fn pmi(&self, a: &EventKind, b: &EventKind) -> f64 {
216 // Total decayed: suma de marginales con decay (no usamos self.total
217 // directo porque debería ser consistente con los decayed values).
218 let total_decayed: f64 = self.marginal.keys()
219 .map(|k| self.marginal_decayed(k))
220 .sum();
221 if total_decayed <= 0.0 { return 0.0; }
222 let joint = self.cooccur_decayed(a, b) / total_decayed;
223 let pa = self.marginal_decayed(a) / total_decayed;
224 let pb = self.marginal_decayed(b) / total_decayed;
225 if joint <= 0.0 || pa <= 0.0 || pb <= 0.0 { return 0.0; }
226 (joint / (pa * pb)).log2()
227 }
228
229 /// Información mutua acumulada de la pareja (a, b) ponderada por su
230 /// probabilidad conjunta. Útil como medida de "interés" del par.
231 pub fn weighted_pmi(&self, a: &EventKind, b: &EventKind) -> f64 {
232 if self.total == 0 { return 0.0; }
233 let joint = self.cooccur
234 .get(&(a.clone(), b.clone()))
235 .copied()
236 .unwrap_or(0) as f64 / self.total as f64;
237 joint * self.pmi(a, b)
238 }
239
240 pub fn marginals(&self) -> &HashMap<EventKind, u64> { &self.marginal }
241
242 /// Última vez que se vio un kind. None si nunca o si fue restaurado
243 /// desde snapshot (los Instants no portables se descartan).
244 pub fn last_seen_marginal(&self, kind: &EventKind) -> Option<Instant> {
245 self.last_seen_marginal.get(kind).copied()
246 }
247 pub fn cooccurrences(&self) -> &HashMap<(EventKind, EventKind), u64> { &self.cooccur }
248 pub fn total(&self) -> u64 { self.total }
249 pub fn window_size(&self) -> usize { self.window_size }
250 pub fn current_window(&self) -> usize { self.window.len() }
251
252 /// Últimos N eventos del window, en orden cronológico (más viejo primero).
253 /// Si N > window.len(), devuelve todo el window.
254 pub fn recent(&self, n: usize) -> impl Iterator<Item = &TimedEvent> {
255 let start = self.window.len().saturating_sub(n);
256 self.window.range(start..)
257 }
258
259 pub fn gap_histograms(&self) -> &HashMap<(EventKind, EventKind), GapHistogram> {
260 &self.gap_histograms
261 }
262
263 /// Top-K pares por count del histograma (más frecuentes primero).
264 /// Útil para limitar cardinalidad de métricas exportadas.
265 pub fn top_gap_pairs(&self, k: usize) -> Vec<(&(EventKind, EventKind), &GapHistogram)> {
266 let mut pairs: Vec<_> = self.gap_histograms.iter().collect();
267 pairs.sort_by(|a, b| b.1.count.cmp(&a.1.count));
268 pairs.truncate(k);
269 pairs
270 }
271
272 /// Snapshot full: estado estadístico completo. Limpia los sets dirty
273 /// como side-effect — los próximos `snapshot_delta()` cubren sólo los
274 /// cambios posteriores.
275 pub fn snapshot(&mut self) -> ObserverSnapshot {
276 self.dirty_marginal.clear();
277 self.dirty_cooccur.clear();
278 ObserverSnapshot {
279 schema_version: OBSERVER_SCHEMA_VERSION,
280 is_delta: false,
281 window_size: self.window_size,
282 half_life_secs: self.half_life_secs,
283 total: self.total,
284 marginal: self.marginal.iter()
285 .map(|(k, v)| (k.clone(), *v))
286 .collect(),
287 cooccur: self.cooccur.iter()
288 .map(|((a, b), c)| (a.clone(), b.clone(), *c))
289 .collect(),
290 gap_histograms: self.gap_histograms.iter()
291 .map(|((a, b), h)| (a.clone(), b.clone(), h.clone()))
292 .collect(),
293 }
294 }
295
296 /// Snapshot incremental: sólo incluye los kinds y pares que cambiaron
297 /// desde el último `snapshot()` o `snapshot_delta()`. Útil para
298 /// checkpoints frecuentes con poco overhead. Limpia los sets dirty.
299 pub fn snapshot_delta(&mut self) -> ObserverSnapshot {
300 let marginal: Vec<_> = self.dirty_marginal.iter()
301 .filter_map(|k| self.marginal.get(k).map(|v| (k.clone(), *v)))
302 .collect();
303 let cooccur: Vec<_> = self.dirty_cooccur.iter()
304 .filter_map(|(a, b)| {
305 self.cooccur.get(&(a.clone(), b.clone()))
306 .map(|c| (a.clone(), b.clone(), *c))
307 })
308 .collect();
309 // Para histogramas: incluimos los pares cuyo cooccur cambió.
310 let gap_histograms: Vec<_> = self.dirty_cooccur.iter()
311 .filter_map(|(a, b)| {
312 self.gap_histograms.get(&(a.clone(), b.clone()))
313 .map(|h| (a.clone(), b.clone(), h.clone()))
314 })
315 .collect();
316 self.dirty_marginal.clear();
317 self.dirty_cooccur.clear();
318 ObserverSnapshot {
319 schema_version: OBSERVER_SCHEMA_VERSION,
320 is_delta: true,
321 window_size: self.window_size,
322 half_life_secs: self.half_life_secs,
323 total: self.total,
324 marginal, cooccur, gap_histograms,
325 }
326 }
327
328 /// Aplica un delta sobre el estado actual. Para `is_delta=true`, los
329 /// valores en marginal/cooccur sobrescriben las entradas existentes.
330 /// Si `is_delta=false`, equivale a `from_snapshot` pero in-place.
331 pub fn apply_delta(&mut self, delta: ObserverSnapshot) {
332 let now = Instant::now();
333 if !delta.is_delta {
334 // Full: reset state.
335 *self = Self::from_snapshot(delta);
336 return;
337 }
338 // Incremental merge.
339 for (k, v) in delta.marginal {
340 self.last_seen_marginal.insert(k.clone(), now);
341 self.marginal.insert(k, v);
342 }
343 for (a, b, c) in delta.cooccur {
344 self.last_seen_cooccur.insert((a.clone(), b.clone()), now);
345 self.cooccur.insert((a, b), c);
346 }
347 for (a, b, h) in delta.gap_histograms {
348 self.gap_histograms.insert((a, b), h);
349 }
350 // total: sólo subimos (el delta podría estar atrasado).
351 if delta.total > self.total { self.total = delta.total; }
352 }
353
354 /// Reconstruye Observer desde un snapshot. El window queda vacío;
355 /// last_seen_* se inicializa en `now()` para que el decay arranque
356 /// "ahora" para todos los counts (aproximación razonable post-reboot).
357 pub fn from_snapshot(snap: ObserverSnapshot) -> Self {
358 let now = Instant::now();
359 let mut marginal = HashMap::new();
360 let mut last_seen_marginal = HashMap::new();
361 for (k, v) in snap.marginal {
362 last_seen_marginal.insert(k.clone(), now);
363 marginal.insert(k, v);
364 }
365 let mut cooccur = HashMap::new();
366 let mut last_seen_cooccur = HashMap::new();
367 for (a, b, c) in snap.cooccur {
368 last_seen_cooccur.insert((a.clone(), b.clone()), now);
369 cooccur.insert((a, b), c);
370 }
371 let gap_histograms = snap.gap_histograms.into_iter()
372 .map(|(a, b, h)| ((a, b), h))
373 .collect();
374 Self {
375 window: VecDeque::with_capacity(snap.window_size),
376 window_size: snap.window_size,
377 marginal,
378 cooccur,
379 total: snap.total,
380 last_seen_marginal,
381 last_seen_cooccur,
382 half_life_secs: snap.half_life_secs,
383 gap_histograms,
384 dirty_marginal: std::collections::HashSet::new(),
385 dirty_cooccur: std::collections::HashSet::new(),
386 }
387 }
388}
389
390const OBSERVER_SCHEMA_VERSION: u16 = 1;
391
392/// Snapshot serializable. Se persiste a JSON en disco y se restaura al
393/// reboot para preservar contadores, co-ocurrencias e histogramas.
394#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
395pub struct ObserverSnapshot {
396 pub schema_version: u16,
397 /// `true` si sólo contiene los cambios desde el último snapshot.
398 /// `false` = full state, sobreescribe el observer al aplicar.
399 #[serde(default)]
400 pub is_delta: bool,
401 pub window_size: usize,
402 pub half_life_secs: Option<f64>,
403 pub total: u64,
404 /// Marginales serializados como Vec porque HashMap<EventKind, _> usa
405 /// EventKind como key — y EventKind tiene variantes con payloads que
406 /// no son JSON-key-serializable (BusInvokeOf, Custom).
407 pub marginal: Vec<(EventKind, u64)>,
408 pub cooccur: Vec<(EventKind, EventKind, u64)>,
409 pub gap_histograms: Vec<(EventKind, EventKind, GapHistogram)>,
410}
411
412#[cfg(test)]
413mod tests {
414 use super::*;
415 use arje_brain_rules::EventKind::*;
416
417 #[test]
418 fn entropy_zero_for_single_event() {
419 let mut obs = Observer::new(10);
420 for _ in 0..5 { obs.record(EnteSpawned); }
421 // Distribución degenerada: una sola observación posible → H = 0.
422 assert!(obs.shannon_entropy() < 1e-9);
423 }
424
425 #[test]
426 fn entropy_one_for_balanced_binary() {
427 let mut obs = Observer::new(100);
428 for _ in 0..10 { obs.record(EnteSpawned); }
429 for _ in 0..10 { obs.record(EnteDied); }
430 // Bernoulli(0.5) → H = 1 bit.
431 assert!((obs.shannon_entropy() - 1.0).abs() < 1e-9);
432 }
433
434 #[test]
435 fn conditional_prob_perfect_dependency() {
436 let mut obs = Observer::new(100);
437 // Spawned siempre seguido por Died.
438 for _ in 0..5 {
439 obs.record(EnteSpawned);
440 obs.record(EnteDied);
441 }
442 let p = obs.conditional_prob(&EnteSpawned, &EnteDied);
443 assert!(p > 0.0, "esperamos correlación positiva, got {p}");
444 }
445}