feat(ente-zero): wire de Arje con brahman-net (red P2P opcional + keypair persistente)

Cierra el ultimo pendiente del plan de red: Arje ahora puede arrancar
opcionalmente con BrahmanNet configurado, persistir su identidad
libp2p entre reboots, y participar en la malla brahman como nodo
publico. Sin breaking changes: usuarios actuales (sin env vars) siguen
viendo el comportamiento Unix-only de antes.

Activacion por env vars:
- BRAHMAN_LISTEN_MULTIADDR — si set, activa la red P2P. Ej:
  /ip4/0.0.0.0/tcp/4101 (publico), /ip4/127.0.0.1/tcp/0 (loopback).
- BRAHMAN_KEYPAIR_PATH — override del path donde se persiste la
  keypair Ed25519. Defaults: PID 1 -> /var/lib/brahman/init-keypair.bin;
  dev mode -> $XDG_DATA_HOME/brahman/init-keypair.bin con fallbacks.
- BRAHMAN_BOOTSTRAP_PEERS — multiaddrs coma-separados a dial-ear al
  arranque para entrar al DHT.

Comportamiento al activarse:
1. keypair_store::load_or_generate carga keypair de disco o genera y
   persiste una nueva (32 bytes raw, 0o600, atomic rename). peer_id
   estable across reboots.
2. BrahmanNet::with_keypair arma el swarm con esa identidad.
3. net.listen() resuelve la addr y se loggea.
4. BRAHMAN_BOOTSTRAP_PEERS si set -> dial cada multiaddr.
5. ServerConfig.net = Some(net), activando announce_outputs automatico
   en el DHT por cada Card con outputs.
6. Ademas del Unix accept loop, se monta libp2p accept loop sobre el
   mismo Server compartido (Arc<Server>). Sesiones locales y remotas
   conviven en las mismas tablas.

Refactor del Unix accept loop: antes consumia server via run().await;
ahora usa Arc<Server>::accept_one().await en loop para coexistir con
el libp2p accept loop sin moverse el server.

Degradacion gracil: si la keypair no carga, multiaddr invalido,
listen falla, bootstrap dial revienta -> log + continuamos en
Unix-only. Doctrina PID 1 ("ningun subsistema opcional rompe el
bucle primordial") preservada.

Tests: 4 unit en keypair_store: generate_persist_and_reload_yields_
same_peer_id (la property fundamental), rejects_corrupted_file,
persisted_file_is_owner_only (0o600 verificados), default_path_
honors_env.

Activacion en una linea:
  BRAHMAN_LISTEN_MULTIADDR=/ip4/0.0.0.0/tcp/4101 ente-zero

Pendientes: stop_providing en cleanup, allowlist/denylist (PKI),
rotacion de keypair sin perder peer_id.

crates/core/ente-zero/src/main.rs

@@ -19,6 +19,7 @@ mod brain_glue;
 mod bus;
 mod events;
 mod graph;
+mod keypair_store;
 mod seed;
 
 use anyhow::Context;
@@ -158,23 +159,66 @@ async fn primordial_loop(
             current_context: broker_context.clone(),
         }),
     ));
