tawasuyu/arje

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}
110
111#[derive(Debug, Serialize, Deserialize)]
112pub enum IntrospectResponse {
113    Rules(Vec<RuleSummary>),
114    Rule(Option<Rule>),
115    Entropy { value_bits: f64, sample_size: u64, distinct_kinds: usize, window_full: bool },
116    Correlations(Vec<CorrelationEntry>),
117    Crystals(Vec<Crystal>),
118    Kcl(String),
119    /// Resultado de PromoteCrystal: id de la regla creada + snippet KCL para
120    /// que el operador lo persista en disco si quiere.
121    Promoted { rule_id: Ulid, kcl_snippet: String },
122    /// Resultado de RemoveRule: true si existía, false si ya no.
123    Removed(bool),
124    /// Entradas del audit log (más recientes al final).
125    AuditEntries(Vec<crate::audit::AuditEntry>),
126    /// Resultado de FlushAudit: cuántas entries se escribieron y SHA del head.
127    Flushed { written: usize, head_sha: Option<[u8; 32]>, total_flushed: u64 },
128    /// Resultado de ReloadRules: número total de reglas tras el reload.
129    Reloaded { count: usize },
130    /// Resultado de VerifyAudit.
131    AuditVerified(crate::audit::VerificationReport),
132    Error(String),
133}
134
135#[derive(Debug, Serialize, Deserialize)]
136pub struct RuleSummary {
137    pub id: Ulid,
138    pub priority: u8,
139    pub event_kind_tag: String,
140    pub action_count: usize,
141    pub scope_wildcard: bool,
142}
143
144#[derive(Debug, Serialize, Deserialize)]
145pub struct CorrelationEntry {
146    pub a: String,
147    pub b: String,
148    pub joint_count: u64,
149    pub conditional_prob: f64,
150    pub pmi_bits: f64,
151}
152
153pub struct IntrospectServer {
154    state: BrainState,
155}
156
157impl IntrospectServer {
158    pub fn new(state: BrainState) -> Self { Self { state } }
159
160    /// Spawn del listener. Devuelve cuando bind() falla; en caso contrario
161    /// corre indefinidamente.
162    pub async fn serve(self, path: &Path) -> anyhow::Result<()> {
163        let _ = std::fs::remove_file(path);
164        if let Some(parent) = path.parent() {
165            let _ = std::fs::create_dir_all(parent);
166        }
167        let listener = UnixListener::bind(path)?;
168        info!(path = %path.display(), "brain introspect escuchando");
169        let arc_self = Arc::new(self);
170        loop {
171            match listener.accept().await {
172                Ok((stream, _)) => {
173                    trace!("introspect conn aceptada");
174                    let me = arc_self.clone();
175                    tokio::spawn(async move {
176                        if let Err(e) = me.handle(stream).await {
177                            warn!(?e, "introspect conn ended");
178                        }
179                    });
180                }
181                Err(e) => {
182                    warn!(?e, "introspect accept failed");
183                    return Ok(());
184                }
185            }
186        }
187    }
188
189    async fn handle(self: Arc<Self>, mut stream: UnixStream) -> anyhow::Result<()> {
190        loop {
191            let mut len_buf = [0u8; 4];
192            if stream.read_exact(&mut len_buf).await.is_err() {
193                return Ok(()); // EOF
194            }
195            let len = u32::from_be_bytes(len_buf) as usize;
196            if len > MAX_FRAME {
197                anyhow::bail!("frame oversize: {len}");
198            }
199            let mut buf = vec![0u8; len];
200            stream.read_exact(&mut buf).await?;
201            let req: IntrospectRequest = bincode::deserialize(&buf)?;
202            debug!(?req, "introspect request");
203
204            let resp = self.dispatch(req).await;
205
206            let out = bincode::serialize(&resp)?;
207            stream.write_u32(out.len() as u32).await?;
208            stream.write_all(&out).await?;
209        }
210    }
211
212    async fn dispatch(&self, req: IntrospectRequest) -> IntrospectResponse {
213        match req {
214            IntrospectRequest::ListRules => {
215                let engine = self.state.engine.read().await;
216                let rules = engine.rules()
217                    .map(|r| RuleSummary {
218                        id: r.id,
219                        priority: r.priority,
220                        event_kind_tag: format!("{:?}", r.when),
221                        action_count: r.then.len(),
222                        scope_wildcard: r.scope.is_wildcard(),
223                    })
224                    .collect();
225                IntrospectResponse::Rules(rules)
226            }
227            IntrospectRequest::GetRule(id) => {
228                let engine = self.state.engine.read().await;
229                let found = engine.rules()
230                    .find(|r| r.id == id)
231                    .map(|r| Rule::clone(r));
232                IntrospectResponse::Rule(found)
233            }
234            IntrospectRequest::EntropySnapshot => {
235                let obs = self.state.observer.read().await;
236                IntrospectResponse::Entropy {
237                    value_bits: obs.shannon_entropy(),
238                    sample_size: obs.total(),
239                    distinct_kinds: obs.marginals().len(),
240                    window_full: obs.current_window() >= obs.window_size(),
241                }
242            }
243            IntrospectRequest::TopCorrelations { n } => {
244                let obs = self.state.observer.read().await;
245                let mut entries: Vec<CorrelationEntry> = obs.cooccurrences().iter()
246                    .map(|((a, b), &joint)| CorrelationEntry {
247                        a: format!("{:?}", a),
248                        b: format!("{:?}", b),
249                        joint_count: joint,
250                        conditional_prob: obs.conditional_prob(a, b),
251                        pmi_bits: obs.pmi(a, b),
252                    })
253                    .collect();
254                entries.sort_by(|x, y| y.joint_count.cmp(&x.joint_count));
255                entries.truncate(n);
256                IntrospectResponse::Correlations(entries)
257            }
258            IntrospectRequest::Crystals => {
259                let obs = self.state.observer.read().await;
260                let crystals = detect_crystals(&obs, &self.state.params);
261                IntrospectResponse::Crystals(crystals)
262            }
263            IntrospectRequest::CrystalKcl { index } => {
264                let obs = self.state.observer.read().await;
265                let crystals = detect_crystals(&obs, &self.state.params);
266                match crystals.get(index) {
267                    Some(c) => IntrospectResponse::Kcl(crate::crystallize::crystal_to_kcl(c)),
268                    None => IntrospectResponse::Error(format!("no crystal at index {index}")),
269                }
270            }
271            IntrospectRequest::PromoteCrystal { index } => {
272                let crystals = {
273                    let obs = self.state.observer.read().await;
274                    detect_crystals(&obs, &self.state.params)
275                };
276                match crystals.get(index) {
277                    Some(c) => {
278                        let rule = crate::crystallize::crystal_to_rule(c);
279                        let snippet = crate::crystallize::crystal_to_kcl(c);
280                        let rule_id = rule.id;
281                        self.state.engine.write().await.insert(rule);
282                        // Persistencia opcional al archivo KCL.
283                        if let Some(path) = self.state.rules_out.as_ref() {
284                            if let Err(e) = append_kcl_snippet(path, &snippet) {
285                                warn!(?e, path = %path.display(), "rules_out append falló");
286                            } else {
287                                info!(path = %path.display(), %rule_id, "regla persistida a .k");
288                            }
289                        }
290                        // Audit entry
291                        self.state.audit.write().await.append(
292                            crate::audit::AuditAction::PromoteCrystal {
293                                rule_id, crystal: c.clone(),
294                            }
295                        );
296                        IntrospectResponse::Promoted { rule_id, kcl_snippet: snippet }
297                    }
298                    None => IntrospectResponse::Error(format!("no crystal at index {index}")),
299                }
300            }
301            IntrospectRequest::RemoveRule { id } => {
302                let removed = self.state.engine.write().await.remove(id);
303                if removed {
304                    self.state.audit.write().await.append(
305                        crate::audit::AuditAction::RemoveRule { rule_id: id }
306                    );
307                }
308                IntrospectResponse::Removed(removed)
309            }
310            IntrospectRequest::ListAudit { limit } => {
311                let audit = self.state.audit.read().await;
312                IntrospectResponse::AuditEntries(audit.recent(limit).cloned().collect())
313            }
314            IntrospectRequest::FlushAudit => {
315                let mut audit = self.state.audit.write().await;
316                match audit.flush_to_cas() {
317                    Ok(written) => IntrospectResponse::Flushed {
318                        written,
319                        head_sha: audit.last_flushed_sha(),
320                        total_flushed: audit.flushed_count(),
321                    },
322                    Err(e) => IntrospectResponse::Error(format!("flush_to_cas: {e}")),
323                }
324            }
325            IntrospectRequest::VerifyAudit => {
326                let head = self.state.audit.read().await.last_flushed_sha();
327                let head = match head {
328                    Some(h) => h,
329                    None => return IntrospectResponse::Error(
330                        "audit log sin entries flushadas — nada que verificar".into()
331                    ),
332                };
333                let report = crate::audit::verify_chain_from_cas(head);
334                IntrospectResponse::AuditVerified(report)
335            }
336            IntrospectRequest::ReloadRules { path } => {
337                // Path explícito gana sobre el rules_out configurado.
338                let resolved = path.map(std::path::PathBuf::from)
339                    .or_else(|| self.state.rules_out.as_ref().map(|p| p.as_path().to_path_buf()));
340                let path = match resolved {
341                    Some(p) => p,
342                    None => return IntrospectResponse::Error(
343                        "ReloadRules sin path y sin rules_out configurado".into()
344                    ),
345                };
346                let rules = match crate::kcl_loader::load_rules_file(&path) {
347                    Ok(r) => r,
348                    Err(e) => return IntrospectResponse::Error(format!("load: {e}")),
349                };
350                // Vaciamos el engine antes de re-cargar — semántica clean-slate.
351                let mut engine = self.state.engine.write().await;
352                *engine = crate::engine::RuleEngine::empty();
353                let count = rules.len();
354                for r in rules { engine.insert(r); }
355                drop(engine);
356                self.state.audit.write().await.append(
357                    crate::audit::AuditAction::LoadRulesFile {
358                        path: path.to_string_lossy().into_owned(),
359                        count,
360                    }
361                );
362                IntrospectResponse::Reloaded { count }
363            }
364        }
365    }
366}
367
368// Cliente helper para tools externos (brainctl).
369pub async fn call(path: &Path, req: IntrospectRequest) -> anyhow::Result<IntrospectResponse> {
370    let mut stream = UnixStream::connect(path).await?;
371    let buf = bincode::serialize(&req)?;
372    stream.write_u32(buf.len() as u32).await?;
373    stream.write_all(&buf).await?;
374
375    let mut len_buf = [0u8; 4];
376    stream.read_exact(&mut len_buf).await?;
377    let len = u32::from_be_bytes(len_buf) as usize;
378    if len > MAX_FRAME {
379        anyhow::bail!("response oversize: {len}");
380    }
381    let mut buf = vec![0u8; len];
382    stream.read_exact(&mut buf).await?;
383    Ok(bincode::deserialize(&buf)?)
384}
385
386/// Consume la lista marginal del observer para humanos. Suprime el detalle
387/// crudo de `EventKind` (ej. payloads largos en BusInvokeOf).
388pub fn marginal_summary(obs: &Observer) -> Vec<(String, u64)> {
389    let mut entries: Vec<(String, u64)> = obs.marginals().iter()
390        .map(|(k, &c)| (format!("{:?}", k), c))
391        .collect();
392    entries.sort_by(|x, y| y.1.cmp(&x.1));
393    entries
394}
395