⎇ d6b8f18b43c9bd9bf7268e48b61392a2d62bb229
crates/ente-brain/src/introspect.rs
Historia
Quién tocó qué
crudo
Pausa: 11 crates del fractal Ente #0 con cerebro completo
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}
101
102#[derive(Debug, Serialize, Deserialize)]
103pub enum IntrospectResponse {
104 Rules(Vec<RuleSummary>),
105 Rule(Option<Rule>),
106 Entropy { value_bits: f64, sample_size: u64, distinct_kinds: usize, window_full: bool },
107 Correlations(Vec<CorrelationEntry>),
108 Crystals(Vec<Crystal>),
109 Kcl(String),
110 /// Resultado de PromoteCrystal: id de la regla creada + snippet KCL para
111 /// que el operador lo persista en disco si quiere.
112 Promoted { rule_id: Ulid, kcl_snippet: String },
113 /// Resultado de RemoveRule: true si existía, false si ya no.
114 Removed(bool),
115 /// Entradas del audit log (más recientes al final).
116 AuditEntries(Vec<crate::audit::AuditEntry>),
117 Error(String),
118}
119
120#[derive(Debug, Serialize, Deserialize)]
121pub struct RuleSummary {
122 pub id: Ulid,
123 pub priority: u8,
124 pub event_kind_tag: String,
125 pub action_count: usize,
126 pub scope_wildcard: bool,
127}
128
129#[derive(Debug, Serialize, Deserialize)]
130pub struct CorrelationEntry {
131 pub a: String,
132 pub b: String,
133 pub joint_count: u64,
134 pub conditional_prob: f64,
135 pub pmi_bits: f64,
136}
137
138pub struct IntrospectServer {
139 state: BrainState,
140}
141
142impl IntrospectServer {
143 pub fn new(state: BrainState) -> Self { Self { state } }
144
145 /// Spawn del listener. Devuelve cuando bind() falla; en caso contrario
146 /// corre indefinidamente.
147 pub async fn serve(self, path: &Path) -> anyhow::Result<()> {
148 let _ = std::fs::remove_file(path);
149 if let Some(parent) = path.parent() {
150 let _ = std::fs::create_dir_all(parent);
151 }
152 let listener = UnixListener::bind(path)?;
153 info!(path = %path.display(), "brain introspect escuchando");
154 let arc_self = Arc::new(self);
155 loop {
156 match listener.accept().await {
157 Ok((stream, _)) => {
158 trace!("introspect conn aceptada");
159 let me = arc_self.clone();
160 tokio::spawn(async move {
161 if let Err(e) = me.handle(stream).await {
162 warn!(?e, "introspect conn ended");
163 }
164 });
165 }
166 Err(e) => {
167 warn!(?e, "introspect accept failed");
168 return Ok(());
169 }
170 }
171 }
172 }
173
174 async fn handle(self: Arc<Self>, mut stream: UnixStream) -> anyhow::Result<()> {
175 loop {
176 let mut len_buf = [0u8; 4];
177 if stream.read_exact(&mut len_buf).await.is_err() {
178 return Ok(()); // EOF
179 }
180 let len = u32::from_be_bytes(len_buf) as usize;
181 if len > MAX_FRAME {
182 anyhow::bail!("frame oversize: {len}");
183 }
184 let mut buf = vec![0u8; len];
185 stream.read_exact(&mut buf).await?;
186 let req: IntrospectRequest = bincode::deserialize(&buf)?;
187 debug!(?req, "introspect request");
188
189 let resp = self.dispatch(req).await;
190
191 let out = bincode::serialize(&resp)?;
192 stream.write_u32(out.len() as u32).await?;
193 stream.write_all(&out).await?;
194 }
195 }
196
197 async fn dispatch(&self, req: IntrospectRequest) -> IntrospectResponse {
198 match req {
199 IntrospectRequest::ListRules => {
200 let engine = self.state.engine.read().await;
201 let rules = engine.rules()
202 .map(|r| RuleSummary {
203 id: r.id,
204 priority: r.priority,
205 event_kind_tag: format!("{:?}", r.when),
206 action_count: r.then.len(),
207 scope_wildcard: r.scope.is_wildcard(),
208 })
209 .collect();
210 IntrospectResponse::Rules(rules)
211 }
212 IntrospectRequest::GetRule(id) => {
213 let engine = self.state.engine.read().await;
214 let found = engine.rules()
215 .find(|r| r.id == id)
216 .map(|r| Rule::clone(r));
217 IntrospectResponse::Rule(found)
218 }
219 IntrospectRequest::EntropySnapshot => {
220 let obs = self.state.observer.read().await;
221 IntrospectResponse::Entropy {
222 value_bits: obs.shannon_entropy(),
223 sample_size: obs.total(),
224 distinct_kinds: obs.marginals().len(),
225 window_full: obs.current_window() >= obs.window_size(),
226 }
227 }
228 IntrospectRequest::TopCorrelations { n } => {
229 let obs = self.state.observer.read().await;
230 let mut entries: Vec<CorrelationEntry> = obs.cooccurrences().iter()
231 .map(|((a, b), &joint)| CorrelationEntry {
232 a: format!("{:?}", a),
233 b: format!("{:?}", b),
234 joint_count: joint,
235 conditional_prob: obs.conditional_prob(a, b),
236 pmi_bits: obs.pmi(a, b),
237 })
238 .collect();
239 entries.sort_by(|x, y| y.joint_count.cmp(&x.joint_count));
240 entries.truncate(n);
241 IntrospectResponse::Correlations(entries)
242 }
243 IntrospectRequest::Crystals => {
244 let obs = self.state.observer.read().await;
245 let crystals = detect_crystals(&obs, &self.state.params);
246 IntrospectResponse::Crystals(crystals)
247 }
248 IntrospectRequest::CrystalKcl { index } => {
249 let obs = self.state.observer.read().await;
250 let crystals = detect_crystals(&obs, &self.state.params);
251 match crystals.get(index) {
252 Some(c) => IntrospectResponse::Kcl(crate::crystallize::crystal_to_kcl(c)),
253 None => IntrospectResponse::Error(format!("no crystal at index {index}")),
254 }
255 }
256 IntrospectRequest::PromoteCrystal { index } => {
257 let crystals = {
258 let obs = self.state.observer.read().await;
259 detect_crystals(&obs, &self.state.params)
260 };
261 match crystals.get(index) {
262 Some(c) => {
263 let rule = crate::crystallize::crystal_to_rule(c);
264 let snippet = crate::crystallize::crystal_to_kcl(c);
265 let rule_id = rule.id;
266 self.state.engine.write().await.insert(rule);
267 // Persistencia opcional al archivo KCL.
268 if let Some(path) = self.state.rules_out.as_ref() {
269 if let Err(e) = append_kcl_snippet(path, &snippet) {
270 warn!(?e, path = %path.display(), "rules_out append falló");
271 } else {
272 info!(path = %path.display(), %rule_id, "regla persistida a .k");
273 }
274 }
275 // Audit entry
276 self.state.audit.write().await.append(
277 crate::audit::AuditAction::PromoteCrystal {
278 rule_id, crystal: c.clone(),
279 }
280 );
281 IntrospectResponse::Promoted { rule_id, kcl_snippet: snippet }
282 }
283 None => IntrospectResponse::Error(format!("no crystal at index {index}")),
284 }
285 }
286 IntrospectRequest::RemoveRule { id } => {
287 let removed = self.state.engine.write().await.remove(id);
288 if removed {
289 self.state.audit.write().await.append(
290 crate::audit::AuditAction::RemoveRule { rule_id: id }
291 );
292 }
293 IntrospectResponse::Removed(removed)
294 }
295 IntrospectRequest::ListAudit { limit } => {
296 let audit = self.state.audit.read().await;
297 IntrospectResponse::AuditEntries(audit.recent(limit).cloned().collect())
298 }
299 }
300 }
301}
302
303// Cliente helper para tools externos (brainctl).
304pub async fn call(path: &Path, req: IntrospectRequest) -> anyhow::Result<IntrospectResponse> {
305 let mut stream = UnixStream::connect(path).await?;
306 let buf = bincode::serialize(&req)?;
307 stream.write_u32(buf.len() as u32).await?;
308 stream.write_all(&buf).await?;
309
310 let mut len_buf = [0u8; 4];
311 stream.read_exact(&mut len_buf).await?;
312 let len = u32::from_be_bytes(len_buf) as usize;
313 if len > MAX_FRAME {
314 anyhow::bail!("response oversize: {len}");
315 }
316 let mut buf = vec![0u8; len];
317 stream.read_exact(&mut buf).await?;
318 Ok(bincode::deserialize(&buf)?)
319}
320
321/// Consume la lista marginal del observer para humanos. Suprime el detalle
322/// crudo de `EventKind` (ej. payloads largos en BusInvokeOf).
323pub fn marginal_summary(obs: &Observer) -> Vec<(String, u64)> {
324 let mut entries: Vec<(String, u64)> = obs.marginals().iter()
325 .map(|(k, &c)| (format!("{:?}", k), c))
326 .collect();
327 entries.sort_by(|x, y| y.1.cmp(&x.1));
328 entries
329}
330