⎇ d7b4886164df5ef048a7c8c0d735fb83555c7990
crates/core/ente-zero/src/main.rs
Historia
Quién tocó qué
crudo
feat(broker): priority contexts — biases per-contexto operativo
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 seed;
23
24use anyhow::Context;
25use ente_brain::{BrainState, IntrospectServer};
26use ente_kernel::{become_child_subreaper, bootstrap_kernel_surface, spawn_sigchld_stream, spawn_uevent_stream};
27use events::{ExitStatus, GraphEvent, ShutdownReason};
28use graph::EnteGraph;
29use nix::errno::Errno;
30use nix::sys::wait::{waitpid, WaitPidFlag, WaitStatus};
31use nix::unistd::{getpid, Pid};
32use std::path::PathBuf;
33use std::time::Duration;
34use tokio::sync::mpsc;
35use tracing::{error, info, warn};
36
37struct CliArgs {
38 checkpoint: Option<PathBuf>,
39 restore: Option<PathBuf>,
40 rules: Option<PathBuf>,
41 rules_out: Option<PathBuf>,
42 audit_head: Option<PathBuf>,
43 metrics_addr: Option<String>,
44 brain_half_life: Option<f64>,
45 autopromote_secs: Option<u64>,
46}
47
48fn parse_args() -> CliArgs {
49 let mut args = std::env::args().skip(1);
50 let mut checkpoint = None;
51 let mut restore = None;
52 let mut rules = None;
53 let mut rules_out = None;
54 let mut audit_head = None;
55 let mut metrics_addr = None;
56 let mut brain_half_life = None;
57 let mut autopromote_secs = None;
58 while let Some(a) = args.next() {
59 match a.as_str() {
60 "--checkpoint" => checkpoint = args.next().map(PathBuf::from),
61 "--restore" => restore = args.next().map(PathBuf::from),
62 "--rules" => rules = args.next().map(PathBuf::from),
63 "--rules-out" => rules_out = args.next().map(PathBuf::from),
64 "--audit-head" => audit_head = args.next().map(PathBuf::from),
65 "--metrics-addr" => metrics_addr = args.next(),
66 "--brain-half-life" => brain_half_life = args.next().and_then(|s| s.parse().ok()),
67 "--autopromote-secs" => autopromote_secs = args.next().and_then(|s| s.parse().ok()),
68 other => warn!(arg = %other, "argumento desconocido, ignorado"),
69 }
70 }
71 CliArgs {
72 checkpoint, restore, rules, rules_out, audit_head,
73 metrics_addr, brain_half_life, autopromote_secs,
74 }
75}
76
77fn main() -> anyhow::Result<()> {
78 init_tracing();
79 let cli = parse_args();
80 let pid = getpid();
81 let dev_mode = pid != Pid::from_raw(1);
82
83 if dev_mode {
84 warn!(?pid, "ente-zero corriendo en DEV MODE (no PID 1) — kernel surface no se monta");
85 } else {
86 info!("ente-zero despierta como PID 1");
87 bootstrap_kernel_surface().context("bootstrap kernel surface")?;
88 become_child_subreaper().context("PR_SET_CHILD_SUBREAPER")?;
89 }
90
91 let card = seed::load(dev_mode, cli.restore.as_deref())?;
92
93 // current_thread runtime: ver doctrina al inicio del módulo.
94 let rt = tokio::runtime::Builder::new_current_thread()
95 .enable_io()
96 .enable_time()
97 .build()?;
98
99 rt.block_on(primordial_loop(
100 card, dev_mode,
101 cli.checkpoint, cli.restore, cli.rules, cli.rules_out,
102 cli.audit_head, cli.metrics_addr, cli.brain_half_life,
103 cli.autopromote_secs,
104 ))
105}
106
107async fn primordial_loop(
108 seed_card: ente_card::EntityCard,
109 dev_mode: bool,
110 checkpoint_path: Option<PathBuf>,
111 restore_path: Option<PathBuf>,
112 rules_path: Option<PathBuf>,
113 rules_out: Option<PathBuf>,
114 audit_head: Option<PathBuf>,
115 metrics_addr: Option<String>,
116 brain_half_life: Option<f64>,
117 autopromote_secs: Option<u64>,
118) -> anyhow::Result<()> {
119 info!(seed_id = %seed_card.id, label = %seed_card.label, "Ente #0 entra al bucle primordial");
120
121 let (graph_tx, mut graph_rx) = mpsc::channel::<GraphEvent>(64);
122 let mut sigchld = spawn_sigchld_stream()?;
123 // Uevents puede fallar en dev (sin CAP_NET_ADMIN). Degradamos a un
124 // canal nunca-listo en lugar de abortar el bucle primordial.
125 let mut uevents = match spawn_uevent_stream() {
126 Ok(rx) => rx,
127 Err(e) => {
128 warn!(?e, "uevents deshabilitados (probablemente falta CAP_NET_ADMIN)");
129 let (_keep_tx, rx) = mpsc::channel::<ente_kernel::UEvent>(1);
130 std::mem::forget(_keep_tx);
131 rx
132 }
133 };
134
135 // Bus interno: listener antes de spawn de hijos para que su Announce
136 // tenga adónde llegar. Su path se inyecta en ENTE_BUS_SOCK por soma.
137 let bus_sock = bus::default_socket_path();
138 let bus_path = bus::spawn_bus(bus_sock, graph_tx.clone())?;
139 ente_soma::set_bus_sock(bus_path.to_string_lossy().into_owned());
140
141 // Brahman protocol: handshake socket + broker compartido.
142 //
143 // Es un canal paralelo al ente-bus, dedicado a módulos "brahman
144 // conscientes" que se presentan con una Card y declaran flujos
145 // tipados. Si el bind falla (socket en uso, FS no escribible),
146 // degradamos a "modo bus-only" — la doctrina de PID 1 no rompe
147 // por subsistemas opcionales.
148 // Contexto operativo del broker: configurable por env var. Útil para
149 // distinguir test/prod/foreground sin recompilar. Sin la var, los
150 // biases per-contexto declarados en las Cards quedan inactivos.
151 let broker_context = std::env::var("BRAHMAN_BROKER_CONTEXT").ok();
152 if let Some(ctx) = &broker_context {
153 info!(context = %ctx, "brahman broker bajo contexto operativo");
154 }
155 let brahman_broker = std::sync::Arc::new(tokio::sync::Mutex::new(
156 brahman_broker::Broker::new(brahman_broker::BrokerConfig {
157 strategy: brahman_broker::MatchStrategy::default(),
158 current_context: broker_context.clone(),
159 }),
160 ));
161 let brahman_sock = brahman_handshake::transport::default_socket_path();
162 match brahman_handshake::server::Server::bind(
163 &brahman_sock,
164 brahman_handshake::server::ServerConfig {
165 init_attached: true,
166 broker: Some(brahman_broker.clone()),
167 },
168 ) {
169 Ok(server) => {
170 info!(socket = %brahman_sock.display(), "brahman handshake escuchando");
171 tokio::spawn(async move {
172 if let Err(e) = server.run().await {
173 warn!(?e, "brahman handshake server cayó");
174 }
175 });
176 }
177 Err(e) => {
178 warn!(?e, socket = %brahman_sock.display(), "brahman handshake deshabilitado");
179 }
180 }
181
182 // Brahman admin: socket separado para snapshots de estado (sesiones +
183 // matches del broker). Misma política de degradación grácil.
184 let admin_sock = brahman_admin::transport::default_socket_path();
185 match brahman_admin::server::AdminServer::bind(
186 &admin_sock,
187 brahman_broker.clone(),
188 brahman_admin::server::AdminConfig {
189 init_attached: true,
190 current_context: broker_context.clone(),
191 },
192 ) {
193 Ok(admin) => {
194 info!(socket = %admin_sock.display(), "brahman admin escuchando");
195 tokio::spawn(async move {
196 if let Err(e) = admin.run().await {
197 warn!(?e, "brahman admin server cayó");
198 }
199 });
200 }
201 Err(e) => {
202 warn!(?e, socket = %admin_sock.display(), "brahman admin deshabilitado");
203 }
204 }
205
206 let mut graph = EnteGraph::new(seed_card);
207 graph.instantiate_seed_dependencies(&graph_tx).await?;
208
209 // Cerebro: BrainState compartido + servidor de introspección.
210 // Window de 1024 eventos — suficiente para correlaciones interesantes
211 // sin gastar memoria de PID 1. En dev bajamos el umbral de cristalización
212 // para que el demo (pocos eventos) produzca cristales observables.
213 let mut brain = if dev_mode {
214 // Umbrales relajados para que el demo (pocos eventos) produzca
215 // cristales observables. Con P(b|a) normalizada a [0,1], los
216 // valores típicos en muestras pequeñas son 0.2-0.5.
217 BrainState::with_params(1024, ente_brain::CrystallizationParams {
218 min_support: 2,
219 min_conditional_prob: 0.3,
220 min_pmi: 1.0,
221 })
222 } else {
223 BrainState::new(1024)
224 };
225 if let Some(out_path) = rules_out {
226 brain = brain.with_rules_out(out_path);
227 }
228 if let Some(hl) = brain_half_life {
229 let mut obs = brain.observer.write().await;
230 // Reemplazar con un observer nuevo que tenga half-life. Estado
231 // anterior (vacío en este punto) descartado.
232 *obs = ente_brain::Observer::new(1024).with_half_life(hl);
233 info!(hl_secs = hl, "observer con time-decay activo");
234 }
235 if let Some(secs) = autopromote_secs {
236 ente_brain::spawn_autopromote_loop(
237 brain.clone(),
238 ente_brain::AutopromoteParams {
239 interval_secs: secs,
240 threshold: brain.params, // mismo threshold que crystals manuales
241 },
242 );
243 }
244
245 // Brain restore: si hay --restore <path>, cargamos el snapshot adjunto
246 // <path>.brain.json. Counters preservados across reboots.
247 if let Some(rpath) = &restore_path {
248 let brain_path = rpath.with_extension("brain.json");
249 if brain_path.exists() {
250 match read_brain_snapshot(&brain_path) {
251 Ok(snap) => {
252 let total = snap.total;
253 let kinds = snap.marginal.len();
254 let restored = ente_brain::Observer::from_snapshot(snap);
255 *brain.observer.write().await = restored;
256 info!(
257 path = %brain_path.display(),
258 total, kinds,
259 "brain snapshot restaurado"
260 );
261 }
262 Err(e) => warn!(?e, path = %brain_path.display(), "brain snapshot read falló"),
263 }
264 }
265 }
266 // Si --audit-head, configuramos el head pointer y arrancamos auto-flush.
267 if let Some(head_path) = audit_head {
268 // Re-creamos el AuditLog con head pointer.
269 let new_audit = ente_brain::audit::AuditLog::new()
270 .with_head_pointer(head_path);
271 *brain.audit.write().await = new_audit;
272 spawn_audit_auto_flush(brain.clone());
273 }
274
275 // Carga inicial de reglas desde JSON/JSONL si --rules path proporcionado.
276 if let Some(path) = &rules_path {
277 match ente_brain::load_rules_file(path) {
278 Ok(rules) => {
279 let mut engine = brain.engine.write().await;
280 for r in rules {
281 engine.insert(r);
282 }
283 info!(count = engine.len(), path = %path.display(), "reglas cargadas");
284 }
285 Err(e) => warn!(?e, path = %path.display(), "carga de reglas falló"),
286 }
287 }
288
289 // Endpoint Prometheus opcional. En dev por defecto en 127.0.0.1:9911 si
290 // el flag no se pasó.
291 let metrics_addr = metrics_addr.or_else(|| {
292 if dev_mode { Some("127.0.0.1:9911".to_string()) } else { None }
293 });
294 if let Some(addr_s) = metrics_addr {
295 match addr_s.parse::<std::net::SocketAddr>() {
296 Ok(addr) => {
297 let s = brain.clone();
298 tokio::spawn(async move {
299 if let Err(e) = ente_brain::serve_metrics(s, addr).await {
300 warn!(?e, "metrics server cayó");
301 }
302 });
303 }
304 Err(e) => warn!(?e, addr = %addr_s, "metrics-addr inválido"),
305 }
306 }
307 spawn_brain_introspect(brain.clone());
308 let brain_sink = brain_glue::GraphSink {
309 graph_tx: graph_tx.clone(),
310 // Spawns auto-disparados desde reglas usan la identidad de la Semilla
311 // (único Ente con Capability::Spawn por construcción).
312 requester: graph.seed_id(),
313 };
314
315 // Demo automático del forwarding (sólo dev, sólo si el binario existe).
316 if dev_mode && std::path::Path::new("target/debug/ente-echo").exists() {
317 spawn_echo_smoke_test(bus_path.clone());
318 }
319
320 // En dev mode no tenemos hijos por defecto y el bucle se quedaría inerte.
321 let dev_exit = if dev_mode {
322 Some(tokio::time::sleep(Duration::from_secs(4)))
323 } else {
324 None
325 };
326 tokio::pin!(dev_exit);
327
328 // GC de capability grants expirados — corre cada 10 segundos.
329 let mut grant_purge = tokio::time::interval(Duration::from_secs(10));
330 grant_purge.tick().await; // descartar primer tick inmediato
331
332 loop {
333 tokio::select! {
334 biased;
335
336 Some(_) = sigchld.recv() => {
337 reap_until_empty(&mut graph, &graph_tx).await;
338 }
339
340 Some(uevt) = uevents.recv() => {
341 graph.on_uevent(uevt, &graph_tx).await;
342 }
343
344 Some(evt) = graph_rx.recv() => {
345 // Cerebro observa antes que el grafo mute. Snapshot del
346 // SubjectInfo se hace contra el estado pre-mutación.
347 feed_brain(&brain, &brain_sink, &graph, &evt).await;
348 if dispatch_graph_event(&mut graph, evt, &graph_tx, &checkpoint_path, &brain).await {
349 return Ok(());
350 }
351 }
352
353 _ = grant_purge.tick() => {
354 let n = graph.purge_expired_grants();
355 if n > 0 {
356 info!(purged = n, active = graph.active_grants_count(), "GC capability grants");
357 }
358 }
359
360 _ = async { dev_exit.as_mut().as_pin_mut().unwrap().await }, if dev_mode => {
361 info!("dev mode: timer expirado, cerrando bucle primordial");
362 let _ = graph_tx.send(GraphEvent::Shutdown {
363 reason: ShutdownReason::SeedRequested,
364 }).await;
365 }
366 }
367 }
368}
369
370/// Devuelve `true` si el bucle primordial debe terminar.
371async fn dispatch_graph_event(
372 graph: &mut EnteGraph,
373 evt: GraphEvent,
374 tx: &mpsc::Sender<GraphEvent>,
375 checkpoint: &Option<PathBuf>,
376 brain: &BrainState,
377) -> bool {
378 match evt {
379 GraphEvent::EnteDied { id, status } => {
380 graph.on_death(id, status, tx).await;
381 }
382 GraphEvent::CapabilityRequested { from, cap, reply } => {
383 graph.mediate_capability(from, cap, reply).await;
384 }
385 GraphEvent::SpawnRequest { card, requester } => {
386 if let Err(e) = graph.authorize_and_spawn(card, requester).await {
387 warn!(?e, "spawn request error");
388 }
389 }
390 GraphEvent::BusRequest { peer, from, request, outbound, reply } => {
391 graph.on_bus_request(peer, from, request, outbound, reply).await;
392 }
393 GraphEvent::BusResponse { seq, response } => {
394 graph.on_bus_response(seq, response).await;
395 }
396 GraphEvent::BusConnClosed { ente_id } => {
397 graph.on_bus_conn_closed(ente_id).await;
398 }
399 GraphEvent::Shutdown { reason } => {
400 warn!(?reason, "shutdown del fractal");
401 if let Some(path) = checkpoint.as_ref() {
402 // Snapshot del grafo
403 let snap = graph.snapshot();
404 match snap.write(path) {
405 Ok(()) => info!(path = %path.display(), entes = snap.entes.len(), "snapshot fractal persistido"),
406 Err(e) => warn!(?e, "snapshot write falló"),
407 }
408 // Snapshot del cerebro (observer state) en archivo adjunto
409 let brain_path = path.with_extension("brain.json");
410 let obs_snap = brain.observer.write().await.snapshot();
411 match write_brain_snapshot(&brain_path, &obs_snap) {
412 Ok(()) => info!(
413 path = %brain_path.display(),
414 total = obs_snap.total,
415 kinds = obs_snap.marginal.len(),
416 "snapshot brain persistido"
417 ),
418 Err(e) => warn!(?e, "brain snapshot write falló"),
419 }
420 }
421 graph.cascade_shutdown().await;
422 return true;
423 }
424 }
425 false
426}
427
428async fn reap_until_empty(graph: &mut EnteGraph, tx: &mpsc::Sender<GraphEvent>) {
429 loop {
430 match waitpid(None, Some(WaitPidFlag::WNOHANG)) {
431 Ok(WaitStatus::StillAlive) => return,
432 Ok(WaitStatus::Exited(pid, code)) => {
433 emit_death(graph, tx, pid, ExitStatus::Exit(code)).await;
434 }
435 Ok(WaitStatus::Signaled(pid, sig, _core)) => {
436 emit_death(graph, tx, pid, ExitStatus::Killed(sig)).await;
437 }
438 Ok(_) => continue, // Stopped/Continued — irrelevantes
439 Err(Errno::ECHILD) => return,
440 Err(e) => {
441 error!(?e, "waitpid fallo no recuperable en bucle de reaping");
442 return;
443 }
444 }
445 }
446}
447
448async fn emit_death(
449 graph: &EnteGraph,
450 tx: &mpsc::Sender<GraphEvent>,
451 pid: Pid,
452 status: ExitStatus,
453) {
454 let id = match graph.lookup_pid(pid) {
455 Some(id) => id,
456 None => {
457 // Proceso adoptado (subreaper): no está en nuestro grafo.
458 info!(?pid, ?status, "huérfano cosechado (no en grafo)");
459 return;
460 }
461 };
462 let _ = tx.send(GraphEvent::EnteDied { id, status }).await;
463}
464
465fn spawn_echo_smoke_test(bus_path: PathBuf) {
466 tokio::spawn(async move {
467 tokio::time::sleep(Duration::from_millis(300)).await;
468 match ente_bus::BusClient::connect(&bus_path).await {
469 Ok(mut client) => {
470 let req = ente_bus::BusRequest::Invoke {
471 cap: ente_echo::echo_capability(),
472 blob: b"hola fractal forwardeado".to_vec(),
473 };
474 match client.call(req).await {
475 Ok(ente_bus::BusResponse::Invoked { result }) => {
476 info!(echo = %String::from_utf8_lossy(&result), "Invoke ECHO round-trip OK");
477 }
478 Ok(other) => warn!(?other, "Invoke ECHO respuesta inesperada"),
479 Err(e) => warn!(?e, "Invoke ECHO falló"),
480 }
481 }
482 Err(e) => warn!(?e, "no se pudo conectar al bus para test"),
483 }
484 });
485}
486
487fn write_brain_snapshot(path: &std::path::Path, snap: &ente_brain::observer::ObserverSnapshot) -> anyhow::Result<()> {
488 let bytes = serde_json::to_vec_pretty(snap)?;
489 if let Some(parent) = path.parent() { let _ = std::fs::create_dir_all(parent); }
490 let tmp = path.with_extension("tmp");
491 std::fs::write(&tmp, &bytes)?;
492 std::fs::rename(&tmp, path)?;
493 Ok(())
494}
495
496fn read_brain_snapshot(path: &std::path::Path) -> anyhow::Result<ente_brain::observer::ObserverSnapshot> {
497 let bytes = std::fs::read(path)?;
498 let snap: ente_brain::observer::ObserverSnapshot = serde_json::from_slice(&bytes)?;
499 Ok(snap)
500}
501
502fn init_tracing() {
503 use tracing_subscriber::{fmt, EnvFilter};
504 let filter = EnvFilter::try_from_default_env()
505 .unwrap_or_else(|_| EnvFilter::new("ente_zero=debug,info"));
506 fmt().with_env_filter(filter).with_target(true).init();
507}
508
509fn brain_introspect_path() -> PathBuf {
510 if let Ok(p) = std::env::var("ENTE_BRAIN_SOCK") {
511 return p.into();
512 }
513 let runtime = std::env::var("XDG_RUNTIME_DIR")
514 .unwrap_or_else(|_| std::env::var("TMPDIR").unwrap_or_else(|_| "/tmp".into()));
515 format!("{runtime}/ente-brain.sock").into()
516}
517
518/// Auto-flush del audit log a CAS cada 10 segundos. Ejecuta best-effort:
519/// si el flush falla lo logeamos pero no abortamos. La integridad del log
520/// queda garantizada por su hash chain — re-flushar es idempotente.
521fn spawn_audit_auto_flush(state: BrainState) {
522 tokio::spawn(async move {
523 let mut tick = tokio::time::interval(std::time::Duration::from_secs(10));
524 tick.tick().await; // descartar primer tick inmediato
525 loop {
526 tick.tick().await;
527 let mut audit = state.audit.write().await;
528 match audit.flush_to_cas() {
529 Ok(0) => {} // nada nuevo
530 Ok(n) => info!(written = n, total = audit.flushed_count(), "audit auto-flush"),
531 Err(e) => warn!(?e, "audit auto-flush falló"),
532 }
533 }
534 });
535}
536
537fn spawn_brain_introspect(state: BrainState) {
538 let path = brain_introspect_path();
539 tokio::spawn(async move {
540 let server = IntrospectServer::new(state);
541 if let Err(e) = server.serve(&path).await {
542 warn!(?e, "introspect server cayó");
543 }
544 });
545}
546
547/// Registra el evento en el observer y dispatcha cualquier regla matched.
548/// Para reglas Sequence: pasamos los últimos N eventos del observer como
549/// history al engine.
550async fn feed_brain(
551 brain: &BrainState,
552 sink: &brain_glue::GraphSink,
553 graph: &EnteGraph,
554 evt: &GraphEvent,
555) {
556 let Some((kind, subj)) = brain_glue::graph_event_to_brain(evt, graph) else { return };
557 let history: Vec<ente_brain::TimedEvent> = {
558 let mut obs = brain.observer.write().await;
559 obs.record(kind.clone());
560 // Snapshot de los últimos 16 eventos — suficiente para cualquier
561 // Sequence pattern razonable. Clone hace una sola alocación.
562 obs.recent(16).cloned().collect()
563 };
564 let rules = {
565 let engine = brain.engine.read().await;
566 engine.dispatch(&kind, &subj, &history)
567 };
568 if !rules.is_empty() {
569 ente_brain::dispatch_actions(&rules, sink).await;
570 }
571}