⎇ 762bf95dfd770cc8bf51f21605bc2cc6cc88ea76
crates/init/arje-zero/src/main.rs
Historia
Quién tocó qué
crudo
refactor(naming): A1 — ente→arje, vista→revista, pluma→fana
1//! Ente #0 — el primer Ente. PID 1 del fractal.
2//!
3//! Reglas no negociables:
4//! 1. NUNCA lógica de servicio aquí. Sólo: leer Semilla, cosechar zombis,
5//! mediar capacidades, propagar eventos.
6//! 2. Single-threaded. Cualquier paralelismo se delega a Entes worker.
7//! Un panic en un thread de PID 1 = kernel panic.
8//! 3. Errores de hijos son *eventos* en `graph_tx`, no `Result` propagado.
9//!
10//! Este archivo es sólo wireup. La lógica vive en:
11//! - `seed` : construcción/restauración de la Tarjeta Semilla
12//! - `bus` : listener Unix + auth via SO_PEERCRED
13//! - `graph::*` : estado del fractal (lifecycle, topology, shutdown,
14//! bus_mediator, devices, capabilities)
15//! - `events` : tipos de eventos del bucle primordial
16//! - crates externos del workspace para CAS, soma, wasm, snapshot, kernel.
17
18mod brain_glue;
19mod bus;
20mod events;
21mod graph;
22mod keypair_store;
23mod seed;
24
25use anyhow::Context;
26use arje_brain::{BrainState, IntrospectServer};
27use arje_kernel::{become_child_subreaper, bootstrap_kernel_surface, spawn_sigchld_stream, spawn_uevent_stream};
28use events::{ExitStatus, GraphEvent, ShutdownReason};
29use graph::EnteGraph;
30use nix::errno::Errno;
31use nix::sys::wait::{waitpid, WaitPidFlag, WaitStatus};
32use nix::unistd::{getpid, Pid};
33use std::path::PathBuf;
34use std::time::Duration;
35use tokio::sync::mpsc;
36use tracing::{error, info, warn};
37
38struct CliArgs {
39 checkpoint: Option<PathBuf>,
40 restore: Option<PathBuf>,
41 rules: Option<PathBuf>,
42 rules_out: Option<PathBuf>,
43 audit_head: Option<PathBuf>,
44 metrics_addr: Option<String>,
45 brain_half_life: Option<f64>,
46 autopromote_secs: Option<u64>,
47}
48
49fn parse_args() -> CliArgs {
50 let mut args = std::env::args().skip(1);
51 let mut checkpoint = None;
52 let mut restore = None;
53 let mut rules = None;
54 let mut rules_out = None;
55 let mut audit_head = None;
56 let mut metrics_addr = None;
57 let mut brain_half_life = None;
58 let mut autopromote_secs = None;
59 while let Some(a) = args.next() {
60 match a.as_str() {
61 "--checkpoint" => checkpoint = args.next().map(PathBuf::from),
62 "--restore" => restore = args.next().map(PathBuf::from),
63 "--rules" => rules = args.next().map(PathBuf::from),
64 "--rules-out" => rules_out = args.next().map(PathBuf::from),
65 "--audit-head" => audit_head = args.next().map(PathBuf::from),
66 "--metrics-addr" => metrics_addr = args.next(),
67 "--brain-half-life" => brain_half_life = args.next().and_then(|s| s.parse().ok()),
68 "--autopromote-secs" => autopromote_secs = args.next().and_then(|s| s.parse().ok()),
69 other => warn!(arg = %other, "argumento desconocido, ignorado"),
70 }
71 }
72 CliArgs {
73 checkpoint, restore, rules, rules_out, audit_head,
74 metrics_addr, brain_half_life, autopromote_secs,
75 }
76}
77
78fn main() -> anyhow::Result<()> {
79 init_tracing();
80 let cli = parse_args();
81 let pid = getpid();
82 let dev_mode = pid != Pid::from_raw(1);
83
84 if dev_mode {
85 warn!(?pid, "ente-zero corriendo en DEV MODE (no PID 1) — kernel surface no se monta");
86 } else {
87 info!("ente-zero despierta como PID 1");
88 bootstrap_kernel_surface().context("bootstrap kernel surface")?;
89 become_child_subreaper().context("PR_SET_CHILD_SUBREAPER")?;
90 }
91
92 let card = seed::load(dev_mode, cli.restore.as_deref())?;
93
94 // current_thread runtime: ver doctrina al inicio del módulo.
95 let rt = tokio::runtime::Builder::new_current_thread()
96 .enable_io()
97 .enable_time()
98 .build()?;
99
100 rt.block_on(primordial_loop(
101 card, dev_mode,
102 cli.checkpoint, cli.restore, cli.rules, cli.rules_out,
103 cli.audit_head, cli.metrics_addr, cli.brain_half_life,
104 cli.autopromote_secs,
105 ))
106}
107
108async fn primordial_loop(
109 seed_card: arje_card::EntityCard,
110 dev_mode: bool,
111 checkpoint_path: Option<PathBuf>,
112 restore_path: Option<PathBuf>,
113 rules_path: Option<PathBuf>,
114 rules_out: Option<PathBuf>,
115 audit_head: Option<PathBuf>,
116 metrics_addr: Option<String>,
117 brain_half_life: Option<f64>,
118 autopromote_secs: Option<u64>,
119) -> anyhow::Result<()> {
120 info!(seed_id = %seed_card.id, label = %seed_card.label, "Ente #0 entra al bucle primordial");
121
122 let (graph_tx, mut graph_rx) = mpsc::channel::<GraphEvent>(64);
123 let mut sigchld = spawn_sigchld_stream()?;
124 // Uevents puede fallar en dev (sin CAP_NET_ADMIN). Degradamos a un
125 // canal nunca-listo en lugar de abortar el bucle primordial.
126 let mut uevents = match spawn_uevent_stream() {
127 Ok(rx) => rx,
128 Err(e) => {
129 warn!(?e, "uevents deshabilitados (probablemente falta CAP_NET_ADMIN)");
130 let (_keep_tx, rx) = mpsc::channel::<arje_kernel::UEvent>(1);
131 std::mem::forget(_keep_tx);
132 rx
133 }
134 };
135
136 // Bus interno: listener antes de spawn de hijos para que su Announce
137 // tenga adónde llegar. Su path se inyecta en ENTE_BUS_SOCK por soma.
138 let bus_sock = bus::default_socket_path();
139 let bus_path = bus::spawn_bus(bus_sock, graph_tx.clone())?;
140 arje_soma::set_bus_sock(bus_path.to_string_lossy().into_owned());
141
142 // Brahman protocol: handshake socket + broker compartido.
143 //
144 // Es un canal paralelo al ente-bus, dedicado a módulos "brahman
145 // conscientes" que se presentan con una Card y declaran flujos
146 // tipados. Si el bind falla (socket en uso, FS no escribible),
147 // degradamos a "modo bus-only" — la doctrina de PID 1 no rompe
148 // por subsistemas opcionales.
149 // Contexto operativo del broker: configurable por env var. Útil para
150 // distinguir test/prod/foreground sin recompilar. Sin la var, los
151 // biases per-contexto declarados en las Cards quedan inactivos.
152 let broker_context = std::env::var("BRAHMAN_BROKER_CONTEXT").ok();
153 if let Some(ctx) = &broker_context {
154 info!(context = %ctx, "brahman broker bajo contexto operativo");
155 }
156 let brahman_broker = std::sync::Arc::new(tokio::sync::Mutex::new(
157 brahman_broker::Broker::new(brahman_broker::BrokerConfig {
158 strategy: brahman_broker::MatchStrategy::default(),
159 current_context: broker_context.clone(),
160 }),
161 ));
162
163 // Brahman-net opcional: si BRAHMAN_LISTEN_MULTIADDR está set,
164 // levantamos la malla P2P y la pasamos como ServerConfig.net (Fase
165 // 2 wire) para que cada Card con outputs se anuncie al DHT y
166 // pueda ser descubierta por nodos remotos. Identidad libp2p
167 // persistida en disco vía keypair_store (peer_id estable across
168 // reboots).
169 let brahman_net = setup_brahman_net(dev_mode).await;
170
171 // Política opcional de peers libp2p: allowlist + denylist + hot
172 // reload. Activada si BRAHMAN_PEER_ALLOWLIST o BRAHMAN_PEER_DENYLIST
173 // están set. Sin ninguna, modo totalmente abierto (Fase 3 sin
174 // restricción adicional). El watcher se queda vivo en background
175 // observando los archivos para hot reload.
176 let (brahman_policy, _policy_watcher) = setup_brahman_policy();
177
178 // Si tenemos AMBOS net y policy, attachamos: el deny de la
179 // policy se proyecta al block_list del swarm para rechazar
180 // conexiones ANTES del Noise handshake (más eficiente que
181 // rechazar en el handshake brahman). Cada hot-reload de la
182 // policy también re-sincroniza vía diff.
183 if let (Some(net), Some(policy)) = (&brahman_net, &brahman_policy) {
184 policy.attach_to_net(net.clone());
185 let (allow, deny) = policy.sizes();
186 info!(
187 allow = ?allow,
188 deny = deny,
189 "policy attached al swarm — denies enforcedeados a nivel libp2p"
190 );
191 }
192
193 let brahman_sock = brahman_handshake::transport::default_socket_path();
194 match brahman_handshake::server::Server::bind(
195 &brahman_sock,
196 brahman_handshake::server::ServerConfig {
197 init_attached: true,
198 broker: Some(brahman_broker.clone()),
199 net: brahman_net.clone(),
200 policy: brahman_policy.clone(),
201 },
202 ) {
203 Ok(server) => {
204 info!(socket = %brahman_sock.display(), "brahman handshake escuchando (Unix)");
205 // Si hay malla P2P, además del Unix accept loop levantamos
206 // el accept loop libp2p sobre el mismo Server compartido.
207 // Las sesiones locales y remotas conviven en las mismas
208 // tablas (sessions, push_table, broker).
209 let server = std::sync::Arc::new(server);
210 if let Some(net) = brahman_net.clone() {
211 let s_libp2p = server.clone();
212 let n_libp2p = net.clone();
213 tokio::spawn(async move {
214 if let Err(e) = brahman_handshake::network::run_libp2p_accept_loop(
215 s_libp2p, n_libp2p,
216 )
217 .await
218 {
219 warn!(?e, "brahman handshake libp2p accept loop cayó");
220 }
221 });
222 info!(
223 "brahman handshake escuchando también vía libp2p (peer_id {})",
224 net.peer_id
225 );
226 }
227 // Unix accept loop: usa Arc<Server> en lugar del consume
228 // de run() para coexistir con el libp2p accept loop.
229 let s_unix = server.clone();
230 tokio::spawn(async move {
231 loop {
232 match s_unix.accept_one().await {
233 Ok(session) => {
234 tokio::spawn(async move {
235 if let Err(e) = session.handle().await {
236 warn!(?e, "session Unix terminó con error");
237 }
238 });
239 }
240 Err(e) => {
241 warn!(?e, "brahman handshake accept_one Unix falló");
242 break;
243 }
244 }
245 }
246 });
247 }
248 Err(e) => {
249 warn!(?e, socket = %brahman_sock.display(), "brahman handshake deshabilitado");
250 }
251 }
252
253 // Brahman admin: socket separado para snapshots de estado (sesiones +
254 // matches del broker). Misma política de degradación grácil.
255 let admin_sock = brahman_admin::transport::default_socket_path();
256 match brahman_admin::server::AdminServer::bind(
257 &admin_sock,
258 brahman_broker.clone(),
259 brahman_admin::server::AdminConfig {
260 init_attached: true,
261 current_context: broker_context.clone(),
262 },
263 ) {
264 Ok(admin) => {
265 info!(socket = %admin_sock.display(), "brahman admin escuchando");
266 tokio::spawn(async move {
267 if let Err(e) = admin.run().await {
268 warn!(?e, "brahman admin server cayó");
269 }
270 });
271 }
272 Err(e) => {
273 warn!(?e, socket = %admin_sock.display(), "brahman admin deshabilitado");
274 }
275 }
276
277 let mut graph = EnteGraph::new(seed_card);
278 graph.instantiate_seed_dependencies(&graph_tx).await?;
279
280 // Cerebro: BrainState compartido + servidor de introspección.
281 // Window de 1024 eventos — suficiente para correlaciones interesantes
282 // sin gastar memoria de PID 1. En dev bajamos el umbral de cristalización
283 // para que el demo (pocos eventos) produzca cristales observables.
284 let mut brain = if dev_mode {
285 // Umbrales relajados para que el demo (pocos eventos) produzca
286 // cristales observables. Con P(b|a) normalizada a [0,1], los
287 // valores típicos en muestras pequeñas son 0.2-0.5.
288 BrainState::with_params(1024, arje_brain::CrystallizationParams {
289 min_support: 2,
290 min_conditional_prob: 0.3,
291 min_pmi: 1.0,
292 })
293 } else {
294 BrainState::new(1024)
295 };
296 if let Some(out_path) = rules_out {
297 brain = brain.with_rules_out(out_path);
298 }
299 if let Some(hl) = brain_half_life {
300 let mut obs = brain.observer.write().await;
301 // Reemplazar con un observer nuevo que tenga half-life. Estado
302 // anterior (vacío en este punto) descartado.
303 *obs = arje_brain::Observer::new(1024).with_half_life(hl);
304 info!(hl_secs = hl, "observer con time-decay activo");
305 }
306 if let Some(secs) = autopromote_secs {
307 arje_brain::spawn_autopromote_loop(
308 brain.clone(),
309 arje_brain::AutopromoteParams {
310 interval_secs: secs,
311 threshold: brain.params, // mismo threshold que crystals manuales
312 },
313 );
314 }
315
316 // Brain restore: si hay --restore <path>, cargamos el snapshot adjunto
317 // <path>.brain.json. Counters preservados across reboots.
318 if let Some(rpath) = &restore_path {
319 let brain_path = rpath.with_extension("brain.json");
320 if brain_path.exists() {
321 match read_brain_snapshot(&brain_path) {
322 Ok(snap) => {
323 let total = snap.total;
324 let kinds = snap.marginal.len();
325 let restored = arje_brain::Observer::from_snapshot(snap);
326 *brain.observer.write().await = restored;
327 info!(
328 path = %brain_path.display(),
329 total, kinds,
330 "brain snapshot restaurado"
331 );
332 }
333 Err(e) => warn!(?e, path = %brain_path.display(), "brain snapshot read falló"),
334 }
335 }
336 }
337 // Si --audit-head, configuramos el head pointer y arrancamos auto-flush.
338 if let Some(head_path) = audit_head {
339 // Re-creamos el AuditLog con head pointer.
340 let new_audit = arje_brain::audit::AuditLog::new()
341 .with_head_pointer(head_path);
342 *brain.audit.write().await = new_audit;
343 spawn_audit_auto_flush(brain.clone());
344 }
345
346 // Carga inicial de reglas desde JSON/JSONL si --rules path proporcionado.
347 if let Some(path) = &rules_path {
348 match arje_brain::load_rules_file(path) {
349 Ok(rules) => {
350 let mut engine = brain.engine.write().await;
351 for r in rules {
352 engine.insert(r);
353 }
354 info!(count = engine.len(), path = %path.display(), "reglas cargadas");
355 }
356 Err(e) => warn!(?e, path = %path.display(), "carga de reglas falló"),
357 }
358 }
359
360 // Endpoint Prometheus opcional. En dev por defecto en 127.0.0.1:9911 si
361 // el flag no se pasó.
362 let metrics_addr = metrics_addr.or_else(|| {
363 if dev_mode { Some("127.0.0.1:9911".to_string()) } else { None }
364 });
365 if let Some(addr_s) = metrics_addr {
366 match addr_s.parse::<std::net::SocketAddr>() {
367 Ok(addr) => {
368 let s = brain.clone();
369 tokio::spawn(async move {
370 if let Err(e) = arje_brain::serve_metrics(s, addr).await {
371 warn!(?e, "metrics server cayó");
372 }
373 });
374 }
375 Err(e) => warn!(?e, addr = %addr_s, "metrics-addr inválido"),
376 }
377 }
378 spawn_brain_introspect(brain.clone());
379 let brain_sink = brain_glue::GraphSink {
380 graph_tx: graph_tx.clone(),
381 // Spawns auto-disparados desde reglas usan la identidad de la Semilla
382 // (único Ente con Capability::Spawn por construcción).
383 requester: graph.seed_id(),
384 };
385
386 // Demo automático del forwarding (sólo dev, sólo si el binario existe).
387 if dev_mode && std::path::Path::new("target/debug/ente-echo").exists() {
388 spawn_echo_smoke_test(bus_path.clone());
389 }
390
391 // En dev mode no tenemos hijos por defecto y el bucle se quedaría inerte.
392 let dev_exit = if dev_mode {
393 Some(tokio::time::sleep(Duration::from_secs(4)))
394 } else {
395 None
396 };
397 tokio::pin!(dev_exit);
398
399 // GC de capability grants expirados — corre cada 10 segundos.
400 let mut grant_purge = tokio::time::interval(Duration::from_secs(10));
401 grant_purge.tick().await; // descartar primer tick inmediato
402
403 loop {
404 tokio::select! {
405 biased;
406
407 Some(_) = sigchld.recv() => {
408 reap_until_empty(&mut graph, &graph_tx).await;
409 }
410
411 Some(uevt) = uevents.recv() => {
412 graph.on_uevent(uevt, &graph_tx).await;
413 }
414
415 Some(evt) = graph_rx.recv() => {
416 // Cerebro observa antes que el grafo mute. Snapshot del
417 // SubjectInfo se hace contra el estado pre-mutación.
418 feed_brain(&brain, &brain_sink, &graph, &evt).await;
419 if dispatch_graph_event(&mut graph, evt, &graph_tx, &checkpoint_path, &brain).await {
420 return Ok(());
421 }
422 }
423
424 _ = grant_purge.tick() => {
425 let n = graph.purge_expired_grants();
426 if n > 0 {
427 info!(purged = n, active = graph.active_grants_count(), "GC capability grants");
428 }
429 }
430
431 _ = async { dev_exit.as_mut().as_pin_mut().unwrap().await }, if dev_mode => {
432 info!("dev mode: timer expirado, cerrando bucle primordial");
433 let _ = graph_tx.send(GraphEvent::Shutdown {
434 reason: ShutdownReason::SeedRequested,
435 }).await;
436 }
437 }
438 }
439}
440
441/// Devuelve `true` si el bucle primordial debe terminar.
442async fn dispatch_graph_event(
443 graph: &mut EnteGraph,
444 evt: GraphEvent,
445 tx: &mpsc::Sender<GraphEvent>,
446 checkpoint: &Option<PathBuf>,
447 brain: &BrainState,
448) -> bool {
449 match evt {
450 GraphEvent::EnteDied { id, status } => {
451 graph.on_death(id, status, tx).await;
452 }
453 GraphEvent::CapabilityRequested { from, cap, reply } => {
454 graph.mediate_capability(from, cap, reply).await;
455 }
456 GraphEvent::SpawnRequest { card, requester } => {
457 if let Err(e) = graph.authorize_and_spawn(card, requester).await {
458 warn!(?e, "spawn request error");
459 }
460 }
461 GraphEvent::BusRequest { peer, from, request, outbound, reply } => {
462 graph.on_bus_request(peer, from, request, outbound, reply).await;
463 }
464 GraphEvent::BusResponse { seq, response } => {
465 graph.on_bus_response(seq, response).await;
466 }
467 GraphEvent::BusConnClosed { ente_id } => {
468 graph.on_bus_conn_closed(ente_id).await;
469 }
470 GraphEvent::Shutdown { reason } => {
471 warn!(?reason, "shutdown del fractal");
472 if let Some(path) = checkpoint.as_ref() {
473 // Snapshot del grafo
474 let snap = graph.snapshot();
475 match snap.write(path) {
476 Ok(()) => info!(path = %path.display(), entes = snap.entes.len(), "snapshot fractal persistido"),
477 Err(e) => warn!(?e, "snapshot write falló"),
478 }
479 // Snapshot del cerebro (observer state) en archivo adjunto
480 let brain_path = path.with_extension("brain.json");
481 let obs_snap = brain.observer.write().await.snapshot();
482 match write_brain_snapshot(&brain_path, &obs_snap) {
483 Ok(()) => info!(
484 path = %brain_path.display(),
485 total = obs_snap.total,
486 kinds = obs_snap.marginal.len(),
487 "snapshot brain persistido"
488 ),
489 Err(e) => warn!(?e, "brain snapshot write falló"),
490 }
491 }
492 graph.cascade_shutdown().await;
493 return true;
494 }
495 }
496 false
497}
498
499async fn reap_until_empty(graph: &mut EnteGraph, tx: &mpsc::Sender<GraphEvent>) {
500 loop {
501 match waitpid(None, Some(WaitPidFlag::WNOHANG)) {
502 Ok(WaitStatus::StillAlive) => return,
503 Ok(WaitStatus::Exited(pid, code)) => {
504 emit_death(graph, tx, pid, ExitStatus::Exit(code)).await;
505 }
506 Ok(WaitStatus::Signaled(pid, sig, _core)) => {
507 emit_death(graph, tx, pid, ExitStatus::Killed(sig)).await;
508 }
509 Ok(_) => continue, // Stopped/Continued — irrelevantes
510 Err(Errno::ECHILD) => return,
511 Err(e) => {
512 error!(?e, "waitpid fallo no recuperable en bucle de reaping");
513 return;
514 }
515 }
516 }
517}
518
519async fn emit_death(
520 graph: &EnteGraph,
521 tx: &mpsc::Sender<GraphEvent>,
522 pid: Pid,
523 status: ExitStatus,
524) {
525 let id = match graph.lookup_pid(pid) {
526 Some(id) => id,
527 None => {
528 // Proceso adoptado (subreaper): no está en nuestro grafo.
529 info!(?pid, ?status, "huérfano cosechado (no en grafo)");
530 return;
531 }
532 };
533 let _ = tx.send(GraphEvent::EnteDied { id, status }).await;
534}
535
536fn spawn_echo_smoke_test(bus_path: PathBuf) {
537 tokio::spawn(async move {
538 tokio::time::sleep(Duration::from_millis(300)).await;
539 match arje_bus::BusClient::connect(&bus_path).await {
540 Ok(mut client) => {
541 let req = arje_bus::BusRequest::Invoke {
542 cap: arje_echo::echo_capability(),
543 blob: b"hola fractal forwardeado".to_vec(),
544 };
545 match client.call(req).await {
546 Ok(arje_bus::BusResponse::Invoked { result }) => {
547 info!(echo = %String::from_utf8_lossy(&result), "Invoke ECHO round-trip OK");
548 }
549 Ok(other) => warn!(?other, "Invoke ECHO respuesta inesperada"),
550 Err(e) => warn!(?e, "Invoke ECHO falló"),
551 }
552 }
553 Err(e) => warn!(?e, "no se pudo conectar al bus para test"),
554 }
555 });
556}
557
558fn write_brain_snapshot(path: &std::path::Path, snap: &arje_brain::observer::ObserverSnapshot) -> anyhow::Result<()> {
559 let bytes = serde_json::to_vec_pretty(snap)?;
560 if let Some(parent) = path.parent() { let _ = std::fs::create_dir_all(parent); }
561 let tmp = path.with_extension("tmp");
562 std::fs::write(&tmp, &bytes)?;
563 std::fs::rename(&tmp, path)?;
564 Ok(())
565}
566
567fn read_brain_snapshot(path: &std::path::Path) -> anyhow::Result<arje_brain::observer::ObserverSnapshot> {
568 let bytes = std::fs::read(path)?;
569 let snap: arje_brain::observer::ObserverSnapshot = serde_json::from_slice(&bytes)?;
570 Ok(snap)
571}
572
573fn init_tracing() {
574 use tracing_subscriber::{fmt, EnvFilter};
575 let filter = EnvFilter::try_from_default_env()
576 .unwrap_or_else(|_| EnvFilter::new("arje_zero=debug,info"));
577 fmt().with_env_filter(filter).with_target(true).init();
578}
579
580fn brain_introspect_path() -> PathBuf {
581 if let Ok(p) = std::env::var("ENTE_BRAIN_SOCK") {
582 return p.into();
583 }
584 let runtime = std::env::var("XDG_RUNTIME_DIR")
585 .unwrap_or_else(|_| std::env::var("TMPDIR").unwrap_or_else(|_| "/tmp".into()));
586 format!("{runtime}/ente-brain.sock").into()
587}
588
589/// Auto-flush del audit log a CAS cada 10 segundos. Ejecuta best-effort:
590/// si el flush falla lo logeamos pero no abortamos. La integridad del log
591/// queda garantizada por su hash chain — re-flushar es idempotente.
592fn spawn_audit_auto_flush(state: BrainState) {
593 tokio::spawn(async move {
594 let mut tick = tokio::time::interval(std::time::Duration::from_secs(10));
595 tick.tick().await; // descartar primer tick inmediato
596 loop {
597 tick.tick().await;
598 let mut audit = state.audit.write().await;
599 match audit.flush_to_cas() {
600 Ok(0) => {} // nada nuevo
601 Ok(n) => info!(written = n, total = audit.flushed_count(), "audit auto-flush"),
602 Err(e) => warn!(?e, "audit auto-flush falló"),
603 }
604 }
605 });
606}
607
608fn spawn_brain_introspect(state: BrainState) {
609 let path = brain_introspect_path();
610 tokio::spawn(async move {
611 let server = IntrospectServer::new(state);
612 if let Err(e) = server.serve(&path).await {
613 warn!(?e, "introspect server cayó");
614 }
615 });
616}
617
618/// Registra el evento en el observer y dispatcha cualquier regla matched.
619/// Para reglas Sequence: pasamos los últimos N eventos del observer como
620/// history al engine.
621async fn feed_brain(
622 brain: &BrainState,
623 sink: &brain_glue::GraphSink,
624 graph: &EnteGraph,
625 evt: &GraphEvent,
626) {
627 let Some((kind, subj)) = brain_glue::graph_event_to_brain(evt, graph) else { return };
628 let history: Vec<arje_brain::TimedEvent> = {
629 let mut obs = brain.observer.write().await;
630 obs.record(kind.clone());
631 // Snapshot de los últimos 16 eventos — suficiente para cualquier
632 // Sequence pattern razonable. Clone hace una sola alocación.
633 obs.recent(16).cloned().collect()
634 };
635 let rules = {
636 let engine = brain.engine.read().await;
637 engine.dispatch(&kind, &subj, &history)
638 };
639 if !rules.is_empty() {
640 arje_brain::dispatch_actions(&rules, sink).await;
641 }
642}
643
644/// Inicializa la malla `brahman-net` opcional. Activa sólo si
645/// `BRAHMAN_LISTEN_MULTIADDR` está set. Identidad libp2p persistente
646/// vía `keypair_store`. Bootstrap del DHT vía `BRAHMAN_BOOTSTRAP_PEERS`
647/// (lista coma-separada de multiaddrs, opcional).
648///
649/// Toda fase de setup degrada grácilmente: si la keypair no carga,
650/// si el listen falla, si bootstrap dial falla — loggea y devuelve
651/// `None`. El Init sigue funcionando en modo Unix-only.
652async fn setup_brahman_net(
653 dev_mode: bool,
654) -> Option<std::sync::Arc<brahman_net::BrahmanNet>> {
655 let listen_addr = match std::env::var("BRAHMAN_LISTEN_MULTIADDR") {
656 Ok(s) if !s.is_empty() => s,
657 _ => {
658 tracing::debug!(
659 "brahman-net deshabilitado (sin BRAHMAN_LISTEN_MULTIADDR)"
660 );
661 return None;
662 }
663 };
664
665 let multiaddr: brahman_net::Multiaddr = match listen_addr.parse() {
666 Ok(m) => m,
667 Err(e) => {
668 warn!(addr = %listen_addr, ?e, "BRAHMAN_LISTEN_MULTIADDR inválido — net deshabilitado");
669 return None;
670 }
671 };
672
673 let keypair_path = keypair_store::default_path(dev_mode);
674 let (keypair, loaded) = match keypair_store::load_or_generate(&keypair_path) {
675 Ok(kp) => kp,
676 Err(e) => {
677 warn!(path = %keypair_path.display(), ?e, "no pude cargar/generar keypair libp2p — net deshabilitado");
678 return None;
679 }
680 };
681 info!(
682 path = %keypair_path.display(),
683 peer_id = %keypair.public().to_peer_id(),
684 loaded = loaded,
685 "identidad libp2p {}",
686 if loaded { "cargada" } else { "generada y persistida" }
687 );
688
689 let net = match brahman_net::BrahmanNet::with_keypair(keypair) {
690 Ok(n) => std::sync::Arc::new(n),
691 Err(e) => {
692 warn!(?e, "BrahmanNet::with_keypair falló — net deshabilitado");
693 return None;
694 }
695 };
696
697 let actual = net.listen(multiaddr).await;
698 info!(addr = %actual, peer_id = %net.peer_id, "brahman-net escuchando");
699
700 // Bootstrap opcional: dial-ar a peers conocidos para entrar al
701 // DHT. Sin bootstrap, el nodo arranca aislado hasta que alguien
702 // se conecte a él.
703 if let Ok(bootstrap) = std::env::var("BRAHMAN_BOOTSTRAP_PEERS") {
704 let mut dialed = 0usize;
705 for entry in bootstrap.split(',').filter(|s| !s.is_empty()) {
706 match entry.parse::<brahman_net::Multiaddr>() {
707 Ok(addr) => {
708 net.dial(addr.clone());
709 dialed += 1;
710 tracing::debug!(peer = %addr, "dial bootstrap");
711 }
712 Err(e) => {
713 warn!(entry = %entry, ?e, "bootstrap multiaddr inválido — saltado");
714 }
715 }
716 }
717 if dialed > 0 {
718 info!(count = dialed, "bootstrap peers dial-eados");
719 }
720 }
721
722 Some(net)
723}
724
725/// Carga la política de peers libp2p desde los archivos apuntados por
726/// `BRAHMAN_PEER_ALLOWLIST` y/o `BRAHMAN_PEER_DENYLIST`, y arranca un
727/// watcher para hot reload sobre cualquier cambio.
728///
729/// - Sin ninguna env: `(None, None)` → modo totalmente abierto.
730/// - Con cualquiera (o ambas) set: política activa + watcher vivo.
731/// - Si los archivos fallan al cargar: degrada a `(None, None)`,
732/// loggea, NO rompe el bucle primordial (doctrina PID 1).
733///
734/// Devuelve la política y el `JoinHandle` del watcher (que el caller
735/// debe mantener para que el thread no se aborte). Si no hay paths,
736/// el watcher es un no-op que termina inmediato.
737fn setup_brahman_policy() -> (
738 Option<brahman_handshake::peer_policy::PeerPolicy>,
739 Option<std::thread::JoinHandle<()>>,
740) {
741 let allow_path = std::env::var("BRAHMAN_PEER_ALLOWLIST")
742 .ok()
743 .filter(|s| !s.is_empty());
744 let deny_path = std::env::var("BRAHMAN_PEER_DENYLIST")
745 .ok()
746 .filter(|s| !s.is_empty());
747
748 if allow_path.is_none() && deny_path.is_none() {
749 tracing::debug!(
750 "BRAHMAN_PEER_ALLOWLIST y BRAHMAN_PEER_DENYLIST no set — modo abierto (todo peer pasa)"
751 );
752 return (None, None);
753 }
754
755 let allow_pb = allow_path.as_deref().map(std::path::Path::new);
756 let deny_pb = deny_path.as_deref().map(std::path::Path::new);
757
758 let policy = match brahman_handshake::peer_policy::PeerPolicy::from_files(allow_pb, deny_pb) {
759 Ok(p) => p,
760 Err(e) => {
761 warn!(
762 ?e,
763 allow = ?allow_path,
764 deny = ?deny_path,
765 "policy de peers inválida — degradando a modo abierto (sin restricción)"
766 );
767 return (None, None);
768 }
769 };
770
771 let (allow_count, deny_count) = policy.sizes();
772 info!(
773 allow = ?allow_count,
774 deny = deny_count,
775 allow_path = ?allow_path,
776 deny_path = ?deny_path,
777 "policy de peers libp2p cargada"
778 );
779
780 // Spawn watcher para hot reload. Errores aquí no son fatales —
781 // tendrías política sin reload, que es razonable.
782 let watcher = match policy.spawn_watcher() {
783 Ok(h) => Some(h),
784 Err(e) => {
785 warn!(?e, "policy watcher no se pudo crear — hot reload deshabilitado");
786 None
787 }
788 };
789
790 (Some(policy), watcher)
791}