refactor(brain): A2 — split arje-brain en 3 sub-crates
DAG de dependencias limpio (modularidad horizontal):
- arje-brain-rules — rules + engine + dispatch (motor determinista)
- arje-brain-cognitive — observer + crystallize (estadística)
- arje-brain-audit — audit chain → CAS (accountability)
- arje-brain — umbrella de integración (introspect +
autopromote + metrics + loader)
Habilitador clave: TimedEvent movido de observer.rs a rules.rs
(engine lo necesitaba, era el único acoplo que rompía el DAG).
arje-brain re-exporta la API de los 3 sub-crates: arje-zero y chasqui
(consumidores) no requieren cambios. cargo check --workspace verde.
24 tests del brain pasan (4 rules + 6 cognitive + 5 audit + 9 umbrella).
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
sergio
@@ -4,8 +4,12 @@ version = "0.0.1"
|
||||
edition.workspace = true
|
||||
license.workspace = true
|
||||
publish.workspace = true
|
||||
description = "Capa de integración del brain: introspect socket + autopromote loop + metrics HTTP + loader. Wirea arje-brain-{rules,cognitive,audit}."
|
||||
|
||||
[dependencies]
|
||||
arje-brain-rules = { path = "../arje-brain-rules" }
|
||||
arje-brain-cognitive = { path = "../arje-brain-cognitive" }
|
||||
arje-brain-audit = { path = "../arje-brain-audit" }
|
||||
arje-card = { path = "../../protocol/arje-card" }
|
||||
arje-cas = { path = "../arje-cas" }
|
||||
serde = { workspace = true }
|
||||
|
||||
@@ -1,550 +0,0 @@
|
||||
//! Audit log: cada acción mutadora del cerebro deja una entry inmutable
|
||||
//! con su predecesor encadenado por SHA256 (estilo Merkle). Verificable a
|
||||
//! posteriori sin confianza en quien escribe.
|
||||
//!
|
||||
//! Los entries viven en memoria. Para persistencia, `flush_to_cas()` los
|
||||
//! escribe al content-addressable store y devuelve el SHA del head, que
|
||||
//! puede guardarse en un archivo de "head pointer" (fuera de scope aquí).
|
||||
|
||||
use crate::crystallize::Crystal;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use std::collections::VecDeque;
|
||||
use ulid::Ulid;
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct AuditEntry {
|
||||
/// Sequence number monotónico desde el inicio del log.
|
||||
pub seq: u64,
|
||||
/// Wall-clock al insertar.
|
||||
pub timestamp_ms: u64,
|
||||
/// SHA256 del entry anterior. None para el primer entry.
|
||||
pub prev_sha: Option<[u8; 32]>,
|
||||
/// SHA256 de este entry (auto-calculado al construir).
|
||||
pub sha: [u8; 32],
|
||||
/// Acción registrada.
|
||||
pub action: AuditAction,
|
||||
}
|
||||
|
||||
/// Sin `#[serde(tag)]`: bincode requiere external tagging (default serde
|
||||
/// para enums) para no usar `deserialize_any`. JSON sigue legible.
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub enum AuditAction {
|
||||
PromoteCrystal { rule_id: Ulid, crystal: Crystal },
|
||||
RemoveRule { rule_id: Ulid },
|
||||
LoadRulesFile { path: String, count: usize },
|
||||
}
|
||||
|
||||
pub struct AuditLog {
|
||||
entries: VecDeque<AuditEntry>,
|
||||
next_seq: u64,
|
||||
/// Cap del log en memoria. Entries más viejos se descartan tras flush.
|
||||
cap: usize,
|
||||
/// Total acumulado de entries flusheadas a CAS.
|
||||
flushed_count: u64,
|
||||
/// SHA del último entry persistido a CAS — el "head pointer" del log.
|
||||
last_flushed_sha: Option<[u8; 32]>,
|
||||
/// Path opcional donde escribir el head pointer tras cada flush.
|
||||
head_pointer_path: Option<std::path::PathBuf>,
|
||||
/// Subscribers a entries en tiempo real. Cada `append` empuja a todos.
|
||||
/// Subscribers cuyo receiver se dropeó se purgan en el siguiente push.
|
||||
subscribers: Vec<tokio::sync::mpsc::UnboundedSender<AuditEntry>>,
|
||||
/// Wall-clock del último flush exitoso a CAS. None si aún no se flush.
|
||||
last_flush_at_ms: Option<u64>,
|
||||
}
|
||||
|
||||
impl AuditLog {
|
||||
pub fn new() -> Self {
|
||||
Self::with_cap(512)
|
||||
}
|
||||
|
||||
pub fn with_cap(cap: usize) -> Self {
|
||||
Self {
|
||||
entries: VecDeque::new(),
|
||||
next_seq: 0,
|
||||
cap,
|
||||
flushed_count: 0,
|
||||
last_flushed_sha: None,
|
||||
head_pointer_path: None,
|
||||
subscribers: Vec::new(),
|
||||
last_flush_at_ms: None,
|
||||
}
|
||||
}
|
||||
|
||||
/// Registra un nuevo subscriber. El receiver recibe cada `AuditEntry`
|
||||
/// futuro hasta que el receiver se dropee (subscriber se purga al
|
||||
/// siguiente `append`).
|
||||
pub fn subscribe(&mut self) -> tokio::sync::mpsc::UnboundedReceiver<AuditEntry> {
|
||||
let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
|
||||
self.subscribers.push(tx);
|
||||
rx
|
||||
}
|
||||
|
||||
pub fn subscriber_count(&self) -> usize { self.subscribers.len() }
|
||||
|
||||
pub fn with_head_pointer(mut self, path: std::path::PathBuf) -> Self {
|
||||
self.head_pointer_path = Some(path);
|
||||
self
|
||||
}
|
||||
|
||||
/// Apendea una acción. Calcula el SHA encadenado contra el último entry.
|
||||
pub fn append(&mut self, action: AuditAction) -> AuditEntry {
|
||||
let prev_sha = self.entries.back().map(|e| e.sha);
|
||||
let timestamp_ms = now_ms();
|
||||
let seq = self.next_seq;
|
||||
self.next_seq += 1;
|
||||
|
||||
// Pre-construct con sha en cero, luego calcular sha sobre el
|
||||
// serializado canónico, luego sobreescribir el campo.
|
||||
let mut entry = AuditEntry {
|
||||
seq, timestamp_ms, prev_sha, sha: [0u8; 32], action,
|
||||
};
|
||||
entry.sha = compute_sha(&entry);
|
||||
|
||||
if self.entries.len() >= self.cap {
|
||||
self.entries.pop_front();
|
||||
}
|
||||
self.entries.push_back(entry.clone());
|
||||
// Empujar a subscribers, purgando los muertos in-place.
|
||||
self.subscribers.retain(|tx| tx.send(entry.clone()).is_ok());
|
||||
entry
|
||||
}
|
||||
|
||||
pub fn recent(&self, limit: usize) -> impl Iterator<Item = &AuditEntry> {
|
||||
let n = if limit == 0 { self.entries.len() } else { limit.min(self.entries.len()) };
|
||||
self.entries.iter().skip(self.entries.len() - n)
|
||||
}
|
||||
|
||||
pub fn len(&self) -> usize { self.entries.len() }
|
||||
pub fn is_empty(&self) -> bool { self.entries.is_empty() }
|
||||
|
||||
pub fn head_sha(&self) -> Option<[u8; 32]> {
|
||||
self.entries.back().map(|e| e.sha)
|
||||
}
|
||||
|
||||
/// Persiste el entry pasado al CAS y devuelve su SHA. Pensado para
|
||||
/// snapshots externos — el log en memoria sigue intacto.
|
||||
pub fn persist_to_cas(entry: &AuditEntry) -> anyhow::Result<[u8; 32]> {
|
||||
let bytes = serde_json::to_vec(entry)?;
|
||||
let sha = arje_cas::store(&bytes)?;
|
||||
Ok(sha)
|
||||
}
|
||||
|
||||
/// Persiste TODOS los entries actuales al CAS y actualiza el head pointer.
|
||||
/// Idempotente: re-flushar dos veces da los mismos SHAs (CAS dedup).
|
||||
/// Devuelve cuántas entries se flushearon en esta pasada.
|
||||
///
|
||||
/// Forma canónica: serializamos `entry` con `sha = [0; 32]` (formato
|
||||
/// pre-hash). El CAS computa sha256 sobre esos bytes y devuelve un SHA
|
||||
/// que por construcción coincide con `entry.sha` calculado al append.
|
||||
pub fn flush_to_cas(&mut self) -> anyhow::Result<usize> {
|
||||
let mut written = 0;
|
||||
let mut last_sha = self.last_flushed_sha;
|
||||
for entry in &self.entries {
|
||||
if entry.seq < self.flushed_count { continue; }
|
||||
let bytes = canonical_bytes(entry);
|
||||
let sha = arje_cas::store(&bytes)?;
|
||||
debug_assert_eq!(sha, entry.sha,
|
||||
"CAS sha != entry.sha — fórmula canónica rota");
|
||||
last_sha = Some(sha);
|
||||
written += 1;
|
||||
}
|
||||
self.flushed_count += written as u64;
|
||||
self.last_flushed_sha = last_sha;
|
||||
if written > 0 {
|
||||
self.last_flush_at_ms = Some(now_ms());
|
||||
}
|
||||
// Persistir head pointer si está configurado.
|
||||
if let (Some(path), Some(sha)) = (&self.head_pointer_path, last_sha) {
|
||||
let pointer = AuditHeadPointer {
|
||||
last_seq: self.next_seq.saturating_sub(1),
|
||||
last_sha: sha,
|
||||
flushed_count: self.flushed_count,
|
||||
timestamp_ms: now_ms(),
|
||||
};
|
||||
let json = serde_json::to_vec_pretty(&pointer)?;
|
||||
// Escritura atómica: tmp + rename
|
||||
let tmp = path.with_extension("tmp");
|
||||
if let Some(parent) = path.parent() { let _ = std::fs::create_dir_all(parent); }
|
||||
std::fs::write(&tmp, json)?;
|
||||
std::fs::rename(&tmp, path)?;
|
||||
}
|
||||
Ok(written)
|
||||
}
|
||||
|
||||
pub fn flushed_count(&self) -> u64 { self.flushed_count }
|
||||
pub fn last_flushed_sha(&self) -> Option<[u8; 32]> { self.last_flushed_sha }
|
||||
pub fn last_flush_at_ms(&self) -> Option<u64> { self.last_flush_at_ms }
|
||||
|
||||
/// Segundos transcurridos desde el último flush. None si nunca se flush.
|
||||
pub fn last_flush_age_secs(&self) -> Option<f64> {
|
||||
let then = self.last_flush_at_ms?;
|
||||
let now = now_ms();
|
||||
Some((now.saturating_sub(then)) as f64 / 1000.0)
|
||||
}
|
||||
}
|
||||
|
||||
/// Pointer al head del audit log — escrito atómicamente en disco tras cada
|
||||
/// flush. Permite verificar la integridad del log sin escanearlo entero:
|
||||
/// el cliente lee el head, recupera el blob desde CAS, valida `prev_sha`
|
||||
/// recursivamente hasta el genesis.
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct AuditHeadPointer {
|
||||
pub last_seq: u64,
|
||||
pub last_sha: [u8; 32],
|
||||
pub flushed_count: u64,
|
||||
pub timestamp_ms: u64,
|
||||
}
|
||||
|
||||
/// Reporte de un replay: número de actions aplicadas + reglas finales.
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct ReplayReport {
|
||||
pub applied: u64,
|
||||
pub final_rule_count: usize,
|
||||
pub error: Option<String>,
|
||||
}
|
||||
|
||||
/// Reporte de verificación de la cadena audit.
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct VerificationReport {
|
||||
/// Cuántas entries se recorrieron y verificaron exitosamente.
|
||||
pub verified: u64,
|
||||
/// Si hubo error, el seq donde se detectó.
|
||||
pub broken_at_seq: Option<u64>,
|
||||
/// Detalles del error si hubo.
|
||||
pub error: Option<String>,
|
||||
/// SHA del genesis (primer entry; prev_sha = None).
|
||||
pub genesis_sha: Option<[u8; 32]>,
|
||||
}
|
||||
|
||||
/// Recorre la cadena del audit log desde `start_sha` hacia atrás vía `prev_sha`
|
||||
/// hasta el genesis. Para cada entry valida:
|
||||
/// 1. CAS contiene un blob bajo ese SHA
|
||||
/// 2. sha256(blob) == SHA esperado (defensa contra tampering del CAS)
|
||||
/// 3. El blob deserializa a AuditEntry con sha=[0;32] (forma canónica)
|
||||
///
|
||||
/// Devuelve un VerificationReport con el conteo, posibles errores y
|
||||
/// el SHA del genesis (útil para clientes que quieren cachearlo).
|
||||
pub fn verify_chain_from_cas(start_sha: [u8; 32]) -> VerificationReport {
|
||||
let mut current = Some(start_sha);
|
||||
let mut verified = 0u64;
|
||||
let mut last_seen: Option<AuditEntry> = None;
|
||||
|
||||
while let Some(sha) = current {
|
||||
let path = arje_cas::cas_root().join(arje_cas::hex(&sha));
|
||||
let bytes = match std::fs::read(&path) {
|
||||
Ok(b) => b,
|
||||
Err(e) => return VerificationReport {
|
||||
verified,
|
||||
broken_at_seq: last_seen.as_ref().map(|e| e.seq),
|
||||
error: Some(format!("CAS read {}: {e}", path.display())),
|
||||
genesis_sha: None,
|
||||
},
|
||||
};
|
||||
// Verificación 1: el blob hashea a la SHA esperada (CAS contract).
|
||||
let actual = arje_cas::sha256_of(&bytes);
|
||||
if actual != sha {
|
||||
return VerificationReport {
|
||||
verified,
|
||||
broken_at_seq: last_seen.as_ref().map(|e| e.seq),
|
||||
error: Some(format!(
|
||||
"CAS tamper en {}: expected {} got {}",
|
||||
path.display(), arje_cas::hex(&sha), arje_cas::hex(&actual)
|
||||
)),
|
||||
genesis_sha: None,
|
||||
};
|
||||
}
|
||||
// Verificación 2: deserialize. El blob canónico tiene sha=[0;32].
|
||||
let mut entry: AuditEntry = match serde_json::from_slice(&bytes) {
|
||||
Ok(e) => e,
|
||||
Err(e) => return VerificationReport {
|
||||
verified,
|
||||
broken_at_seq: last_seen.as_ref().map(|e| e.seq),
|
||||
error: Some(format!("deserialize: {e}")),
|
||||
genesis_sha: None,
|
||||
},
|
||||
};
|
||||
// Re-poblar el sha en el entry para reportar coherentemente.
|
||||
entry.sha = sha;
|
||||
verified += 1;
|
||||
|
||||
let prev = entry.prev_sha;
|
||||
last_seen = Some(entry);
|
||||
current = prev;
|
||||
}
|
||||
|
||||
VerificationReport {
|
||||
verified,
|
||||
broken_at_seq: None,
|
||||
error: None,
|
||||
genesis_sha: last_seen.as_ref().map(|e| e.sha),
|
||||
}
|
||||
}
|
||||
|
||||
/// Devuelve el set de SHAs alcanzables desde `start_sha` siguiendo
|
||||
/// `prev_sha` hasta el genesis. Usado por el GC del CAS para construir
|
||||
/// las "raíces vivas" del audit log.
|
||||
pub fn reachable_from_head(start_sha: [u8; 32]) -> std::collections::HashSet<[u8; 32]> {
|
||||
let mut set = std::collections::HashSet::new();
|
||||
let mut current = Some(start_sha);
|
||||
while let Some(sha) = current {
|
||||
if !set.insert(sha) { break; } // ciclo (no debería pasar) — corta
|
||||
let path = arje_cas::cas_root().join(arje_cas::hex(&sha));
|
||||
let bytes = match std::fs::read(&path) { Ok(b) => b, Err(_) => break };
|
||||
let entry: AuditEntry = match serde_json::from_slice(&bytes) {
|
||||
Ok(e) => e, Err(_) => break,
|
||||
};
|
||||
current = entry.prev_sha;
|
||||
}
|
||||
set
|
||||
}
|
||||
|
||||
/// Recorre la cadena entera (head→genesis) y reconstruye la lista de
|
||||
/// actions en orden cronológico (oldest first). Útil tanto para replay
|
||||
/// como para auditoría retrospectiva.
|
||||
pub fn collect_chain_from_cas(start_sha: [u8; 32]) -> anyhow::Result<Vec<AuditEntry>> {
|
||||
let mut entries = Vec::new();
|
||||
let mut current = Some(start_sha);
|
||||
while let Some(sha) = current {
|
||||
let path = arje_cas::cas_root().join(arje_cas::hex(&sha));
|
||||
let bytes = std::fs::read(&path)?;
|
||||
let mut entry: AuditEntry = serde_json::from_slice(&bytes)?;
|
||||
entry.sha = sha;
|
||||
let prev = entry.prev_sha;
|
||||
entries.push(entry);
|
||||
current = prev;
|
||||
}
|
||||
// entries está en orden head→genesis. Reverse para chronological.
|
||||
entries.reverse();
|
||||
Ok(entries)
|
||||
}
|
||||
|
||||
/// Aplica las actions de la cadena en orden cronológico contra un engine
|
||||
/// fresco. PromoteCrystal → insert. RemoveRule → remove. LoadRulesFile →
|
||||
/// log informativo (los archivos pueden no existir en el ambiente actual).
|
||||
pub fn replay_chain(
|
||||
start_sha: [u8; 32],
|
||||
engine: &mut crate::engine::RuleEngine,
|
||||
) -> ReplayReport {
|
||||
let entries = match collect_chain_from_cas(start_sha) {
|
||||
Ok(es) => es,
|
||||
Err(e) => return ReplayReport {
|
||||
applied: 0, final_rule_count: engine.len(),
|
||||
error: Some(format!("collect chain: {e}")),
|
||||
},
|
||||
};
|
||||
let mut applied = 0u64;
|
||||
for entry in &entries {
|
||||
match &entry.action {
|
||||
AuditAction::PromoteCrystal { rule_id, crystal } => {
|
||||
let mut rule = crate::crystallize::crystal_to_rule(crystal);
|
||||
rule.id = *rule_id; // preservar identidad histórica
|
||||
engine.insert(rule);
|
||||
}
|
||||
AuditAction::RemoveRule { rule_id } => {
|
||||
engine.remove(*rule_id);
|
||||
}
|
||||
AuditAction::LoadRulesFile { path: _, count: _ } => {
|
||||
// Los archivos referenciados por path pueden haber cambiado
|
||||
// o no existir. Log y skip — el replay sólo reconstruye
|
||||
// promotes/removes que tienen estado en CAS.
|
||||
}
|
||||
}
|
||||
applied += 1;
|
||||
}
|
||||
ReplayReport {
|
||||
applied,
|
||||
final_rule_count: engine.len(),
|
||||
error: None,
|
||||
}
|
||||
}
|
||||
|
||||
impl Default for AuditLog {
|
||||
fn default() -> Self { Self::new() }
|
||||
}
|
||||
|
||||
fn now_ms() -> u64 {
|
||||
std::time::SystemTime::now()
|
||||
.duration_since(std::time::UNIX_EPOCH)
|
||||
.map(|d| d.as_millis() as u64)
|
||||
.unwrap_or(0)
|
||||
}
|
||||
|
||||
/// SHA256 sobre el entry en forma canónica (sha=[0;32]). Hash y CAS storage
|
||||
/// ven los mismos bytes, así que `arje_cas::store(canonical)` devuelve el
|
||||
/// mismo SHA que `compute_sha(entry)`.
|
||||
fn compute_sha(entry: &AuditEntry) -> [u8; 32] {
|
||||
let bytes = canonical_bytes(entry);
|
||||
arje_cas::sha256_of(&bytes)
|
||||
}
|
||||
|
||||
/// Forma canónica: el entry serializado JSON con `sha = [0; 32]`.
|
||||
/// JSON sin pretty-print es determinístico para nuestros tipos.
|
||||
fn canonical_bytes(entry: &AuditEntry) -> Vec<u8> {
|
||||
let canonical = AuditEntry {
|
||||
sha: [0u8; 32],
|
||||
..entry.clone()
|
||||
};
|
||||
serde_json::to_vec(&canonical).unwrap_or_default()
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn chain_links_consecutive_entries() {
|
||||
let mut log = AuditLog::new();
|
||||
let e1 = log.append(AuditAction::RemoveRule { rule_id: Ulid::new() });
|
||||
let e2 = log.append(AuditAction::RemoveRule { rule_id: Ulid::new() });
|
||||
assert!(e1.prev_sha.is_none());
|
||||
assert_eq!(e2.prev_sha, Some(e1.sha));
|
||||
assert_ne!(e1.sha, e2.sha);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn seq_monotonic() {
|
||||
let mut log = AuditLog::new();
|
||||
let e1 = log.append(AuditAction::RemoveRule { rule_id: Ulid::new() });
|
||||
let e2 = log.append(AuditAction::RemoveRule { rule_id: Ulid::new() });
|
||||
assert_eq!(e2.seq, e1.seq + 1);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn cap_evicts_oldest() {
|
||||
let mut log = AuditLog::with_cap(3);
|
||||
for _ in 0..5 {
|
||||
log.append(AuditAction::RemoveRule { rule_id: Ulid::new() });
|
||||
}
|
||||
assert_eq!(log.len(), 3);
|
||||
// El primer seq superviviente debe ser 2.
|
||||
assert_eq!(log.recent(0).next().unwrap().seq, 2);
|
||||
}
|
||||
|
||||
// ---------- Tests de integración con CAS real (en directorio temporal) ----------
|
||||
|
||||
use crate::engine::RuleEngine;
|
||||
use std::sync::Mutex;
|
||||
|
||||
/// Lock para serializar tests que mutan ENTE_CAS_ROOT (test threads
|
||||
/// comparten env vars). Sin esto, dos tests en paralelo pisan el path.
|
||||
static CAS_TEST_LOCK: Mutex<()> = Mutex::new(());
|
||||
|
||||
fn with_temp_cas<F: FnOnce()>(f: F) {
|
||||
let _guard = CAS_TEST_LOCK.lock().unwrap();
|
||||
let dir = std::env::temp_dir().join(format!("ente-cas-test-{}", Ulid::new()));
|
||||
std::env::set_var("ENTE_CAS_ROOT", &dir);
|
||||
let _cleanup = scopeguard(&dir);
|
||||
f();
|
||||
}
|
||||
|
||||
fn scopeguard(dir: &std::path::Path) -> impl Drop + '_ {
|
||||
struct G<'a>(&'a std::path::Path);
|
||||
impl<'a> Drop for G<'a> {
|
||||
fn drop(&mut self) {
|
||||
std::env::remove_var("ENTE_CAS_ROOT");
|
||||
let _ = std::fs::remove_dir_all(self.0);
|
||||
}
|
||||
}
|
||||
G(dir)
|
||||
}
|
||||
|
||||
fn dummy_crystal(ant: EventKind, con: EventKind) -> Crystal {
|
||||
Crystal {
|
||||
antecedent: ant,
|
||||
consequent: con,
|
||||
conditional_prob: 0.9,
|
||||
pmi: 1.5,
|
||||
support: 7,
|
||||
gap_stats: None,
|
||||
}
|
||||
}
|
||||
|
||||
use crate::rules::EventKind;
|
||||
|
||||
#[test]
|
||||
fn flush_round_trip_preserves_chain() {
|
||||
with_temp_cas(|| {
|
||||
let mut log = AuditLog::new();
|
||||
let id1 = Ulid::new();
|
||||
let id2 = Ulid::new();
|
||||
log.append(AuditAction::PromoteCrystal {
|
||||
rule_id: id1,
|
||||
crystal: dummy_crystal(EventKind::EnteSpawned, EventKind::EnteDied),
|
||||
});
|
||||
log.append(AuditAction::PromoteCrystal {
|
||||
rule_id: id2,
|
||||
crystal: dummy_crystal(EventKind::BusAnnounce, EventKind::BusInvoke),
|
||||
});
|
||||
log.append(AuditAction::RemoveRule { rule_id: id1 });
|
||||
|
||||
assert_eq!(log.flush_to_cas().unwrap(), 3);
|
||||
let head = log.last_flushed_sha().expect("head set");
|
||||
let report = verify_chain_from_cas(head);
|
||||
assert!(report.error.is_none(), "verification failed: {:?}", report.error);
|
||||
assert_eq!(report.verified, 3);
|
||||
});
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn replay_reconstructs_engine_state() {
|
||||
with_temp_cas(|| {
|
||||
let mut log = AuditLog::new();
|
||||
let id1: Ulid = "01KQR3000000000000000000A1".parse().unwrap();
|
||||
let id2: Ulid = "01KQR3000000000000000000A2".parse().unwrap();
|
||||
let id3: Ulid = "01KQR3000000000000000000A3".parse().unwrap();
|
||||
log.append(AuditAction::PromoteCrystal {
|
||||
rule_id: id1,
|
||||
crystal: dummy_crystal(EventKind::EnteSpawned, EventKind::EnteDied),
|
||||
});
|
||||
log.append(AuditAction::PromoteCrystal {
|
||||
rule_id: id2,
|
||||
crystal: dummy_crystal(EventKind::BusAnnounce, EventKind::BusInvoke),
|
||||
});
|
||||
log.append(AuditAction::PromoteCrystal {
|
||||
rule_id: id3,
|
||||
crystal: dummy_crystal(EventKind::DeviceAdded, EventKind::DeviceRemoved),
|
||||
});
|
||||
log.append(AuditAction::RemoveRule { rule_id: id2 });
|
||||
log.flush_to_cas().unwrap();
|
||||
let head = log.last_flushed_sha().unwrap();
|
||||
|
||||
let mut engine = RuleEngine::empty();
|
||||
let rep = replay_chain(head, &mut engine);
|
||||
assert!(rep.error.is_none(), "replay error: {:?}", rep.error);
|
||||
assert_eq!(rep.applied, 4);
|
||||
assert_eq!(engine.len(), 2, "id2 should be removed, id1 + id3 remain");
|
||||
// Ulids preservados
|
||||
let ids: Vec<Ulid> = engine.rules().map(|r| r.id).collect();
|
||||
assert!(ids.contains(&id1));
|
||||
assert!(!ids.contains(&id2));
|
||||
assert!(ids.contains(&id3));
|
||||
});
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn replay_after_eviction_still_works() {
|
||||
with_temp_cas(|| {
|
||||
// Cap pequeño: la mayoría de entries se evictan de memoria pero
|
||||
// siguen en CAS. Replay debe poder reconstruir desde CAS solo.
|
||||
let mut log = AuditLog::with_cap(2);
|
||||
let mut ids = Vec::new();
|
||||
for _ in 0..6 {
|
||||
let id = Ulid::new();
|
||||
ids.push(id);
|
||||
log.append(AuditAction::PromoteCrystal {
|
||||
rule_id: id,
|
||||
crystal: dummy_crystal(EventKind::EnteSpawned, EventKind::EnteDied),
|
||||
});
|
||||
log.flush_to_cas().unwrap();
|
||||
}
|
||||
assert_eq!(log.len(), 2, "cap eviction limita memoria");
|
||||
let head = log.last_flushed_sha().unwrap();
|
||||
|
||||
let mut engine = RuleEngine::empty();
|
||||
let rep = replay_chain(head, &mut engine);
|
||||
assert!(rep.error.is_none());
|
||||
assert_eq!(rep.applied, 6);
|
||||
assert_eq!(engine.len(), 6);
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -6,10 +6,10 @@
|
||||
//! no exista ya una regla con el mismo trigger_kind (heurística simple —
|
||||
//! evita ráfagas de duplicados de la misma estadística).
|
||||
|
||||
use crate::audit::AuditAction;
|
||||
use crate::crystallize::{crystal_to_rule, detect_crystals, Crystal, CrystallizationParams};
|
||||
use arje_brain_audit::audit::AuditAction;
|
||||
use arje_brain_cognitive::crystallize::{crystal_to_rule, detect_crystals, Crystal, CrystallizationParams};
|
||||
use crate::introspect::{append_rule_jsonl, BrainState};
|
||||
use crate::rules::EventKind;
|
||||
use arje_brain_rules::rules::EventKind;
|
||||
use std::collections::HashSet;
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
|
||||
@@ -1,244 +0,0 @@
|
||||
//! Cristalización: del flujo observado a reglas explícitas.
|
||||
//!
|
||||
//! Detecta pares (a, b) donde:
|
||||
//! - support(a, b) ≥ min_support (suficientes muestras para no ser ruido)
|
||||
//! - P(b|a) ≥ min_conditional_prob (a predice b con confianza)
|
||||
//! - PMI(a; b) ≥ min_pmi (más correlacionados que random)
|
||||
//!
|
||||
//! Cada cristal se materializa como `Rule` ejecutable (`crystal_to_rule`).
|
||||
//! Para persistencia/transporte, `crystal_to_json_pretty` serializa la Rule
|
||||
//! resultante con serde — sin formatos intermedios.
|
||||
|
||||
use crate::observer::{GapStats, Observer};
|
||||
use crate::rules::{Action, EventKind, EventPattern, LogLevel, Rule, Scope};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use std::time::Instant;
|
||||
use ulid::Ulid;
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct Crystal {
|
||||
pub antecedent: EventKind,
|
||||
pub consequent: EventKind,
|
||||
pub conditional_prob: f64,
|
||||
pub pmi: f64,
|
||||
pub support: u64,
|
||||
/// Estadísticas del gap temporal entre antecedent → consequent.
|
||||
/// None si no hay histograma. Habilita generación de reglas Sequence
|
||||
/// con `within_ms = (mean + 2σ) * 1000`.
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub gap_stats: Option<GapStats>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Copy)]
|
||||
pub struct CrystallizationParams {
|
||||
pub min_support: u64,
|
||||
pub min_conditional_prob: f64,
|
||||
pub min_pmi: f64,
|
||||
}
|
||||
|
||||
impl Default for CrystallizationParams {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
min_support: 5,
|
||||
min_conditional_prob: 0.7,
|
||||
min_pmi: 0.5,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub fn detect_crystals(obs: &Observer, params: &CrystallizationParams) -> Vec<Crystal> {
|
||||
let mut out = Vec::new();
|
||||
for ((a, b), &count) in obs.cooccurrences() {
|
||||
if count < params.min_support { continue; }
|
||||
let cp = obs.conditional_prob(a, b);
|
||||
if cp < params.min_conditional_prob { continue; }
|
||||
let mi = obs.pmi(a, b);
|
||||
if mi < params.min_pmi { continue; }
|
||||
// Stats del histograma si existen para este par.
|
||||
let gap_stats = obs.gap_histograms()
|
||||
.get(&(a.clone(), b.clone()))
|
||||
.map(|h| h.stats());
|
||||
out.push(Crystal {
|
||||
antecedent: a.clone(),
|
||||
consequent: b.clone(),
|
||||
conditional_prob: cp,
|
||||
pmi: mi,
|
||||
support: count,
|
||||
gap_stats,
|
||||
});
|
||||
}
|
||||
out.sort_by(|x, y| y.conditional_prob.partial_cmp(&x.conditional_prob).unwrap_or(std::cmp::Ordering::Equal));
|
||||
out
|
||||
}
|
||||
|
||||
/// Serializa la `Rule` derivada del cristal como JSON pretty-printed. Ese
|
||||
/// JSON es el formato canónico de persistencia: el loader lo lee como una
|
||||
/// línea de JSONL o como elemento de un array. Los stats del cristal (P, PMI,
|
||||
/// support) viven en el audit log vía `AuditAction::PromoteCrystal`, no se
|
||||
/// duplican aquí.
|
||||
pub fn crystal_to_json_pretty(c: &Crystal) -> String {
|
||||
serde_json::to_string_pretty(&crystal_to_rule(c))
|
||||
.expect("Rule serialize should never fail")
|
||||
}
|
||||
|
||||
/// Convierte un cristal a una `Rule` ejecutable. Si hay gap_stats con
|
||||
/// muestras suficientes (≥ 4), genera una regla `Sequence` con
|
||||
/// `within_ms = (mean + 2σ) * 1000`. 2σ cubre ~95% de la distribución
|
||||
/// asumiendo normalidad — captura el "tiempo típico de respuesta" del
|
||||
/// patrón observado. Si no hay stats, fallback a `Single { antecedent }`.
|
||||
pub fn crystal_to_rule(c: &Crystal) -> Rule {
|
||||
let when = match &c.gap_stats {
|
||||
Some(s) if s.count >= 4 => {
|
||||
// Mínimo 1ms para evitar within_ms=0 cuando varianza colapsa.
|
||||
let bound_secs = (s.mean_secs + 2.0 * s.stddev_secs).max(0.001);
|
||||
EventPattern::Sequence {
|
||||
kinds: vec![c.antecedent.clone(), c.consequent.clone()],
|
||||
within_ms: (bound_secs * 1000.0).ceil() as u64,
|
||||
}
|
||||
}
|
||||
_ => EventPattern::Single { kind: c.antecedent.clone() },
|
||||
};
|
||||
let message = match &c.gap_stats {
|
||||
Some(s) if s.count >= 4 => format!(
|
||||
"crystal seq: {:?} → {:?} (P={:.2}, PMI={:.2}, gap={:.3}±{:.3}s)",
|
||||
c.antecedent, c.consequent, c.conditional_prob, c.pmi,
|
||||
s.mean_secs, s.stddev_secs,
|
||||
),
|
||||
_ => format!(
|
||||
"crystal: {:?} → {:?} (P={:.2}, PMI={:.2}, n={})",
|
||||
c.antecedent, c.consequent, c.conditional_prob, c.pmi, c.support
|
||||
),
|
||||
};
|
||||
Rule {
|
||||
id: Ulid::new(),
|
||||
priority: 5,
|
||||
when,
|
||||
scope: Scope::default(),
|
||||
then: vec![Action::Log { level: LogLevel::Info, message }],
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
// ============================================================================
|
||||
// Patrones extendidos: Burst (alta frecuencia) y Silence (ausencia prolongada).
|
||||
// Estos cristales son sobre un único kind, no pares — capturan dinámicas
|
||||
// temporales de eventos individuales.
|
||||
// ============================================================================
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub enum PatternCrystal {
|
||||
/// Mismo evento aparece con frecuencia alta. `frequency_per_sec` se
|
||||
/// estima sobre el window de observación.
|
||||
Burst {
|
||||
kind: EventKind,
|
||||
count: u64,
|
||||
frequency_per_sec: f64,
|
||||
},
|
||||
/// Evento que dejó de aparecer. `since_secs` es el tiempo desde la
|
||||
/// última observación.
|
||||
Silence {
|
||||
kind: EventKind,
|
||||
last_count: u64,
|
||||
since_secs: f64,
|
||||
},
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Copy)]
|
||||
pub struct PatternParams {
|
||||
/// Mínimo de ocurrencias para considerar Burst.
|
||||
pub burst_min_count: u64,
|
||||
/// Frecuencia mínima (eventos por segundo) para considerar Burst.
|
||||
pub burst_min_freq_hz: f64,
|
||||
/// Tiempo desde última ocurrencia para considerar Silence.
|
||||
pub silence_min_secs: f64,
|
||||
/// Mínimo total previo para considerar Silence (eventos < N son ruido).
|
||||
pub silence_min_prior_count: u64,
|
||||
}
|
||||
|
||||
impl Default for PatternParams {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
burst_min_count: 10,
|
||||
burst_min_freq_hz: 5.0,
|
||||
silence_min_secs: 30.0,
|
||||
silence_min_prior_count: 3,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Detecta Bursts y Silences sobre la distribución marginal del observer.
|
||||
/// La frecuencia de un Burst se aproxima asumiendo que la observación cubre
|
||||
/// el rango entre `last_seen` y `Instant::now()` para ese kind.
|
||||
pub fn detect_pattern_crystals(obs: &Observer, params: &PatternParams) -> Vec<PatternCrystal> {
|
||||
let mut out = Vec::new();
|
||||
let now = Instant::now();
|
||||
for (kind, &count) in obs.marginals() {
|
||||
let last_seen = obs.last_seen_marginal(kind);
|
||||
// ---- Burst ----
|
||||
if count >= params.burst_min_count {
|
||||
// Aproximación: si vimos `count` eventos hasta `last_seen`, y el
|
||||
// primer evento sucedió en algún momento del window, la freq es
|
||||
// count / window_age. Sin tiempo del primer evento, usamos
|
||||
// last_seen → now como denominador (subestima freq) o asumimos
|
||||
// ventana fija de 60s. Usamos la última como aproximación.
|
||||
let elapsed = last_seen
|
||||
.map(|t| now.saturating_duration_since(t).as_secs_f64().max(0.001))
|
||||
.unwrap_or(60.0);
|
||||
// Estimación conservadora: count / max(window_age, 1s).
|
||||
// Si tenemos histograma, podríamos refinar — TODO.
|
||||
let freq = count as f64 / elapsed.max(1.0);
|
||||
if freq >= params.burst_min_freq_hz {
|
||||
out.push(PatternCrystal::Burst {
|
||||
kind: kind.clone(),
|
||||
count,
|
||||
frequency_per_sec: freq,
|
||||
});
|
||||
}
|
||||
}
|
||||
// ---- Silence ----
|
||||
if count >= params.silence_min_prior_count {
|
||||
if let Some(t) = last_seen {
|
||||
let since = now.saturating_duration_since(t).as_secs_f64();
|
||||
if since >= params.silence_min_secs {
|
||||
out.push(PatternCrystal::Silence {
|
||||
kind: kind.clone(),
|
||||
last_count: count,
|
||||
since_secs: since,
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
out
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::rules::EventKind::*;
|
||||
|
||||
#[test]
|
||||
fn detects_perfect_correlation() {
|
||||
let mut obs = Observer::new(100);
|
||||
for _ in 0..10 {
|
||||
obs.record(EnteSpawned);
|
||||
obs.record(EnteDied);
|
||||
}
|
||||
let crystals = detect_crystals(&obs, &CrystallizationParams {
|
||||
min_support: 3,
|
||||
min_conditional_prob: 0.5,
|
||||
min_pmi: 0.0,
|
||||
});
|
||||
assert!(crystals.iter().any(|c| matches!(c.antecedent, EnteSpawned)
|
||||
&& matches!(c.consequent, EnteDied)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rejects_below_threshold() {
|
||||
let mut obs = Observer::new(100);
|
||||
// Sin co-ocurrencia significativa.
|
||||
for _ in 0..3 { obs.record(EnteSpawned); }
|
||||
let crystals = detect_crystals(&obs, &CrystallizationParams::default());
|
||||
assert!(crystals.is_empty(), "no debería haber cristales: {:?}", crystals);
|
||||
}
|
||||
}
|
||||
@@ -1,73 +0,0 @@
|
||||
//! Despacho asíncrono de Actions. El motor entrega `Vec<Arc<Rule>>` matched;
|
||||
//! este módulo las traduce a efectos del fractal vía un `ActionSink` trait.
|
||||
//!
|
||||
//! Esto invierte la dependencia: ente-brain no conoce a ente-zero. El init
|
||||
//! implementa `ActionSink` y wirea spawn/invoke/log a sus propias estructuras.
|
||||
|
||||
use crate::rules::{Action, LogLevel, Rule};
|
||||
use std::sync::Arc;
|
||||
use tracing::{debug, error, info, trace, warn};
|
||||
|
||||
/// Backend de ejecución de Actions. ente-zero implementa esto delegando a
|
||||
/// graph_tx (Spawn → SpawnRequest, Invoke → bus call, etc.).
|
||||
pub trait ActionSink: Send + Sync {
|
||||
/// Spawn una Card decodificada. Implementación: GraphEvent::SpawnRequest.
|
||||
fn spawn(&self, card_blob: &str);
|
||||
/// Invoke por bus. blob crudo; el sink lo enruta vía bus_mediator.
|
||||
fn invoke(&self, target_cap: arje_card::Capability, blob: Vec<u8>);
|
||||
/// Notifica a un Ente específico (target_id). Implementación: forward por bus.
|
||||
fn notify(&self, target_id: ulid::Ulid, message: &str);
|
||||
/// Inhibe un comportamiento (placeholder; semántica depende del sink).
|
||||
fn inhibit(&self, reason: &str);
|
||||
}
|
||||
|
||||
/// Sink por defecto que sólo logea. Útil para tests y dev sin runtime.
|
||||
pub struct NullSink;
|
||||
|
||||
impl ActionSink for NullSink {
|
||||
fn spawn(&self, card_blob: &str) {
|
||||
info!(blob_len = card_blob.len(), "NullSink::spawn (no-op)");
|
||||
}
|
||||
fn invoke(&self, target_cap: arje_card::Capability, blob: Vec<u8>) {
|
||||
info!(?target_cap, blob_len = blob.len(), "NullSink::invoke (no-op)");
|
||||
}
|
||||
fn notify(&self, target_id: ulid::Ulid, message: &str) {
|
||||
info!(%target_id, %message, "NullSink::notify (no-op)");
|
||||
}
|
||||
fn inhibit(&self, reason: &str) {
|
||||
info!(%reason, "NullSink::inhibit (no-op)");
|
||||
}
|
||||
}
|
||||
|
||||
/// Ejecuta las reglas matched. Cada Rule puede tener N Actions; ejecutamos
|
||||
/// todas. Las acciones de Log se evalúan inline (tracing es async-safe).
|
||||
/// Las acciones de Spawn/Invoke/Notify se delegan al sink — el sink decide
|
||||
/// si procesarlas sincrónica o asincrónicamente.
|
||||
pub async fn dispatch_actions(rules: &[Arc<Rule>], sink: &dyn ActionSink) {
|
||||
for rule in rules {
|
||||
trace!(id = %rule.id, priority = rule.priority, n = rule.then.len(), "dispatching rule");
|
||||
for action in &rule.then {
|
||||
execute_action(action, sink, rule.id).await;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn execute_action(action: &Action, sink: &dyn ActionSink, rule_id: ulid::Ulid) {
|
||||
match action {
|
||||
Action::Log { level, message } => emit_log(level, message, rule_id),
|
||||
Action::Notify { target_id, message } => sink.notify(*target_id, message),
|
||||
Action::Spawn { card_blob } => sink.spawn(card_blob),
|
||||
Action::Invoke { target_cap, blob } => sink.invoke(target_cap.clone(), blob.clone()),
|
||||
Action::Inhibit { reason } => sink.inhibit(reason),
|
||||
}
|
||||
}
|
||||
|
||||
fn emit_log(level: &LogLevel, message: &str, rule_id: ulid::Ulid) {
|
||||
match level {
|
||||
LogLevel::Trace => trace!(rule = %rule_id, "{}", message),
|
||||
LogLevel::Debug => debug!(rule = %rule_id, "{}", message),
|
||||
LogLevel::Info => info! (rule = %rule_id, "{}", message),
|
||||
LogLevel::Warn => warn! (rule = %rule_id, "{}", message),
|
||||
LogLevel::Error => error!(rule = %rule_id, "{}", message),
|
||||
}
|
||||
}
|
||||
@@ -1,399 +0,0 @@
|
||||
//! Motor de inferencia. HashMap<EventKindDiscriminant, Vec<Arc<Rule>>> para
|
||||
//! lookup O(1) por tipo de evento, luego filter lineal por scope + filtros
|
||||
//! del payload (BusInvokeOf, Custom).
|
||||
//!
|
||||
//! Inmutabilidad fractal: `Arc<Rule>` es el unit de compartición. Clonar una
|
||||
//! regla del motor para entregarla al dispatcher es un refcount bump, no copia.
|
||||
|
||||
use crate::observer::TimedEvent;
|
||||
use crate::rules::{EventKind, EventPattern, Rule, Scope};
|
||||
use arje_card::Capability;
|
||||
use std::collections::HashMap;
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
use ulid::Ulid;
|
||||
|
||||
/// Discriminante barato de `EventKind` para indexar el HashMap. Sin payload —
|
||||
/// el match de payload se hace en una segunda pasada lineal en O(k) donde k
|
||||
/// es el número de reglas para ese tag.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
|
||||
pub enum EventKindDiscriminant {
|
||||
EnteSpawned,
|
||||
EnteDied,
|
||||
BusAnnounce,
|
||||
BusInvoke,
|
||||
BusInvokeOf,
|
||||
DeviceAdded,
|
||||
DeviceRemoved,
|
||||
Custom,
|
||||
}
|
||||
|
||||
impl From<&EventKind> for EventKindDiscriminant {
|
||||
fn from(k: &EventKind) -> Self {
|
||||
match k {
|
||||
EventKind::EnteSpawned => Self::EnteSpawned,
|
||||
EventKind::EnteDied => Self::EnteDied,
|
||||
EventKind::BusAnnounce => Self::BusAnnounce,
|
||||
EventKind::BusInvoke => Self::BusInvoke,
|
||||
EventKind::BusInvokeOf(_) => Self::BusInvokeOf,
|
||||
EventKind::DeviceAdded => Self::DeviceAdded,
|
||||
EventKind::DeviceRemoved => Self::DeviceRemoved,
|
||||
EventKind::Custom(_) => Self::Custom,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Snapshot del Ente que disparó el evento. Necesario para evaluar `Scope`.
|
||||
#[derive(Debug, Clone, Default)]
|
||||
pub struct SubjectInfo {
|
||||
pub id: Option<Ulid>,
|
||||
pub label: Option<String>,
|
||||
pub capabilities: Vec<Capability>,
|
||||
}
|
||||
|
||||
pub struct RuleEngine {
|
||||
rules: Vec<Arc<Rule>>,
|
||||
/// Reglas atómicas (Single, Sequence) indexadas por discriminante del
|
||||
/// kind que las dispara. Lookup O(1).
|
||||
by_kind: HashMap<EventKindDiscriminant, Vec<Arc<Rule>>>,
|
||||
/// Reglas compuestas (Either, All): se evalúan contra cada evento.
|
||||
/// Para fractales con N pequeño no afecta perf; con N grande, optimizar
|
||||
/// emitiendo a múltiples buckets en insert (fan-out).
|
||||
compound: Vec<Arc<Rule>>,
|
||||
}
|
||||
|
||||
impl Default for RuleEngine {
|
||||
fn default() -> Self { Self::empty() }
|
||||
}
|
||||
|
||||
impl RuleEngine {
|
||||
pub fn empty() -> Self {
|
||||
Self { rules: Vec::new(), by_kind: HashMap::new(), compound: Vec::new() }
|
||||
}
|
||||
|
||||
/// Carga reglas desde JSON (lista de Rule).
|
||||
pub fn load_json(json: &str) -> anyhow::Result<Self> {
|
||||
let rules: Vec<Rule> = serde_json::from_str(json)?;
|
||||
let mut engine = Self::empty();
|
||||
for r in rules {
|
||||
r.validate().map_err(|e| anyhow::anyhow!("regla inválida: {e}"))?;
|
||||
engine.insert(r);
|
||||
}
|
||||
Ok(engine)
|
||||
}
|
||||
|
||||
pub fn insert(&mut self, rule: Rule) {
|
||||
let arc = Arc::new(rule);
|
||||
// Atómicas → bucket por discriminante. Compuestas → bucket fallback.
|
||||
if let Some(trigger) = arc.when.trigger_kind() {
|
||||
let disc = EventKindDiscriminant::from(trigger);
|
||||
self.by_kind.entry(disc).or_default().push(arc.clone());
|
||||
} else {
|
||||
self.compound.push(arc.clone());
|
||||
}
|
||||
self.rules.push(arc);
|
||||
}
|
||||
|
||||
pub fn remove(&mut self, id: Ulid) -> bool {
|
||||
let before = self.rules.len();
|
||||
self.rules.retain(|r| r.id != id);
|
||||
for v in self.by_kind.values_mut() {
|
||||
v.retain(|r| r.id != id);
|
||||
}
|
||||
self.compound.retain(|r| r.id != id);
|
||||
before != self.rules.len()
|
||||
}
|
||||
|
||||
pub fn rules(&self) -> impl Iterator<Item = &Arc<Rule>> { self.rules.iter() }
|
||||
|
||||
pub fn len(&self) -> usize { self.rules.len() }
|
||||
pub fn is_empty(&self) -> bool { self.rules.is_empty() }
|
||||
|
||||
/// Despacho determinista. Devuelve reglas que matchean, ordenadas por
|
||||
/// prioridad descendente. Cada Arc<Rule> se clona (refcount) — sin copiar
|
||||
/// los datos.
|
||||
///
|
||||
/// `history` es el slice de eventos recientes (en orden cronológico,
|
||||
/// más reciente al final) usado para evaluar Sequence patterns.
|
||||
/// Para reglas Single, history se ignora.
|
||||
///
|
||||
/// Si el evento es `BusInvokeOf(_)`, también consultamos el bucket
|
||||
/// `BusInvoke` (regla genérica que ignora la cap).
|
||||
pub fn dispatch(
|
||||
&self,
|
||||
event: &EventKind,
|
||||
subject: &SubjectInfo,
|
||||
history: &[TimedEvent],
|
||||
) -> Vec<Arc<Rule>> {
|
||||
let primary = EventKindDiscriminant::from(event);
|
||||
let mut buckets: Vec<&Vec<Arc<Rule>>> = Vec::with_capacity(2);
|
||||
if let Some(v) = self.by_kind.get(&primary) {
|
||||
buckets.push(v);
|
||||
}
|
||||
if matches!(event, EventKind::BusInvokeOf(_)) {
|
||||
if let Some(v) = self.by_kind.get(&EventKindDiscriminant::BusInvoke) {
|
||||
buckets.push(v);
|
||||
}
|
||||
}
|
||||
let mut hits: Vec<Arc<Rule>> = buckets.into_iter()
|
||||
.flat_map(|v| v.iter())
|
||||
.filter(|r| matches_pattern(&r.when, event, history))
|
||||
.filter(|r| matches_scope(&r.scope, subject))
|
||||
.cloned()
|
||||
.collect();
|
||||
// Fallback: reglas compuestas (Either/All) se evalúan siempre.
|
||||
for r in &self.compound {
|
||||
if matches_pattern(&r.when, event, history) && matches_scope(&r.scope, subject) {
|
||||
hits.push(r.clone());
|
||||
}
|
||||
}
|
||||
hits.sort_by(|a, b| b.priority.cmp(&a.priority));
|
||||
hits
|
||||
}
|
||||
}
|
||||
|
||||
/// Match recursivo del pattern. Atomic patterns evalúan contra el evento
|
||||
/// actual + history. Compuestos (Either/All) recursan sobre sus children.
|
||||
fn matches_pattern(pattern: &EventPattern, event: &EventKind, history: &[TimedEvent]) -> bool {
|
||||
match pattern {
|
||||
EventPattern::Single { kind } => matches_event_payload(kind, event),
|
||||
EventPattern::Sequence { kinds, within_ms } => {
|
||||
if kinds.is_empty() { return false; }
|
||||
let last_kind = kinds.last().unwrap();
|
||||
if !matches_event_payload(last_kind, event) { return false; }
|
||||
if history.len() < kinds.len() { return false; }
|
||||
let tail = &history[history.len() - kinds.len()..];
|
||||
for (t, k) in tail.iter().zip(kinds) {
|
||||
if !matches_event_payload(k, &t.kind) { return false; }
|
||||
}
|
||||
if *within_ms > 0 {
|
||||
let span = tail.last().unwrap().at.duration_since(tail.first().unwrap().at);
|
||||
if span > Duration::from_millis(*within_ms) { return false; }
|
||||
}
|
||||
true
|
||||
}
|
||||
EventPattern::Either { patterns } => {
|
||||
patterns.iter().any(|p| matches_pattern(p, event, history))
|
||||
}
|
||||
EventPattern::All { patterns } => {
|
||||
patterns.iter().all(|p| matches_pattern(p, event, history))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn matches_event_payload(rule_kind: &EventKind, evt: &EventKind) -> bool {
|
||||
use EventKind::*;
|
||||
match (rule_kind, evt) {
|
||||
(EnteSpawned, EnteSpawned) => true,
|
||||
(EnteDied, EnteDied) => true,
|
||||
(BusAnnounce, BusAnnounce) => true,
|
||||
(BusInvoke, BusInvoke) | (BusInvoke, BusInvokeOf(_)) => true,
|
||||
(BusInvokeOf(want), BusInvokeOf(got)) => want == got,
|
||||
(DeviceAdded, DeviceAdded) => true,
|
||||
(DeviceRemoved, DeviceRemoved) => true,
|
||||
(Custom(want), Custom(got)) => want == got,
|
||||
_ => false,
|
||||
}
|
||||
}
|
||||
|
||||
fn matches_scope(scope: &Scope, subj: &SubjectInfo) -> bool {
|
||||
if scope.is_wildcard() { return true; }
|
||||
if let Some(id) = scope.subject_id {
|
||||
if subj.id != Some(id) { return false; }
|
||||
}
|
||||
if let Some(lbl) = &scope.subject_label {
|
||||
if subj.label.as_ref() != Some(lbl) { return false; }
|
||||
}
|
||||
if let Some(cap) = &scope.subject_has_cap {
|
||||
if !subj.capabilities.contains(cap) { return false; }
|
||||
}
|
||||
true
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::rules::{Action, EventPattern, LogLevel};
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
fn rule_single(id_str: &str, kind: EventKind, prio: u8) -> Rule {
|
||||
Rule {
|
||||
id: id_str.parse().unwrap(),
|
||||
priority: prio,
|
||||
when: EventPattern::Single { kind },
|
||||
then: vec![Action::Log {
|
||||
level: LogLevel::Info,
|
||||
message: id_str.into(),
|
||||
}],
|
||||
scope: Scope::default(),
|
||||
}
|
||||
}
|
||||
|
||||
fn empty_history() -> Vec<TimedEvent> { Vec::new() }
|
||||
|
||||
#[test]
|
||||
fn dispatch_picks_only_matching_kind() {
|
||||
let mut e = RuleEngine::empty();
|
||||
e.insert(rule_single("01KQQ100000000000000000001", EventKind::EnteSpawned, 5));
|
||||
e.insert(rule_single("01KQQ100000000000000000002", EventKind::EnteDied, 5));
|
||||
let hits = e.dispatch(&EventKind::EnteSpawned, &SubjectInfo::default(), &empty_history());
|
||||
assert_eq!(hits.len(), 1);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn priority_orders_descending() {
|
||||
let mut e = RuleEngine::empty();
|
||||
e.insert(rule_single("01KQQ100000000000000000003", EventKind::EnteSpawned, 1));
|
||||
e.insert(rule_single("01KQQ100000000000000000004", EventKind::EnteSpawned, 9));
|
||||
let hits = e.dispatch(&EventKind::EnteSpawned, &SubjectInfo::default(), &empty_history());
|
||||
assert_eq!(hits[0].priority, 9);
|
||||
assert_eq!(hits[1].priority, 1);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn scope_filters_by_label() {
|
||||
let mut e = RuleEngine::empty();
|
||||
let mut r = rule_single("01KQQ100000000000000000005", EventKind::EnteSpawned, 5);
|
||||
r.scope = Scope { subject_label: Some("foo".into()), ..Default::default() };
|
||||
e.insert(r);
|
||||
let foo = SubjectInfo { label: Some("foo".into()), ..Default::default() };
|
||||
let bar = SubjectInfo { label: Some("bar".into()), ..Default::default() };
|
||||
assert_eq!(e.dispatch(&EventKind::EnteSpawned, &foo, &empty_history()).len(), 1);
|
||||
assert_eq!(e.dispatch(&EventKind::EnteSpawned, &bar, &empty_history()).len(), 0);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn bus_invoke_generic_matches_specific() {
|
||||
let mut e = RuleEngine::empty();
|
||||
e.insert(rule_single("01KQQ100000000000000000006", EventKind::BusInvoke, 5));
|
||||
let hits = e.dispatch(
|
||||
&EventKind::BusInvokeOf(Capability::LegacyLogind),
|
||||
&SubjectInfo::default(),
|
||||
&empty_history(),
|
||||
);
|
||||
assert_eq!(hits.len(), 1);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn sequence_pattern_matches_with_history() {
|
||||
let mut e = RuleEngine::empty();
|
||||
let r = Rule {
|
||||
id: "01KQQ100000000000000000007".parse().unwrap(),
|
||||
priority: 5,
|
||||
when: EventPattern::Sequence {
|
||||
kinds: vec![EventKind::EnteSpawned, EventKind::BusAnnounce],
|
||||
within_ms: 1000,
|
||||
},
|
||||
then: vec![Action::Log { level: LogLevel::Info, message: "seq".into() }],
|
||||
scope: Scope::default(),
|
||||
};
|
||||
e.insert(r);
|
||||
|
||||
let now = Instant::now();
|
||||
let history = vec![
|
||||
TimedEvent { kind: EventKind::EnteSpawned, at: now },
|
||||
TimedEvent { kind: EventKind::BusAnnounce, at: now + Duration::from_millis(50) },
|
||||
];
|
||||
let hits = e.dispatch(&EventKind::BusAnnounce, &SubjectInfo::default(), &history);
|
||||
assert_eq!(hits.len(), 1, "esperaba match secuencia, got {}", hits.len());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn sequence_rejects_outside_time_window() {
|
||||
let mut e = RuleEngine::empty();
|
||||
let r = Rule {
|
||||
id: "01KQQ100000000000000000008".parse().unwrap(),
|
||||
priority: 5,
|
||||
when: EventPattern::Sequence {
|
||||
kinds: vec![EventKind::EnteSpawned, EventKind::BusAnnounce],
|
||||
within_ms: 100,
|
||||
},
|
||||
then: vec![Action::Log { level: LogLevel::Info, message: "seq".into() }],
|
||||
scope: Scope::default(),
|
||||
};
|
||||
e.insert(r);
|
||||
let now = Instant::now();
|
||||
let history = vec![
|
||||
TimedEvent { kind: EventKind::EnteSpawned, at: now },
|
||||
TimedEvent { kind: EventKind::BusAnnounce, at: now + Duration::from_millis(500) },
|
||||
];
|
||||
let hits = e.dispatch(&EventKind::BusAnnounce, &SubjectInfo::default(), &history);
|
||||
assert!(hits.is_empty(), "no debería matchear fuera de la ventana");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn either_matches_any_branch() {
|
||||
let mut e = RuleEngine::empty();
|
||||
let r = Rule {
|
||||
id: "01KQQ100000000000000000010".parse().unwrap(),
|
||||
priority: 5,
|
||||
when: EventPattern::Either { patterns: vec![
|
||||
EventPattern::Single { kind: EventKind::EnteSpawned },
|
||||
EventPattern::Single { kind: EventKind::EnteDied },
|
||||
]},
|
||||
then: vec![Action::Log { level: LogLevel::Info, message: "either".into() }],
|
||||
scope: Scope::default(),
|
||||
};
|
||||
e.insert(r);
|
||||
assert_eq!(e.dispatch(&EventKind::EnteSpawned, &SubjectInfo::default(), &[]).len(), 1);
|
||||
assert_eq!(e.dispatch(&EventKind::EnteDied, &SubjectInfo::default(), &[]).len(), 1);
|
||||
assert_eq!(e.dispatch(&EventKind::BusAnnounce, &SubjectInfo::default(), &[]).len(), 0);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn all_requires_every_branch() {
|
||||
let mut e = RuleEngine::empty();
|
||||
// All: matchear sólo si el evento actual es BusAnnounce Y la
|
||||
// secuencia EnteSpawned→BusAnnounce ocurrió en history.
|
||||
let r = Rule {
|
||||
id: "01KQQ100000000000000000011".parse().unwrap(),
|
||||
priority: 5,
|
||||
when: EventPattern::All { patterns: vec![
|
||||
EventPattern::Single { kind: EventKind::BusAnnounce },
|
||||
EventPattern::Sequence {
|
||||
kinds: vec![EventKind::EnteSpawned, EventKind::BusAnnounce],
|
||||
within_ms: 0,
|
||||
},
|
||||
]},
|
||||
then: vec![Action::Log { level: LogLevel::Info, message: "all".into() }],
|
||||
scope: Scope::default(),
|
||||
};
|
||||
e.insert(r);
|
||||
|
||||
let now = Instant::now();
|
||||
let history = vec![
|
||||
TimedEvent { kind: EventKind::EnteSpawned, at: now },
|
||||
TimedEvent { kind: EventKind::BusAnnounce, at: now + Duration::from_millis(10) },
|
||||
];
|
||||
// Single y Sequence ambos matchean → All matches.
|
||||
assert_eq!(e.dispatch(&EventKind::BusAnnounce, &SubjectInfo::default(), &history).len(), 1);
|
||||
// Sólo Single matchea (history vacío) → All no matches.
|
||||
assert!(e.dispatch(&EventKind::BusAnnounce, &SubjectInfo::default(), &[]).is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn sequence_requires_correct_order() {
|
||||
let mut e = RuleEngine::empty();
|
||||
let r = Rule {
|
||||
id: "01KQQ100000000000000000009".parse().unwrap(),
|
||||
priority: 5,
|
||||
when: EventPattern::Sequence {
|
||||
kinds: vec![EventKind::EnteSpawned, EventKind::BusAnnounce],
|
||||
within_ms: 0,
|
||||
},
|
||||
then: vec![Action::Log { level: LogLevel::Info, message: "seq".into() }],
|
||||
scope: Scope::default(),
|
||||
};
|
||||
e.insert(r);
|
||||
let now = Instant::now();
|
||||
// Orden invertido en el history.
|
||||
let history = vec![
|
||||
TimedEvent { kind: EventKind::BusAnnounce, at: now },
|
||||
TimedEvent { kind: EventKind::EnteSpawned, at: now + Duration::from_millis(10) },
|
||||
];
|
||||
// El evento actual es EnteSpawned, pero el último de la secuencia
|
||||
// requerida es BusAnnounce — no debería matchear.
|
||||
let hits = e.dispatch(&EventKind::EnteSpawned, &SubjectInfo::default(), &history);
|
||||
assert!(hits.is_empty());
|
||||
}
|
||||
}
|
||||
@@ -4,10 +4,10 @@
|
||||
//! cerebro sin tocar el bus interno del fractal. Esto separa observación de
|
||||
//! ejecución — la introspección es read-only por diseño.
|
||||
|
||||
use crate::crystallize::{detect_crystals, Crystal, CrystallizationParams};
|
||||
use crate::engine::RuleEngine;
|
||||
use crate::observer::Observer;
|
||||
use crate::rules::Rule;
|
||||
use arje_brain_cognitive::crystallize::{detect_crystals, Crystal, CrystallizationParams};
|
||||
use arje_brain_rules::engine::RuleEngine;
|
||||
use arje_brain_cognitive::observer::Observer;
|
||||
use arje_brain_rules::rules::Rule;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use std::io::Write;
|
||||
use std::path::{Path, PathBuf};
|
||||
@@ -32,7 +32,7 @@ pub struct BrainState {
|
||||
/// cada PromoteCrystal añade una línea (append-only) con la Rule serializada.
|
||||
pub rules_out: Option<Arc<PathBuf>>,
|
||||
/// Audit log en memoria. Cada promote/remove deja huella aquí.
|
||||
pub audit: Arc<RwLock<crate::audit::AuditLog>>,
|
||||
pub audit: Arc<RwLock<arje_brain_audit::audit::AuditLog>>,
|
||||
}
|
||||
|
||||
impl BrainState {
|
||||
@@ -46,7 +46,7 @@ impl BrainState {
|
||||
observer: Arc::new(RwLock::new(Observer::new(window_size))),
|
||||
params,
|
||||
rules_out: None,
|
||||
audit: Arc::new(RwLock::new(crate::audit::AuditLog::new())),
|
||||
audit: Arc::new(RwLock::new(arje_brain_audit::audit::AuditLog::new())),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -135,21 +135,21 @@ pub enum IntrospectResponse {
|
||||
/// Resultado de RemoveRule: true si existía, false si ya no.
|
||||
Removed(bool),
|
||||
/// Entradas del audit log (más recientes al final).
|
||||
AuditEntries(Vec<crate::audit::AuditEntry>),
|
||||
AuditEntries(Vec<arje_brain_audit::audit::AuditEntry>),
|
||||
/// Resultado de FlushAudit: cuántas entries se escribieron y SHA del head.
|
||||
Flushed { written: usize, head_sha: Option<[u8; 32]>, total_flushed: u64 },
|
||||
/// Resultado de ReloadRules: número total de reglas tras el reload.
|
||||
Reloaded { count: usize },
|
||||
/// Resultado de VerifyAudit.
|
||||
AuditVerified(crate::audit::VerificationReport),
|
||||
AuditVerified(arje_brain_audit::audit::VerificationReport),
|
||||
/// Resultado de ReplayAudit.
|
||||
Replayed(crate::audit::ReplayReport),
|
||||
Replayed(arje_brain_audit::audit::ReplayReport),
|
||||
/// Frame de streaming. El cliente lee estos en bucle hasta EOF.
|
||||
AuditStreamFrame(crate::audit::AuditEntry),
|
||||
AuditStreamFrame(arje_brain_audit::audit::AuditEntry),
|
||||
/// Resultado de GcCas: cuántos blobs eliminados y bytes liberados.
|
||||
GcResult { deleted: usize, freed_bytes: u64 },
|
||||
/// Cristales de Burst/Silence detectados.
|
||||
Patterns(Vec<crate::crystallize::PatternCrystal>),
|
||||
Patterns(Vec<arje_brain_cognitive::crystallize::PatternCrystal>),
|
||||
Error(String),
|
||||
}
|
||||
|
||||
@@ -306,7 +306,7 @@ impl IntrospectServer {
|
||||
let obs = self.state.observer.read().await;
|
||||
let crystals = detect_crystals(&obs, &self.state.params);
|
||||
match crystals.get(index) {
|
||||
Some(c) => IntrospectResponse::Json(crate::crystallize::crystal_to_json_pretty(c)),
|
||||
Some(c) => IntrospectResponse::Json(arje_brain_cognitive::crystallize::crystal_to_json_pretty(c)),
|
||||
None => IntrospectResponse::Error(format!("no crystal at index {index}")),
|
||||
}
|
||||
}
|
||||
@@ -317,7 +317,7 @@ impl IntrospectServer {
|
||||
};
|
||||
match crystals.get(index) {
|
||||
Some(c) => {
|
||||
let rule = crate::crystallize::crystal_to_rule(c);
|
||||
let rule = arje_brain_cognitive::crystallize::crystal_to_rule(c);
|
||||
let rule_id = rule.id;
|
||||
let rule_json = serde_json::to_string_pretty(&rule)
|
||||
.unwrap_or_else(|_| "<serialize failed>".into());
|
||||
@@ -332,7 +332,7 @@ impl IntrospectServer {
|
||||
}
|
||||
// Audit entry
|
||||
self.state.audit.write().await.append(
|
||||
crate::audit::AuditAction::PromoteCrystal {
|
||||
arje_brain_audit::audit::AuditAction::PromoteCrystal {
|
||||
rule_id, crystal: c.clone(),
|
||||
}
|
||||
);
|
||||
@@ -345,7 +345,7 @@ impl IntrospectServer {
|
||||
let removed = self.state.engine.write().await.remove(id);
|
||||
if removed {
|
||||
self.state.audit.write().await.append(
|
||||
crate::audit::AuditAction::RemoveRule { rule_id: id }
|
||||
arje_brain_audit::audit::AuditAction::RemoveRule { rule_id: id }
|
||||
);
|
||||
}
|
||||
IntrospectResponse::Removed(removed)
|
||||
@@ -373,7 +373,7 @@ impl IntrospectServer {
|
||||
"audit log sin entries flushadas — nada que verificar".into()
|
||||
),
|
||||
};
|
||||
let report = crate::audit::verify_chain_from_cas(head);
|
||||
let report = arje_brain_audit::audit::verify_chain_from_cas(head);
|
||||
IntrospectResponse::AuditVerified(report)
|
||||
}
|
||||
IntrospectRequest::StreamAudit => {
|
||||
@@ -385,15 +385,15 @@ impl IntrospectServer {
|
||||
}
|
||||
IntrospectRequest::PatternCrystals => {
|
||||
let obs = self.state.observer.read().await;
|
||||
let params = crate::crystallize::PatternParams::default();
|
||||
let patterns = crate::crystallize::detect_pattern_crystals(&obs, ¶ms);
|
||||
let params = arje_brain_cognitive::crystallize::PatternParams::default();
|
||||
let patterns = arje_brain_cognitive::crystallize::detect_pattern_crystals(&obs, ¶ms);
|
||||
IntrospectResponse::Patterns(patterns)
|
||||
}
|
||||
IntrospectRequest::GcCas { extra_roots } => {
|
||||
// Reachable = audit chain desde head + extra_roots provistos.
|
||||
let mut reachable = std::collections::HashSet::new();
|
||||
if let Some(head) = self.state.audit.read().await.last_flushed_sha() {
|
||||
reachable.extend(crate::audit::reachable_from_head(head));
|
||||
reachable.extend(arje_brain_audit::audit::reachable_from_head(head));
|
||||
}
|
||||
reachable.extend(extra_roots);
|
||||
match arje_cas::gc(&reachable) {
|
||||
@@ -410,8 +410,8 @@ impl IntrospectServer {
|
||||
),
|
||||
};
|
||||
let mut engine = self.state.engine.write().await;
|
||||
*engine = crate::engine::RuleEngine::empty();
|
||||
let report = crate::audit::replay_chain(head, &mut engine);
|
||||
*engine = arje_brain_rules::engine::RuleEngine::empty();
|
||||
let report = arje_brain_audit::audit::replay_chain(head, &mut engine);
|
||||
IntrospectResponse::Replayed(report)
|
||||
}
|
||||
IntrospectRequest::ReloadRules { path } => {
|
||||
@@ -430,12 +430,12 @@ impl IntrospectServer {
|
||||
};
|
||||
// Vaciamos el engine antes de re-cargar — semántica clean-slate.
|
||||
let mut engine = self.state.engine.write().await;
|
||||
*engine = crate::engine::RuleEngine::empty();
|
||||
*engine = arje_brain_rules::engine::RuleEngine::empty();
|
||||
let count = rules.len();
|
||||
for r in rules { engine.insert(r); }
|
||||
drop(engine);
|
||||
self.state.audit.write().await.append(
|
||||
crate::audit::AuditAction::LoadRulesFile {
|
||||
arje_brain_audit::audit::AuditAction::LoadRulesFile {
|
||||
path: path.to_string_lossy().into_owned(),
|
||||
count,
|
||||
}
|
||||
|
||||
@@ -1,38 +1,34 @@
|
||||
//! ente-brain: motor de reglas determinista + observador estadístico.
|
||||
//! arje-brain — capa de integración del brain.
|
||||
//!
|
||||
//! Tres capas:
|
||||
//! 1. `rules` — tipos de regla (Triplet: Subject + Event + Action)
|
||||
//! 2. `engine` — RuleEngine con HashMap<EventKindDiscriminant, Vec<Arc<Rule>>>
|
||||
//! para dispatch O(1)
|
||||
//! 3. `dispatch` — ejecutor async de Actions (vía tokio)
|
||||
//! 4. `observer` — sliding window + marginales + co-ocurrencias
|
||||
//! + Shannon entropy + información mutua
|
||||
//! 5. `crystallize` — detección de patrones estadísticamente significativos
|
||||
//! y materialización en `Rule` ejecutables
|
||||
//! 6. `introspect` — Unix socket bincode API para tools externos
|
||||
//! El brain se divide en tres sub-crates con un DAG de dependencias limpio:
|
||||
//! - `arje-brain-rules` — motor determinista (rules + engine + dispatch)
|
||||
//! - `arje-brain-cognitive` — estadística (observer + crystallize)
|
||||
//! - `arje-brain-audit` — accountability (audit chain → CAS)
|
||||
//!
|
||||
//! Diseño de inmutabilidad:
|
||||
//! - Rules son `Arc<Rule>` — clonar es zero-copy (refcount bump).
|
||||
//! - El motor expone sólo lecturas; mutaciones pasan por `insert/remove`.
|
||||
//! - Observer mantiene contadores incrementales — sin recomputación.
|
||||
//! Este crate es la capa que los wirea: `introspect` (socket API),
|
||||
//! `autopromote` (loop de promoción de cristales), `metrics` (HTTP) y
|
||||
//! `loader` (carga de cards/rules). Re-exporta la API de los tres
|
||||
//! sub-crates para compatibilidad de los consumidores históricos.
|
||||
|
||||
pub mod audit;
|
||||
pub mod autopromote;
|
||||
pub mod crystallize;
|
||||
pub mod dispatch;
|
||||
pub mod engine;
|
||||
pub mod introspect;
|
||||
pub mod loader;
|
||||
pub mod autopromote;
|
||||
pub mod metrics;
|
||||
pub mod observer;
|
||||
pub mod rules;
|
||||
pub mod loader;
|
||||
|
||||
// --- Re-export de los módulos de las 3 sub-crates ---
|
||||
pub use arje_brain_rules::{dispatch, engine, rules};
|
||||
pub use arje_brain_cognitive::{crystallize, observer};
|
||||
pub use arje_brain_audit::audit;
|
||||
|
||||
// --- Re-exports planos (API histórica que consumen arje-zero, chasqui) ---
|
||||
pub use rules::{Action, EventKind, EventPattern, LogLevel, Rule, Scope, TimedEvent};
|
||||
pub use engine::{EventKindDiscriminant, RuleEngine, SubjectInfo};
|
||||
pub use dispatch::{dispatch_actions, ActionSink, NullSink};
|
||||
pub use crystallize::{detect_crystals, Crystal, CrystallizationParams};
|
||||
pub use observer::Observer;
|
||||
pub use audit::AuditLog;
|
||||
|
||||
pub use autopromote::{spawn_autopromote_loop, AutopromoteParams};
|
||||
pub use crystallize::{detect_crystals, Crystal, CrystallizationParams};
|
||||
pub use dispatch::{dispatch_actions, ActionSink, NullSink};
|
||||
pub use engine::{EventKindDiscriminant, RuleEngine, SubjectInfo};
|
||||
pub use introspect::{IntrospectRequest, IntrospectResponse, IntrospectServer, BrainState};
|
||||
pub use introspect::{BrainState, IntrospectRequest, IntrospectResponse, IntrospectServer};
|
||||
pub use loader::{load_card_file, load_rules_file};
|
||||
pub use metrics::serve_metrics;
|
||||
pub use observer::{Observer, TimedEvent};
|
||||
pub use rules::{Action, EventKind, EventPattern, LogLevel, Rule, Scope};
|
||||
|
||||
@@ -8,7 +8,7 @@
|
||||
//! Ergonomía de autoría futura (RON, Dhall, etc.) se añade como ramas
|
||||
//! adicionales aquí cuando duela escribir JSON a mano. Hoy: una sola rama.
|
||||
|
||||
use crate::rules::Rule;
|
||||
use arje_brain_rules::rules::Rule;
|
||||
use arje_card::EntityCard;
|
||||
use std::path::Path;
|
||||
use tracing::info;
|
||||
@@ -102,7 +102,7 @@ pub fn extract_rules_from_json(raw: &str) -> anyhow::Result<Vec<Rule>> {
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::introspect::append_rule_jsonl;
|
||||
use crate::rules::{Action, EventKind, EventPattern, LogLevel, Rule, Scope};
|
||||
use arje_brain_rules::rules::{Action, EventKind, EventPattern, LogLevel, Rule, Scope};
|
||||
use ulid::Ulid;
|
||||
|
||||
fn sample_rule() -> Rule {
|
||||
|
||||
@@ -5,7 +5,7 @@
|
||||
//! `prometheus` con su Registry + encoders.
|
||||
|
||||
use crate::introspect::BrainState;
|
||||
use crate::rules::EventKind;
|
||||
use arje_brain_rules::rules::EventKind;
|
||||
use std::net::SocketAddr;
|
||||
use tokio::io::{AsyncReadExt, AsyncWriteExt};
|
||||
use tokio::net::{TcpListener, TcpStream};
|
||||
@@ -98,7 +98,7 @@ async fn format_metrics(state: &BrainState) -> String {
|
||||
}
|
||||
|
||||
// ---- Cristales detectados (con params actuales) ----
|
||||
let crystals = crate::detect_crystals(&obs, &state.params);
|
||||
let crystals = arje_brain_cognitive::detect_crystals(&obs, &state.params);
|
||||
out.push_str("# HELP ente_brain_crystals_total Number of crystals detected with current params.\n");
|
||||
out.push_str("# TYPE ente_brain_crystals_total gauge\n");
|
||||
out.push_str(&format!("ente_brain_crystals_total {}\n", crystals.len()));
|
||||
@@ -135,7 +135,7 @@ async fn format_metrics(state: &BrainState) -> String {
|
||||
// ---- Histogramas de gaps temporales (top-32 pares más frecuentes) ----
|
||||
out.push_str("# HELP ente_brain_pair_gap_seconds Time gap between correlated events.\n");
|
||||
out.push_str("# TYPE ente_brain_pair_gap_seconds histogram\n");
|
||||
let limits = crate::observer::GapHistogram::bucket_limits();
|
||||
let limits = arje_brain_cognitive::observer::GapHistogram::bucket_limits();
|
||||
for ((a, b), hist) in obs.top_gap_pairs(32) {
|
||||
let labels = format!(r#"a="{}",b="{}""#, kind_label(a), kind_label(b));
|
||||
for (i, &limit) in limits.iter().enumerate() {
|
||||
|
||||
@@ -1,453 +0,0 @@
|
||||
//! Observador estadístico. Mantiene marginales y co-ocurrencias dentro de una
|
||||
//! ventana deslizante. Calcula entropía de Shannon e información mutua para
|
||||
//! identificar correlaciones significativas.
|
||||
//!
|
||||
//! Diseño:
|
||||
//! - Counters incrementales: cada `record()` es O(window_size) en el peor
|
||||
//! caso (actualiza co-ocurrencias con cada evento del window).
|
||||
//! - Sin recomputaciones globales: marginales y joint counts son state.
|
||||
//! - El cálculo de H(X), P(B|A), I(A;B) es O(|distinct events|).
|
||||
|
||||
use crate::rules::EventKind;
|
||||
use std::collections::{HashMap, VecDeque};
|
||||
use std::time::Instant;
|
||||
|
||||
/// Evento timestamped. El timestamp se conserva para futuras políticas de
|
||||
/// expiración por tiempo (no sólo por count).
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct TimedEvent {
|
||||
pub kind: EventKind,
|
||||
pub at: Instant,
|
||||
}
|
||||
|
||||
/// Histograma de gaps temporales con buckets exponenciales en segundos.
|
||||
/// Cubre 6 órdenes de magnitud: 1ms hasta 1000s.
|
||||
#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
|
||||
pub struct GapHistogram {
|
||||
/// Buckets cumulativos (Prometheus-style): cada índice cuenta eventos
|
||||
/// con gap ≤ ese límite. Limites: 1ms, 10ms, 100ms, 1s, 10s, 100s, 1000s.
|
||||
pub buckets: [u64; 7],
|
||||
pub count: u64,
|
||||
pub sum_secs: f64,
|
||||
/// Suma de cuadrados — permite calcular varianza/stddev en O(1).
|
||||
pub sum_squares_secs: f64,
|
||||
pub max_secs: f64,
|
||||
}
|
||||
|
||||
/// Estadísticas resumidas de un GapHistogram, usables en cristales temporales.
|
||||
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
|
||||
pub struct GapStats {
|
||||
pub count: u64,
|
||||
pub mean_secs: f64,
|
||||
pub stddev_secs: f64,
|
||||
pub max_secs: f64,
|
||||
}
|
||||
|
||||
const GAP_BUCKET_LIMITS_SECS: [f64; 7] = [
|
||||
0.001, 0.01, 0.1, 1.0, 10.0, 100.0, 1000.0,
|
||||
];
|
||||
|
||||
impl GapHistogram {
|
||||
pub fn observe(&mut self, gap_secs: f64) {
|
||||
for (i, &limit) in GAP_BUCKET_LIMITS_SECS.iter().enumerate() {
|
||||
if gap_secs <= limit {
|
||||
self.buckets[i] += 1;
|
||||
}
|
||||
}
|
||||
self.count += 1;
|
||||
self.sum_secs += gap_secs;
|
||||
self.sum_squares_secs += gap_secs * gap_secs;
|
||||
if gap_secs > self.max_secs { self.max_secs = gap_secs; }
|
||||
}
|
||||
|
||||
pub fn mean_secs(&self) -> f64 {
|
||||
if self.count == 0 { 0.0 } else { self.sum_secs / self.count as f64 }
|
||||
}
|
||||
|
||||
/// Desviación estándar muestral. Computada vía `sum_squares - n*mean²`
|
||||
/// para precisión razonable sin almacenar las muestras.
|
||||
pub fn stddev_secs(&self) -> f64 {
|
||||
if self.count < 2 { return 0.0; }
|
||||
let n = self.count as f64;
|
||||
let mean = self.mean_secs();
|
||||
let var = (self.sum_squares_secs - n * mean * mean) / (n - 1.0);
|
||||
// Numerical floor: var puede ser ligeramente negativo por float ε.
|
||||
if var <= 0.0 { 0.0 } else { var.sqrt() }
|
||||
}
|
||||
|
||||
pub fn stats(&self) -> GapStats {
|
||||
GapStats {
|
||||
count: self.count,
|
||||
mean_secs: self.mean_secs(),
|
||||
stddev_secs: self.stddev_secs(),
|
||||
max_secs: self.max_secs,
|
||||
}
|
||||
}
|
||||
|
||||
pub fn bucket_limits() -> &'static [f64; 7] { &GAP_BUCKET_LIMITS_SECS }
|
||||
}
|
||||
|
||||
pub struct Observer {
|
||||
window: VecDeque<TimedEvent>,
|
||||
window_size: usize,
|
||||
marginal: HashMap<EventKind, u64>,
|
||||
cooccur: HashMap<(EventKind, EventKind), u64>,
|
||||
total: u64,
|
||||
/// Last-seen timestamps para aplicar decay en query time. None = sin
|
||||
/// time-decay (modo tradicional).
|
||||
last_seen_marginal: HashMap<EventKind, Instant>,
|
||||
last_seen_cooccur: HashMap<(EventKind, EventKind), Instant>,
|
||||
/// Half-life del decay exponencial en segundos. None = sin decay
|
||||
/// (las consultas devuelven los counts crudos).
|
||||
half_life_secs: Option<f64>,
|
||||
/// Histograma de gaps temporales por par (a, b). Capturado al `record()`.
|
||||
gap_histograms: HashMap<(EventKind, EventKind), GapHistogram>,
|
||||
/// Sets de "qué cambió desde el último snapshot". Se vacían en
|
||||
/// `snapshot()` y `snapshot_delta()`. Usado para escritura incremental.
|
||||
dirty_marginal: std::collections::HashSet<EventKind>,
|
||||
dirty_cooccur: std::collections::HashSet<(EventKind, EventKind)>,
|
||||
}
|
||||
|
||||
impl Observer {
|
||||
pub fn new(window_size: usize) -> Self {
|
||||
Self {
|
||||
window: VecDeque::with_capacity(window_size),
|
||||
window_size,
|
||||
marginal: HashMap::new(),
|
||||
cooccur: HashMap::new(),
|
||||
total: 0,
|
||||
last_seen_marginal: HashMap::new(),
|
||||
last_seen_cooccur: HashMap::new(),
|
||||
half_life_secs: None,
|
||||
gap_histograms: HashMap::new(),
|
||||
dirty_marginal: std::collections::HashSet::new(),
|
||||
dirty_cooccur: std::collections::HashSet::new(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Activa decay exponencial con half-life en segundos. λ = ln(2)/half_life.
|
||||
/// Aplicado en query time sobre los counts crudos usando last_seen.
|
||||
pub fn with_half_life(mut self, half_life_secs: f64) -> Self {
|
||||
if half_life_secs > 0.0 {
|
||||
self.half_life_secs = Some(half_life_secs);
|
||||
}
|
||||
self
|
||||
}
|
||||
|
||||
pub fn half_life(&self) -> Option<f64> { self.half_life_secs }
|
||||
|
||||
/// Registra un evento. Actualiza marginales y co-ocurrencias contra todo
|
||||
/// evento aún en la ventana.
|
||||
pub fn record(&mut self, kind: EventKind) {
|
||||
let now = Instant::now();
|
||||
let timed = TimedEvent { kind: kind.clone(), at: now };
|
||||
|
||||
// Co-ocurrencias: este evento con cada uno previo en ventana.
|
||||
// Capturamos también el gap temporal (now - w.at) para histograma.
|
||||
for w in &self.window {
|
||||
let key = (w.kind.clone(), kind.clone());
|
||||
*self.cooccur.entry(key.clone()).or_insert(0) += 1;
|
||||
self.last_seen_cooccur.insert(key.clone(), now);
|
||||
let gap_secs = now.duration_since(w.at).as_secs_f64();
|
||||
self.gap_histograms.entry(key.clone()).or_default().observe(gap_secs);
|
||||
self.dirty_cooccur.insert(key);
|
||||
}
|
||||
|
||||
self.window.push_back(timed);
|
||||
if self.window.len() > self.window_size {
|
||||
self.window.pop_front();
|
||||
}
|
||||
|
||||
*self.marginal.entry(kind.clone()).or_insert(0) += 1;
|
||||
self.last_seen_marginal.insert(kind.clone(), now);
|
||||
self.dirty_marginal.insert(kind);
|
||||
self.total += 1;
|
||||
}
|
||||
|
||||
/// Aplica el decay sobre un count crudo dado el `last_seen` correspondiente.
|
||||
/// Si half_life es None, devuelve el count tal cual (sin decay).
|
||||
fn decay(&self, count: u64, last_seen: Option<Instant>) -> f64 {
|
||||
let raw = count as f64;
|
||||
let (hl, last) = match (self.half_life_secs, last_seen) {
|
||||
(Some(hl), Some(t)) => (hl, t),
|
||||
_ => return raw,
|
||||
};
|
||||
let age_secs = Instant::now().duration_since(last).as_secs_f64();
|
||||
raw * 0.5_f64.powf(age_secs / hl)
|
||||
}
|
||||
|
||||
/// Marginal con decay aplicado.
|
||||
pub fn marginal_decayed(&self, k: &EventKind) -> f64 {
|
||||
let raw = self.marginal.get(k).copied().unwrap_or(0);
|
||||
let last = self.last_seen_marginal.get(k).copied();
|
||||
self.decay(raw, last)
|
||||
}
|
||||
|
||||
/// Cooccurrence con decay aplicado.
|
||||
pub fn cooccur_decayed(&self, a: &EventKind, b: &EventKind) -> f64 {
|
||||
let raw = self.cooccur.get(&(a.clone(), b.clone())).copied().unwrap_or(0);
|
||||
let last = self.last_seen_cooccur.get(&(a.clone(), b.clone())).copied();
|
||||
self.decay(raw, last)
|
||||
}
|
||||
|
||||
/// Entropía de Shannon de la distribución marginal de eventos.
|
||||
/// H(X) = −Σ p(x) log₂ p(x). Unidad: bits.
|
||||
pub fn shannon_entropy(&self) -> f64 {
|
||||
if self.total == 0 { return 0.0; }
|
||||
let total = self.total as f64;
|
||||
self.marginal.values()
|
||||
.map(|&c| {
|
||||
let p = c as f64 / total;
|
||||
if p > 0.0 { -p * p.log2() } else { 0.0 }
|
||||
})
|
||||
.sum()
|
||||
}
|
||||
|
||||
/// P(b | a) = "dado que algo siguió a `a` dentro del window, qué fracción
|
||||
/// fue `b`". Suma 1 sobre todos los b posibles para un a fijo.
|
||||
///
|
||||
/// Implementación: cooccur_decayed(a, b) / Σ_x cooccur_decayed(a, x).
|
||||
/// Si half_life is None, los decayed values son los counts crudos.
|
||||
pub fn conditional_prob(&self, a: &EventKind, b: &EventKind) -> f64 {
|
||||
let joint = self.cooccur_decayed(a, b);
|
||||
let row_total: f64 = self.cooccur.keys()
|
||||
.filter(|(x, _)| x == a)
|
||||
.map(|(x, y)| self.cooccur_decayed(x, y))
|
||||
.sum();
|
||||
if row_total <= 0.0 { 0.0 } else { joint / row_total }
|
||||
}
|
||||
|
||||
/// Información mutua puntual entre `a` y `b` con decay aplicado:
|
||||
/// PMI(a, b) = log₂( P(a, b) / (P(a) · P(b)) ).
|
||||
/// Positivo → más correlacionados de lo que sugiere independencia.
|
||||
pub fn pmi(&self, a: &EventKind, b: &EventKind) -> f64 {
|
||||
// Total decayed: suma de marginales con decay (no usamos self.total
|
||||
// directo porque debería ser consistente con los decayed values).
|
||||
let total_decayed: f64 = self.marginal.keys()
|
||||
.map(|k| self.marginal_decayed(k))
|
||||
.sum();
|
||||
if total_decayed <= 0.0 { return 0.0; }
|
||||
let joint = self.cooccur_decayed(a, b) / total_decayed;
|
||||
let pa = self.marginal_decayed(a) / total_decayed;
|
||||
let pb = self.marginal_decayed(b) / total_decayed;
|
||||
if joint <= 0.0 || pa <= 0.0 || pb <= 0.0 { return 0.0; }
|
||||
(joint / (pa * pb)).log2()
|
||||
}
|
||||
|
||||
/// Información mutua acumulada de la pareja (a, b) ponderada por su
|
||||
/// probabilidad conjunta. Útil como medida de "interés" del par.
|
||||
pub fn weighted_pmi(&self, a: &EventKind, b: &EventKind) -> f64 {
|
||||
if self.total == 0 { return 0.0; }
|
||||
let joint = self.cooccur
|
||||
.get(&(a.clone(), b.clone()))
|
||||
.copied()
|
||||
.unwrap_or(0) as f64 / self.total as f64;
|
||||
joint * self.pmi(a, b)
|
||||
}
|
||||
|
||||
pub fn marginals(&self) -> &HashMap<EventKind, u64> { &self.marginal }
|
||||
|
||||
/// Última vez que se vio un kind. None si nunca o si fue restaurado
|
||||
/// desde snapshot (los Instants no portables se descartan).
|
||||
pub fn last_seen_marginal(&self, kind: &EventKind) -> Option<Instant> {
|
||||
self.last_seen_marginal.get(kind).copied()
|
||||
}
|
||||
pub fn cooccurrences(&self) -> &HashMap<(EventKind, EventKind), u64> { &self.cooccur }
|
||||
pub fn total(&self) -> u64 { self.total }
|
||||
pub fn window_size(&self) -> usize { self.window_size }
|
||||
pub fn current_window(&self) -> usize { self.window.len() }
|
||||
|
||||
/// Últimos N eventos del window, en orden cronológico (más viejo primero).
|
||||
/// Si N > window.len(), devuelve todo el window.
|
||||
pub fn recent(&self, n: usize) -> impl Iterator<Item = &TimedEvent> {
|
||||
let start = self.window.len().saturating_sub(n);
|
||||
self.window.range(start..)
|
||||
}
|
||||
|
||||
pub fn gap_histograms(&self) -> &HashMap<(EventKind, EventKind), GapHistogram> {
|
||||
&self.gap_histograms
|
||||
}
|
||||
|
||||
/// Top-K pares por count del histograma (más frecuentes primero).
|
||||
/// Útil para limitar cardinalidad de métricas exportadas.
|
||||
pub fn top_gap_pairs(&self, k: usize) -> Vec<(&(EventKind, EventKind), &GapHistogram)> {
|
||||
let mut pairs: Vec<_> = self.gap_histograms.iter().collect();
|
||||
pairs.sort_by(|a, b| b.1.count.cmp(&a.1.count));
|
||||
pairs.truncate(k);
|
||||
pairs
|
||||
}
|
||||
|
||||
/// Snapshot full: estado estadístico completo. Limpia los sets dirty
|
||||
/// como side-effect — los próximos `snapshot_delta()` cubren sólo los
|
||||
/// cambios posteriores.
|
||||
pub fn snapshot(&mut self) -> ObserverSnapshot {
|
||||
self.dirty_marginal.clear();
|
||||
self.dirty_cooccur.clear();
|
||||
ObserverSnapshot {
|
||||
schema_version: OBSERVER_SCHEMA_VERSION,
|
||||
is_delta: false,
|
||||
window_size: self.window_size,
|
||||
half_life_secs: self.half_life_secs,
|
||||
total: self.total,
|
||||
marginal: self.marginal.iter()
|
||||
.map(|(k, v)| (k.clone(), *v))
|
||||
.collect(),
|
||||
cooccur: self.cooccur.iter()
|
||||
.map(|((a, b), c)| (a.clone(), b.clone(), *c))
|
||||
.collect(),
|
||||
gap_histograms: self.gap_histograms.iter()
|
||||
.map(|((a, b), h)| (a.clone(), b.clone(), h.clone()))
|
||||
.collect(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Snapshot incremental: sólo incluye los kinds y pares que cambiaron
|
||||
/// desde el último `snapshot()` o `snapshot_delta()`. Útil para
|
||||
/// checkpoints frecuentes con poco overhead. Limpia los sets dirty.
|
||||
pub fn snapshot_delta(&mut self) -> ObserverSnapshot {
|
||||
let marginal: Vec<_> = self.dirty_marginal.iter()
|
||||
.filter_map(|k| self.marginal.get(k).map(|v| (k.clone(), *v)))
|
||||
.collect();
|
||||
let cooccur: Vec<_> = self.dirty_cooccur.iter()
|
||||
.filter_map(|(a, b)| {
|
||||
self.cooccur.get(&(a.clone(), b.clone()))
|
||||
.map(|c| (a.clone(), b.clone(), *c))
|
||||
})
|
||||
.collect();
|
||||
// Para histogramas: incluimos los pares cuyo cooccur cambió.
|
||||
let gap_histograms: Vec<_> = self.dirty_cooccur.iter()
|
||||
.filter_map(|(a, b)| {
|
||||
self.gap_histograms.get(&(a.clone(), b.clone()))
|
||||
.map(|h| (a.clone(), b.clone(), h.clone()))
|
||||
})
|
||||
.collect();
|
||||
self.dirty_marginal.clear();
|
||||
self.dirty_cooccur.clear();
|
||||
ObserverSnapshot {
|
||||
schema_version: OBSERVER_SCHEMA_VERSION,
|
||||
is_delta: true,
|
||||
window_size: self.window_size,
|
||||
half_life_secs: self.half_life_secs,
|
||||
total: self.total,
|
||||
marginal, cooccur, gap_histograms,
|
||||
}
|
||||
}
|
||||
|
||||
/// Aplica un delta sobre el estado actual. Para `is_delta=true`, los
|
||||
/// valores en marginal/cooccur sobrescriben las entradas existentes.
|
||||
/// Si `is_delta=false`, equivale a `from_snapshot` pero in-place.
|
||||
pub fn apply_delta(&mut self, delta: ObserverSnapshot) {
|
||||
let now = Instant::now();
|
||||
if !delta.is_delta {
|
||||
// Full: reset state.
|
||||
*self = Self::from_snapshot(delta);
|
||||
return;
|
||||
}
|
||||
// Incremental merge.
|
||||
for (k, v) in delta.marginal {
|
||||
self.last_seen_marginal.insert(k.clone(), now);
|
||||
self.marginal.insert(k, v);
|
||||
}
|
||||
for (a, b, c) in delta.cooccur {
|
||||
self.last_seen_cooccur.insert((a.clone(), b.clone()), now);
|
||||
self.cooccur.insert((a, b), c);
|
||||
}
|
||||
for (a, b, h) in delta.gap_histograms {
|
||||
self.gap_histograms.insert((a, b), h);
|
||||
}
|
||||
// total: sólo subimos (el delta podría estar atrasado).
|
||||
if delta.total > self.total { self.total = delta.total; }
|
||||
}
|
||||
|
||||
/// Reconstruye Observer desde un snapshot. El window queda vacío;
|
||||
/// last_seen_* se inicializa en `now()` para que el decay arranque
|
||||
/// "ahora" para todos los counts (aproximación razonable post-reboot).
|
||||
pub fn from_snapshot(snap: ObserverSnapshot) -> Self {
|
||||
let now = Instant::now();
|
||||
let mut marginal = HashMap::new();
|
||||
let mut last_seen_marginal = HashMap::new();
|
||||
for (k, v) in snap.marginal {
|
||||
last_seen_marginal.insert(k.clone(), now);
|
||||
marginal.insert(k, v);
|
||||
}
|
||||
let mut cooccur = HashMap::new();
|
||||
let mut last_seen_cooccur = HashMap::new();
|
||||
for (a, b, c) in snap.cooccur {
|
||||
last_seen_cooccur.insert((a.clone(), b.clone()), now);
|
||||
cooccur.insert((a, b), c);
|
||||
}
|
||||
let gap_histograms = snap.gap_histograms.into_iter()
|
||||
.map(|(a, b, h)| ((a, b), h))
|
||||
.collect();
|
||||
Self {
|
||||
window: VecDeque::with_capacity(snap.window_size),
|
||||
window_size: snap.window_size,
|
||||
marginal,
|
||||
cooccur,
|
||||
total: snap.total,
|
||||
last_seen_marginal,
|
||||
last_seen_cooccur,
|
||||
half_life_secs: snap.half_life_secs,
|
||||
gap_histograms,
|
||||
dirty_marginal: std::collections::HashSet::new(),
|
||||
dirty_cooccur: std::collections::HashSet::new(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const OBSERVER_SCHEMA_VERSION: u16 = 1;
|
||||
|
||||
/// Snapshot serializable. Se persiste a JSON en disco y se restaura al
|
||||
/// reboot para preservar contadores, co-ocurrencias e histogramas.
|
||||
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
|
||||
pub struct ObserverSnapshot {
|
||||
pub schema_version: u16,
|
||||
/// `true` si sólo contiene los cambios desde el último snapshot.
|
||||
/// `false` = full state, sobreescribe el observer al aplicar.
|
||||
#[serde(default)]
|
||||
pub is_delta: bool,
|
||||
pub window_size: usize,
|
||||
pub half_life_secs: Option<f64>,
|
||||
pub total: u64,
|
||||
/// Marginales serializados como Vec porque HashMap<EventKind, _> usa
|
||||
/// EventKind como key — y EventKind tiene variantes con payloads que
|
||||
/// no son JSON-key-serializable (BusInvokeOf, Custom).
|
||||
pub marginal: Vec<(EventKind, u64)>,
|
||||
pub cooccur: Vec<(EventKind, EventKind, u64)>,
|
||||
pub gap_histograms: Vec<(EventKind, EventKind, GapHistogram)>,
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::rules::EventKind::*;
|
||||
|
||||
#[test]
|
||||
fn entropy_zero_for_single_event() {
|
||||
let mut obs = Observer::new(10);
|
||||
for _ in 0..5 { obs.record(EnteSpawned); }
|
||||
// Distribución degenerada: una sola observación posible → H = 0.
|
||||
assert!(obs.shannon_entropy() < 1e-9);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn entropy_one_for_balanced_binary() {
|
||||
let mut obs = Observer::new(100);
|
||||
for _ in 0..10 { obs.record(EnteSpawned); }
|
||||
for _ in 0..10 { obs.record(EnteDied); }
|
||||
// Bernoulli(0.5) → H = 1 bit.
|
||||
assert!((obs.shannon_entropy() - 1.0).abs() < 1e-9);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn conditional_prob_perfect_dependency() {
|
||||
let mut obs = Observer::new(100);
|
||||
// Spawned siempre seguido por Died.
|
||||
for _ in 0..5 {
|
||||
obs.record(EnteSpawned);
|
||||
obs.record(EnteDied);
|
||||
}
|
||||
let p = obs.conditional_prob(&EnteSpawned, &EnteDied);
|
||||
assert!(p > 0.0, "esperamos correlación positiva, got {p}");
|
||||
}
|
||||
}
|
||||
@@ -1,206 +0,0 @@
|
||||
//! Tipos de regla. La fuente de verdad del shape es esta definición Rust;
|
||||
//! `schema/rule.k` queda como referencia de diseño no cargada.
|
||||
//!
|
||||
//! Cargables desde JSON (array, objeto-con-array, o JSONL). El motor no
|
||||
//! acepta Rules construidas a mano sin pasar por validate() — ver
|
||||
//! `engine::insert`.
|
||||
|
||||
use arje_card::Capability;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use ulid::Ulid;
|
||||
|
||||
/// Triplet [Sujeto + Evento + Acción]. Inmutable tras carga.
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct Rule {
|
||||
pub id: Ulid,
|
||||
#[serde(default = "default_priority")]
|
||||
pub priority: u8,
|
||||
pub when: EventPattern,
|
||||
pub then: Vec<Action>,
|
||||
#[serde(default)]
|
||||
pub scope: Scope,
|
||||
}
|
||||
|
||||
fn default_priority() -> u8 { 5 }
|
||||
|
||||
impl Rule {
|
||||
pub fn validate(&self) -> Result<(), RuleError> {
|
||||
if self.then.is_empty() {
|
||||
return Err(RuleError::EmptyActions);
|
||||
}
|
||||
self.when.validate_recursive()
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
pub enum RuleError {
|
||||
EmptyActions,
|
||||
EmptySequence,
|
||||
EmptyCompound,
|
||||
InvalidPriority,
|
||||
}
|
||||
|
||||
impl std::fmt::Display for RuleError {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
match self {
|
||||
Self::EmptyActions => write!(f, "regla sin acciones"),
|
||||
Self::EmptySequence => write!(f, "Sequence pattern con kinds vacío"),
|
||||
Self::EmptyCompound => write!(f, "Either/All con patterns vacío"),
|
||||
Self::InvalidPriority => write!(f, "prioridad fuera de rango"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl std::error::Error for RuleError {}
|
||||
|
||||
/// Match del sujeto. Vacío en todos los campos = match cualquier Ente.
|
||||
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
|
||||
pub struct Scope {
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub subject_id: Option<Ulid>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub subject_label: Option<String>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub subject_has_cap: Option<Capability>,
|
||||
}
|
||||
|
||||
impl Scope {
|
||||
pub fn is_wildcard(&self) -> bool {
|
||||
self.subject_id.is_none()
|
||||
&& self.subject_label.is_none()
|
||||
&& self.subject_has_cap.is_none()
|
||||
}
|
||||
}
|
||||
|
||||
/// Patrón de evento que dispara una regla. Tagged union — JSON con campo
|
||||
/// `type`. Soporta composición recursiva (Either/All) sobre Single y
|
||||
/// Sequence atómicos.
|
||||
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Hash)]
|
||||
#[serde(tag = "type")]
|
||||
pub enum EventPattern {
|
||||
/// Match un único evento por kind.
|
||||
Single { kind: EventKind },
|
||||
/// Match si los últimos N eventos del history coinciden en orden con
|
||||
/// `kinds`, todos dentro de `within_ms` (0 = sin límite temporal).
|
||||
Sequence {
|
||||
kinds: Vec<EventKind>,
|
||||
#[serde(default)]
|
||||
within_ms: u64,
|
||||
},
|
||||
/// OR: match si cualquier sub-pattern matchea.
|
||||
Either { patterns: Vec<EventPattern> },
|
||||
/// AND: match si todos los sub-patterns matchean simultáneamente
|
||||
/// contra el mismo (event, history).
|
||||
All { patterns: Vec<EventPattern> },
|
||||
}
|
||||
|
||||
impl EventPattern {
|
||||
/// `true` si el pattern es atómico (no recursivo) y puede ser indexado
|
||||
/// por discriminante de `EventKind` para dispatch O(1). Compuestos
|
||||
/// (Either/All) se evalúan en un bucket de fallback.
|
||||
pub fn is_simple(&self) -> bool {
|
||||
matches!(self, Self::Single { .. } | Self::Sequence { .. })
|
||||
}
|
||||
|
||||
/// Última `EventKind` que dispara la evaluación de un pattern atómico.
|
||||
/// Devuelve None para compuestos.
|
||||
pub fn trigger_kind(&self) -> Option<&EventKind> {
|
||||
match self {
|
||||
Self::Single { kind } => Some(kind),
|
||||
Self::Sequence { kinds, .. } => kinds.last(),
|
||||
Self::Either { .. } | Self::All { .. } => None,
|
||||
}
|
||||
}
|
||||
|
||||
/// Validación recursiva. Compuestos vacíos se rechazan al cargar.
|
||||
pub fn validate_recursive(&self) -> Result<(), RuleError> {
|
||||
match self {
|
||||
Self::Single { .. } => Ok(()),
|
||||
Self::Sequence { kinds, .. } => {
|
||||
if kinds.is_empty() { Err(RuleError::EmptySequence) } else { Ok(()) }
|
||||
}
|
||||
Self::Either { patterns } | Self::All { patterns } => {
|
||||
if patterns.is_empty() {
|
||||
return Err(RuleError::EmptyCompound);
|
||||
}
|
||||
for p in patterns { p.validate_recursive()?; }
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Tipo de evento que dispara reglas. Indexado por discriminante en el motor.
|
||||
/// Diseñado para que `EventKindDiscriminant::from(&kind)` sea barato (no hash
|
||||
/// del payload, sólo del tag).
|
||||
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Hash)]
|
||||
pub enum EventKind {
|
||||
EnteSpawned,
|
||||
EnteDied,
|
||||
BusAnnounce,
|
||||
BusInvoke,
|
||||
BusInvokeOf(Capability),
|
||||
DeviceAdded,
|
||||
DeviceRemoved,
|
||||
Custom(String),
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
#[serde(rename_all = "lowercase")]
|
||||
pub enum LogLevel { Trace, Debug, Info, Warn, Error }
|
||||
|
||||
impl LogLevel {
|
||||
pub fn as_str(&self) -> &'static str {
|
||||
match self {
|
||||
Self::Trace => "trace",
|
||||
Self::Debug => "debug",
|
||||
Self::Info => "info",
|
||||
Self::Warn => "warn",
|
||||
Self::Error => "error",
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
#[serde(tag = "kind", rename_all = "PascalCase")]
|
||||
pub enum Action {
|
||||
Log {
|
||||
#[serde(default = "default_log_level")]
|
||||
level: LogLevel,
|
||||
message: String,
|
||||
},
|
||||
Notify {
|
||||
target_id: Ulid,
|
||||
message: String,
|
||||
},
|
||||
/// `card_blob` es JSON del EntityCard codificado en base64. El motor lo
|
||||
/// decodifica y forwarda como SpawnRequest al graph.
|
||||
Spawn {
|
||||
card_blob: String,
|
||||
},
|
||||
Invoke {
|
||||
target_cap: Capability,
|
||||
/// blob crudo (en JSON viaja como base64 vía `blob_b64`).
|
||||
#[serde(with = "blob_b64")]
|
||||
blob: Vec<u8>,
|
||||
},
|
||||
Inhibit {
|
||||
reason: String,
|
||||
},
|
||||
}
|
||||
|
||||
fn default_log_level() -> LogLevel { LogLevel::Info }
|
||||
|
||||
mod blob_b64 {
|
||||
use base64::{engine::general_purpose::STANDARD, Engine};
|
||||
use serde::{Deserialize, Deserializer, Serializer};
|
||||
|
||||
pub fn serialize<S: Serializer>(bytes: &[u8], s: S) -> Result<S::Ok, S::Error> {
|
||||
s.serialize_str(&STANDARD.encode(bytes))
|
||||
}
|
||||
|
||||
pub fn deserialize<'de, D: Deserializer<'de>>(d: D) -> Result<Vec<u8>, D::Error> {
|
||||
let s = String::deserialize(d)?;
|
||||
STANDARD.decode(&s).map_err(serde::de::Error::custom)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user