⎇ c715ee2dee341c730cd2206797b8bec989f7ac5c
crates/init/arje-zero/src/main.rs
Historia
Quién tocó qué
crudo
fix(init): la salida de arje-zero ahora se ve en VGA Y serial
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 return run(cli, true);
87 }
88
89 info!("ente-zero despierta como PID 1");
90 // Eco temprano al ring buffer del kernel: garantiza que el «estoy
91 // vivo» aparezca en TODAS las consolas (VGA + serial), no sólo en
92 // el `/dev/console` apuntado por el último `console=` del cmdline.
93 write_to_kmsg("despierta como PID 1");
94 // Doctrina dura: PID 1 NUNCA puede salir — el kernel haría panic
95 // ("Attempted to kill init") y, con `panic=N` en el cmdline, la
96 // máquina cae en un reboot-loop. Por eso cualquier fallo de arranque
97 // se desvía a una shell de rescate: deja diagnosticar y reparar en
98 // vez de reiniciar a ciegas cada diez segundos.
99 match run(cli, false) {
100 Ok(()) => emergency_shell(
101 "el bucle primordial terminó — el fractal pidió shutdown",
102 ),
103 Err(e) => emergency_shell(&format!("{e:#}")),
104 }
105}
106
107/// Arranque + bucle primordial. En PID 1, cualquier `Err` que devuelva
108/// lo intercepta `main` y lo convierte en shell de rescate.
109fn run(cli: CliArgs, dev_mode: bool) -> anyhow::Result<()> {
110 if !dev_mode {
111 bootstrap_kernel_surface().context("bootstrap kernel surface")?;
112 become_child_subreaper().context("PR_SET_CHILD_SUBREAPER")?;
113 }
114
115 let card = seed::load(dev_mode, cli.restore.as_deref())?;
116
117 // current_thread runtime: ver doctrina al inicio del módulo.
118 let rt = tokio::runtime::Builder::new_current_thread()
119 .enable_io()
120 .enable_time()
121 .build()?;
122
123 rt.block_on(primordial_loop(
124 card, dev_mode,
125 cli.checkpoint, cli.restore, cli.rules, cli.rules_out,
126 cli.audit_head, cli.metrics_addr, cli.brain_half_life,
127 cli.autopromote_secs,
128 ))
129}
130
131/// Último recurso de PID 1: imprime el diagnóstico en la consola y abre
132/// una shell de rescate. **Nunca retorna** — si lo hiciera, el proceso
133/// saldría y el kernel haría panic.
134fn emergency_shell(reason: &str) -> ! {
135 let banner = format!(
136 "\n\n\
137 =============== arje-zero — ARRANQUE FALLIDO ================\n\
138 {reason}\n\
139 ---------------------------------------------------------------\n\
140 Se abre una shell de rescate sobre la consola. Revisá el sistema\n\
141 (p. ej. /ente/seed.card.json y los binarios en /usr/sbin) y\n\
142 reiniciá con `reboot -f`. Salir de la shell la vuelve a abrir.\n\n",
143 );
144 write_to_console(&banner);
145 error!(reason = %reason.replace('\n', " "), "arranque de PID 1 fallido");
146 loop {
147 match spawn_console_shell() {
148 Ok(status) => write_to_console(&format!(
149 "\n[arje-zero] la shell de rescate terminó ({status}) — reabriendo.\n",
150 )),
151 Err(e) => {
152 write_to_console(&format!(
153 "\n[arje-zero] no hay shell de rescate disponible: {e}\n\
154 PID 1 queda en espera pasiva — usá la consola del proveedor.\n",
155 ));
156 loop {
157 std::thread::sleep(Duration::from_secs(3600));
158 }
159 }
160 }
161 }
162}
163
164/// Abre una shell interactiva con stdin/stdout/stderr sobre
165/// `/dev/console` y espera a que termine.
166fn spawn_console_shell() -> std::io::Result<std::process::ExitStatus> {
167 let shell = ["/bin/sh", "/bin/bash", "/usr/bin/sh", "/usr/bin/bash"]
168 .into_iter()
169 .find(|p| std::path::Path::new(p).exists())
170 .ok_or_else(|| {
171 std::io::Error::new(
172 std::io::ErrorKind::NotFound,
173 "ninguna shell en /bin ni /usr/bin",
174 )
175 })?;
176 let console = std::fs::OpenOptions::new()
177 .read(true)
178 .write(true)
179 .open("/dev/console")?;
180 std::process::Command::new(shell)
181 .stdin(console.try_clone()?)
182 .stdout(console.try_clone()?)
183 .stderr(console)
184 .env("HOME", "/root")
185 .env("PATH", "/usr/sbin:/usr/bin:/sbin:/bin")
186 .env("PS1", "arje-rescate# ")
187 .env("TERM", "linux")
188 .status()
189}
190
191/// Escribe un mensaje directo a `/dev/console`, **a `/dev/kmsg`**
192/// (ring buffer del kernel, que se hace eco a todas las consolas
193/// registradas), y a stderr de respaldo. Así el banner se ve en VGA y
194/// serial sin importar cuál sea el último `console=` del cmdline.
195fn write_to_console(msg: &str) {
196 use std::io::Write;
197 if let Ok(mut f) = std::fs::OpenOptions::new().write(true).open("/dev/console") {
198 let _ = f.write_all(msg.as_bytes());
199 }
200 write_to_kmsg(msg);
201 eprint!("{msg}");
202}
203
204/// Escribe un mensaje al `/dev/kmsg` del kernel — éste lo replica a
205/// todas las consolas registradas (`console=` del cmdline). Es el
206/// canal que usa systemd para que sus avisos se vean tanto en la VGA
207/// como en el serial.
208fn write_to_kmsg(msg: &str) {
209 use std::io::Write;
210 let Ok(mut f) = std::fs::OpenOptions::new().write(true).open("/dev/kmsg") else {
211 return;
212 };
213 // `<1>` = ALERT, prioridad alta — aparece incluso con loglevel bajo.
214 for line in msg.lines() {
215 let trimmed = line.trim_end();
216 if !trimmed.is_empty() {
217 let _ = writeln!(f, "<1>arje-zero: {trimmed}");
218 }
219 }
220}
221
222async fn primordial_loop(
223 seed_card: arje_card::EntityCard,
224 dev_mode: bool,
225 checkpoint_path: Option<PathBuf>,
226 restore_path: Option<PathBuf>,
227 rules_path: Option<PathBuf>,
228 rules_out: Option<PathBuf>,
229 audit_head: Option<PathBuf>,
230 metrics_addr: Option<String>,
231 brain_half_life: Option<f64>,
232 autopromote_secs: Option<u64>,
233) -> anyhow::Result<()> {
234 info!(seed_id = %seed_card.id, label = %seed_card.label, "Ente #0 entra al bucle primordial");
235
236 let (graph_tx, mut graph_rx) = mpsc::channel::<GraphEvent>(64);
237 let mut sigchld = spawn_sigchld_stream()?;
238 // Uevents puede fallar en dev (sin CAP_NET_ADMIN). Degradamos a un
239 // canal nunca-listo en lugar de abortar el bucle primordial.
240 let mut uevents = match spawn_uevent_stream() {
241 Ok(rx) => rx,
242 Err(e) => {
243 warn!(?e, "uevents deshabilitados (probablemente falta CAP_NET_ADMIN)");
244 let (_keep_tx, rx) = mpsc::channel::<arje_kernel::UEvent>(1);
245 std::mem::forget(_keep_tx);
246 rx
247 }
248 };
249
250 // Bus interno: listener antes de spawn de hijos para que su Announce
251 // tenga adónde llegar. Su path se inyecta en ENTE_BUS_SOCK por soma.
252 let bus_sock = bus::default_socket_path();
253 let bus_path = bus::spawn_bus(bus_sock, graph_tx.clone())?;
254 arje_soma::set_bus_sock(bus_path.to_string_lossy().into_owned());
255
256 // Brahman protocol: handshake socket + broker compartido.
257 //
258 // Es un canal paralelo al ente-bus, dedicado a módulos "brahman
259 // conscientes" que se presentan con una Card y declaran flujos
260 // tipados. Si el bind falla (socket en uso, FS no escribible),
261 // degradamos a "modo bus-only" — la doctrina de PID 1 no rompe
262 // por subsistemas opcionales.
263 // Contexto operativo del broker: configurable por env var. Útil para
264 // distinguir test/prod/foreground sin recompilar. Sin la var, los
265 // biases per-contexto declarados en las Cards quedan inactivos.
266 let broker_context = std::env::var("BRAHMAN_BROKER_CONTEXT").ok();
267 if let Some(ctx) = &broker_context {
268 info!(context = %ctx, "brahman broker bajo contexto operativo");
269 }
270 let brahman_broker = std::sync::Arc::new(tokio::sync::Mutex::new(
271 brahman_broker::Broker::new(brahman_broker::BrokerConfig {
272 strategy: brahman_broker::MatchStrategy::default(),
273 current_context: broker_context.clone(),
274 }),
275 ));
276
277 // Brahman-net opcional: si BRAHMAN_LISTEN_MULTIADDR está set,
278 // levantamos la malla P2P y la pasamos como ServerConfig.net (Fase
279 // 2 wire) para que cada Card con outputs se anuncie al DHT y
280 // pueda ser descubierta por nodos remotos. Identidad libp2p
281 // persistida en disco vía keypair_store (peer_id estable across
282 // reboots).
283 let brahman_net = setup_brahman_net(dev_mode).await;
284
285 // Política opcional de peers libp2p: allowlist + denylist + hot
286 // reload. Activada si BRAHMAN_PEER_ALLOWLIST o BRAHMAN_PEER_DENYLIST
287 // están set. Sin ninguna, modo totalmente abierto (Fase 3 sin
288 // restricción adicional). El watcher se queda vivo en background
289 // observando los archivos para hot reload.
290 let (brahman_policy, _policy_watcher) = setup_brahman_policy();
291
292 // Si tenemos AMBOS net y policy, attachamos: el deny de la
293 // policy se proyecta al block_list del swarm para rechazar
294 // conexiones ANTES del Noise handshake (más eficiente que
295 // rechazar en el handshake brahman). Cada hot-reload de la
296 // policy también re-sincroniza vía diff.
297 if let (Some(net), Some(policy)) = (&brahman_net, &brahman_policy) {
298 policy.attach_to_net(net.clone());
299 let (allow, deny) = policy.sizes();
300 info!(
301 allow = ?allow,
302 deny = deny,
303 "policy attached al swarm — denies enforcedeados a nivel libp2p"
304 );
305 }
306
307 let brahman_sock = brahman_handshake::transport::default_socket_path();
308 match brahman_handshake::server::Server::bind(
309 &brahman_sock,
310 brahman_handshake::server::ServerConfig {
311 init_attached: true,
312 broker: Some(brahman_broker.clone()),
313 net: brahman_net.clone(),
314 policy: brahman_policy.clone(),
315 },
316 ) {
317 Ok(server) => {
318 info!(socket = %brahman_sock.display(), "brahman handshake escuchando (Unix)");
319 // Si hay malla P2P, además del Unix accept loop levantamos
320 // el accept loop libp2p sobre el mismo Server compartido.
321 // Las sesiones locales y remotas conviven en las mismas
322 // tablas (sessions, push_table, broker).
323 let server = std::sync::Arc::new(server);
324 if let Some(net) = brahman_net.clone() {
325 let s_libp2p = server.clone();
326 let n_libp2p = net.clone();
327 tokio::spawn(async move {
328 if let Err(e) = brahman_handshake::network::run_libp2p_accept_loop(
329 s_libp2p, n_libp2p,
330 )
331 .await
332 {
333 warn!(?e, "brahman handshake libp2p accept loop cayó");
334 }
335 });
336 info!(
337 "brahman handshake escuchando también vía libp2p (peer_id {})",
338 net.peer_id
339 );
340 }
341 // Unix accept loop: usa Arc<Server> en lugar del consume
342 // de run() para coexistir con el libp2p accept loop.
343 let s_unix = server.clone();
344 tokio::spawn(async move {
345 loop {
346 match s_unix.accept_one().await {
347 Ok(session) => {
348 tokio::spawn(async move {
349 if let Err(e) = session.handle().await {
350 warn!(?e, "session Unix terminó con error");
351 }
352 });
353 }
354 Err(e) => {
355 warn!(?e, "brahman handshake accept_one Unix falló");
356 break;
357 }
358 }
359 }
360 });
361 }
362 Err(e) => {
363 warn!(?e, socket = %brahman_sock.display(), "brahman handshake deshabilitado");
364 }
365 }
366
367 // Brahman admin: socket separado para snapshots de estado (sesiones +
368 // matches del broker). Misma política de degradación grácil.
369 let admin_sock = brahman_admin::transport::default_socket_path();
370 match brahman_admin::server::AdminServer::bind(
371 &admin_sock,
372 brahman_broker.clone(),
373 brahman_admin::server::AdminConfig {
374 init_attached: true,
375 current_context: broker_context.clone(),
376 },
377 ) {
378 Ok(admin) => {
379 info!(socket = %admin_sock.display(), "brahman admin escuchando");
380 tokio::spawn(async move {
381 if let Err(e) = admin.run().await {
382 warn!(?e, "brahman admin server cayó");
383 }
384 });
385 }
386 Err(e) => {
387 warn!(?e, socket = %admin_sock.display(), "brahman admin deshabilitado");
388 }
389 }
390
391 let mut graph = EnteGraph::new(seed_card);
392 graph.instantiate_seed_dependencies(&graph_tx).await?;
393
394 // Cerebro: BrainState compartido + servidor de introspección.
395 // Window de 1024 eventos — suficiente para correlaciones interesantes
396 // sin gastar memoria de PID 1. En dev bajamos el umbral de cristalización
397 // para que el demo (pocos eventos) produzca cristales observables.
398 let mut brain = if dev_mode {
399 // Umbrales relajados para que el demo (pocos eventos) produzca
400 // cristales observables. Con P(b|a) normalizada a [0,1], los
401 // valores típicos en muestras pequeñas son 0.2-0.5.
402 BrainState::with_params(1024, arje_brain::CrystallizationParams {
403 min_support: 2,
404 min_conditional_prob: 0.3,
405 min_pmi: 1.0,
406 })
407 } else {
408 BrainState::new(1024)
409 };
410 if let Some(out_path) = rules_out {
411 brain = brain.with_rules_out(out_path);
412 }
413 if let Some(hl) = brain_half_life {
414 let mut obs = brain.observer.write().await;
415 // Reemplazar con un observer nuevo que tenga half-life. Estado
416 // anterior (vacío en este punto) descartado.
417 *obs = arje_brain::Observer::new(1024).with_half_life(hl);
418 info!(hl_secs = hl, "observer con time-decay activo");
419 }
420 if let Some(secs) = autopromote_secs {
421 arje_brain::spawn_autopromote_loop(
422 brain.clone(),
423 arje_brain::AutopromoteParams {
424 interval_secs: secs,
425 threshold: brain.params, // mismo threshold que crystals manuales
426 },
427 );
428 }
429
430 // Brain restore: si hay --restore <path>, cargamos el snapshot adjunto
431 // <path>.brain.json. Counters preservados across reboots.
432 if let Some(rpath) = &restore_path {
433 let brain_path = rpath.with_extension("brain.json");
434 if brain_path.exists() {
435 match read_brain_snapshot(&brain_path) {
436 Ok(snap) => {
437 let total = snap.total;
438 let kinds = snap.marginal.len();
439 let restored = arje_brain::Observer::from_snapshot(snap);
440 *brain.observer.write().await = restored;
441 info!(
442 path = %brain_path.display(),
443 total, kinds,
444 "brain snapshot restaurado"
445 );
446 }
447 Err(e) => warn!(?e, path = %brain_path.display(), "brain snapshot read falló"),
448 }
449 }
450 }
451 // Si --audit-head, configuramos el head pointer y arrancamos auto-flush.
452 if let Some(head_path) = audit_head {
453 // Re-creamos el AuditLog con head pointer.
454 let new_audit = arje_brain::audit::AuditLog::new()
455 .with_head_pointer(head_path);
456 *brain.audit.write().await = new_audit;
457 spawn_audit_auto_flush(brain.clone());
458 }
459
460 // Carga inicial de reglas desde JSON/JSONL si --rules path proporcionado.
461 if let Some(path) = &rules_path {
462 match arje_brain::load_rules_file(path) {
463 Ok(rules) => {
464 let mut engine = brain.engine.write().await;
465 for r in rules {
466 engine.insert(r);
467 }
468 info!(count = engine.len(), path = %path.display(), "reglas cargadas");
469 }
470 Err(e) => warn!(?e, path = %path.display(), "carga de reglas falló"),
471 }
472 }
473
474 // Endpoint Prometheus opcional. En dev por defecto en 127.0.0.1:9911 si
475 // el flag no se pasó.
476 let metrics_addr = metrics_addr.or_else(|| {
477 if dev_mode { Some("127.0.0.1:9911".to_string()) } else { None }
478 });
479 if let Some(addr_s) = metrics_addr {
480 match addr_s.parse::<std::net::SocketAddr>() {
481 Ok(addr) => {
482 let s = brain.clone();
483 tokio::spawn(async move {
484 if let Err(e) = arje_brain::serve_metrics(s, addr).await {
485 warn!(?e, "metrics server cayó");
486 }
487 });
488 }
489 Err(e) => warn!(?e, addr = %addr_s, "metrics-addr inválido"),
490 }
491 }
492 spawn_brain_introspect(brain.clone());
493 let brain_sink = brain_glue::GraphSink {
494 graph_tx: graph_tx.clone(),
495 // Spawns auto-disparados desde reglas usan la identidad de la Semilla
496 // (único Ente con Capability::Spawn por construcción).
497 requester: graph.seed_id(),
498 };
499
500 // Demo automático del forwarding (sólo dev, sólo si el binario existe).
501 if dev_mode && std::path::Path::new("target/debug/ente-echo").exists() {
502 spawn_echo_smoke_test(bus_path.clone());
503 }
504
505 // En dev mode no tenemos hijos por defecto y el bucle se quedaría inerte.
506 let dev_exit = if dev_mode {
507 Some(tokio::time::sleep(Duration::from_secs(4)))
508 } else {
509 None
510 };
511 tokio::pin!(dev_exit);
512
513 // GC de capability grants expirados — corre cada 10 segundos.
514 let mut grant_purge = tokio::time::interval(Duration::from_secs(10));
515 grant_purge.tick().await; // descartar primer tick inmediato
516
517 loop {
518 tokio::select! {
519 biased;
520
521 Some(_) = sigchld.recv() => {
522 reap_until_empty(&mut graph, &graph_tx).await;
523 }
524
525 Some(uevt) = uevents.recv() => {
526 graph.on_uevent(uevt, &graph_tx).await;
527 }
528
529 Some(evt) = graph_rx.recv() => {
530 // Cerebro observa antes que el grafo mute. Snapshot del
531 // SubjectInfo se hace contra el estado pre-mutación.
532 feed_brain(&brain, &brain_sink, &graph, &evt).await;
533 if dispatch_graph_event(&mut graph, evt, &graph_tx, &checkpoint_path, &brain).await {
534 return Ok(());
535 }
536 }
537
538 _ = grant_purge.tick() => {
539 let n = graph.purge_expired_grants();
540 if n > 0 {
541 info!(purged = n, active = graph.active_grants_count(), "GC capability grants");
542 }
543 }
544
545 _ = async { dev_exit.as_mut().as_pin_mut().unwrap().await }, if dev_mode => {
546 info!("dev mode: timer expirado, cerrando bucle primordial");
547 let _ = graph_tx.send(GraphEvent::Shutdown {
548 reason: ShutdownReason::SeedRequested,
549 }).await;
550 }
551 }
552 }
553}
554
555/// Devuelve `true` si el bucle primordial debe terminar.
556async fn dispatch_graph_event(
557 graph: &mut EnteGraph,
558 evt: GraphEvent,
559 tx: &mpsc::Sender<GraphEvent>,
560 checkpoint: &Option<PathBuf>,
561 brain: &BrainState,
562) -> bool {
563 match evt {
564 GraphEvent::EnteDied { id, status } => {
565 graph.on_death(id, status, tx).await;
566 }
567 GraphEvent::CapabilityRequested { from, cap, reply } => {
568 graph.mediate_capability(from, cap, reply).await;
569 }
570 GraphEvent::SpawnRequest { card, requester } => {
571 if let Err(e) = graph.authorize_and_spawn(card, requester).await {
572 warn!(?e, "spawn request error");
573 }
574 }
575 GraphEvent::BusRequest { peer, from, request, outbound, reply } => {
576 graph.on_bus_request(peer, from, request, outbound, reply).await;
577 }
578 GraphEvent::BusResponse { seq, response } => {
579 graph.on_bus_response(seq, response).await;
580 }
581 GraphEvent::BusConnClosed { ente_id } => {
582 graph.on_bus_conn_closed(ente_id).await;
583 }
584 GraphEvent::Shutdown { reason } => {
585 warn!(?reason, "shutdown del fractal");
586 if let Some(path) = checkpoint.as_ref() {
587 // Snapshot del grafo
588 let snap = graph.snapshot();
589 match snap.write(path) {
590 Ok(()) => info!(path = %path.display(), entes = snap.entes.len(), "snapshot fractal persistido"),
591 Err(e) => warn!(?e, "snapshot write falló"),
592 }
593 // Snapshot del cerebro (observer state) en archivo adjunto
594 let brain_path = path.with_extension("brain.json");
595 let obs_snap = brain.observer.write().await.snapshot();
596 match write_brain_snapshot(&brain_path, &obs_snap) {
597 Ok(()) => info!(
598 path = %brain_path.display(),
599 total = obs_snap.total,
600 kinds = obs_snap.marginal.len(),
601 "snapshot brain persistido"
602 ),
603 Err(e) => warn!(?e, "brain snapshot write falló"),
604 }
605 }
606 graph.cascade_shutdown().await;
607 return true;
608 }
609 }
610 false
611}
612
613async fn reap_until_empty(graph: &mut EnteGraph, tx: &mpsc::Sender<GraphEvent>) {
614 loop {
615 match waitpid(None, Some(WaitPidFlag::WNOHANG)) {
616 Ok(WaitStatus::StillAlive) => return,
617 Ok(WaitStatus::Exited(pid, code)) => {
618 emit_death(graph, tx, pid, ExitStatus::Exit(code)).await;
619 }
620 Ok(WaitStatus::Signaled(pid, sig, _core)) => {
621 emit_death(graph, tx, pid, ExitStatus::Killed(sig)).await;
622 }
623 Ok(_) => continue, // Stopped/Continued — irrelevantes
624 Err(Errno::ECHILD) => return,
625 Err(e) => {
626 error!(?e, "waitpid fallo no recuperable en bucle de reaping");
627 return;
628 }
629 }
630 }
631}
632
633async fn emit_death(
634 graph: &EnteGraph,
635 tx: &mpsc::Sender<GraphEvent>,
636 pid: Pid,
637 status: ExitStatus,
638) {
639 let id = match graph.lookup_pid(pid) {
640 Some(id) => id,
641 None => {
642 // Proceso adoptado (subreaper): no está en nuestro grafo.
643 info!(?pid, ?status, "huérfano cosechado (no en grafo)");
644 return;
645 }
646 };
647 let _ = tx.send(GraphEvent::EnteDied { id, status }).await;
648}
649
650fn spawn_echo_smoke_test(bus_path: PathBuf) {
651 tokio::spawn(async move {
652 tokio::time::sleep(Duration::from_millis(300)).await;
653 match arje_bus::BusClient::connect(&bus_path).await {
654 Ok(mut client) => {
655 let req = arje_bus::BusRequest::Invoke {
656 cap: arje_echo::echo_capability(),
657 blob: b"hola fractal forwardeado".to_vec(),
658 };
659 match client.call(req).await {
660 Ok(arje_bus::BusResponse::Invoked { result }) => {
661 info!(echo = %String::from_utf8_lossy(&result), "Invoke ECHO round-trip OK");
662 }
663 Ok(other) => warn!(?other, "Invoke ECHO respuesta inesperada"),
664 Err(e) => warn!(?e, "Invoke ECHO falló"),
665 }
666 }
667 Err(e) => warn!(?e, "no se pudo conectar al bus para test"),
668 }
669 });
670}
671
672fn write_brain_snapshot(path: &std::path::Path, snap: &arje_brain::observer::ObserverSnapshot) -> anyhow::Result<()> {
673 let bytes = serde_json::to_vec_pretty(snap)?;
674 if let Some(parent) = path.parent() { let _ = std::fs::create_dir_all(parent); }
675 let tmp = path.with_extension("tmp");
676 std::fs::write(&tmp, &bytes)?;
677 std::fs::rename(&tmp, path)?;
678 Ok(())
679}
680
681fn read_brain_snapshot(path: &std::path::Path) -> anyhow::Result<arje_brain::observer::ObserverSnapshot> {
682 let bytes = std::fs::read(path)?;
683 let snap: arje_brain::observer::ObserverSnapshot = serde_json::from_slice(&bytes)?;
684 Ok(snap)
685}
686
687fn init_tracing() {
688 use tracing_subscriber::{fmt, EnvFilter};
689 let filter = EnvFilter::try_from_default_env()
690 .unwrap_or_else(|_| EnvFilter::new("arje_zero=debug,info"));
691 fmt().with_env_filter(filter).with_target(true).init();
692}
693
694fn brain_introspect_path() -> PathBuf {
695 if let Ok(p) = std::env::var("ENTE_BRAIN_SOCK") {
696 return p.into();
697 }
698 let runtime = std::env::var("XDG_RUNTIME_DIR")
699 .unwrap_or_else(|_| std::env::var("TMPDIR").unwrap_or_else(|_| "/tmp".into()));
700 format!("{runtime}/ente-brain.sock").into()
701}
702
703/// Auto-flush del audit log a CAS cada 10 segundos. Ejecuta best-effort:
704/// si el flush falla lo logeamos pero no abortamos. La integridad del log
705/// queda garantizada por su hash chain — re-flushar es idempotente.
706fn spawn_audit_auto_flush(state: BrainState) {
707 tokio::spawn(async move {
708 let mut tick = tokio::time::interval(std::time::Duration::from_secs(10));
709 tick.tick().await; // descartar primer tick inmediato
710 loop {
711 tick.tick().await;
712 let mut audit = state.audit.write().await;
713 match audit.flush_to_cas() {
714 Ok(0) => {} // nada nuevo
715 Ok(n) => info!(written = n, total = audit.flushed_count(), "audit auto-flush"),
716 Err(e) => warn!(?e, "audit auto-flush falló"),
717 }
718 }
719 });
720}
721
722fn spawn_brain_introspect(state: BrainState) {
723 let path = brain_introspect_path();
724 tokio::spawn(async move {
725 let server = IntrospectServer::new(state);
726 if let Err(e) = server.serve(&path).await {
727 warn!(?e, "introspect server cayó");
728 }
729 });
730}
731
732/// Registra el evento en el observer y dispatcha cualquier regla matched.
733/// Para reglas Sequence: pasamos los últimos N eventos del observer como
734/// history al engine.
735async fn feed_brain(
736 brain: &BrainState,
737 sink: &brain_glue::GraphSink,
738 graph: &EnteGraph,
739 evt: &GraphEvent,
740) {
741 let Some((kind, subj)) = brain_glue::graph_event_to_brain(evt, graph) else { return };
742 let history: Vec<arje_brain::TimedEvent> = {
743 let mut obs = brain.observer.write().await;
744 obs.record(kind.clone());
745 // Snapshot de los últimos 16 eventos — suficiente para cualquier
746 // Sequence pattern razonable. Clone hace una sola alocación.
747 obs.recent(16).cloned().collect()
748 };
749 let rules = {
750 let engine = brain.engine.read().await;
751 engine.dispatch(&kind, &subj, &history)
752 };
753 if !rules.is_empty() {
754 arje_brain::dispatch_actions(&rules, sink).await;
755 }
756}
757
758/// Inicializa la malla `brahman-net` opcional. Activa sólo si
759/// `BRAHMAN_LISTEN_MULTIADDR` está set. Identidad libp2p persistente
760/// vía `keypair_store`. Bootstrap del DHT vía `BRAHMAN_BOOTSTRAP_PEERS`
761/// (lista coma-separada de multiaddrs, opcional).
762///
763/// Toda fase de setup degrada grácilmente: si la keypair no carga,
764/// si el listen falla, si bootstrap dial falla — loggea y devuelve
765/// `None`. El Init sigue funcionando en modo Unix-only.
766async fn setup_brahman_net(
767 dev_mode: bool,
768) -> Option<std::sync::Arc<brahman_net::BrahmanNet>> {
769 let listen_addr = match std::env::var("BRAHMAN_LISTEN_MULTIADDR") {
770 Ok(s) if !s.is_empty() => s,
771 _ => {
772 tracing::debug!(
773 "brahman-net deshabilitado (sin BRAHMAN_LISTEN_MULTIADDR)"
774 );
775 return None;
776 }
777 };
778
779 let multiaddr: brahman_net::Multiaddr = match listen_addr.parse() {
780 Ok(m) => m,
781 Err(e) => {
782 warn!(addr = %listen_addr, ?e, "BRAHMAN_LISTEN_MULTIADDR inválido — net deshabilitado");
783 return None;
784 }
785 };
786
787 let keypair_path = keypair_store::default_path(dev_mode);
788 let (keypair, loaded) = match keypair_store::load_or_generate(&keypair_path) {
789 Ok(kp) => kp,
790 Err(e) => {
791 warn!(path = %keypair_path.display(), ?e, "no pude cargar/generar keypair libp2p — net deshabilitado");
792 return None;
793 }
794 };
795 info!(
796 path = %keypair_path.display(),
797 peer_id = %keypair.public().to_peer_id(),
798 loaded = loaded,
799 "identidad libp2p {}",
800 if loaded { "cargada" } else { "generada y persistida" }
801 );
802
803 let net = match brahman_net::BrahmanNet::with_keypair(keypair) {
804 Ok(n) => std::sync::Arc::new(n),
805 Err(e) => {
806 warn!(?e, "BrahmanNet::with_keypair falló — net deshabilitado");
807 return None;
808 }
809 };
810
811 let actual = net.listen(multiaddr).await;
812 info!(addr = %actual, peer_id = %net.peer_id, "brahman-net escuchando");
813
814 // Bootstrap opcional: dial-ar a peers conocidos para entrar al
815 // DHT. Sin bootstrap, el nodo arranca aislado hasta que alguien
816 // se conecte a él.
817 if let Ok(bootstrap) = std::env::var("BRAHMAN_BOOTSTRAP_PEERS") {
818 let mut dialed = 0usize;
819 for entry in bootstrap.split(',').filter(|s| !s.is_empty()) {
820 match entry.parse::<brahman_net::Multiaddr>() {
821 Ok(addr) => {
822 net.dial(addr.clone());
823 dialed += 1;
824 tracing::debug!(peer = %addr, "dial bootstrap");
825 }
826 Err(e) => {
827 warn!(entry = %entry, ?e, "bootstrap multiaddr inválido — saltado");
828 }
829 }
830 }
831 if dialed > 0 {
832 info!(count = dialed, "bootstrap peers dial-eados");
833 }
834 }
835
836 Some(net)
837}
838
839/// Carga la política de peers libp2p desde los archivos apuntados por
840/// `BRAHMAN_PEER_ALLOWLIST` y/o `BRAHMAN_PEER_DENYLIST`, y arranca un
841/// watcher para hot reload sobre cualquier cambio.
842///
843/// - Sin ninguna env: `(None, None)` → modo totalmente abierto.
844/// - Con cualquiera (o ambas) set: política activa + watcher vivo.
845/// - Si los archivos fallan al cargar: degrada a `(None, None)`,
846/// loggea, NO rompe el bucle primordial (doctrina PID 1).
847///
848/// Devuelve la política y el `JoinHandle` del watcher (que el caller
849/// debe mantener para que el thread no se aborte). Si no hay paths,
850/// el watcher es un no-op que termina inmediato.
851fn setup_brahman_policy() -> (
852 Option<brahman_handshake::peer_policy::PeerPolicy>,
853 Option<std::thread::JoinHandle<()>>,
854) {
855 let allow_path = std::env::var("BRAHMAN_PEER_ALLOWLIST")
856 .ok()
857 .filter(|s| !s.is_empty());
858 let deny_path = std::env::var("BRAHMAN_PEER_DENYLIST")
859 .ok()
860 .filter(|s| !s.is_empty());
861
862 if allow_path.is_none() && deny_path.is_none() {
863 tracing::debug!(
864 "BRAHMAN_PEER_ALLOWLIST y BRAHMAN_PEER_DENYLIST no set — modo abierto (todo peer pasa)"
865 );
866 return (None, None);
867 }
868
869 let allow_pb = allow_path.as_deref().map(std::path::Path::new);
870 let deny_pb = deny_path.as_deref().map(std::path::Path::new);
871
872 let policy = match brahman_handshake::peer_policy::PeerPolicy::from_files(allow_pb, deny_pb) {
873 Ok(p) => p,
874 Err(e) => {
875 warn!(
876 ?e,
877 allow = ?allow_path,
878 deny = ?deny_path,
879 "policy de peers inválida — degradando a modo abierto (sin restricción)"
880 );
881 return (None, None);
882 }
883 };
884
885 let (allow_count, deny_count) = policy.sizes();
886 info!(
887 allow = ?allow_count,
888 deny = deny_count,
889 allow_path = ?allow_path,
890 deny_path = ?deny_path,
891 "policy de peers libp2p cargada"
892 );
893
894 // Spawn watcher para hot reload. Errores aquí no son fatales —
895 // tendrías política sin reload, que es razonable.
896 let watcher = match policy.spawn_watcher() {
897 Ok(h) => Some(h),
898 Err(e) => {
899 warn!(?e, "policy watcher no se pudo crear — hot reload deshabilitado");
900 None
901 }
902 };
903
904 (Some(policy), watcher)
905}