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