⎇ d6b8f18b43c9bd9bf7268e48b61392a2d62bb229
crates/ente-zero/src/main.rs
Historia
Quién tocó qué
crudo
Pausa: 11 crates del fractal Ente #0 con cerebro completo
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 metrics_addr: Option<String>,
43}
44
45fn parse_args() -> CliArgs {
46 let mut args = std::env::args().skip(1);
47 let mut checkpoint = None;
48 let mut restore = None;
49 let mut rules = None;
50 let mut rules_out = None;
51 let mut metrics_addr = None;
52 while let Some(a) = args.next() {
53 match a.as_str() {
54 "--checkpoint" => checkpoint = args.next().map(PathBuf::from),
55 "--restore" => restore = args.next().map(PathBuf::from),
56 "--rules" => rules = args.next().map(PathBuf::from),
57 "--rules-out" => rules_out = args.next().map(PathBuf::from),
58 "--metrics-addr" => metrics_addr = args.next(),
59 other => warn!(arg = %other, "argumento desconocido, ignorado"),
60 }
61 }
62 CliArgs { checkpoint, restore, rules, rules_out, metrics_addr }
63}
64
65fn main() -> anyhow::Result<()> {
66 init_tracing();
67 let cli = parse_args();
68 let pid = getpid();
69 let dev_mode = pid != Pid::from_raw(1);
70
71 if dev_mode {
72 warn!(?pid, "ente-zero corriendo en DEV MODE (no PID 1) — kernel surface no se monta");
73 } else {
74 info!("ente-zero despierta como PID 1");
75 bootstrap_kernel_surface().context("bootstrap kernel surface")?;
76 become_child_subreaper().context("PR_SET_CHILD_SUBREAPER")?;
77 }
78
79 let card = seed::load(dev_mode, cli.restore.as_deref())?;
80
81 // current_thread runtime: ver doctrina al inicio del módulo.
82 let rt = tokio::runtime::Builder::new_current_thread()
83 .enable_io()
84 .enable_time()
85 .build()?;
86
87 rt.block_on(primordial_loop(card, dev_mode, cli.checkpoint, cli.rules, cli.rules_out, cli.metrics_addr))
88}
89
90async fn primordial_loop(
91 seed_card: ente_card::EntityCard,
92 dev_mode: bool,
93 checkpoint_path: Option<PathBuf>,
94 rules_path: Option<PathBuf>,
95 rules_out: Option<PathBuf>,
96 metrics_addr: Option<String>,
97) -> anyhow::Result<()> {
98 info!(seed_id = %seed_card.id, label = %seed_card.label, "Ente #0 entra al bucle primordial");
99
100 let (graph_tx, mut graph_rx) = mpsc::channel::<GraphEvent>(64);
101 let mut sigchld = spawn_sigchld_stream()?;
102 // Uevents puede fallar en dev (sin CAP_NET_ADMIN). Degradamos a un
103 // canal nunca-listo en lugar de abortar el bucle primordial.
104 let mut uevents = match spawn_uevent_stream() {
105 Ok(rx) => rx,
106 Err(e) => {
107 warn!(?e, "uevents deshabilitados (probablemente falta CAP_NET_ADMIN)");
108 let (_keep_tx, rx) = mpsc::channel::<ente_kernel::UEvent>(1);
109 std::mem::forget(_keep_tx);
110 rx
111 }
112 };
113
114 // Bus interno: listener antes de spawn de hijos para que su Announce
115 // tenga adónde llegar. Su path se inyecta en ENTE_BUS_SOCK por soma.
116 let bus_sock = bus::default_socket_path();
117 let bus_path = bus::spawn_bus(bus_sock, graph_tx.clone())?;
118 ente_soma::set_bus_sock(bus_path.to_string_lossy().into_owned());
119
120 let mut graph = EnteGraph::new(seed_card);
121 graph.instantiate_seed_dependencies(&graph_tx).await?;
122
123 // Cerebro: BrainState compartido + servidor de introspección.
124 // Window de 1024 eventos — suficiente para correlaciones interesantes
125 // sin gastar memoria de PID 1. En dev bajamos el umbral de cristalización
126 // para que el demo (pocos eventos) produzca cristales observables.
127 let mut brain = if dev_mode {
128 // Umbrales relajados para que el demo (pocos eventos) produzca
129 // cristales observables. Con P(b|a) normalizada a [0,1], los
130 // valores típicos en muestras pequeñas son 0.2-0.5.
131 BrainState::with_params(1024, ente_brain::CrystallizationParams {
132 min_support: 2,
133 min_conditional_prob: 0.3,
134 min_pmi: 1.0,
135 })
136 } else {
137 BrainState::new(1024)
138 };
139 if let Some(out_path) = rules_out {
140 brain = brain.with_rules_out(out_path);
141 }
142
143 // Carga inicial de reglas vía KCL o JSON, si --rules path proporcionado.
144 if let Some(path) = &rules_path {
145 match ente_brain::load_rules_file(path) {
146 Ok(rules) => {
147 let mut engine = brain.engine.write().await;
148 for r in rules {
149 engine.insert(r);
150 }
151 info!(count = engine.len(), path = %path.display(), "reglas cargadas");
152 }
153 Err(e) => warn!(?e, path = %path.display(), "carga de reglas falló"),
154 }
155 }
156
157 // Endpoint Prometheus opcional. En dev por defecto en 127.0.0.1:9911 si
158 // el flag no se pasó.
159 let metrics_addr = metrics_addr.or_else(|| {
160 if dev_mode { Some("127.0.0.1:9911".to_string()) } else { None }
161 });
162 if let Some(addr_s) = metrics_addr {
163 match addr_s.parse::<std::net::SocketAddr>() {
164 Ok(addr) => {
165 let s = brain.clone();
166 tokio::spawn(async move {
167 if let Err(e) = ente_brain::serve_metrics(s, addr).await {
168 warn!(?e, "metrics server cayó");
169 }
170 });
171 }
172 Err(e) => warn!(?e, addr = %addr_s, "metrics-addr inválido"),
173 }
174 }
175 spawn_brain_introspect(brain.clone());
176 let brain_sink = brain_glue::GraphSink {
177 graph_tx: graph_tx.clone(),
178 // Spawns auto-disparados desde reglas usan la identidad de la Semilla
179 // (único Ente con Capability::Spawn por construcción).
180 requester: graph.seed_id(),
181 };
182
183 // Demo automático del forwarding (sólo dev, sólo si el binario existe).
184 if dev_mode && std::path::Path::new("target/debug/ente-echo").exists() {
185 spawn_echo_smoke_test(bus_path.clone());
186 }
187
188 // En dev mode no tenemos hijos por defecto y el bucle se quedaría inerte.
189 let dev_exit = if dev_mode {
190 Some(tokio::time::sleep(Duration::from_secs(4)))
191 } else {
192 None
193 };
194 tokio::pin!(dev_exit);
195
196 loop {
197 tokio::select! {
198 biased;
199
200 Some(_) = sigchld.recv() => {
201 reap_until_empty(&mut graph, &graph_tx).await;
202 }
203
204 Some(uevt) = uevents.recv() => {
205 graph.on_uevent(uevt, &graph_tx).await;
206 }
207
208 Some(evt) = graph_rx.recv() => {
209 // Cerebro observa antes que el grafo mute. Snapshot del
210 // SubjectInfo se hace contra el estado pre-mutación.
211 feed_brain(&brain, &brain_sink, &graph, &evt).await;
212 if dispatch_graph_event(&mut graph, evt, &graph_tx, &checkpoint_path).await {
213 return Ok(());
214 }
215 }
216
217 _ = async { dev_exit.as_mut().as_pin_mut().unwrap().await }, if dev_mode => {
218 info!("dev mode: timer expirado, cerrando bucle primordial");
219 let _ = graph_tx.send(GraphEvent::Shutdown {
220 reason: ShutdownReason::SeedRequested,
221 }).await;
222 }
223 }
224 }
225}
226
227/// Devuelve `true` si el bucle primordial debe terminar.
228async fn dispatch_graph_event(
229 graph: &mut EnteGraph,
230 evt: GraphEvent,
231 tx: &mpsc::Sender<GraphEvent>,
232 checkpoint: &Option<PathBuf>,
233) -> bool {
234 match evt {
235 GraphEvent::EnteDied { id, status } => {
236 graph.on_death(id, status, tx).await;
237 }
238 GraphEvent::CapabilityRequested { from, cap, reply } => {
239 graph.mediate_capability(from, cap, reply).await;
240 }
241 GraphEvent::SpawnRequest { card, requester } => {
242 if let Err(e) = graph.authorize_and_spawn(card, requester).await {
243 warn!(?e, "spawn request error");
244 }
245 }
246 GraphEvent::BusRequest { peer, from, request, outbound, reply } => {
247 graph.on_bus_request(peer, from, request, outbound, reply).await;
248 }
249 GraphEvent::BusResponse { seq, response } => {
250 graph.on_bus_response(seq, response).await;
251 }
252 GraphEvent::BusConnClosed { ente_id } => {
253 graph.on_bus_conn_closed(ente_id).await;
254 }
255 GraphEvent::Shutdown { reason } => {
256 warn!(?reason, "shutdown del fractal");
257 if let Some(path) = checkpoint.as_ref() {
258 let snap = graph.snapshot();
259 match snap.write(path) {
260 Ok(()) => info!(path = %path.display(), entes = snap.entes.len(), "snapshot persistido"),
261 Err(e) => warn!(?e, "snapshot write falló"),
262 }
263 }
264 graph.cascade_shutdown().await;
265 return true;
266 }
267 }
268 false
269}
270
271async fn reap_until_empty(graph: &mut EnteGraph, tx: &mpsc::Sender<GraphEvent>) {
272 loop {
273 match waitpid(None, Some(WaitPidFlag::WNOHANG)) {
274 Ok(WaitStatus::StillAlive) => return,
275 Ok(WaitStatus::Exited(pid, code)) => {
276 emit_death(graph, tx, pid, ExitStatus::Exit(code)).await;
277 }
278 Ok(WaitStatus::Signaled(pid, sig, _core)) => {
279 emit_death(graph, tx, pid, ExitStatus::Killed(sig)).await;
280 }
281 Ok(_) => continue, // Stopped/Continued — irrelevantes
282 Err(Errno::ECHILD) => return,
283 Err(e) => {
284 error!(?e, "waitpid fallo no recuperable en bucle de reaping");
285 return;
286 }
287 }
288 }
289}
290
291async fn emit_death(
292 graph: &EnteGraph,
293 tx: &mpsc::Sender<GraphEvent>,
294 pid: Pid,
295 status: ExitStatus,
296) {
297 let id = match graph.lookup_pid(pid) {
298 Some(id) => id,
299 None => {
300 // Proceso adoptado (subreaper): no está en nuestro grafo.
301 info!(?pid, ?status, "huérfano cosechado (no en grafo)");
302 return;
303 }
304 };
305 let _ = tx.send(GraphEvent::EnteDied { id, status }).await;
306}
307
308fn spawn_echo_smoke_test(bus_path: PathBuf) {
309 tokio::spawn(async move {
310 tokio::time::sleep(Duration::from_millis(300)).await;
311 match ente_bus::BusClient::connect(&bus_path).await {
312 Ok(mut client) => {
313 let req = ente_bus::BusRequest::Invoke {
314 cap: ente_echo::echo_capability(),
315 blob: b"hola fractal forwardeado".to_vec(),
316 };
317 match client.call(req).await {
318 Ok(ente_bus::BusResponse::Invoked { result }) => {
319 info!(echo = %String::from_utf8_lossy(&result), "Invoke ECHO round-trip OK");
320 }
321 Ok(other) => warn!(?other, "Invoke ECHO respuesta inesperada"),
322 Err(e) => warn!(?e, "Invoke ECHO falló"),
323 }
324 }
325 Err(e) => warn!(?e, "no se pudo conectar al bus para test"),
326 }
327 });
328}
329
330fn init_tracing() {
331 use tracing_subscriber::{fmt, EnvFilter};
332 let filter = EnvFilter::try_from_default_env()
333 .unwrap_or_else(|_| EnvFilter::new("ente_zero=debug,info"));
334 fmt().with_env_filter(filter).with_target(true).init();
335}
336
337fn brain_introspect_path() -> PathBuf {
338 if let Ok(p) = std::env::var("ENTE_BRAIN_SOCK") {
339 return p.into();
340 }
341 let runtime = std::env::var("XDG_RUNTIME_DIR")
342 .unwrap_or_else(|_| std::env::var("TMPDIR").unwrap_or_else(|_| "/tmp".into()));
343 format!("{runtime}/ente-brain.sock").into()
344}
345
346fn spawn_brain_introspect(state: BrainState) {
347 let path = brain_introspect_path();
348 tokio::spawn(async move {
349 let server = IntrospectServer::new(state);
350 if let Err(e) = server.serve(&path).await {
351 warn!(?e, "introspect server cayó");
352 }
353 });
354}
355
356/// Registra el evento en el observer y dispatcha cualquier regla matched.
357/// Para reglas Sequence: pasamos los últimos N eventos del observer como
358/// history al engine.
359async fn feed_brain(
360 brain: &BrainState,
361 sink: &brain_glue::GraphSink,
362 graph: &EnteGraph,
363 evt: &GraphEvent,
364) {
365 let Some((kind, subj)) = brain_glue::graph_event_to_brain(evt, graph) else { return };
366 let history: Vec<ente_brain::TimedEvent> = {
367 let mut obs = brain.observer.write().await;
368 obs.record(kind.clone());
369 // Snapshot de los últimos 16 eventos — suficiente para cualquier
370 // Sequence pattern razonable. Clone hace una sola alocación.
371 obs.recent(16).cloned().collect()
372 };
373 let rules = {
374 let engine = brain.engine.read().await;
375 engine.dispatch(&kind, &subj, &history)
376 };
377 if !rules.is_empty() {
378 ente_brain::dispatch_actions(&rules, sink).await;
379 }
380}