tawasuyu/arje

1//! brainctl: cliente CLI del introspect API.
2//!
3//! Uso:
4//!   cargo run --example brainctl -p ente-brain -- list-rules
5//!   cargo run --example brainctl -p ente-brain -- entropy
6//!   cargo run --example brainctl -p ente-brain -- top 10
7//!   cargo run --example brainctl -p ente-brain -- crystals
8//!   cargo run --example brainctl -p ente-brain -- crystal-kcl 0
9//!
10//! Path del socket: $ENTE_BRAIN_SOCK o $XDG_RUNTIME_DIR/ente-brain.sock
11
12use ente_brain::introspect::{call, IntrospectRequest, IntrospectResponse};
13use std::path::{Path, PathBuf};
14use tokio::io::{AsyncReadExt, AsyncWriteExt};
15use tokio::net::UnixStream;
16
17fn socket_path() -> PathBuf {
18    if let Ok(p) = std::env::var("ENTE_BRAIN_SOCK") {
19        return p.into();
20    }
21    let runtime = std::env::var("XDG_RUNTIME_DIR")
22        .unwrap_or_else(|_| std::env::var("TMPDIR").unwrap_or_else(|_| "/tmp".into()));
23    format!("{runtime}/ente-brain.sock").into()
24}
25
26#[tokio::main(flavor = "current_thread")]
27async fn main() -> anyhow::Result<()> {
28    let args: Vec<String> = std::env::args().collect();
29    let cmd = args.get(1).map(|s| s.as_str()).unwrap_or("entropy");
30
31    // Comando especial: streaming. Mantiene la conn abierta y lee frames
32    // hasta Ctrl-C o EOF del servidor.
33    if cmd == "stream-audit" || cmd == "stream" {
34        return run_stream_audit(socket_path()).await;
35    }
36
37    let req = match cmd {
38        "list-rules" | "rules" => IntrospectRequest::ListRules,
39        "entropy" => IntrospectRequest::EntropySnapshot,
40        "top" => {
41            let n: usize = args.get(2).and_then(|s| s.parse().ok()).unwrap_or(10);
42            IntrospectRequest::TopCorrelations { n }
43        }
44        "crystals" => IntrospectRequest::Crystals,
45        "crystal-kcl" => {
46            let i: usize = args.get(2).and_then(|s| s.parse().ok()).unwrap_or(0);
47            IntrospectRequest::CrystalKcl { index: i }
48        }
49        "promote" => {
50            let i: usize = args.get(2).and_then(|s| s.parse().ok()).unwrap_or(0);
51            IntrospectRequest::PromoteCrystal { index: i }
52        }
53        "remove" => {
54            let id_s = args.get(2).ok_or_else(|| anyhow::anyhow!("se requiere <ulid>"))?;
55            let id: ulid::Ulid = id_s.parse()?;
56            IntrospectRequest::RemoveRule { id }
57        }
58        "audit" => {
59            let limit: usize = args.get(2).and_then(|s| s.parse().ok()).unwrap_or(20);
60            IntrospectRequest::ListAudit { limit }
61        }
62        "flush-audit" => IntrospectRequest::FlushAudit,
63        "audit-verify" | "verify" => IntrospectRequest::VerifyAudit,
64        "replay" => IntrospectRequest::ReplayAudit,
65        "gc-cas" => IntrospectRequest::GcCas { extra_roots: Vec::new() },
66        "patterns" => IntrospectRequest::PatternCrystals,
67        "reload" => {
68            let path = args.get(2).cloned();
69            IntrospectRequest::ReloadRules { path }
70        }
71        other => {
72            eprintln!("subcomando desconocido: {other}");
73            eprintln!("válidos: list-rules | entropy | top <n> | crystals | crystal-kcl <i> | promote <i> | remove <ulid> | audit <limit> | flush-audit | reload [path]");
74            std::process::exit(2);
75        }
76    };
77
78    let path = socket_path();
79    let resp = call(&path, req).await?;
80    print_response(&resp);
81    Ok(())
82}
83
84fn print_response(r: &IntrospectResponse) {
85    match r {
86        IntrospectResponse::Rules(rs) => {
87            println!("{} reglas vivas:", rs.len());
88            for r in rs {
89                println!("  {} prio={} kind={} actions={} wildcard={}",
90                    r.id, r.priority, r.event_kind_tag, r.action_count, r.scope_wildcard);
91            }
92        }
93        IntrospectResponse::Rule(rule) => match rule {
94            Some(r) => println!("{r:#?}"),
95            None => println!("regla no encontrada"),
96        },
97        IntrospectResponse::Entropy { value_bits, sample_size, distinct_kinds, window_full } => {
98            println!("Shannon entropy : {value_bits:.4} bits");
99            println!("Sample size     : {sample_size}");
100            println!("Distinct kinds  : {distinct_kinds}");
101            println!("Window full     : {window_full}");
102        }
103        IntrospectResponse::Correlations(entries) => {
104            println!("{} pares (top, ordenado por co-ocurrencia):", entries.len());
105            for e in entries {
106                println!("  n={:>4}  P(b|a)={:.3}  PMI={:>6.3}b  {} → {}",
107                    e.joint_count, e.conditional_prob, e.pmi_bits, e.a, e.b);
108            }
109        }
110        IntrospectResponse::Crystals(cs) => {
111            println!("{} cristales detectados:", cs.len());
112            for (i, c) in cs.iter().enumerate() {
113                println!("  [{i}] {:?} → {:?}  P={:.3}  PMI={:.3}b  n={}",
114                    c.antecedent, c.consequent, c.conditional_prob, c.pmi, c.support);
115            }
116        }
117        IntrospectResponse::Kcl(s) => println!("{s}"),
118        IntrospectResponse::Promoted { rule_id, kcl_snippet } => {
119            println!("regla creada: {rule_id}");
120            println!("--- KCL para auditoría / persistencia ---");
121            println!("{kcl_snippet}");
122        }
123        IntrospectResponse::Removed(was_present) => {
124            if *was_present { println!("regla eliminada"); }
125            else { println!("regla no encontrada"); }
126        }
127        IntrospectResponse::AuditEntries(entries) => {
128            println!("{} entries de audit log:", entries.len());
129            for e in entries {
130                let prev = e.prev_sha.map(hex_short).unwrap_or_else(|| "—".into());
131                let sha = hex_short(e.sha);
132                println!("  seq={:>4} t={} prev={} sha={}  {:?}",
133                    e.seq, e.timestamp_ms, prev, sha, e.action);
134            }
135        }
136        IntrospectResponse::Flushed { written, head_sha, total_flushed } => {
137            println!("flushed: {written} entries esta pasada, total acumulado: {total_flushed}");
138            if let Some(sha) = head_sha {
139                println!("head sha: {}", hex_long(*sha));
140            }
141        }
142        IntrospectResponse::Reloaded { count } => {
143            println!("reload OK: {count} reglas activas tras reload");
144        }
145        IntrospectResponse::Replayed(rep) => {
146            if let Some(e) = &rep.error {
147                println!("✗ replay falló: {e}");
148            } else {
149                println!("✓ replay completo — {} actions aplicadas, {} reglas finales",
150                    rep.applied, rep.final_rule_count);
151            }
152        }
153        IntrospectResponse::AuditVerified(rep) => {
154            if let Some(seq) = rep.broken_at_seq {
155                println!("✗ verificación FALLÓ tras seq={seq}");
156                if let Some(e) = &rep.error { println!("  motivo: {e}"); }
157                println!("  entries verificadas: {}", rep.verified);
158            } else {
159                println!("✓ chain verificada — {} entries íntegras", rep.verified);
160                if let Some(g) = rep.genesis_sha { println!("  genesis: {}", hex_long(g)); }
161            }
162        }
163        IntrospectResponse::Patterns(ps) => {
164            println!("{} cristales pattern detectados:", ps.len());
165            for p in ps {
166                match p {
167                    ente_brain::crystallize::PatternCrystal::Burst { kind, count, frequency_per_sec } => {
168                        println!("  burst: {kind:?}  count={count}  freq={frequency_per_sec:.2} Hz");
169                    }
170                    ente_brain::crystallize::PatternCrystal::Silence { kind, last_count, since_secs } => {
171                        println!("  silence: {kind:?}  last_count={last_count}  ausente={since_secs:.1}s");
172                    }
173                }
174            }
175        }
176        IntrospectResponse::GcResult { deleted, freed_bytes } => {
177            println!("CAS gc: {deleted} blobs eliminados, {freed_bytes} bytes liberados");
178        }
179        IntrospectResponse::AuditStreamFrame(_) => {
180            // En modo request/response no debería llegar; solo aparece en
181            // run_stream_audit. Si llega aquí es un bug del servidor.
182            eprintln!("frame de stream recibido fuera de stream-audit (bug)");
183        }
184        IntrospectResponse::Error(e) => eprintln!("error: {e}"),
185    }
186}
187
188fn hex_short(sha: [u8; 32]) -> String {
189    sha[..4].iter().map(|b| format!("{:02x}", b)).collect::<String>() + ".."
190}
191
192fn hex_long(sha: [u8; 32]) -> String {
193    sha.iter().map(|b| format!("{:02x}", b)).collect()
194}
195
196async fn run_stream_audit(path: PathBuf) -> anyhow::Result<()> {
197    let mut stream = UnixStream::connect(&path).await?;
198    let req = IntrospectRequest::StreamAudit;
199    let buf = bincode::serialize(&req)?;
200    stream.write_u32(buf.len() as u32).await?;
201    stream.write_all(&buf).await?;
202    eprintln!("audit stream conectado a {} — Ctrl-C para salir", path.display());
203
204    loop {
205        let mut len_buf = [0u8; 4];
206        if stream.read_exact(&mut len_buf).await.is_err() {
207            eprintln!("\nstream cerrado por el servidor");
208            return Ok(());
209        }
210        let len = u32::from_be_bytes(len_buf) as usize;
211        if len > 4 * 1024 * 1024 { anyhow::bail!("frame oversize"); }
212        let mut buf = vec![0u8; len];
213        stream.read_exact(&mut buf).await?;
214        let resp: IntrospectResponse = bincode::deserialize(&buf)?;
215        match resp {
216            IntrospectResponse::AuditStreamFrame(entry) => {
217                let prev = entry.prev_sha
218                    .map(|s| s[..4].iter().map(|b| format!("{:02x}", b)).collect::<String>() + "..")
219                    .unwrap_or_else(|| "—".into());
220                let sha = entry.sha[..4].iter().map(|b| format!("{:02x}", b))
221                    .collect::<String>() + "..";
222                println!("[stream] seq={} prev={} sha={}  {:?}",
223                    entry.seq, prev, sha, entry.action);
224            }
225            other => {
226                eprintln!("frame no esperado en stream: {other:?}");
227                return Ok(());
228            }
229        }
230    }
231}
232
233#[allow(dead_code)]
234fn _suppress(_: &Path) {} // mantener Path import si compilador se queja