+
+    // Brahman-net opcional: si BRAHMAN_LISTEN_MULTIADDR está set,
+    // levantamos la malla P2P y la pasamos como ServerConfig.net (Fase
+    // 2 wire) para que cada Card con outputs se anuncie al DHT y
+    // pueda ser descubierta por nodos remotos. Identidad libp2p
+    // persistida en disco vía keypair_store (peer_id estable across
+    // reboots).
+    let brahman_net = setup_brahman_net(dev_mode).await;
+
     let brahman_sock = brahman_handshake::transport::default_socket_path();
     match brahman_handshake::server::Server::bind(
         &brahman_sock,
         brahman_handshake::server::ServerConfig {
             init_attached: true,
             broker: Some(brahman_broker.clone()),
-            // Fase 2: el Init aún no expone red P2P por default. Cuando
-            // arje quiera publicar Cards al DHT remoto, pasar aquí un
-            // `Some(Arc<BrahmanNet>)` con la malla configurada.
-            net: None,
+            net: brahman_net.clone(),
         },
     ) {
         Ok(server) => {
-            info!(socket = %brahman_sock.display(), "brahman handshake escuchando");
+            info!(socket = %brahman_sock.display(), "brahman handshake escuchando (Unix)");
+            // Si hay malla P2P, además del Unix accept loop levantamos
+            // el accept loop libp2p sobre el mismo Server compartido.
+            // Las sesiones locales y remotas conviven en las mismas
+            // tablas (sessions, push_table, broker).
+            let server = std::sync::Arc::new(server);
+            if let Some(net) = brahman_net.clone() {
+                let s_libp2p = server.clone();
+                let n_libp2p = net.clone();
+                tokio::spawn(async move {
+                    if let Err(e) = brahman_handshake::network::run_libp2p_accept_loop(
+                        s_libp2p, n_libp2p,
+                    )
+                    .await
+                    {
+                        warn!(?e, "brahman handshake libp2p accept loop cayó");
+                    }
+                });
+                info!(
+                    "brahman handshake escuchando también vía libp2p (peer_id {})",
+                    net.peer_id
+                );
+            }
+            // Unix accept loop: usa Arc<Server> en lugar del consume
+            // de run() para coexistir con el libp2p accept loop.
+            let s_unix = server.clone();
             tokio::spawn(async move {
-                if let Err(e) = server.run().await {
-                    warn!(?e, "brahman handshake server cayó");
+                loop {
+                    match s_unix.accept_one().await {
+                        Ok(session) => {
+                            tokio::spawn(async move {
+                                if let Err(e) = session.handle().await {
+                                    warn!(?e, "session Unix terminó con error");
+                                }
+                            });
+                        }
+                        Err(e) => {
+                            warn!(?e, "brahman handshake accept_one Unix falló");
+                            break;
+                        }
+                    }
                 }
             });
         }
@@ -573,3 +617,84 @@ async fn feed_brain(
         ente_brain::dispatch_actions(&rules, sink).await;
     }
 }
+
+/// Inicializa la malla `brahman-net` opcional. Activa sólo si
+/// `BRAHMAN_LISTEN_MULTIADDR` está set. Identidad libp2p persistente
+/// vía `keypair_store`. Bootstrap del DHT vía `BRAHMAN_BOOTSTRAP_PEERS`
+/// (lista coma-separada de multiaddrs, opcional).
+///
+/// Toda fase de setup degrada grácilmente: si la keypair no carga,
+/// si el listen falla, si bootstrap dial falla — loggea y devuelve
+/// `None`. El Init sigue funcionando en modo Unix-only.
+async fn setup_brahman_net(
+    dev_mode: bool,
+) -> Option<std::sync::Arc<brahman_net::BrahmanNet>> {
+    let listen_addr = match std::env::var("BRAHMAN_LISTEN_MULTIADDR") {
+        Ok(s) if !s.is_empty() => s,
+        _ => {
+            tracing::debug!(
+                "brahman-net deshabilitado (sin BRAHMAN_LISTEN_MULTIADDR)"
+            );
+            return None;
+        }
+    };
+
+    let multiaddr: brahman_net::Multiaddr = match listen_addr.parse() {
+        Ok(m) => m,
+        Err(e) => {
+            warn!(addr = %listen_addr, ?e, "BRAHMAN_LISTEN_MULTIADDR inválido — net deshabilitado");
+            return None;
+        }
+    };
+
+    let keypair_path = keypair_store::default_path(dev_mode);
+    let (keypair, loaded) = match keypair_store::load_or_generate(&keypair_path) {
+        Ok(kp) => kp,
+        Err(e) => {
+            warn!(path = %keypair_path.display(), ?e, "no pude cargar/generar keypair libp2p — net deshabilitado");
+            return None;
+        }
+    };
+    info!(
+        path = %keypair_path.display(),
+        peer_id = %keypair.public().to_peer_id(),
+        loaded = loaded,
+        "identidad libp2p {}",
+        if loaded { "cargada" } else { "generada y persistida" }
+    );
+
+    let net = match brahman_net::BrahmanNet::with_keypair(keypair) {
+        Ok(n) => std::sync::Arc::new(n),
+        Err(e) => {
+            warn!(?e, "BrahmanNet::with_keypair falló — net deshabilitado");
+            return None;
+        }
+    };
+
+    let actual = net.listen(multiaddr).await;
+    info!(addr = %actual, peer_id = %net.peer_id, "brahman-net escuchando");
+
+    // Bootstrap opcional: dial-ar a peers conocidos para entrar al
+    // DHT. Sin bootstrap, el nodo arranca aislado hasta que alguien
+    // se conecte a él.
+    if let Ok(bootstrap) = std::env::var("BRAHMAN_BOOTSTRAP_PEERS") {
+        let mut dialed = 0usize;
+        for entry in bootstrap.split(',').filter(|s| !s.is_empty()) {
+            match entry.parse::<brahman_net::Multiaddr>() {
+                Ok(addr) => {
+                    net.dial(addr.clone());
+                    dialed += 1;
+                    tracing::debug!(peer = %addr, "dial bootstrap");
+                }
+                Err(e) => {
+                    warn!(entry = %entry, ?e, "bootstrap multiaddr inválido — saltado");
+                }
+            }
+        }
+        if dialed > 0 {
+            info!(count = dialed, "bootstrap peers dial-eados");
+        }
+    }
+
+    Some(net)
+}