⎇ d270c5e67457195cd845f783c0b3b7bd66b9add0
crates/ente-brain/src/introspect.rs
History
Who touched what
raw
1//! Introspect API. Unix Domain Socket + framing length-prefijo + bincode.
2//!
3//! Una herramienta externa (ej. `brainctl`) puede consultar el estado del
4//! cerebro sin tocar el bus interno del fractal. Esto separa observación de
5//! ejecución — la introspección es read-only por diseño.
6
7use crate::crystallize::{detect_crystals, Crystal, CrystallizationParams};
8use crate::engine::RuleEngine;
9use crate::observer::Observer;
10use crate::rules::Rule;
11use serde::{Deserialize, Serialize};
12use std::io::Write;
13use std::path::{Path, PathBuf};
14use std::sync::Arc;
15use tokio::io::{AsyncReadExt, AsyncWriteExt};
16use tokio::net::{UnixListener, UnixStream};
17use tokio::sync::RwLock;
18use tracing::{debug, info, trace, warn};
19use ulid::Ulid;
20
21const MAX_FRAME: usize = 4 * 1024 * 1024; // 4 MiB — correlation matrices crecen
22
23/// Estado compartido entre el bucle del Init y el servidor de introspección.
24/// `Arc<RwLock<...>>` permite muchos lectores concurrentes (introspect) y
25/// un escritor (el dispatcher de eventos en el bucle primordial).
26#[derive(Clone)]
27pub struct BrainState {
28 pub engine: Arc<RwLock<RuleEngine>>,
29 pub observer: Arc<RwLock<Observer>>,
30 pub params: CrystallizationParams,
31 /// Path opcional donde apendear reglas promovidas como KCL. Si Some,
32 /// cada PromoteCrystal añade el snippet al archivo (append-only).
33 pub rules_out: Option<Arc<PathBuf>>,
34 /// Audit log en memoria. Cada promote/remove deja huella aquí.
35 pub audit: Arc<RwLock<crate::audit::AuditLog>>,
36}
37
38impl BrainState {
39 pub fn new(window_size: usize) -> Self {
40 Self::with_params(window_size, CrystallizationParams::default())
41 }
42
43 pub fn with_params(window_size: usize, params: CrystallizationParams) -> Self {
44 Self {
45 engine: Arc::new(RwLock::new(RuleEngine::empty())),
46 observer: Arc::new(RwLock::new(Observer::new(window_size))),
47 params,
48 rules_out: None,
49 audit: Arc::new(RwLock::new(crate::audit::AuditLog::new())),
50 }
51 }
52
53 pub fn with_rules_out(mut self, path: PathBuf) -> Self {
54 self.rules_out = Some(Arc::new(path));
55 self
56 }
57}
58
59/// Append-only writer del KCL snippet a `rules_out`. Crea el archivo con
60/// header si no existe; en caso contrario sólo apendea.
61pub fn append_kcl_snippet(path: &Path, snippet: &str) -> std::io::Result<()> {
62 if let Some(parent) = path.parent() {
63 std::fs::create_dir_all(parent)?;
64 }
65 let exists = path.exists();
66 let mut file = std::fs::OpenOptions::new()
67 .create(true)
68 .append(true)
69 .open(path)?;
70 if !exists {
71 writeln!(file, "# Reglas promovidas automáticamente desde cristales.")?;
72 writeln!(file, "# Cada bloque proviene de PromoteCrystal vía brainctl.")?;
73 writeln!(file)?;
74 }
75 writeln!(file, "{snippet}")?;
76 Ok(())
77}
78
79#[derive(Debug, Serialize, Deserialize)]
80pub enum IntrospectRequest {
81 /// Lista resumida de reglas vivas.
82 ListRules,
83 /// Detalle de una regla concreta.
84 GetRule(Ulid),
85 /// Snapshot de la entropía y conteos básicos.
86 EntropySnapshot,
87 /// Top N pares (a, b) por co-ocurrencia.
88 TopCorrelations { n: usize },
89 /// Cristales detectados con los parámetros del BrainState.
90 Crystals,
91 /// Genera el snippet KCL de un cristal específico (índice tras Crystals).
92 CrystalKcl { index: usize },
93 /// Promueve el cristal #index a regla viva en el motor. Devuelve el
94 /// rule_id asignado y el snippet KCL para auditoría/persistencia.
95 PromoteCrystal { index: usize },
96 /// Elimina una regla viva por id. Útil para revertir un promote.
97 RemoveRule { id: Ulid },
98 /// Lista las últimas N entradas del audit log. limit=0 = todas.
99 ListAudit { limit: usize },
100 /// Persiste todas las entries pendientes al CAS y actualiza el head
101 /// pointer si el log lo tiene configurado.
102 FlushAudit,
103 /// Recarga reglas desde el archivo configurado por --rules-out (o el
104 /// path provisto). Vacía el engine antes de cargar.
105 ReloadRules { path: Option<String> },
106 /// Verifica la cadena audit recorriendo prev_sha hasta el genesis,
107 /// validando integridad de cada entry contra el CAS.
108 VerifyAudit,
109 /// Reconstruye el engine desde la cadena audit. Vacía engine y aplica
110 /// PromoteCrystal/RemoveRule en orden cronológico.
111 ReplayAudit,
112 /// Mantiene la conexión abierta y empuja cada `AuditEntry` nuevo en
113 /// frames `IntrospectResponse::AuditStreamFrame` hasta que el cliente
114 /// cierra. Tras esta request no se aceptan más requests en la misma conn.
115 StreamAudit,
116 /// Garbage-collect el CAS. Considera reachable: todo lo alcanzable desde
117 /// el head del audit log. Cualquier blob extra (Wasm modules referenciados
118 /// por Cards) debe haberse pasado en `extra_roots` por el caller.
119 GcCas { extra_roots: Vec<[u8; 32]> },
120 /// Detecta cristales de patrones temporales (Burst, Silence).
121 PatternCrystals,
122}
123
124#[derive(Debug, Serialize, Deserialize)]
125pub enum IntrospectResponse {
126 Rules(Vec<RuleSummary>),
127 Rule(Option<Rule>),
128 Entropy { value_bits: f64, sample_size: u64, distinct_kinds: usize, window_full: bool },
129 Correlations(Vec<CorrelationEntry>),
130 Crystals(Vec<Crystal>),
131 Kcl(String),
132 /// Resultado de PromoteCrystal: id de la regla creada + snippet KCL para
133 /// que el operador lo persista en disco si quiere.
134 Promoted { rule_id: Ulid, kcl_snippet: String },
135 /// Resultado de RemoveRule: true si existía, false si ya no.
136 Removed(bool),
137 /// Entradas del audit log (más recientes al final).
138 AuditEntries(Vec<crate::audit::AuditEntry>),
139 /// Resultado de FlushAudit: cuántas entries se escribieron y SHA del head.
140 Flushed { written: usize, head_sha: Option<[u8; 32]>, total_flushed: u64 },
141 /// Resultado de ReloadRules: número total de reglas tras el reload.
142 Reloaded { count: usize },
143 /// Resultado de VerifyAudit.
144 AuditVerified(crate::audit::VerificationReport),
145 /// Resultado de ReplayAudit.
146 Replayed(crate::audit::ReplayReport),
147 /// Frame de streaming. El cliente lee estos en bucle hasta EOF.
148 AuditStreamFrame(crate::audit::AuditEntry),
149 /// Resultado de GcCas: cuántos blobs eliminados y bytes liberados.
150 GcResult { deleted: usize, freed_bytes: u64 },
151 /// Cristales de Burst/Silence detectados.
152 Patterns(Vec<crate::crystallize::PatternCrystal>),
153 Error(String),
154}
155
156#[derive(Debug, Serialize, Deserialize)]
157pub struct RuleSummary {
158 pub id: Ulid,
159 pub priority: u8,
160 pub event_kind_tag: String,
161 pub action_count: usize,
162 pub scope_wildcard: bool,
163}
164
165#[derive(Debug, Serialize, Deserialize)]
166pub struct CorrelationEntry {
167 pub a: String,
168 pub b: String,
169 pub joint_count: u64,
170 pub conditional_prob: f64,
171 pub pmi_bits: f64,
172}
173
174pub struct IntrospectServer {
175 state: BrainState,
176}
177
178impl IntrospectServer {
179 pub fn new(state: BrainState) -> Self { Self { state } }
180
181 /// Spawn del listener. Devuelve cuando bind() falla; en caso contrario
182 /// corre indefinidamente.
183 pub async fn serve(self, path: &Path) -> anyhow::Result<()> {
184 let _ = std::fs::remove_file(path);
185 if let Some(parent) = path.parent() {
186 let _ = std::fs::create_dir_all(parent);
187 }
188 let listener = UnixListener::bind(path)?;
189 info!(path = %path.display(), "brain introspect escuchando");
190 let arc_self = Arc::new(self);
191 loop {
192 match listener.accept().await {
193 Ok((stream, _)) => {
194 trace!("introspect conn aceptada");
195 let me = arc_self.clone();
196 tokio::spawn(async move {
197 if let Err(e) = me.handle(stream).await {
198 warn!(?e, "introspect conn ended");
199 }
200 });
201 }
202 Err(e) => {
203 warn!(?e, "introspect accept failed");
204 return Ok(());
205 }
206 }
207 }
208 }
209
210 async fn handle(self: Arc<Self>, mut stream: UnixStream) -> anyhow::Result<()> {
211 loop {
212 let mut len_buf = [0u8; 4];
213 if stream.read_exact(&mut len_buf).await.is_err() {
214 return Ok(()); // EOF
215 }
216 let len = u32::from_be_bytes(len_buf) as usize;
217 if len > MAX_FRAME {
218 anyhow::bail!("frame oversize: {len}");
219 }
220 let mut buf = vec![0u8; len];
221 stream.read_exact(&mut buf).await?;
222 let req: IntrospectRequest = bincode::deserialize(&buf)?;
223 debug!(?req, "introspect request");
224
225 // StreamAudit toma posesión de la conn — no más requests aquí.
226 if matches!(req, IntrospectRequest::StreamAudit) {
227 return self.stream_audit(stream).await;
228 }
229
230 let resp = self.dispatch(req).await;
231
232 let out = bincode::serialize(&resp)?;
233 stream.write_u32(out.len() as u32).await?;
234 stream.write_all(&out).await?;
235 }
236 }
237
238 /// Modo streaming: subscribe al audit log y empuja cada entry como
239 /// frame `AuditStreamFrame`. La función retorna cuando el cliente
240 /// cierra (write falla) o el subscriber se desconecta.
241 async fn stream_audit(self: Arc<Self>, mut stream: UnixStream) -> anyhow::Result<()> {
242 let mut rx = self.state.audit.write().await.subscribe();
243 info!("audit stream client conectado");
244 while let Some(entry) = rx.recv().await {
245 let frame = IntrospectResponse::AuditStreamFrame(entry);
246 let bytes = bincode::serialize(&frame)?;
247 if stream.write_u32(bytes.len() as u32).await.is_err() { break; }
248 if stream.write_all(&bytes).await.is_err() { break; }
249 }
250 info!("audit stream client desconectado");
251 Ok(())
252 }
253
254 async fn dispatch(&self, req: IntrospectRequest) -> IntrospectResponse {
255 match req {
256 IntrospectRequest::ListRules => {
257 let engine = self.state.engine.read().await;
258 let rules = engine.rules()
259 .map(|r| RuleSummary {
260 id: r.id,
261 priority: r.priority,
262 event_kind_tag: format!("{:?}", r.when),
263 action_count: r.then.len(),
264 scope_wildcard: r.scope.is_wildcard(),
265 })
266 .collect();
267 IntrospectResponse::Rules(rules)
268 }
269 IntrospectRequest::GetRule(id) => {
270 let engine = self.state.engine.read().await;
271 let found = engine.rules()
272 .find(|r| r.id == id)
273 .map(|r| Rule::clone(r));
274 IntrospectResponse::Rule(found)
275 }
276 IntrospectRequest::EntropySnapshot => {
277 let obs = self.state.observer.read().await;
278 IntrospectResponse::Entropy {
279 value_bits: obs.shannon_entropy(),
280 sample_size: obs.total(),
281 distinct_kinds: obs.marginals().len(),
282 window_full: obs.current_window() >= obs.window_size(),
283 }
284 }
285 IntrospectRequest::TopCorrelations { n } => {
286 let obs = self.state.observer.read().await;
287 let mut entries: Vec<CorrelationEntry> = obs.cooccurrences().iter()
288 .map(|((a, b), &joint)| CorrelationEntry {
289 a: format!("{:?}", a),
290 b: format!("{:?}", b),
291 joint_count: joint,
292 conditional_prob: obs.conditional_prob(a, b),
293 pmi_bits: obs.pmi(a, b),
294 })
295 .collect();
296 entries.sort_by(|x, y| y.joint_count.cmp(&x.joint_count));
297 entries.truncate(n);
298 IntrospectResponse::Correlations(entries)
299 }
300 IntrospectRequest::Crystals => {
301 let obs = self.state.observer.read().await;
302 let crystals = detect_crystals(&obs, &self.state.params);
303 IntrospectResponse::Crystals(crystals)
304 }
305 IntrospectRequest::CrystalKcl { index } => {
306 let obs = self.state.observer.read().await;
307 let crystals = detect_crystals(&obs, &self.state.params);
308 match crystals.get(index) {
309 Some(c) => IntrospectResponse::Kcl(crate::crystallize::crystal_to_kcl(c)),
310 None => IntrospectResponse::Error(format!("no crystal at index {index}")),
311 }
312 }
313 IntrospectRequest::PromoteCrystal { index } => {
314 let crystals = {
315 let obs = self.state.observer.read().await;
316 detect_crystals(&obs, &self.state.params)
317 };
318 match crystals.get(index) {
319 Some(c) => {
320 let rule = crate::crystallize::crystal_to_rule(c);
321 let snippet = crate::crystallize::crystal_to_kcl(c);
322 let rule_id = rule.id;
323 self.state.engine.write().await.insert(rule);
324 // Persistencia opcional al archivo KCL.
325 if let Some(path) = self.state.rules_out.as_ref() {
326 if let Err(e) = append_kcl_snippet(path, &snippet) {
327 warn!(?e, path = %path.display(), "rules_out append falló");
328 } else {
329 info!(path = %path.display(), %rule_id, "regla persistida a .k");
330 }
331 }
332 // Audit entry
333 self.state.audit.write().await.append(
334 crate::audit::AuditAction::PromoteCrystal {
335 rule_id, crystal: c.clone(),
336 }
337 );
338 IntrospectResponse::Promoted { rule_id, kcl_snippet: snippet }
339 }
340 None => IntrospectResponse::Error(format!("no crystal at index {index}")),
341 }
342 }
343 IntrospectRequest::RemoveRule { id } => {
344 let removed = self.state.engine.write().await.remove(id);
345 if removed {
346 self.state.audit.write().await.append(
347 crate::audit::AuditAction::RemoveRule { rule_id: id }
348 );
349 }
350 IntrospectResponse::Removed(removed)
351 }
352 IntrospectRequest::ListAudit { limit } => {
353 let audit = self.state.audit.read().await;
354 IntrospectResponse::AuditEntries(audit.recent(limit).cloned().collect())
355 }
356 IntrospectRequest::FlushAudit => {
357 let mut audit = self.state.audit.write().await;
358 match audit.flush_to_cas() {
359 Ok(written) => IntrospectResponse::Flushed {
360 written,
361 head_sha: audit.last_flushed_sha(),
362 total_flushed: audit.flushed_count(),
363 },
364 Err(e) => IntrospectResponse::Error(format!("flush_to_cas: {e}")),
365 }
366 }
367 IntrospectRequest::VerifyAudit => {
368 let head = self.state.audit.read().await.last_flushed_sha();
369 let head = match head {
370 Some(h) => h,
371 None => return IntrospectResponse::Error(
372 "audit log sin entries flushadas — nada que verificar".into()
373 ),
374 };
375 let report = crate::audit::verify_chain_from_cas(head);
376 IntrospectResponse::AuditVerified(report)
377 }
378 IntrospectRequest::StreamAudit => {
379 // Inalcanzable por construcción: handle() detecta StreamAudit
380 // antes de llamar a dispatch(). Pero el match exige cubrir.
381 IntrospectResponse::Error(
382 "StreamAudit no debe llegar a dispatch — bug del handler".into()
383 )
384 }
385 IntrospectRequest::PatternCrystals => {
386 let obs = self.state.observer.read().await;
387 let params = crate::crystallize::PatternParams::default();
388 let patterns = crate::crystallize::detect_pattern_crystals(&obs, ¶ms);
389 IntrospectResponse::Patterns(patterns)
390 }
391 IntrospectRequest::GcCas { extra_roots } => {
392 // Reachable = audit chain desde head + extra_roots provistos.
393 let mut reachable = std::collections::HashSet::new();
394 if let Some(head) = self.state.audit.read().await.last_flushed_sha() {
395 reachable.extend(crate::audit::reachable_from_head(head));
396 }
397 reachable.extend(extra_roots);
398 match ente_cas::gc(&reachable) {
399 Ok((deleted, freed_bytes)) => IntrospectResponse::GcResult { deleted, freed_bytes },
400 Err(e) => IntrospectResponse::Error(format!("gc: {e}")),
401 }
402 }
403 IntrospectRequest::ReplayAudit => {
404 let head = self.state.audit.read().await.last_flushed_sha();
405 let head = match head {
406 Some(h) => h,
407 None => return IntrospectResponse::Error(
408 "audit log sin entries flushadas — nada que replayar".into()
409 ),
410 };
411 let mut engine = self.state.engine.write().await;
412 *engine = crate::engine::RuleEngine::empty();
413 let report = crate::audit::replay_chain(head, &mut engine);
414 IntrospectResponse::Replayed(report)
415 }
416 IntrospectRequest::ReloadRules { path } => {
417 // Path explícito gana sobre el rules_out configurado.
418 let resolved = path.map(std::path::PathBuf::from)
419 .or_else(|| self.state.rules_out.as_ref().map(|p| p.as_path().to_path_buf()));
420 let path = match resolved {
421 Some(p) => p,
422 None => return IntrospectResponse::Error(
423 "ReloadRules sin path y sin rules_out configurado".into()
424 ),
425 };
426 let rules = match crate::loader::load_rules_file(&path) {
427 Ok(r) => r,
428 Err(e) => return IntrospectResponse::Error(format!("load: {e}")),
429 };
430 // Vaciamos el engine antes de re-cargar — semántica clean-slate.
431 let mut engine = self.state.engine.write().await;
432 *engine = crate::engine::RuleEngine::empty();
433 let count = rules.len();
434 for r in rules { engine.insert(r); }
435 drop(engine);
436 self.state.audit.write().await.append(
437 crate::audit::AuditAction::LoadRulesFile {
438 path: path.to_string_lossy().into_owned(),
439 count,
440 }
441 );
442 IntrospectResponse::Reloaded { count }
443 }
444 }
445 }
446}
447
448// Cliente helper para tools externos (brainctl).
449pub async fn call(path: &Path, req: IntrospectRequest) -> anyhow::Result<IntrospectResponse> {
450 let mut stream = UnixStream::connect(path).await?;
451 let buf = bincode::serialize(&req)?;
452 stream.write_u32(buf.len() as u32).await?;
453 stream.write_all(&buf).await?;
454
455 let mut len_buf = [0u8; 4];
456 stream.read_exact(&mut len_buf).await?;
457 let len = u32::from_be_bytes(len_buf) as usize;
458 if len > MAX_FRAME {
459 anyhow::bail!("response oversize: {len}");
460 }
461 let mut buf = vec![0u8; len];
462 stream.read_exact(&mut buf).await?;
463 Ok(bincode::deserialize(&buf)?)
464}
465
466/// Consume la lista marginal del observer para humanos. Suprime el detalle
467/// crudo de `EventKind` (ej. payloads largos en BusInvokeOf).
468pub fn marginal_summary(obs: &Observer) -> Vec<(String, u64)> {
469 let mut entries: Vec<(String, u64)> = obs.marginals().iter()
470 .map(|(k, &c)| (format!("{:?}", k), c))
471 .collect();
472 entries.sort_by(|x, y| y.1.cmp(&x.1));
473 entries
474}
475