chore: monorepo inicial con arje + minga + yahweh absorbidos

Workspace en 4 ejes (core/modules/apps/shared):

- core/: 24 crates de arje (Init systemd-compatible: ente-card, ente-zero,
  ente-kernel, ente-bus, ente-cas, ente-soma, ente-wasm, ente-snapshot,
  ente-brain, ente-echo, ente-policy-provider, + 12 crates *-compat)
- modules/semantic_dht/: 5 crates de minga (minga-core con AST/CAS/MST,
  minga-p2p con libp2p Kad, minga-store, minga-vfs, minga-cli)
- modules/ui_engine/: 11 crates de yahweh (libs/{core,theme,bus,providers},
  widgets/{tree,splitter,tabs,tiled,container_core,text_input})
- apps/: 5 crates de yahweh (file_explorer, database_explorer, text_viewer,
  image_viewer, yahweh-shell)
- shared_wit/protocol.wit: handshake/lifecycle inicial

Cargo.toml unificado: thiserror bumped a 2 (transparente para arje), tokio
"full", paths intra-workspace de yahweh redirigidos a su nueva ubicación.

cargo check --workspace: 0 errores, 17 warnings (dead code preexistente).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

crates/modules/semantic_dht/minga-p2p/Cargo.toml

@@ -0,0 +1,24 @@
+[package]
+name = "minga-p2p"
+version.workspace = true
+edition.workspace = true
+license.workspace = true
+authors.workspace = true
+description = "Minga P2P: protocolo de sincronización entre repositorios. Lógica pura; el transporte (libp2p) se monta encima."
+
+[dependencies]
+minga-core = { path = "../minga-core" }
+minga-store = { path = "../minga-store" }
+serde = { workspace = true }
+postcard = { workspace = true }
+thiserror = { workspace = true }
+tokio = { workspace = true }
+tokio-util = { workspace = true }
+futures = { workspace = true }
+libp2p = { workspace = true }
+libp2p-stream = { workspace = true }
+rand = { workspace = true }
+
+[dev-dependencies]
+tokio = { workspace = true }
+tempfile = { workspace = true }

crates/modules/semantic_dht/minga-p2p/src/async_driver.rs

@@ -0,0 +1,100 @@
+//! Driver de sincronización sobre I/O asíncrona.
+//!
+//! Bridge entre la `SyncSession` puramente lógica y cualquier
+//! transporte que implemente `AsyncRead + AsyncWrite`. Encuadre
+//! length-prefixed: cada `Message` se serializa con postcard y se
+//! envía precedido de un `u32 LE` con su longitud en bytes.
+//!
+//! La estructura del bucle es:
+//! 1. Drenar todos los `Message`s pendientes a la salida.
+//! 2. Si la sesión declara `is_done`, salir.
+//! 3. Bloquear esperando un `Message` entrante; alimentarlo a la
+//!    sesión y volver al paso 1.
+//!
+//! Esto funciona porque cada paso del state machine emite los
+//! mensajes que necesita inmediatamente — nunca quedan colgados
+//! mensajes por un `Message` futuro. La única espera real ocurre en
+//! el paso 3, cuando estamos esperando que el peer responda.
+
+use std::collections::VecDeque;
+use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
+
+use crate::message::Message;
+use crate::session::SyncSession;
+
+/// Cota dura sobre el tamaño de un frame, para evitar que un peer
+/// malicioso (o un bug) cause asignaciones desbocadas. 16 MB es de
+/// sobra para mensajes de sync — un `AttestPush` de cien mil
+/// atestaciones cabe en ~13 MB.
+const MAX_FRAME_SIZE: u32 = 16 * 1024 * 1024;
+
+#[derive(Debug, thiserror::Error)]
+pub enum AsyncSyncError {
+    #[error("io: {0}")]
+    Io(#[from] std::io::Error),
+
+    #[error("decode postcard: {0}")]
+    Decode(#[from] postcard::Error),
+
+    #[error("frame demasiado grande: {0} bytes")]
+    FrameTooLarge(u32),
+
+    #[error("la sesión cerró sin alcanzar `is_done`")]
+    UnexpectedClose,
+}
+
+/// Ejecuta una sesión de sincronización completa sobre una stream
+/// duplex. Devuelve la `SyncSession` resultante (con el `Mst`,
+/// `MemStore` y `AttestationStore` ya mergeados con el peer).
+pub async fn run_sync_async<S>(
+    mut session: SyncSession,
+    mut stream: S,
+) -> Result<SyncSession, AsyncSyncError>
+where
+    S: AsyncRead + AsyncWrite + Unpin,
+{
+    let mut outbound: VecDeque<Message> = session.start().into();
+
+    loop {
+        while let Some(msg) = outbound.pop_front() {
+            send_frame(&mut stream, &msg).await?;
+        }
+
+        if session.is_done() {
+            return Ok(session);
+        }
+
+        let msg = recv_frame(&mut stream).await?;
+        outbound.extend(session.handle(msg));
+    }
+}
+
+async fn send_frame<S>(stream: &mut S, msg: &Message) -> Result<(), AsyncSyncError>
+where
+    S: AsyncWrite + Unpin,
+{
+    let bytes = msg.encode();
+    let len = bytes.len() as u32;
+    if len > MAX_FRAME_SIZE {
+        return Err(AsyncSyncError::FrameTooLarge(len));
+    }
+    stream.write_all(&len.to_le_bytes()).await?;
+    stream.write_all(&bytes).await?;
+    stream.flush().await?;
+    Ok(())
+}
+
+async fn recv_frame<S>(stream: &mut S) -> Result<Message, AsyncSyncError>
+where
+    S: AsyncRead + Unpin,
+{
+    let mut len_buf = [0u8; 4];
+    stream.read_exact(&mut len_buf).await?;
+    let len = u32::from_le_bytes(len_buf);
+    if len > MAX_FRAME_SIZE {
+        return Err(AsyncSyncError::FrameTooLarge(len));
+    }
+    let mut buf = vec![0u8; len as usize];
+    stream.read_exact(&mut buf).await?;
+    Ok(Message::decode(&buf)?)
+}

crates/modules/semantic_dht/minga-p2p/src/harness.rs

@@ -0,0 +1,89 @@
+//! Harness in-memory determinístico para correr dos `SyncSession`s
+//! una contra la otra y verificar invariantes del protocolo.
+
+use std::collections::VecDeque;
+
+use crate::message::Message;
+use crate::session::SyncSession;
+
+#[derive(Debug, Default, Clone, PartialEq, Eq)]
+pub struct SyncStats {
+    pub challenges: usize,
+    pub hellos: usize,
+    pub probe_reqs: usize,
+    pub probe_ress: usize,
+    pub fetches: usize,
+    pub delivers: usize,
+    pub attest_pushes: usize,
+    pub dones: usize,
+}
+
+impl SyncStats {
+    fn record(&mut self, m: &Message) {
+        match m {
+            Message::Challenge { .. } => self.challenges += 1,
+            Message::Hello { .. } => self.hellos += 1,
+            Message::ProbeReq { .. } => self.probe_reqs += 1,
+            Message::ProbeRes { .. } => self.probe_ress += 1,
+            Message::Fetch { .. } => self.fetches += 1,
+            Message::Deliver { .. } => self.delivers += 1,
+            Message::AttestPush { .. } => self.attest_pushes += 1,
+            Message::Done => self.dones += 1,
+        }
+    }
+
+    pub fn total(&self) -> usize {
+        self.challenges
+            + self.hellos
+            + self.probe_reqs
+            + self.probe_ress
+            + self.fetches
+            + self.delivers
+            + self.attest_pushes
+            + self.dones
+    }
+}
+
+/// Ejecuta la sincronización entre dos sesiones hasta convergencia.
+///
+/// Pánico si la conversación termina sin que ambas partes alcancen
+/// `is_done()` — eso sería un deadlock del protocolo y una regresión.
+pub fn run_sync(a: &mut SyncSession, b: &mut SyncSession) -> SyncStats {
+    let mut from_a: VecDeque<Message> = VecDeque::new();
+    let mut from_b: VecDeque<Message> = VecDeque::new();
+    let mut stats = SyncStats::default();
+
+    from_a.extend(a.start());
+    from_b.extend(b.start());
+
+    loop {
+        let mut progress = false;
+
+        if let Some(msg) = from_a.pop_front() {
+            stats.record(&msg);
+            for out in b.handle(msg) {
+                from_b.push_back(out);
+            }
+            progress = true;
+        }
+
+        if let Some(msg) = from_b.pop_front() {
+            stats.record(&msg);
+            for out in a.handle(msg) {
+                from_a.push_back(out);
+            }
+            progress = true;
+        }
+
+        if !progress {
+            break;
+        }
+    }
+
+    assert!(
+        a.is_done() && b.is_done(),
+        "deadlock: sync terminó sin que ambos peers cerraran"
+    );
+
+    stats
+}

crates/modules/semantic_dht/minga-p2p/src/lib.rs

@@ -0,0 +1,26 @@
+//! minga-p2p: protocolo de sincronización entre repositorios Minga.
+//!
+//! Este crate define el **protocolo** y la **máquina de estados** de la
+//! sincronización P2P, sin acoplarse a un transporte concreto. Un peer
+//! manipula una `SyncSession` (puramente lógica) que consume mensajes
+//! entrantes y produce mensajes salientes; el transporte real —libp2p,
+//! HTTP, in-memory, lo que sea— se reduce a serializar/deserializar y
+//! mover bytes.
+//!
+//! Este orden refleja el principio bottom-up del proyecto: validamos la
+//! convergencia del protocolo con un `harness` in-memory determinístico
+//! antes de invertir en async runtime + libp2p.
+
+pub mod async_driver;
+pub mod harness;
+pub mod message;
+pub mod network;
+pub mod peer;
+pub mod session;
+
+pub use async_driver::{run_sync_async, AsyncSyncError};
+pub use harness::{run_sync, SyncStats};
+pub use message::Message;
+pub use network::{DiscoveredPeer, LibP2pNode, NodeError, SYNC_PROTOCOL};
+pub use peer::{MingaPeer, PeerOpenError, PeerSyncError};
+pub use session::SyncSession;

crates/modules/semantic_dht/minga-p2p/src/message.rs

@@ -0,0 +1,94 @@
+//! Mensajes del protocolo de sincronización (versión recursiva sobre
+//! la estructura del MST).
+//!
+//! El protocolo es simétrico — ambos peers ejecutan el mismo rol y
+//! emiten los mismos mensajes — y consta de seis tipos:
+//!
+//! 1. `Hello { root_subtree_hash }` anuncia el hash Merkle del MST raíz
+//!    del emisor. Si ambos hashes coinciden, los dos repos son idénticos
+//!    y la sincronización termina sin un solo byte adicional.
+//!
+//! 2. `ProbeReq { subtree_hash }` solicita la **estructura** (level +
+//!    keys + child_hashes) de un subárbol previamente anunciado por el
+//!    otro peer. Es lo que permite descender el árbol del peer paso a
+//!    paso, podando ramas idénticas por igualdad de hash.
+//!
+//! 3. `ProbeRes { subtree_hash, probe }` responde con el `NodeProbe`,
+//!    o `None` si el subárbol era el vacío. Cada subárbol que el peer
+//!    no reconoce dispara un `ProbeReq` recursivo; cuando el peer ya
+//!    tiene un subárbol con el mismo hash, la rama se poda.
+//!
+//! 4. `Fetch { hash }` y `Deliver { hash, stored }` mueven los nodos
+//!    propiamente dichos. El receptor del `Deliver` **verifica
+//!    criptográficamente** que `hash_stored(stored) == hash` antes de
+//!    insertar — un peer malicioso no puede colar un `StoredNode`
+//!    distinto bajo un hash anunciado.
+//!
+//! 5. `Done` cierra el lado del emisor: ya recibió el `Hello` del otro,
+//!    no tiene probes ni fetches pendientes. Cuando ambos `Done`s han
+//!    cruzado, la sesión termina con ambos repos convergentes.
+
+use minga_core::{Attestation, ContentHash, Did, NodeProbe, Signature, StoredNode};
+
+#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
+pub enum Message {
+    /// Reto de session-handshake: 32 bytes aleatorios. Cada peer envía
+    /// uno al inicio. El otro lado lo incrustará en el payload del
+    /// `Hello` que firme con su llave privada — así un `Hello`
+    /// capturado en una sesión no puede replayearse en otra (que
+    /// tendrá un nonce distinto).
+    Challenge {
+        nonce: [u8; 32],
+    },
+
+    /// Saludo autenticado anti-replay: el emisor presenta su DID, el
+    /// hash del subárbol raíz de su MST, y una firma sobre el payload
+    /// `(peer_did || root_subtree_hash || nonce_recibido_del_peer)`.
+    /// El receptor reconstruye el payload con su PROPIO nonce (el que
+    /// envió en su Challenge) y verifica con la llave pública del
+    /// peer. Sin Challenge previo no hay Hello válido posible.
+    Hello {
+        peer_did: Did,
+        root_subtree_hash: ContentHash,
+        signature: Signature,
+    },
+    ProbeReq {
+        subtree_hash: ContentHash,
+    },
+    ProbeRes {
+        subtree_hash: ContentHash,
+        probe: Option<NodeProbe>,
+    },
+    Fetch {
+        hash: ContentHash,
+    },
+    Deliver {
+        hash: ContentHash,
+        stored: StoredNode,
+    },
+    /// Empuje de atestaciones: el emisor entrega al peer las pruebas
+    /// criptográficas de autoría que conoce. Cada `Attestation` es
+    /// auto-verificable (firma + autor + contenido), así que el
+    /// receptor puede validar y mezclar sin confiar en la palabra del
+    /// remitente. Se envían tras el `Hello` autenticado para que el
+    /// peer verifique la identidad del remitente antes de procesarlas.
+    AttestPush {
+        attestations: Vec<Attestation>,
+    },
+    Done,
+}
+
+impl Message {
+    /// Codifica el mensaje a bytes vía postcard. Diseñado para
+    /// transferir sobre cualquier transporte que mueva `Vec<u8>`.
+    /// Postcard es compacto, sin overhead de schema runtime.
+    pub fn encode(&self) -> Vec<u8> {
+        postcard::to_allocvec(self).expect("postcard encoding cannot fail for our types")
+    }
+
+    /// Decodifica bytes a un `Message`. `Err` si los bytes son
+    /// malformados o no representan un `Message` válido.
+    pub fn decode(bytes: &[u8]) -> Result<Self, postcard::Error> {
+        postcard::from_bytes(bytes)
+    }
+}

crates/modules/semantic_dht/minga-p2p/src/network.rs

@@ -0,0 +1,282 @@
+//! Integración libp2p con behaviour compuesto: streams Minga +
+//! Kademlia DHT.
+//!
+//! - **TCP + Noise + Yamux**: transporte autenticado y multiplexado.
+//! - **`stream::Behaviour`**: streams bidireccionales para el
+//!   protocolo `/minga/sync/1.0.0`.
+//! - **`kad::Behaviour<MemoryStore>`**: tabla de routing distribuida
+//!   para descubrimiento. Cada nodo arranca en modo `Server` y
+//!   responde a queries del DHT.
+//!
+//! El swarm corre en una task tokio dedicada que procesa comandos
+//! externos (Dial, Listen, AddDhtPeer, FindClosestPeers) y eventos
+//! del swarm (NewListenAddr para señalar address resuelto, eventos
+//! Kad para completar queries). Los métodos públicos solo envían
+//! comandos por canal.
+
+use std::collections::HashMap;
+use std::time::Duration;
+
+use futures::StreamExt;
+use libp2p::{
+    identify, identity, kad, noise,
+    swarm::{NetworkBehaviour, SwarmEvent},
+    tcp, yamux, Multiaddr, PeerId, StreamProtocol, Swarm, SwarmBuilder,
+};
+use libp2p_stream as stream;
+use tokio::sync::{mpsc, oneshot, Mutex};
+
+pub const SYNC_PROTOCOL: StreamProtocol = StreamProtocol::new("/minga/sync/1.0.0");
+const IDENTIFY_PROTOCOL: &str = "/minga/0.1.0";
+
+#[derive(NetworkBehaviour)]
+struct MingaBehaviour {
+    stream: stream::Behaviour,
+    kad: kad::Behaviour<kad::store::MemoryStore>,
+    identify: identify::Behaviour,
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum NodeError {
+    #[error("transport build failed: {0}")]
+    Build(String),
+}
+
+#[derive(Debug)]
+enum Command {
+    Dial(Multiaddr),
+    Listen(Multiaddr),
+    AddDhtPeer(PeerId, Multiaddr),
+    FindClosestPeers(PeerId, oneshot::Sender<Vec<DiscoveredPeer>>),
+    StartProviding(Vec<u8>),
+    GetProviders(Vec<u8>, oneshot::Sender<Vec<PeerId>>),
+}
+
+/// Peer descubierto vía DHT: identidad + direcciones conocidas.
+#[derive(Debug, Clone)]
+pub struct DiscoveredPeer {
+    pub peer_id: PeerId,
+    pub addrs: Vec<Multiaddr>,
+}
+
+pub struct LibP2pNode {
+    pub peer_id: PeerId,
+    cmd_tx: mpsc::UnboundedSender<Command>,
+    listen_rx: Mutex<mpsc::UnboundedReceiver<Multiaddr>>,
+    /// Control para abrir/aceptar streams.
+    pub control: stream::Control,
+}
+
+impl LibP2pNode {
+    pub fn new() -> Result<Self, NodeError> {
+        let id = identity::Keypair::generate_ed25519();
+        let peer_id = id.public().to_peer_id();
+
+        let mut swarm: Swarm<MingaBehaviour> = SwarmBuilder::with_existing_identity(id)
+            .with_tokio()
+            .with_tcp(
+                tcp::Config::default(),
+                noise::Config::new,
+                yamux::Config::default,
+            )
+            .map_err(|e| NodeError::Build(format!("{e}")))?
+            .with_behaviour(|key| {
+                let local = key.public().to_peer_id();
+                let mut kad =
+                    kad::Behaviour::new(local, kad::store::MemoryStore::new(local));
+                // Modo Server: respondemos a queries del DHT. Por
+                // defecto kad arranca en Auto, que requiere detectar
+                // reachability. Para tests en localhost forzamos Server.
+                kad.set_mode(Some(kad::Mode::Server));
+                let identify = identify::Behaviour::new(
+                    identify::Config::new(IDENTIFY_PROTOCOL.to_string(), key.public())
+                        .with_agent_version(format!("minga/{}", env!("CARGO_PKG_VERSION"))),
+                );
+                MingaBehaviour {
+                    stream: stream::Behaviour::new(),
+                    kad,
+                    identify,
+                }
+            })
+            .map_err(|e| NodeError::Build(format!("{e}")))?
+            .with_swarm_config(|c| c.with_idle_connection_timeout(Duration::from_secs(60)))
+            .build();
+
+        let control = swarm.behaviour().stream.new_control();
+
+        let (cmd_tx, mut cmd_rx) = mpsc::unbounded_channel::<Command>();
+        let (listen_tx, listen_rx) = mpsc::unbounded_channel::<Multiaddr>();
+
+        tokio::spawn(async move {
+            let mut pending_finds: HashMap<
+                kad::QueryId,
+                oneshot::Sender<Vec<DiscoveredPeer>>,
+            > = HashMap::new();
+            let mut pending_providers: HashMap<
+                kad::QueryId,
+                (Vec<PeerId>, oneshot::Sender<Vec<PeerId>>),
+            > = HashMap::new();
+
+            loop {
+                tokio::select! {
+                    Some(cmd) = cmd_rx.recv() => {
+                        match cmd {
+                            Command::Dial(addr) => {
+                                let _ = swarm.dial(addr);
+                            }
+                            Command::Listen(addr) => {
+                                let _ = swarm.listen_on(addr);
+                            }
+                            Command::AddDhtPeer(peer, addr) => {
+                                swarm.behaviour_mut().kad.add_address(&peer, addr);
+                            }
+                            Command::FindClosestPeers(target, tx) => {
+                                let qid = swarm.behaviour_mut().kad.get_closest_peers(target);
+                                pending_finds.insert(qid, tx);
+                            }
+                            Command::StartProviding(key) => {
+                                // Best-effort: si falla (sin peers cercanos para
+                                // replicar), seguirá viviendo en el local store
+                                // y se servirá vía get_providers de quien
+                                // tenga conexión con nosotros.
+                                let _ = swarm.behaviour_mut().kad.start_providing(key.into());
+                            }
+                            Command::GetProviders(key, tx) => {
+                                let qid = swarm.behaviour_mut().kad.get_providers(key.into());
+                                pending_providers.insert(qid, (Vec::new(), tx));
+                            }
+                        }
+                    }
+                    event = swarm.select_next_some() => {
+                        match event {
+                            SwarmEvent::NewListenAddr { address, .. } => {
+                                let _ = listen_tx.send(address);
+                            }
+                            // Identify nos dice las listen-addrs reales del
+                            // peer. Las inyectamos a Kad para poblar el
+                            // routing table sin necesidad de add_dht_peer
+                            // manual — la propagación pasa a ser automática.
+                            SwarmEvent::Behaviour(MingaBehaviourEvent::Identify(
+                                identify::Event::Received { peer_id, info, .. }
+                            )) => {
+                                for addr in info.listen_addrs {
+                                    swarm.behaviour_mut().kad.add_address(&peer_id, addr);
+                                }
+                            }
+                            SwarmEvent::Behaviour(MingaBehaviourEvent::Kad(
+                                kad::Event::OutboundQueryProgressed { id, result, step, .. }
+                            )) => {
+                                match result {
+                                    kad::QueryResult::GetClosestPeers(Ok(ok)) if step.last => {
+                                        if let Some(tx) = pending_finds.remove(&id) {
+                                            let infos = ok.peers.into_iter()
+                                                .map(|p| DiscoveredPeer {
+                                                    peer_id: p.peer_id,
+                                                    addrs: p.addrs,
+                                                })
+                                                .collect();
+                                            let _ = tx.send(infos);
+                                        }
+                                    }
+                                    kad::QueryResult::GetClosestPeers(Err(_)) if step.last => {
+                                        if let Some(tx) = pending_finds.remove(&id) {
+                                            let _ = tx.send(Vec::new());
+                                        }
+                                    }
+                                    kad::QueryResult::GetProviders(Ok(ok)) => {
+                                        if let Some((collected, _)) =
+                                            pending_providers.get_mut(&id)
+                                        {
+                                            if let kad::GetProvidersOk::FoundProviders {
+                                                providers, ..
+                                            } = ok
+                                            {
+                                                for p in providers {
+                                                    if !collected.contains(&p) {
+                                                        collected.push(p);
+                                                    }
+                                                }
+                                            }
+                                        }
+                                        if step.last {
+                                            if let Some((providers, tx)) =
+                                                pending_providers.remove(&id)
+                                            {
+                                                let _ = tx.send(providers);
+                                            }
+                                        }
+                                    }
+                                    kad::QueryResult::GetProviders(Err(_)) if step.last => {
+                                        if let Some((providers, tx)) =
+                                            pending_providers.remove(&id)
+                                        {
+                                            let _ = tx.send(providers);
+                                        }
+                                    }
+                                    _ => {}
+                                }
+                            }
+                            _ => {}
+                        }
+                    }
+                }
+            }
+        });
+
+        Ok(Self {
+            peer_id,
+            cmd_tx,
+            listen_rx: Mutex::new(listen_rx),
+            control,
+        })
+    }
+
+    pub async fn listen(&self, addr: Multiaddr) -> Multiaddr {
+        self.cmd_tx
+            .send(Command::Listen(addr))
+            .expect("swarm task alive");
+        let mut rx = self.listen_rx.lock().await;
+        rx.recv().await.expect("listen address arrives")
+    }
+
+    pub fn dial(&self, addr: Multiaddr) {
+        let _ = self.cmd_tx.send(Command::Dial(addr));
+    }
+
+    /// Añade un peer al routing table de Kademlia. Punto de entrada
+    /// para bootstrap: tras esto, el nodo puede dirigir queries DHT
+    /// a través de este peer.
+    pub fn add_dht_peer(&self, peer: PeerId, addr: Multiaddr) {
+        let _ = self.cmd_tx.send(Command::AddDhtPeer(peer, addr));
+    }
+
+    /// Consulta el DHT por los peers más cercanos al `target` PeerId.
+    /// Devuelve la lista resuelta (vacía si la query falla o si no
+    /// hay peers conocidos). Bloquea hasta que la query completa.
+    pub async fn find_closest_peers(&self, target: PeerId) -> Vec<DiscoveredPeer> {
+        let (tx, rx) = oneshot::channel();
+        let _ = self
+            .cmd_tx
+            .send(Command::FindClosestPeers(target, tx));
+        rx.await.unwrap_or_default()
+    }
+
+    /// Anuncia en el DHT que este peer tiene el contenido identificado
+    /// por `key`. Otros peers pueden luego descubrirlo vía
+    /// `find_providers(key)`. Best-effort: si la replicación falla
+    /// inicialmente, el record vive en el store local.
+    pub fn start_providing(&self, key: &[u8]) {
+        let _ = self.cmd_tx.send(Command::StartProviding(key.to_vec()));
+    }
+
+    /// Consulta el DHT por peers que han anunciado proveer `key`.
+    /// Devuelve la lista de `PeerId`s que se reportan como providers.
+    /// Lista vacía si nadie anuncia.
+    pub async fn find_providers(&self, key: &[u8]) -> Vec<PeerId> {
+        let (tx, rx) = oneshot::channel();
+        let _ = self
+            .cmd_tx
+            .send(Command::GetProviders(key.to_vec(), tx));
+        rx.await.unwrap_or_default()
+    }
+}

crates/modules/semantic_dht/minga-p2p/src/peer.rs

@@ -0,0 +1,313 @@
+//! `MingaPeer`: API de alto nivel para un nodo Minga "always-on".
+//!
+//! Envuelve `LibP2pNode` con estado compartido (`Mst` + `MemStore` +
+//! `AttestationStore` + `Keypair`) protegido por un `Mutex` async, y
+//! expone:
+//! - `run_passive_accept()`: lanza un bucle que acepta streams de
+//!   sync continuamente, procesa cada uno en una task paralela, y
+//!   mergea el resultado al estado compartido.
+//! - `sync_with(peer_id)`: inicia un sync activo con un peer conocido.
+//! - `snapshot()`: instantánea del estado actual.
+//!
+//! Modelo de concurrencia: cada sync entrante toma un *clone* del
+//! estado, ejecuta la sesión sobre la copia, y al terminar mergea las
+//! novedades al estado compartido. Múltiples syncs pueden correr en
+//! paralelo; el merge final adquiere el lock brevemente. Eventualmente
+//! consistente: un sync que empezó antes que un merge terminado puede
+//! no ver esas novedades, pero el siguiente sync sí.
+
+use std::path::Path;
+use std::sync::Arc;
+
+use futures::StreamExt;
+use libp2p::{Multiaddr, PeerId, Stream};
+use tokio::sync::Mutex;
+use tokio_util::compat::FuturesAsyncReadCompatExt;
+
+use minga_core::{AttestationStore, ContentHash, Keypair, MemStore, Mst, NodeStore, SemanticNode};
+use minga_store::{PersistentRepo, StoreError};
+
+use crate::async_driver::{run_sync_async, AsyncSyncError};
+use crate::network::{DiscoveredPeer, LibP2pNode, NodeError, SYNC_PROTOCOL};
+use crate::session::SyncSession;
+
+#[derive(Debug, thiserror::Error)]
+pub enum PeerSyncError {
+    #[error("open stream: {0}")]
+    OpenStream(#[from] libp2p_stream::OpenStreamError),
+
+    #[error("sync: {0}")]
+    AsyncSync(#[from] AsyncSyncError),
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum PeerOpenError {
+    #[error("network: {0}")]
+    Network(#[from] NodeError),
+
+    #[error("store: {0}")]
+    Store(#[from] StoreError),
+}
+
+struct PeerState {
+    mst: Mst,
+    store: MemStore,
+    attestations: AttestationStore,
+    keypair: Keypair,
+    /// Backing persistente opcional. Si está presente, todo cambio
+    /// de estado escribe a disco vía write-through.
+    persistent: Option<Arc<PersistentRepo>>,
+}
+
+pub struct MingaPeer {
+    node: LibP2pNode,
+    state: Arc<Mutex<PeerState>>,
+}
+
+impl MingaPeer {
+    pub fn new(
+        keypair: Keypair,
+        mst: Mst,
+        store: MemStore,
+        attestations: AttestationStore,
+    ) -> Result<Self, NodeError> {
+        let node = LibP2pNode::new()?;
+        let state = Arc::new(Mutex::new(PeerState {
+            mst,
+            store,
+            attestations,
+            keypair,
+            persistent: None,
+        }));
+        Ok(Self { node, state })
+    }
+
+    /// Abre o crea un peer persistente sobre `path`. Si el directorio
+    /// no contiene un repo, se crea vacío. Si lo contiene, se carga
+    /// el estado completo (MST, nodos, atestaciones) en memoria.
+    /// Cualquier cambio posterior se escribe a disco vía write-through.
+    pub fn open(keypair: Keypair, path: impl AsRef<Path>) -> Result<Self, PeerOpenError> {
+        let repo = Arc::new(PersistentRepo::open(path)?);
+
+        // Cargar MST desde disco.
+        let mut mst = Mst::new();
+        for r in repo.mst.iter() {
+            mst.insert(r?);
+        }
+
+        // Cargar nodos desde disco.
+        let mut store = MemStore::new();
+        for r in repo.nodes.iter() {
+            let (h, node) = r?;
+            store.put_chunked(h, node);
+        }
+
+        // Cargar atestaciones desde disco.
+        let mut attestations = AttestationStore::new();
+        for r in repo.attestations.iter() {
+            let att = r?;
+            // `add` re-verifica criptográficamente. Lo persistido ya
+            // estaba verificado, pero re-validar es cheap insurance.
+            let _ = attestations.add(att);
+        }
+
+        let node = LibP2pNode::new()?;
+        let state = Arc::new(Mutex::new(PeerState {
+            mst,
+            store,
+            attestations,
+            keypair,
+            persistent: Some(repo),
+        }));
+        Ok(Self { node, state })
+    }
+
+    pub fn peer_id(&self) -> PeerId {
+        self.node.peer_id
+    }
+
+    pub async fn listen(&self, addr: Multiaddr) -> Multiaddr {
+        self.node.listen(addr).await
+    }
+
+    pub fn dial(&self, addr: Multiaddr) {
+        self.node.dial(addr);
+    }
+
+    /// Añade un peer al routing table de Kademlia (bootstrap).
+    pub fn add_dht_peer(&self, peer: PeerId, addr: Multiaddr) {
+        self.node.add_dht_peer(peer, addr);
+    }
+
+    /// Consulta DHT por los peers más cercanos al `target`.
+    pub async fn find_closest_peers(&self, target: PeerId) -> Vec<DiscoveredPeer> {
+        self.node.find_closest_peers(target).await
+    }
+
+    /// Anuncia en el DHT que este peer provee el contenido `hash`.
+    /// Otros peers podrán descubrirlo vía `find_providers(hash)`.
+    pub fn announce_provider(&self, hash: ContentHash) {
+        self.node.start_providing(&hash.0);
+    }
+
+    /// Consulta el DHT por peers que han anunciado proveer este
+    /// contenido. La unión de los `PeerId`s permite a quien busque
+    /// `hash` decidir a quién dial directamente para sincronizar.
+    pub async fn find_providers(&self, hash: ContentHash) -> Vec<PeerId> {
+        self.node.find_providers(&hash.0).await
+    }
+
+    /// Lanza el bucle de aceptación pasiva. Devuelve un `JoinHandle`
+    /// que el caller puede mantener vivo (o ignorar — la task se
+    /// aborta al cerrar el runtime).
+    ///
+    /// Cada stream entrante dispara un sync en una task aislada que
+    /// trabaja sobre un clone del estado y mergea al final.
+    pub fn run_passive_accept(&self) -> tokio::task::JoinHandle<()> {
+        let mut control = self.node.control.clone();
+        let state = Arc::clone(&self.state);
+        tokio::spawn(async move {
+            let mut incoming = control
+                .accept(SYNC_PROTOCOL)
+                .expect("only one accept handle per protocol");
+            while let Some((_peer, stream)) = incoming.next().await {
+                let state = Arc::clone(&state);
+                tokio::spawn(handle_incoming(stream, state));
+            }
+        })
+    }
+
+    /// Inicia un sync activo con un peer del que ya tenemos conexión
+    /// (vía `dial` previo). Toma un snapshot del estado, corre la
+    /// sesión, y mergea novedades al volver.
+    pub async fn sync_with(&self, peer_id: PeerId) -> Result<(), PeerSyncError> {
+        let mut control = self.node.control.clone();
+        let stream = control.open_stream(peer_id, SYNC_PROTOCOL).await?;
+        let session = self.snapshot_session().await;
+        let result = run_sync_async(session, stream.compat()).await?;
+        self.merge_back(result).await;
+        Ok(())
+    }
+
+    async fn snapshot_session(&self) -> SyncSession {
+        let s = self.state.lock().await;
+        SyncSession::new(
+            s.mst.clone(),
+            s.store.clone(),
+            s.attestations.clone(),
+            s.keypair.clone(),
+        )
+    }
+
+    async fn merge_back(&self, session: SyncSession) {
+        let (new_mst, new_store, new_atts) = session.into_parts();
+        let mut s = self.state.lock().await;
+        merge_into_state(&mut s, new_mst, new_store, new_atts);
+    }
+
+    /// Instantánea del estado actual (mst + store + attestations).
+    pub async fn snapshot(&self) -> (Mst, MemStore, AttestationStore) {
+        let s = self.state.lock().await;
+        (s.mst.clone(), s.store.clone(), s.attestations.clone())
+    }
+
+    /// Inserta un árbol directamente en el estado del peer (sin sync).
+    /// Si el peer está respaldado por disco, también lo persiste.
+    /// Anuncia automáticamente al peer como proveedor del contenido en
+    /// el DHT — de esa forma cualquier otro peer puede descubrirlo
+    /// preguntando "¿quién tiene este hash?".
+    /// Devuelve el `ContentHash` raíz del árbol.
+    pub async fn ingest(&self, node: &SemanticNode) -> ContentHash {
+        let mut s = self.state.lock().await;
+        let h = s.store.put(node);
+        s.mst.insert(h);
+        if let Some(repo) = &s.persistent {
+            let _ = repo.nodes.put(node);
+            let _ = repo.mst.insert(h);
+        }
+        drop(s);
+
+        // Anunciamos como proveedores en el DHT. Best-effort: si no
+        // hay peers cercanos para replicar, el record vive local hasta
+        // que llegue una conexión.
+        self.node.start_providing(&h.0);
+
+        h
+    }
+
+    /// Inserta una atestación en el peer. Si el peer es persistente,
+    /// también la escribe a disco. Falla si la firma no verifica.
+    pub async fn ingest_attestation(
+        &self,
+        att: minga_core::Attestation,
+    ) -> Result<(), minga_core::AttestationError> {
+        let mut s = self.state.lock().await;
+        s.attestations.add(att.clone())?;
+        if let Some(repo) = &s.persistent {
+            let _ = repo.attestations.add(att);
+        }
+        Ok(())
+    }
+
+    /// Fuerza un flush del backing persistente a disco. No hace nada
+    /// si el peer es solo en memoria.
+    pub async fn flush(&self) -> Result<(), StoreError> {
+        let s = self.state.lock().await;
+        if let Some(repo) = &s.persistent {
+            repo.flush()?;
+        }
+        Ok(())
+    }
+}
+
+async fn handle_incoming(stream: Stream, state: Arc<Mutex<PeerState>>) {
+    let session = {
+        let s = state.lock().await;
+        SyncSession::new(
+            s.mst.clone(),
+            s.store.clone(),
+            s.attestations.clone(),
+            s.keypair.clone(),
+        )
+    };
+    if let Ok(result) = run_sync_async(session, stream.compat()).await {
+        let (new_mst, new_store, new_atts) = result.into_parts();
+        let mut s = state.lock().await;
+        merge_into_state(&mut s, new_mst, new_store, new_atts);
+    }
+    // Errores de sync se ignoran: cada sesión es independiente, una
+    // sesión rota no debería tumbar el peer entero. Una iteración
+    // futura puede contar errores para telemetría.
+}
+
+fn merge_into_state(
+    state: &mut PeerState,
+    new_mst: Mst,
+    new_store: MemStore,
+    new_atts: AttestationStore,
+) {
+    // Write-through: cada inserción en memoria también va al backing
+    // persistente si existe. Errores de IO se ignoran (best-effort);
+    // el estado en memoria sigue siendo la fuente de verdad inmediata
+    // y un siguiente sync re-popula lo que se haya perdido.
+    for h in new_mst.iter() {
+        state.mst.insert(*h);
+        if let Some(repo) = &state.persistent {
+            let _ = repo.mst.insert(*h);
+        }
+    }
+    for (h, node) in new_store.iter() {
+        state.store.put_chunked(*h, node.clone());
+        if let Some(repo) = &state.persistent {
+            let _ = repo.nodes.put_chunked(*h, node);
+        }
+    }
+    for att in new_atts.all() {
+        if state.attestations.add(att.clone()).is_ok() {
+            // Solo persistimos las que pasaron verificación en memoria.
+            if let Some(repo) = &state.persistent {
+                let _ = repo.attestations.add(att.clone());
+            }
+        }
+    }
+}

crates/modules/semantic_dht/minga-p2p/src/session.rs

@@ -0,0 +1,461 @@
+//! Máquina de estados de sincronización recursiva sobre la estructura
+//! del MST, con verificación criptográfica de cada nodo entregado.
+//!
+//! La sesión es **pura**: no hace IO, no toca la red, no usa async. El
+//! transporte la alimenta vía `handle(msg)` y consume sus salidas como
+//! `Vec<Message>`.
+//!
+//! ## Algoritmo
+//!
+//! 1. Cada peer construye al inicio un `own_probes: HashMap<ContentHash,
+//!    NodeProbe>` que indexa cada nodo interno de su MST por su hash
+//!    Merkle de subárbol. Es la tabla con la que respondemos
+//!    `ProbeReq`s en `O(1)`.
+//!
+//! 2. Cada peer envía `Hello` con el hash de su raíz. Si el peer
+//!    contrario reconoce ese hash en su propio `own_probes` (o coincide
+//!    con su propia raíz, o es la raíz vacía), no hay nada estructural
+//!    que descubrir — la rama está ya alineada.
+//!
+//! 3. Si el hash no se reconoce, el peer emite un `ProbeReq` para
+//!    pedirle al otro la estructura de ese subárbol. Cuando llega el
+//!    `ProbeRes`, el peer:
+//!    - Para cada **clave** del probe que no tiene en su MST, programa
+//!      un `Fetch` (la clave entrará al MST cuando llegue su `Deliver`).
+//!    - Para cada **child_hash** del probe que no aparece en
+//!      `own_probes`, recurre con un nuevo `ProbeReq`. Si el child_hash
+//!      ya está en `own_probes`, la rama se poda — toda esa subestructura
+//!      es idéntica a la nuestra.
+//!
+//! 4. Cuando un peer recibe un `Deliver`, verifica que el hash
+//!    anunciado coincida con el `hash_stored` real del nodo. Si no,
+//!    descarta. Si sí, inserta en el `MemStore` y, si el hash venía de
+//!    la raíz del MST del peer (no de un descendiente), también lo
+//!    inserta en su MST.
+//!
+//! 5. Cada `StoredNode` recibido contiene los hashes de sus hijos. Si
+//!    el receptor no los tiene, los pide vía `Fetch` (sync transitivo).
+//!
+//! 6. Un peer envía `Done` cuando: emitió y recibió `Hello`, no tiene
+//!    probes pendientes, ni fetches pendientes (raíz o hijo). La sesión
+//!    cierra cuando ambos `Done`s han cruzado.
+
+use minga_core::{
+    cas::ContentHash, empty_subtree_hash, hash_stored, AttestationStore, Did, Keypair, MemStore,
+    Mst, NodeProbe, NodeStore,
+};
+use rand::rngs::OsRng;
+use rand::RngCore;
+use std::collections::{HashMap, HashSet};
+
+use crate::message::Message;
+
+/// Construye el payload firmado del `Hello` con orden fijo:
+/// `verifier_nonce(32) || peer_did(32) || root_subtree_hash(32) = 96 bytes`.
+/// El `verifier_nonce` es el nonce que emitió el peer que verificará
+/// la firma; al firmar sobre él se vincula la firma a esta sesión.
+/// Cualquier cambio al formato es incompatible al protocolo.
+pub(crate) fn hello_payload(
+    verifier_nonce: &[u8; 32],
+    did: &Did,
+    root: &ContentHash,
+) -> [u8; 96] {
+    let mut p = [0u8; 96];
+    p[..32].copy_from_slice(verifier_nonce);
+    p[32..64].copy_from_slice(&did.0);
+    p[64..].copy_from_slice(&root.0);
+    p
+}
+
+pub struct SyncSession {
+    mst: Mst,
+    store: MemStore,
+    attestations: AttestationStore,
+
+    /// Llave del peer local: firma el `Hello` y queda asociada al
+    /// `Did` que el peer remoto verá.
+    keypair: Keypair,
+
+    /// Identidad del peer remoto, capturada tras verificar la firma
+    /// de su `Hello`.
+    peer_did: Option<Did>,
+
+    own_probes: HashMap<ContentHash, NodeProbe>,
+    own_root_subtree_hash: ContentHash,
+
+    awaited_probes: HashSet<ContentHash>,
+    seen_probes: HashSet<ContentHash>,
+    awaiting_root: HashSet<ContentHash>,
+    awaiting_child: HashSet<ContentHash>,
+
+    rejected_hellos: usize,
+    rejected_delivers: usize,
+    /// Contador de atestaciones rechazadas: firma rota, llegada antes
+    /// de autenticar al peer, o cualquier otra inconsistencia que el
+    /// `AttestationStore` rechace.
+    rejected_attests: usize,
+
+    /// Nonce aleatorio que **nosotros** emitimos en `Challenge`. La
+    /// firma del `Hello` del peer debe ser sobre este nonce.
+    self_nonce: [u8; 32],
+    /// Nonce que el peer publicó en su `Challenge` — sobre este
+    /// nonce firmamos nosotros nuestro `Hello`.
+    peer_nonce: Option<[u8; 32]>,
+
+    sent_challenge: bool,
+    received_challenge: bool,
+    sent_hello: bool,
+    received_hello: bool,
+    sent_attestations: bool,
+    sent_done: bool,
+    received_done: bool,
+}
+
+impl SyncSession {
+    pub fn new(
+        mst: Mst,
+        store: MemStore,
+        attestations: AttestationStore,
+        keypair: Keypair,
+    ) -> Self {
+        let own_probes = mst.build_probe_index();
+        let own_root_subtree_hash = mst.root_hash();
+        let mut self_nonce = [0u8; 32];
+        OsRng.fill_bytes(&mut self_nonce);
+        Self {
+            mst,
+            store,
+            attestations,
+            keypair,
+            peer_did: None,
+            own_probes,
+            own_root_subtree_hash,
+            awaited_probes: HashSet::new(),
+            seen_probes: HashSet::new(),
+            awaiting_root: HashSet::new(),
+            awaiting_child: HashSet::new(),
+            rejected_hellos: 0,
+            rejected_delivers: 0,
+            rejected_attests: 0,
+            self_nonce,
+            peer_nonce: None,
+            sent_challenge: false,
+            received_challenge: false,
+            sent_hello: false,
+            received_hello: false,
+            sent_attestations: false,
+            sent_done: false,
+            received_done: false,
+        }
+    }
+
+    /// Conveniencia para sesiones sin atestaciones previas. Equivalente
+    /// a `new(mst, store, AttestationStore::new(), keypair)`.
+    pub fn without_attestations(mst: Mst, store: MemStore, keypair: Keypair) -> Self {
+        Self::new(mst, store, AttestationStore::new(), keypair)
+    }
+
+    /// Mensaje inicial: `Challenge` con un nonce aleatorio. El `Hello`
+    /// y las atestaciones llegarán como respuesta al `Challenge` del
+    /// otro peer (cuando lo recibamos, ya tendremos su nonce sobre el
+    /// que firmar nuestra identidad).
+    pub fn start(&mut self) -> Vec<Message> {
+        if self.sent_challenge {
+            return Vec::new();
+        }
+        self.sent_challenge = true;
+        let mut out = vec![Message::Challenge {
+            nonce: self.self_nonce,
+        }];
+        out.extend(self.maybe_done());
+        out
+    }
+
+    pub fn handle(&mut self, msg: Message) -> Vec<Message> {
+        let mut out = Vec::new();
+        match msg {
+            Message::Challenge { nonce } => {
+                if self.received_challenge {
+                    // Challenge duplicado: ignoramos. Un peer
+                    // legítimo no debería enviar dos.
+                    return out;
+                }
+                self.received_challenge = true;
+                self.peer_nonce = Some(nonce);
+
+                // Ahora podemos firmar nuestro Hello sobre el nonce
+                // del peer — lo que ata la firma a esta sesión.
+                let payload =
+                    hello_payload(&nonce, &self.keypair.did(), &self.own_root_subtree_hash);
+                let signature = self.keypair.sign(&payload);
+                self.sent_hello = true;
+                out.push(Message::Hello {
+                    peer_did: self.keypair.did(),
+                    root_subtree_hash: self.own_root_subtree_hash,
+                    signature,
+                });
+
+                // Empuje de atestaciones: el peer ya nos verificará
+                // como remitente cuando reciba nuestro Hello.
+                let atts: Vec<_> = self.attestations.all().cloned().collect();
+                if !atts.is_empty() {
+                    out.push(Message::AttestPush { attestations: atts });
+                }
+                self.sent_attestations = true;
+            }
+
+            Message::Hello {
+                peer_did,
+                root_subtree_hash,
+                signature,
+            } => {
+                // ── Autenticación del peer + anti-replay ─────────
+                // La firma debe ser sobre nuestro `self_nonce` (que
+                // emitimos en nuestro Challenge), atándola a esta
+                // sesión. Un Hello capturado de otra sesión tendría
+                // un nonce distinto y la verificación fallaría.
+                let payload = hello_payload(&self.self_nonce, &peer_did, &root_subtree_hash);
+                if !peer_did.verify(&payload, &signature) {
+                    self.rejected_hellos += 1;
+                    return out;
+                }
+                self.peer_did = Some(peer_did);
+                self.received_hello = true;
+                if self.should_probe(&root_subtree_hash) {
+                    self.awaited_probes.insert(root_subtree_hash);
+                    out.push(Message::ProbeReq {
+                        subtree_hash: root_subtree_hash,
+                    });
+                }
+            }
+
+            Message::ProbeReq { subtree_hash } => {
+                let probe = self.own_probes.get(&subtree_hash).cloned();
+                // Si el subárbol pedido era vacío (o desconocido para
+                // nosotros), respondemos con `None` — el peer lo
+                // tratará como un punto sin descendientes que descubrir.
+                out.push(Message::ProbeRes {
+                    subtree_hash,
+                    probe,
+                });
+            }
+
+            Message::ProbeRes {
+                subtree_hash,
+                probe,
+            } => {
+                self.awaited_probes.remove(&subtree_hash);
+                self.seen_probes.insert(subtree_hash);
+                if let Some(probe) = probe {
+                    out.extend(self.process_probe(&probe));
+                }
+            }
+
+            Message::Fetch { hash } => {
+                if let Some(stored) = self.store.get(&hash).cloned() {
+                    out.push(Message::Deliver { hash, stored });
+                }
+                // Si no lo tenemos, callamos. El peer no debería estar
+                // pidiéndonos algo que no le hayamos anunciado.
+            }
+
+            Message::Deliver { hash, stored } => {
+                // ── Verificación criptográfica ────────────────────
+                // Recomputamos el hash del nodo entregado a partir de
+                // sus componentes. Si no coincide con el anunciado,
+                // alguien (peer malicioso o ruido en transporte) está
+                // intentando colar contenido distinto bajo un hash que
+                // no le corresponde. Descartamos silenciosamente y
+                // contamos para diagnóstico.
+                if hash_stored(&stored) != hash {
+                    self.rejected_delivers += 1;
+                    // No tocamos awaiting_*: la solicitud sigue
+                    // pendiente y el peer (legítimo o no) puede
+                    // reintentarla.
+                    return out;
+                }
+
+                let was_root = self.awaiting_root.remove(&hash);
+                self.awaiting_child.remove(&hash);
+
+                // Antes de mover `stored`, descubrimos qué hijos
+                // faltan y los pedimos.
+                let mut new_fetches = Vec::new();
+                for ch in &stored.children {
+                    if !self.store.contains(ch)
+                        && !self.awaiting_root.contains(ch)
+                        && !self.awaiting_child.contains(ch)
+                    {
+                        self.awaiting_child.insert(*ch);
+                        new_fetches.push(*ch);
+                    }
+                }
+
+                self.store.put_chunked(hash, stored);
+                if was_root {
+                    self.mst.insert(hash);
+                }
+
+                for h in new_fetches {
+                    out.push(Message::Fetch { hash: h });
+                }
+            }
+
+            Message::AttestPush { attestations } => {
+                // Antes de procesar atestaciones del peer, exigimos
+                // haber autenticado su identidad. Un push antes del
+                // `Hello` es protocolo malformado o ataque — todas las
+                // atestaciones se cuentan como rechazadas.
+                if !self.received_hello {
+                    self.rejected_attests += attestations.len();
+                    return out;
+                }
+                for att in attestations {
+                    // `AttestationStore::add` re-verifica cada firma.
+                    // Una sola atestación corrupta no contamina las
+                    // demás del lote.
+                    if self.attestations.add(att).is_err() {
+                        self.rejected_attests += 1;
+                    }
+                }
+            }
+
+            Message::Done => {
+                self.received_done = true;
+            }
+        }
+        out.extend(self.maybe_done());
+        out
+    }
+
+    fn process_probe(&mut self, probe: &NodeProbe) -> Vec<Message> {
+        let mut out = Vec::new();
+
+        // Cada clave del probe que no tenemos pasa a `awaiting_root` y
+        // generamos un Fetch. Si ya está en el store (sin estar aún en
+        // el MST), simplemente la promovemos al MST sin pedirla.
+        for k in &probe.keys {
+            if self.mst.contains(k) {
+                continue;
+            }
+            if self.store.contains(k) {
+                self.mst.insert(*k);
+                continue;
+            }
+            if self.awaiting_root.contains(k) {
+                continue;
+            }
+            self.awaiting_root.insert(*k);
+            out.push(Message::Fetch { hash: *k });
+        }
+
+        // Para cada subárbol hijo, decidimos si recurrir o podar:
+        //   - el vacío se reconoce por hash sin red,
+        //   - los que ya tenemos en `own_probes` (igualdad de hash =
+        //     subestructura idéntica) se podan,
+        //   - los ya vistos o solicitados no se duplican,
+        //   - el resto dispara un `ProbeReq` recursivo.
+        for ch in &probe.child_hashes {
+            if self.should_probe(ch) {
+                self.awaited_probes.insert(*ch);
+                out.push(Message::ProbeReq { subtree_hash: *ch });
+            }
+        }
+
+        out
+    }
+
+    /// Decide si vale la pena solicitar un probe sobre `h`. Cuatro
+    /// razones para NO pedirlo:
+    /// - es el subárbol vacío (lo conocemos por convención),
+    /// - coincide con nuestra propia raíz (igualdad estructural),
+    /// - aparece en `own_probes` (ya tenemos un subárbol idéntico),
+    /// - ya lo solicitamos o ya lo recibimos.
+    fn should_probe(&self, h: &ContentHash) -> bool {
+        if *h == empty_subtree_hash() {
+            return false;
+        }
+        if *h == self.own_root_subtree_hash {
+            return false;
+        }
+        if self.own_probes.contains_key(h) {
+            return false;
+        }
+        if self.awaited_probes.contains(h) || self.seen_probes.contains(h) {
+            return false;
+        }
+        true
+    }
+
+    fn maybe_done(&mut self) -> Vec<Message> {
+        if self.sent_done {
+            return Vec::new();
+        }
+        if !self.sent_challenge || !self.received_challenge {
+            return Vec::new();
+        }
+        if !self.sent_hello || !self.received_hello {
+            return Vec::new();
+        }
+        if !self.sent_attestations {
+            return Vec::new();
+        }
+        if !self.awaited_probes.is_empty() {
+            return Vec::new();
+        }
+        if !self.awaiting_root.is_empty() || !self.awaiting_child.is_empty() {
+            return Vec::new();
+        }
+        self.sent_done = true;
+        vec![Message::Done]
+    }
+
+    pub fn is_done(&self) -> bool {
+        self.sent_done && self.received_done
+    }
+
+    pub fn rejected_delivers(&self) -> usize {
+        self.rejected_delivers
+    }
+
+    pub fn rejected_hellos(&self) -> usize {
+        self.rejected_hellos
+    }
+
+    pub fn rejected_attests(&self) -> usize {
+        self.rejected_attests
+    }
+
+    pub fn attestations(&self) -> &AttestationStore {
+        &self.attestations
+    }
+
+    /// Identidad del peer remoto, capturada tras verificar su `Hello`.
+    /// `None` si todavía no llegó un `Hello` válido.
+    pub fn peer_did(&self) -> Option<Did> {
+        self.peer_did
+    }
+
+    pub fn local_did(&self) -> Did {
+        self.keypair.did()
+    }
+
+    /// Nonce aleatorio que esta sesión emitió en su `Challenge`.
+    /// Expuesto principalmente para tests y debugging — el nonce
+    /// viaja en claro por el wire y no es secreto.
+    pub fn self_nonce(&self) -> [u8; 32] {
+        self.self_nonce
+    }
+
+    pub fn mst(&self) -> &Mst {
+        &self.mst
+    }
+
+    pub fn store(&self) -> &MemStore {
+        &self.store
+    }
+
+    pub fn into_parts(self) -> (Mst, MemStore, AttestationStore) {
+        (self.mst, self.store, self.attestations)
+    }
+}

crates/modules/semantic_dht/minga-p2p/tests/async_driver_invariants.rs

@@ -0,0 +1,161 @@
+//! Tests del `run_sync_async` sobre canales async in-memory.
+//!
+//! Equivalentes a los del harness síncrono pero ejecutados sobre
+//! `tokio::io::duplex` — la misma lógica protocolar viajando sobre
+//! bytes serializados con postcard, encuadrados con length-prefix, y
+//! transportados por una pipa async. Si esto pasa, lo único que falta
+//! para el sync sobre TCP/QUIC/libp2p es enchufar el transporte real.
+
+use minga_core::{parse, ContentHash, Keypair, MemStore, Mst, NodeStore};
+use minga_p2p::{run_sync_async, SyncSession};
+
+fn kp(seed: u8) -> Keypair {
+    Keypair::from_seed(&[seed; 32])
+}
+
+fn build_repo(sources: &[&str]) -> (Mst, MemStore, Vec<ContentHash>) {
+    let mut mst = Mst::new();
+    let mut store = MemStore::new();
+    let mut roots = Vec::new();
+    for src in sources {
+        let n = parse::rust(src).unwrap();
+        let h = store.put(&n);
+        mst.insert(h);
+        roots.push(h);
+    }
+    (mst, store, roots)
+}
+
+#[tokio::test]
+async fn async_sync_identical_repos() {
+    let sources = &["fn add(x: i32, y: i32) -> i32 { x + y }"];
+    let (mst_a, store_a, _) = build_repo(sources);
+    let (mst_b, store_b, _) = build_repo(sources);
+
+    let session_a = SyncSession::without_attestations(mst_a, store_a, kp(1));
+    let session_b = SyncSession::without_attestations(mst_b, store_b, kp(2));
+
+    let (a_stream, b_stream) = tokio::io::duplex(64 * 1024);
+
+    let task_a = tokio::spawn(run_sync_async(session_a, a_stream));
+    let task_b = tokio::spawn(run_sync_async(session_b, b_stream));
+
+    let a = task_a.await.unwrap().unwrap();
+    let b = task_b.await.unwrap().unwrap();
+
+    assert_eq!(a.mst().root_hash(), b.mst().root_hash());
+}
+
+#[tokio::test]
+async fn async_sync_one_empty_pulls_everything() {
+    let sources = &["fn complex(x: i32) -> i32 { let y = x * 2; y + 1 }"];
+    let (mst_a, store_a, _) = build_repo(sources);
+    let (mst_b, store_b, _) = build_repo(&[]);
+    let store_a_size = store_a.len();
+
+    let session_a = SyncSession::without_attestations(mst_a, store_a, kp(1));
+    let session_b = SyncSession::without_attestations(mst_b, store_b, kp(2));
+
+    let (a_stream, b_stream) = tokio::io::duplex(64 * 1024);
+
+    let task_a = tokio::spawn(run_sync_async(session_a, a_stream));
+    let task_b = tokio::spawn(run_sync_async(session_b, b_stream));
+
+    let a = task_a.await.unwrap().unwrap();
+    let b = task_b.await.unwrap().unwrap();
+
+    assert_eq!(a.mst().root_hash(), b.mst().root_hash());
+    assert_eq!(a.store().len(), b.store().len());
+    assert_eq!(b.store().len(), store_a_size);
+}
+
+#[tokio::test]
+async fn async_sync_disjoint_sets_merge() {
+    let only_a = &[
+        "fn alpha() -> i32 { 1 }",
+        "fn beta(x: i32) -> i32 { x + 1 }",
+    ];
+    let only_b = &[
+        "fn gamma(y: i32) -> bool { y > 0 }",
+        "fn delta() -> &'static str { \"hello\" }",
+    ];
+
+    let (mst_a, store_a, _) = build_repo(only_a);
+    let (mst_b, store_b, _) = build_repo(only_b);
+
+    let session_a = SyncSession::without_attestations(mst_a, store_a, kp(1));
+    let session_b = SyncSession::without_attestations(mst_b, store_b, kp(2));
+
+    let (a_stream, b_stream) = tokio::io::duplex(64 * 1024);
+
+    let task_a = tokio::spawn(run_sync_async(session_a, a_stream));
+    let task_b = tokio::spawn(run_sync_async(session_b, b_stream));
+
+    let a = task_a.await.unwrap().unwrap();
+    let b = task_b.await.unwrap().unwrap();
+
+    assert_eq!(a.mst().root_hash(), b.mst().root_hash());
+    assert_eq!(a.mst().len(), 4);
+}
+
+#[tokio::test]
+async fn async_sync_propagates_authenticated_identity() {
+    // Cada peer debe acabar conociendo el DID verificado del otro,
+    // exactamente como en el harness síncrono.
+    let kp_a = kp(10);
+    let kp_b = kp(20);
+    let did_a = kp_a.did();
+    let did_b = kp_b.did();
+
+    let session_a = SyncSession::without_attestations(Mst::new(), MemStore::new(), kp_a);
+    let session_b = SyncSession::without_attestations(Mst::new(), MemStore::new(), kp_b);
+
+    let (a_stream, b_stream) = tokio::io::duplex(64 * 1024);
+
+    let task_a = tokio::spawn(run_sync_async(session_a, a_stream));
+    let task_b = tokio::spawn(run_sync_async(session_b, b_stream));
+
+    let a = task_a.await.unwrap().unwrap();
+    let b = task_b.await.unwrap().unwrap();
+
+    assert_eq!(a.peer_did(), Some(did_b));
+    assert_eq!(b.peer_did(), Some(did_a));
+}
+
+#[tokio::test]
+async fn async_sync_propagates_attestations() {
+    use minga_core::{Attestation, AttestationStore};
+
+    let kp_a = kp(30);
+    let kp_b = kp(40);
+
+    let (mst_a, store_a, roots_a) = build_repo(&["fn from_a() -> i32 { 1 }"]);
+    let (mst_b, store_b, roots_b) = build_repo(&["fn from_b() -> i32 { 2 }"]);
+
+    let mut atts_a = AttestationStore::new();
+    atts_a
+        .add(Attestation::create(&kp_a, roots_a[0]))
+        .unwrap();
+
+    let mut atts_b = AttestationStore::new();
+    atts_b
+        .add(Attestation::create(&kp_b, roots_b[0]))
+        .unwrap();
+
+    let session_a = SyncSession::new(mst_a, store_a, atts_a, kp_a.clone());
+    let session_b = SyncSession::new(mst_b, store_b, atts_b, kp_b.clone());
+
+    let (a_stream, b_stream) = tokio::io::duplex(128 * 1024);
+
+    let task_a = tokio::spawn(run_sync_async(session_a, a_stream));
+    let task_b = tokio::spawn(run_sync_async(session_b, b_stream));
+
+    let a = task_a.await.unwrap().unwrap();
+    let b = task_b.await.unwrap().unwrap();
+
+    // Los DIDs y atestaciones cruzaron correctamente sobre el wire.
+    assert_eq!(a.attestations().authors_of(&roots_a[0]), vec![kp_a.did()]);
+    assert_eq!(a.attestations().authors_of(&roots_b[0]), vec![kp_b.did()]);
+    assert_eq!(b.attestations().authors_of(&roots_a[0]), vec![kp_a.did()]);
+    assert_eq!(b.attestations().authors_of(&roots_b[0]), vec![kp_b.did()]);
+}

crates/modules/semantic_dht/minga-p2p/tests/kad_discovery.rs

@@ -0,0 +1,189 @@
+//! Tests de descubrimiento vía Kademlia DHT.
+
+use std::time::Duration;
+
+use minga_core::{parse, AttestationStore, Keypair, MemStore, Mst, NodeStore};
+use minga_p2p::{LibP2pNode, MingaPeer};
+
+#[tokio::test]
+async fn identify_auto_populates_kad_routing_table() {
+    // Sin `add_dht_peer` manual: solo dial. Identify intercambia
+    // direcciones automáticamente y poblamos Kad con ellas. Tras
+    // unos cientos de ms, A puede consultar B vía DHT.
+    let a = LibP2pNode::new().unwrap();
+    let b = LibP2pNode::new().unwrap();
+
+    let addr_b = b.listen("/ip4/127.0.0.1/tcp/0".parse().unwrap()).await;
+
+    a.dial(addr_b);
+
+    // Margen para handshake Noise + Yamux + Identify.
+    tokio::time::sleep(Duration::from_millis(500)).await;
+
+    let result = a.find_closest_peers(b.peer_id).await;
+    assert!(
+        result.iter().any(|p| p.peer_id == b.peer_id),
+        "tras Identify, B debe estar en el routing de A. Obtuvo: {:?}",
+        result.iter().map(|p| p.peer_id).collect::<Vec<_>>()
+    );
+}
+
+#[tokio::test]
+async fn kad_two_node_basic_discovery() {
+    // A escucha. B dializa, añade A al routing table de Kad.
+    // Tras el handshake Kad, B puede consultar el DHT y encontrar A.
+    let a = LibP2pNode::new().unwrap();
+    let b = LibP2pNode::new().unwrap();
+
+    let addr_a = a.listen("/ip4/127.0.0.1/tcp/0".parse().unwrap()).await;
+
+    b.add_dht_peer(a.peer_id, addr_a.clone());
+    b.dial(addr_a.clone());
+
+    // Damos margen para handshake Noise+Yamux+Kad.
+    tokio::time::sleep(Duration::from_millis(300)).await;
+
+    let result = b.find_closest_peers(a.peer_id).await;
+    assert!(
+        result.iter().any(|p| p.peer_id == a.peer_id),
+        "B debe encontrar A vía DHT, obtuvo {:?}",
+        result
+    );
+}
+
+#[tokio::test]
+async fn kad_three_node_discovery_via_rendezvous() {
+    // Test canónico de descubrimiento DHT:
+    // - A es un peer "rendezvous" que pre-conoce a B y C (en una red
+    //   real, A los aprendería de los handshakes Kad cuando B y C se
+    //   conectan; aquí lo seedeamos explícitamente para no depender
+    //   de timing de propagación).
+    // - B solo conoce a A.
+    // - B pregunta al DHT por C: la query va a A, A responde con C,
+    //   B aprende la dirección de C sin haberle hablado nunca.
+    //
+    // Este es exactamente el patrón de IPFS, libp2p bootstrap nodes
+    // y cualquier P2P descentralizado real.
+
+    let a = LibP2pNode::new().unwrap(); // rendezvous
+    let b = LibP2pNode::new().unwrap();
+    let c = LibP2pNode::new().unwrap();
+
+    let addr_a = a.listen("/ip4/127.0.0.1/tcp/0".parse().unwrap()).await;
+    let addr_b = b.listen("/ip4/127.0.0.1/tcp/0".parse().unwrap()).await;
+    let addr_c = c.listen("/ip4/127.0.0.1/tcp/0".parse().unwrap()).await;
+
+    // A (el rendezvous) tiene a B y C en su routing table.
+    a.add_dht_peer(b.peer_id, addr_b);
+    a.add_dht_peer(c.peer_id, addr_c);
+
+    // B solo conoce a A.
+    b.add_dht_peer(a.peer_id, addr_a.clone());
+    b.dial(addr_a.clone());
+
+    // Margen para que la conexión Kad B↔A se establezca.
+    tokio::time::sleep(Duration::from_millis(300)).await;
+
+    // B pregunta al DHT por C. Su routing table solo tiene A; la
+    // query va a A; A responde con C de su table. B descubre.
+    let result = b.find_closest_peers(c.peer_id).await;
+    assert!(
+        result.iter().any(|p| p.peer_id == c.peer_id),
+        "B debe descubrir C vía A; obtuvo: {:?}",
+        result.iter().map(|p| p.peer_id).collect::<Vec<_>>()
+    );
+
+    // Y la dirección de C debe haber viajado en el resultado, así
+    // que B podría dialarlo directamente sin pasar por A.
+    let c_entry = result.iter().find(|p| p.peer_id == c.peer_id).unwrap();
+    assert!(!c_entry.addrs.is_empty(), "C debe venir con address resoluble");
+}
+
+#[tokio::test]
+async fn kad_discovery_then_sync() {
+    // Cierre del bucle: B descubre C vía DHT a través de A, y luego
+    // sincroniza directamente con C. Discovery + transport + sync
+    // protocolar autenticado, todo end-to-end sobre red real.
+
+    fn singleton(seed: u8, src: &str) -> MingaPeer {
+        let mut mst = Mst::new();
+        let mut store = MemStore::new();
+        let h = store.put(&parse::rust(src).unwrap());
+        mst.insert(h);
+        MingaPeer::new(
+            Keypair::from_seed(&[seed; 32]),
+            mst,
+            store,
+            AttestationStore::new(),
+        )
+        .unwrap()
+    }
+
+    // A: rendezvous puro, solo Kad (no MingaPeer, no necesita estado).
+    let a = LibP2pNode::new().unwrap();
+    let addr_a = a.listen("/ip4/127.0.0.1/tcp/0".parse().unwrap()).await;
+
+    // C: tiene una función que B querrá. Pasivo para aceptar el sync.
+    let c = singleton(3, "fn from_c(x: i32) -> i32 { x + 100 }");
+    let addr_c = c.listen("/ip4/127.0.0.1/tcp/0".parse().unwrap()).await;
+    let _accept_c = c.run_passive_accept();
+
+    // A pre-conoce a C en su routing table (rendezvous comportándose
+    // como tal).
+    a.add_dht_peer(c.peer_id(), addr_c);
+
+    // B: tiene su propia función. Solo conoce A.
+    let b = singleton(2, "fn from_b() -> i32 { 0 }");
+    b.add_dht_peer(a.peer_id, addr_a.clone());
+    b.dial(addr_a.clone());
+
+    tokio::time::sleep(Duration::from_millis(300)).await;
+
+    // B descubre a C vía DHT.
+    let discovered = b.find_closest_peers(c.peer_id()).await;
+    let c_entry = discovered
+        .iter()
+        .find(|p| p.peer_id == c.peer_id())
+        .unwrap_or_else(|| {
+            panic!(
+                "B no descubrió C; encontró: {:?}",
+                discovered.iter().map(|p| p.peer_id).collect::<Vec<_>>()
+            )
+        });
+
+    // B usa la dirección descubierta para dial directo y sync.
+    let addr_c_via_dht = c_entry.addrs[0].clone();
+    b.dial(addr_c_via_dht);
+
+    // Reintentamos sync hasta que la conexión esté arriba.
+    let deadline = std::time::Instant::now() + Duration::from_secs(5);
+    loop {
+        if b.sync_with(c.peer_id()).await.is_ok() {
+            break;
+        }
+        if std::time::Instant::now() >= deadline {
+            panic!("sync no completó en 5s");
+        }
+        tokio::time::sleep(Duration::from_millis(50)).await;
+    }
+
+    // Tras el sync, B y C tienen el mismo MST (unión). El merge de
+    // C sucede en su task de accept (paralela a B); esperamos a que
+    // ese merge se vea reflejado en su state.
+    let deadline = std::time::Instant::now() + Duration::from_secs(2);
+    loop {
+        let (mst_b, _, _) = b.snapshot().await;
+        let (mst_c, _, _) = c.snapshot().await;
+        if mst_b.root_hash() == mst_c.root_hash() && mst_b.len() == 2 {
+            break;
+        }
+        if std::time::Instant::now() >= deadline {
+            panic!(
+                "no convergencia tras 2s: |B|={}, |C|={}",
+                mst_b.len(),
+                mst_c.len()
+            );
+        }
+        tokio::time::sleep(Duration::from_millis(20)).await;
+    }
+}

crates/modules/semantic_dht/minga-p2p/tests/kad_providers.rs

@@ -0,0 +1,98 @@
+//! Tests de Provider Records vía Kademlia DHT.
+//!
+//! Discovery a nivel de **contenido**: en lugar de "¿quién está
+//! cerca?", la pregunta es "¿quién tiene el hash X?". Cuando un peer
+//! ingresa contenido, se anuncia como provider; otros peers consultan
+//! el DHT para encontrar a quién dial directamente.
+
+use std::time::Duration;
+
+use minga_core::{parse, AttestationStore, ContentHash, Keypair, MemStore, Mst};
+use minga_p2p::{LibP2pNode, MingaPeer};
+
+fn kp(seed: u8) -> Keypair {
+    Keypair::from_seed(&[seed; 32])
+}
+
+#[tokio::test]
+async fn provider_announce_and_lookup_two_nodes() {
+    let a = LibP2pNode::new().unwrap();
+    let b = LibP2pNode::new().unwrap();
+
+    let addr_b = b.listen("/ip4/127.0.0.1/tcp/0".parse().unwrap()).await;
+
+    // A conoce a B y dializa para establecer conexión Kad.
+    a.add_dht_peer(b.peer_id, addr_b.clone());
+    a.dial(addr_b);
+
+    tokio::time::sleep(Duration::from_millis(300)).await;
+
+    // A anuncia que tiene `content`.
+    let content = ContentHash([0x42; 32]);
+    a.start_providing(&content.0);
+
+    // Margen para que el ADD_PROVIDER se replique a B.
+    tokio::time::sleep(Duration::from_millis(500)).await;
+
+    // B consulta — debe encontrar A.
+    let providers = b.find_providers(&content.0).await;
+    assert!(
+        providers.iter().any(|p| *p == a.peer_id),
+        "B debe descubrir a A como provider, obtuvo: {:?}",
+        providers
+    );
+}
+
+#[tokio::test]
+async fn provider_lookup_returns_empty_for_unknown_content() {
+    let a = LibP2pNode::new().unwrap();
+    let b = LibP2pNode::new().unwrap();
+
+    let addr_b = b.listen("/ip4/127.0.0.1/tcp/0".parse().unwrap()).await;
+    a.add_dht_peer(b.peer_id, addr_b.clone());
+    a.dial(addr_b);
+    tokio::time::sleep(Duration::from_millis(300)).await;
+
+    // Nadie ha anunciado este hash.
+    let unknown = ContentHash([0xFF; 32]);
+    let providers = b.find_providers(&unknown.0).await;
+    assert!(providers.is_empty());
+}
+
+#[tokio::test]
+async fn minga_peer_ingest_auto_announces_provider() {
+    // El test de integración del flujo "fase de salida al mundo real":
+    // un peer hace ingest de un archivo y, sin acción adicional, otro
+    // peer puede descubrirlo vía DHT como provider.
+
+    let a_kp = kp(1);
+    let b_kp = kp(2);
+
+    let a = MingaPeer::new(a_kp, Mst::new(), MemStore::new(), AttestationStore::new()).unwrap();
+    let b = MingaPeer::new(b_kp, Mst::new(), MemStore::new(), AttestationStore::new()).unwrap();
+
+    let addr_a = a.listen("/ip4/127.0.0.1/tcp/0".parse().unwrap()).await;
+    let _addr_b = b.listen("/ip4/127.0.0.1/tcp/0".parse().unwrap()).await;
+
+    // Conectar B a A vía Kad (rendezvous bidireccional).
+    a.add_dht_peer(b.peer_id(), _addr_b);
+    b.add_dht_peer(a.peer_id(), addr_a.clone());
+    b.dial(addr_a);
+
+    tokio::time::sleep(Duration::from_millis(300)).await;
+
+    // A ingresa una función. Esto debe anunciarla automáticamente.
+    let n = parse::rust("fn discover_me() -> i32 { 7 }").unwrap();
+    let h = a.ingest(&n).await;
+
+    // Margen para la replicación del provider record.
+    tokio::time::sleep(Duration::from_millis(500)).await;
+
+    // B busca quién tiene `h` y debe encontrar A.
+    let providers = b.find_providers(h).await;
+    assert!(
+        providers.iter().any(|p| *p == a.peer_id()),
+        "B debe descubrir a A como provider del contenido recién ingerido. Obtuvo: {:?}",
+        providers,
+    );
+}

crates/modules/semantic_dht/minga-p2p/tests/libp2p_integration.rs

@@ -0,0 +1,161 @@
+//! Test de integración real con libp2p.
+//!
+//! Dos `LibP2pNode`s independientes en localhost:
+//! - cada uno con su propia identidad libp2p,
+//! - conectados por TCP (con cifrado Noise + multiplexado Yamux),
+//! - intercambiando una sesión completa de sync vía bidirectional
+//!   streams sobre el protocolo `/minga/sync/1.0.0`.
+//!
+//! Lo único que el wire añade respecto al harness in-memory es el
+//! transporte. La lógica del protocolo y el state machine son los
+//! mismos — eso es exactamente lo que queríamos demostrar.
+
+use std::time::Duration;
+
+use futures::StreamExt;
+use minga_core::{parse, ContentHash, Keypair, MemStore, Mst, NodeStore};
+use minga_p2p::{run_sync_async, LibP2pNode, SyncSession, SYNC_PROTOCOL};
+use tokio_util::compat::FuturesAsyncReadCompatExt;
+
+fn kp(seed: u8) -> Keypair {
+    Keypair::from_seed(&[seed; 32])
+}
+
+fn build_repo(sources: &[&str]) -> (Mst, MemStore, Vec<ContentHash>) {
+    let mut mst = Mst::new();
+    let mut store = MemStore::new();
+    let mut roots = Vec::new();
+    for src in sources {
+        let n = parse::rust(src).unwrap();
+        let h = store.put(&n);
+        mst.insert(h);
+        roots.push(h);
+    }
+    (mst, store, roots)
+}
+
+#[tokio::test]
+async fn libp2p_sync_two_peers_over_tcp() {
+    let node_a = LibP2pNode::new().unwrap();
+    let node_b = LibP2pNode::new().unwrap();
+    let peer_b = node_b.peer_id;
+
+    // Solo B necesita escuchar; A inicia el dial.
+    let addr_b = node_b
+        .listen("/ip4/127.0.0.1/tcp/0".parse().unwrap())
+        .await;
+
+    // B acepta streams del protocolo Minga en una tarea.
+    let only_b_sources = &["fn from_b(x: i32) -> i32 { x + 1 }"];
+    let (mst_b, store_b, _) = build_repo(only_b_sources);
+    let session_b = SyncSession::without_attestations(mst_b, store_b, kp(2));
+
+    let mut control_b = node_b.control.clone();
+    let task_b = tokio::spawn(async move {
+        let mut incoming = control_b.accept(SYNC_PROTOCOL).unwrap();
+        let (_peer, stream) = incoming.next().await.expect("incoming stream");
+        run_sync_async(session_b, stream.compat()).await
+    });
+
+    // A dializa B y abre stream. Reintenta hasta que la conexión esté
+    // arriba (puede tardar unos ms el handshake Noise+Yamux).
+    node_a.dial(addr_b);
+    let mut control_a = node_a.control.clone();
+    let stream_a = {
+        let deadline = std::time::Instant::now() + Duration::from_secs(5);
+        loop {
+            match control_a.open_stream(peer_b, SYNC_PROTOCOL).await {
+                Ok(s) => break s,
+                Err(_) if std::time::Instant::now() < deadline => {
+                    tokio::time::sleep(Duration::from_millis(50)).await;
+                }
+                Err(e) => panic!("no se pudo abrir stream tras 5s: {e:?}"),
+            }
+        }
+    };
+
+    let only_a_sources = &["fn from_a() -> i32 { 0 }"];
+    let (mst_a, store_a, _) = build_repo(only_a_sources);
+    let session_a = SyncSession::without_attestations(mst_a, store_a, kp(1));
+
+    let task_a = tokio::spawn(async move { run_sync_async(session_a, stream_a.compat()).await });
+
+    let result_a = task_a.await.expect("task A").expect("sync A");
+    let result_b = task_b.await.expect("task B").expect("sync B");
+
+    // Convergencia tras viajar sobre TCP real.
+    assert_eq!(result_a.mst().root_hash(), result_b.mst().root_hash());
+    assert_eq!(result_a.mst().len(), 2);
+    assert_eq!(result_b.mst().len(), 2);
+
+    // Cada peer terminó con la identidad libp2p del otro autenticada.
+    // (Las identidades libp2p no son las mismas que los DIDs Minga —
+    // las primeras autentican el canal, los segundos firman contenido.)
+    assert!(result_a.peer_did().is_some());
+    assert!(result_b.peer_did().is_some());
+}
+
+#[tokio::test]
+async fn libp2p_sync_with_attestations() {
+    use minga_core::{Attestation, AttestationStore};
+
+    let node_a = LibP2pNode::new().unwrap();
+    let node_b = LibP2pNode::new().unwrap();
+    let peer_b = node_b.peer_id;
+
+    let addr_b = node_b
+        .listen("/ip4/127.0.0.1/tcp/0".parse().unwrap())
+        .await;
+
+    let kp_a = kp(10);
+    let kp_b = kp(20);
+
+    let (mst_a, store_a, roots_a) = build_repo(&["fn signed_by_a() -> i32 { 1 }"]);
+    let (mst_b, store_b, roots_b) = build_repo(&["fn signed_by_b() -> i32 { 2 }"]);
+
+    let mut atts_a = AttestationStore::new();
+    atts_a.add(Attestation::create(&kp_a, roots_a[0])).unwrap();
+
+    let mut atts_b = AttestationStore::new();
+    atts_b.add(Attestation::create(&kp_b, roots_b[0])).unwrap();
+
+    let session_a = SyncSession::new(mst_a, store_a, atts_a, kp_a.clone());
+    let session_b = SyncSession::new(mst_b, store_b, atts_b, kp_b.clone());
+
+    let mut control_b = node_b.control.clone();
+    let task_b = tokio::spawn(async move {
+        let mut incoming = control_b.accept(SYNC_PROTOCOL).unwrap();
+        let (_peer, stream) = incoming.next().await.expect("incoming stream");
+        run_sync_async(session_b, stream.compat()).await
+    });
+
+    node_a.dial(addr_b);
+    let mut control_a = node_a.control.clone();
+    let stream_a = {
+        let deadline = std::time::Instant::now() + Duration::from_secs(5);
+        loop {
+            match control_a.open_stream(peer_b, SYNC_PROTOCOL).await {
+                Ok(s) => break s,
+                Err(_) if std::time::Instant::now() < deadline => {
+                    tokio::time::sleep(Duration::from_millis(50)).await;
+                }
+                Err(e) => panic!("no se pudo abrir stream: {e:?}"),
+            }
+        }
+    };
+
+    let task_a = tokio::spawn(async move { run_sync_async(session_a, stream_a.compat()).await });
+
+    let result_a = task_a.await.unwrap().unwrap();
+    let result_b = task_b.await.unwrap().unwrap();
+
+    // Atestaciones cruzaron criptográficamente verificadas.
+    assert_eq!(
+        result_a.attestations().authors_of(&roots_b[0]),
+        vec![kp_b.did()]
+    );
+    assert_eq!(
+        result_b.attestations().authors_of(&roots_a[0]),
+        vec![kp_a.did()]
+    );
+}

crates/modules/semantic_dht/minga-p2p/tests/passive_listener.rs

@@ -0,0 +1,128 @@
+//! Tests del passive listener.
+//!
+//! Un peer "always-on" que acepta sincronizaciones continuamente:
+//! cada peer entrante mergea sus contribuciones al estado compartido.
+//! El test demuestra que dos peers consecutivos (B luego C) se
+//! sincronizan independientemente con A, y A acaba con la unión de
+//! ambos estados.
+
+use std::time::Duration;
+
+use minga_core::{parse, AttestationStore, Keypair, MemStore, Mst, NodeStore};
+use minga_p2p::MingaPeer;
+
+fn kp(seed: u8) -> Keypair {
+    Keypair::from_seed(&[seed; 32])
+}
+
+fn singleton_repo(src: &str) -> (Mst, MemStore, minga_core::ContentHash) {
+    let mut mst = Mst::new();
+    let mut store = MemStore::new();
+    let h = store.put(&parse::rust(src).unwrap());
+    mst.insert(h);
+    (mst, store, h)
+}
+
+async fn sync_with_retry(peer: &MingaPeer, target: libp2p::PeerId) {
+    let deadline = std::time::Instant::now() + Duration::from_secs(5);
+    loop {
+        if peer.sync_with(target).await.is_ok() {
+            return;
+        }
+        if std::time::Instant::now() >= deadline {
+            panic!("sync no completó en 5s");
+        }
+        tokio::time::sleep(Duration::from_millis(50)).await;
+    }
+}
+
+#[tokio::test]
+async fn passive_listener_serves_two_consecutive_peers() {
+    // ── Peer A: vacío, escucha pasivamente ─────────────────────────
+    let a = MingaPeer::new(
+        kp(1),
+        Mst::new(),
+        MemStore::new(),
+        AttestationStore::new(),
+    )
+    .unwrap();
+    let addr_a = a.listen("/ip4/127.0.0.1/tcp/0".parse().unwrap()).await;
+    let _accept = a.run_passive_accept();
+
+    // ── Peer B: tiene función X. Sincroniza con A ─────────────────
+    let (mst_b, store_b, h_x) = singleton_repo("fn x() -> i32 { 1 }");
+    let b = MingaPeer::new(kp(2), mst_b, store_b, AttestationStore::new()).unwrap();
+
+    b.dial(addr_a.clone());
+    sync_with_retry(&b, a.peer_id()).await;
+
+    // A debe haber absorbido X.
+    let (mst_a_mid, _, _) = a.snapshot().await;
+    assert!(mst_a_mid.contains(&h_x), "A no aprendió X de B");
+
+    // ── Peer C: tiene función Y. Sincroniza con A ─────────────────
+    let (mst_c, store_c, h_y) = singleton_repo("fn y(z: i32) -> i32 { z * 2 }");
+    let c = MingaPeer::new(kp(3), mst_c, store_c, AttestationStore::new()).unwrap();
+
+    c.dial(addr_a.clone());
+    sync_with_retry(&c, a.peer_id()).await;
+
+    // ── Verificación: A acumuló X (de B) e Y (de C) ──────────────
+    let (mst_a_final, _, _) = a.snapshot().await;
+    assert!(mst_a_final.contains(&h_x), "A perdió X");
+    assert!(mst_a_final.contains(&h_y), "A no aprendió Y");
+    assert_eq!(mst_a_final.len(), 2);
+
+    // C también tiene ambas: la suya y X que recibió de A durante el sync.
+    let (mst_c_final, _, _) = c.snapshot().await;
+    assert!(mst_c_final.contains(&h_x), "C no recibió X transitivamente");
+    assert!(mst_c_final.contains(&h_y));
+    assert_eq!(mst_c_final.len(), 2);
+}
+
+#[tokio::test]
+async fn passive_listener_propagates_attestations() {
+    use minga_core::Attestation;
+
+    let kp_a = kp(10);
+    let kp_b = kp(20);
+    let kp_c = kp(30);
+
+    // A pasivo, sin contenido.
+    let a = MingaPeer::new(
+        kp_a.clone(),
+        Mst::new(),
+        MemStore::new(),
+        AttestationStore::new(),
+    )
+    .unwrap();
+    let addr_a = a.listen("/ip4/127.0.0.1/tcp/0".parse().unwrap()).await;
+    let _accept = a.run_passive_accept();
+
+    // B con contenido firmado por kp_b.
+    let (mst_b, store_b, h_b) = singleton_repo("fn from_b() -> i32 { 1 }");
+    let mut atts_b = AttestationStore::new();
+    atts_b.add(Attestation::create(&kp_b, h_b)).unwrap();
+    let b = MingaPeer::new(kp_b.clone(), mst_b, store_b, atts_b).unwrap();
+    b.dial(addr_a.clone());
+    sync_with_retry(&b, a.peer_id()).await;
+
+    // C con contenido firmado por kp_c. Sincroniza con A: aprende
+    // tanto el contenido de B como su atestación.
+    let (mst_c, store_c, h_c) = singleton_repo("fn from_c() -> i32 { 2 }");
+    let mut atts_c = AttestationStore::new();
+    atts_c.add(Attestation::create(&kp_c, h_c)).unwrap();
+    let c = MingaPeer::new(kp_c.clone(), mst_c, store_c, atts_c).unwrap();
+    c.dial(addr_a.clone());
+    sync_with_retry(&c, a.peer_id()).await;
+
+    // C ahora ve la atestación de B sobre h_b — sin haber hablado
+    // nunca con B directamente. La transitividad funciona.
+    let (_, _, atts_c_final) = c.snapshot().await;
+    let authors_b = atts_c_final.authors_of(&h_b);
+    assert_eq!(authors_b, vec![kp_b.did()]);
+
+    // Y C tiene su propia atestación intacta.
+    let authors_c = atts_c_final.authors_of(&h_c);
+    assert_eq!(authors_c, vec![kp_c.did()]);
+}

crates/modules/semantic_dht/minga-p2p/tests/persistent_peer.rs

@@ -0,0 +1,239 @@
+//! Tests del `MingaPeer` con backing persistente.
+//!
+//! Verifica que:
+//! - Abrir un path nuevo crea un repo vacío.
+//! - Datos ingresados a un peer abierto se persisten a disco.
+//! - Tras cerrar y reabrir el mismo path, el estado completo se
+//!   recupera (MST con mismo `root_hash`, store con todos los nodos
+//!   reconstruibles, atestaciones intactas y verificables).
+//! - El sync sobre red poblando un peer persistente sobrevive
+//!   reinicio.
+
+use std::time::Duration;
+
+use minga_core::{parse, Attestation, AttestationStore, Keypair, MemStore, Mst, NodeStore};
+use minga_p2p::{MingaPeer, SyncSession};
+use tempfile::TempDir;
+
+fn kp(seed: u8) -> Keypair {
+    Keypair::from_seed(&[seed; 32])
+}
+
+#[tokio::test]
+async fn open_creates_empty_repo_at_new_path() {
+    let dir = TempDir::new().unwrap();
+    let peer = MingaPeer::open(kp(1), dir.path()).unwrap();
+    let (mst, store, atts) = peer.snapshot().await;
+    assert!(mst.is_empty());
+    assert!(store.is_empty());
+    assert!(atts.is_empty());
+}
+
+#[tokio::test]
+async fn ingest_persists_across_restart() {
+    let dir = TempDir::new().unwrap();
+    let kp_a = kp(1);
+
+    let n = parse::rust("fn add(x: i32, y: i32) -> i32 { x + y }").unwrap();
+    let h_expected = minga_core::hash_node(&n);
+
+    // Sesión 1: abrir, ingerir, flush, drop.
+    {
+        let peer = MingaPeer::open(kp_a.clone(), dir.path()).unwrap();
+        let h = peer.ingest(&n).await;
+        assert_eq!(h, h_expected);
+        peer.flush().await.unwrap();
+    }
+
+    // Sesión 2: reabrir, verificar que todo está intacto.
+    {
+        let peer = MingaPeer::open(kp_a, dir.path()).unwrap();
+        let (mst, store, _) = peer.snapshot().await;
+        assert_eq!(mst.len(), 1);
+        assert!(mst.contains(&h_expected));
+        assert!(store.contains(&h_expected));
+
+        // Reconstrucción exacta del árbol original.
+        let reconstructed = store.reconstruct(&h_expected).unwrap();
+        assert_eq!(reconstructed, n);
+    }
+}
+
+#[tokio::test]
+async fn ingest_attestation_persists_across_restart() {
+    let dir = TempDir::new().unwrap();
+    let kp_owner = kp(1);
+    let kp_signer = kp(2);
+
+    let n = parse::rust("fn signed_function() -> i32 { 42 }").unwrap();
+    let h = minga_core::hash_node(&n);
+
+    {
+        let peer = MingaPeer::open(kp_owner.clone(), dir.path()).unwrap();
+        peer.ingest(&n).await;
+        let att = Attestation::create(&kp_signer, h);
+        peer.ingest_attestation(att).await.unwrap();
+        peer.flush().await.unwrap();
+    }
+
+    {
+        let peer = MingaPeer::open(kp_owner, dir.path()).unwrap();
+        let (_, _, atts) = peer.snapshot().await;
+        let authors = atts.authors_of(&h);
+        assert_eq!(authors, vec![kp_signer.did()]);
+
+        // La firma sigue verificando tras viajar disco→memoria.
+        let stored_atts = atts.get(&h);
+        assert_eq!(stored_atts.len(), 1);
+        assert!(stored_atts[0].verify());
+    }
+}
+
+#[tokio::test]
+async fn ingest_multiple_authors_for_same_content_persist() {
+    let dir = TempDir::new().unwrap();
+    let kp_owner = kp(1);
+    let alice = kp(10);
+    let bob = kp(20);
+    let carol = kp(30);
+
+    let n = parse::rust("fn shared() -> i32 { 0 }").unwrap();
+    let h = minga_core::hash_node(&n);
+
+    {
+        let peer = MingaPeer::open(kp_owner.clone(), dir.path()).unwrap();
+        peer.ingest(&n).await;
+        peer.ingest_attestation(Attestation::create(&alice, h))
+            .await
+            .unwrap();
+        peer.ingest_attestation(Attestation::create(&bob, h))
+            .await
+            .unwrap();
+        peer.ingest_attestation(Attestation::create(&carol, h))
+            .await
+            .unwrap();
+        peer.flush().await.unwrap();
+    }
+
+    {
+        let peer = MingaPeer::open(kp_owner, dir.path()).unwrap();
+        let (_, _, atts) = peer.snapshot().await;
+        let mut authors = atts.authors_of(&h);
+        authors.sort_by_key(|d| d.0);
+        assert_eq!(authors.len(), 3);
+        let mut expected = vec![alice.did(), bob.did(), carol.did()];
+        expected.sort_by_key(|d| d.0);
+        assert_eq!(authors, expected);
+    }
+}
+
+#[tokio::test]
+async fn root_hash_stable_across_restart() {
+    // El `root_hash` del MST es función pura del set de claves. Tras
+    // reabrir desde disco, debe ser idéntico.
+    let dir = TempDir::new().unwrap();
+    let kp_a = kp(1);
+
+    let target_root_hash;
+    {
+        let peer = MingaPeer::open(kp_a.clone(), dir.path()).unwrap();
+        for src in &[
+            "fn one() -> i32 { 1 }",
+            "fn two() -> i32 { 2 }",
+            "fn three(x: i32) -> i32 { x * x }",
+        ] {
+            peer.ingest(&parse::rust(src).unwrap()).await;
+        }
+        target_root_hash = peer.snapshot().await.0.root_hash();
+        peer.flush().await.unwrap();
+    }
+
+    {
+        let peer = MingaPeer::open(kp_a, dir.path()).unwrap();
+        let (mst, _, _) = peer.snapshot().await;
+        assert_eq!(mst.root_hash(), target_root_hash);
+        assert_eq!(mst.len(), 3);
+    }
+}
+
+#[tokio::test]
+async fn sync_into_persistent_peer_survives_restart() {
+    // Caso end-to-end: peer A pasivo y persistente. B sincroniza con
+    // A. A persiste lo que recibió. Cerramos A. Reabrimos. El estado
+    // sincronizado sigue ahí.
+    let dir = TempDir::new().unwrap();
+    let kp_a = kp(1);
+
+    let n = parse::rust("fn from_b(z: i32) -> i32 { z + 7 }").unwrap();
+    let h_b = minga_core::hash_node(&n);
+
+    // ── Sesión 1: A persistente acepta sync de B ─────────────────
+    {
+        let a = MingaPeer::open(kp_a.clone(), dir.path()).unwrap();
+        let addr_a = a.listen("/ip4/127.0.0.1/tcp/0".parse().unwrap()).await;
+        let accept = a.run_passive_accept();
+
+        // B en memoria, le sincroniza su contenido.
+        let mut store_b = MemStore::new();
+        let mut mst_b = Mst::new();
+        let h = store_b.put(&n);
+        mst_b.insert(h);
+        let b = MingaPeer::new(kp(2), mst_b, store_b, AttestationStore::new()).unwrap();
+        b.dial(addr_a);
+
+        // Reintentar sync hasta éxito.
+        let deadline = std::time::Instant::now() + Duration::from_secs(5);
+        loop {
+            if b.sync_with(a.peer_id()).await.is_ok() {
+                break;
+            }
+            if std::time::Instant::now() >= deadline {
+                panic!("sync no completó en 5s");
+            }
+            tokio::time::sleep(Duration::from_millis(50)).await;
+        }
+
+        // Esperar a que A's accept handler haya mergeado.
+        let deadline = std::time::Instant::now() + Duration::from_secs(2);
+        loop {
+            let (mst_a, _, _) = a.snapshot().await;
+            if mst_a.contains(&h_b) {
+                break;
+            }
+            if std::time::Instant::now() >= deadline {
+                panic!("merge en A no se vio en 2s");
+            }
+            tokio::time::sleep(Duration::from_millis(20)).await;
+        }
+
+        a.flush().await.unwrap();
+
+        // Cleanup explícito: abort la accept task y espera a que
+        // termine para liberar el lock de sled.
+        accept.abort();
+        let _ = accept.await;
+    }
+
+    // Pequeño margen para que tasks spawneadas terminen y los Arc
+    // se liberen.
+    tokio::time::sleep(Duration::from_millis(200)).await;
+
+    // ── Sesión 2: reabrir A, verificar contenido sincronizado ────
+    {
+        let a = MingaPeer::open(kp_a, dir.path()).unwrap();
+        let (mst_a, store_a, _) = a.snapshot().await;
+        assert!(
+            mst_a.contains(&h_b),
+            "el contenido de B no sobrevivió al reinicio"
+        );
+        assert!(store_a.contains(&h_b));
+
+        // Reconstruimos: lo que B firmó sigue ahí íntegro.
+        let reconstructed = store_a.reconstruct(&h_b).unwrap();
+        assert_eq!(reconstructed, n);
+    }
+}
+
+// Helper: silencia un warning si SyncSession se importa pero no se usa.
+#[allow(dead_code)]
+fn _session_marker(_: SyncSession) {}

crates/modules/semantic_dht/minga-p2p/tests/sync_invariants.rs

@@ -0,0 +1,798 @@
+//! Invariantes del protocolo de sincronización recursivo.
+//!
+//! Tres familias de tests:
+//! - **Convergencia funcional**: tras `run_sync`, ambos peers tienen
+//!   el mismo `root_hash`, `MemStore` equivalente, y reconstruyen los
+//!   árboles bit a bit.
+//! - **Eficiencia estructural**: el short-circuit por hash de subárbol
+//!   reduce probes y delivers cuando los repos comparten ramas.
+//! - **Seguridad**: el receptor verifica `hash_stored(stored) == hash`
+//!   y rechaza nodos manipulados.
+
+use minga_core::{
+    cas::hash_components, hash_node, hash_stored, parse, ContentHash, Keypair, MemStore, Mst,
+    NodeStore, Signature, StoredNode,
+};
+use minga_p2p::{run_sync, Message, SyncSession};
+
+fn kp(seed: u8) -> Keypair {
+    Keypair::from_seed(&[seed; 32])
+}
+
+/// Helper que replica la construcción del payload firmado del `Hello`
+/// dentro del protocolo Minga. Usado por los tests que inyectan
+/// mensajes manualmente.
+fn hello_payload(nonce: &[u8; 32], did: &minga_core::Did, root: &ContentHash) -> [u8; 96] {
+    let mut p = [0u8; 96];
+    p[..32].copy_from_slice(nonce);
+    p[32..64].copy_from_slice(&did.0);
+    p[64..96].copy_from_slice(&root.0);
+    p
+}
+
+fn build_repo(sources: &[&str]) -> (Mst, MemStore, Vec<ContentHash>) {
+    let mut mst = Mst::new();
+    let mut store = MemStore::new();
+    let mut roots = Vec::new();
+    for src in sources {
+        let n = parse::rust(src).unwrap();
+        let h = store.put(&n);
+        mst.insert(h);
+        roots.push(h);
+    }
+    (mst, store, roots)
+}
+
+// ─── Convergencia funcional ────────────────────────────────────────
+
+#[test]
+fn sync_identical_is_noop() {
+    let sources = &[
+        "fn add(x: i32, y: i32) -> i32 { x + y }",
+        "fn neg(x: i32) -> i32 { -x }",
+    ];
+    let (mst_a, store_a, _) = build_repo(sources);
+    let (mst_b, store_b, _) = build_repo(sources);
+
+    let mut a = SyncSession::without_attestations(mst_a, store_a, kp(1));
+    let mut b = SyncSession::without_attestations(mst_b, store_b, kp(2));
+    let stats = run_sync(&mut a, &mut b);
+
+    // Mismas raíces de MST: el short-circuit en Hello evita cualquier
+    // probe o transferencia. Solo cruzan los 2 Hellos y los 2 Dones.
+    assert_eq!(stats.hellos, 2);
+    assert_eq!(stats.probe_reqs, 0);
+    assert_eq!(stats.probe_ress, 0);
+    assert_eq!(stats.fetches, 0);
+    assert_eq!(stats.delivers, 0);
+    assert_eq!(stats.dones, 2);
+
+    assert_eq!(a.mst().root_hash(), b.mst().root_hash());
+}
+
+#[test]
+fn sync_one_empty_pulls_everything() {
+    let sources = &["fn f(x: i32) -> i32 { x * 2 }"];
+    let (mst_a, store_a, _) = build_repo(sources);
+    let (mst_b, store_b, _) = build_repo(&[]);
+
+    let mut a = SyncSession::without_attestations(mst_a, store_a, kp(1));
+    let mut b = SyncSession::without_attestations(mst_b, store_b, kp(2));
+    run_sync(&mut a, &mut b);
+
+    assert_eq!(a.mst().root_hash(), b.mst().root_hash());
+    assert_eq!(a.store().len(), b.store().len());
+
+    for h in a.mst().iter() {
+        assert!(b.store().contains(h));
+        let a_tree = a.store().reconstruct(h).unwrap();
+        let b_tree = b.store().reconstruct(h).unwrap();
+        assert_eq!(a_tree, b_tree);
+    }
+}
+
+#[test]
+fn sync_disjoint_sets_merge() {
+    let only_a = &[
+        "fn alpha() -> i32 { 1 }",
+        "fn beta(x: i32) -> i32 { x + 1 }",
+    ];
+    let only_b = &[
+        "fn gamma(y: i32) -> bool { y > 0 }",
+        "fn delta() -> &'static str { \"hello\" }",
+    ];
+
+    let (mst_a, store_a, _) = build_repo(only_a);
+    let (mst_b, store_b, _) = build_repo(only_b);
+
+    let mut a = SyncSession::without_attestations(mst_a, store_a, kp(1));
+    let mut b = SyncSession::without_attestations(mst_b, store_b, kp(2));
+    run_sync(&mut a, &mut b);
+
+    assert_eq!(a.mst().root_hash(), b.mst().root_hash());
+    assert_eq!(a.mst().len(), 4);
+    assert_eq!(b.mst().len(), 4);
+}
+
+#[test]
+fn sync_partial_overlap_converges() {
+    let common = &[
+        "fn shared_one() -> i32 { 42 }",
+        "fn shared_two(n: i32) -> i32 { n + 1 }",
+    ];
+    let extra_a = &["fn only_in_a() -> bool { true }"];
+    let extra_b = &["fn only_in_b(s: &str) -> usize { s.len() }"];
+
+    let mut sources_a: Vec<&str> = common.to_vec();
+    sources_a.extend_from_slice(extra_a);
+    let mut sources_b: Vec<&str> = common.to_vec();
+    sources_b.extend_from_slice(extra_b);
+
+    let (mst_a, store_a, _) = build_repo(&sources_a);
+    let (mst_b, store_b, _) = build_repo(&sources_b);
+
+    let mut a = SyncSession::without_attestations(mst_a, store_a, kp(1));
+    let mut b = SyncSession::without_attestations(mst_b, store_b, kp(2));
+    run_sync(&mut a, &mut b);
+
+    assert_eq!(a.mst().root_hash(), b.mst().root_hash());
+    assert_eq!(a.mst().len(), 4);
+}
+
+#[test]
+fn sync_transitive_children_pulled() {
+    let big_src = r#"
+        fn complicated(x: i32, y: i32) -> i32 {
+            let a = x + y;
+            let b = a * 2;
+            match b {
+                n if n > 100 => n - 50,
+                n if n < 0 => -n,
+                _ => b,
+            }
+        }
+    "#;
+    let (mst_a, store_a, roots) = build_repo(&[big_src]);
+    let store_a_size = store_a.len();
+    let root_hash = roots[0];
+
+    let (mst_b, store_b, _) = build_repo(&[]);
+
+    let mut a = SyncSession::without_attestations(mst_a, store_a, kp(1));
+    let mut b = SyncSession::without_attestations(mst_b, store_b, kp(2));
+    run_sync(&mut a, &mut b);
+
+    assert!(b.store().contains(&root_hash));
+    assert_eq!(b.store().len(), store_a_size);
+
+    let a_tree = a.store().reconstruct(&root_hash).unwrap();
+    let b_tree = b.store().reconstruct(&root_hash).unwrap();
+    assert_eq!(a_tree, b_tree);
+}
+
+#[test]
+fn sync_idempotent_after_convergence() {
+    let sources = &["fn p() -> i32 { 1 }", "fn q(x: i32) -> i32 { x + 1 }"];
+    let (mst_a, store_a, _) = build_repo(sources);
+    let (mst_b, store_b, _) = build_repo(&["fn r(y: i32) -> i32 { y - 1 }"]);
+
+    let mut a = SyncSession::without_attestations(mst_a, store_a, kp(1));
+    let mut b = SyncSession::without_attestations(mst_b, store_b, kp(2));
+    run_sync(&mut a, &mut b);
+
+    let (mst_a, store_a, _) = a.into_parts();
+    let (mst_b, store_b, _) = b.into_parts();
+    let mut a2 = SyncSession::without_attestations(mst_a, store_a, kp(1));
+    let mut b2 = SyncSession::without_attestations(mst_b, store_b, kp(2));
+    let stats = run_sync(&mut a2, &mut b2);
+
+    // Tras converger, la segunda corrida es 2 Hellos + 2 Dones, nada
+    // estructural ni transferencias.
+    assert_eq!(stats.probe_reqs, 0);
+    assert_eq!(stats.probe_ress, 0);
+    assert_eq!(stats.fetches, 0);
+    assert_eq!(stats.delivers, 0);
+    assert_eq!(stats.hellos, 2);
+    assert_eq!(stats.dones, 2);
+}
+
+#[test]
+fn sync_both_empty_terminates() {
+    let mut a = SyncSession::without_attestations(Mst::new(), MemStore::new(), kp(1));
+    let mut b = SyncSession::without_attestations(Mst::new(), MemStore::new(), kp(2));
+    let stats = run_sync(&mut a, &mut b);
+    assert_eq!(stats.hellos, 2);
+    assert_eq!(stats.probe_reqs, 0);
+    assert_eq!(stats.dones, 2);
+    assert!(a.mst().is_empty());
+    assert!(b.mst().is_empty());
+}
+
+#[test]
+fn sync_three_way_via_pairwise_runs() {
+    let sources_a = &["fn a1() -> i32 { 1 }", "fn shared() -> i32 { 0 }"];
+    let sources_b = &["fn b1(x: i32) -> i32 { x }", "fn shared() -> i32 { 0 }"];
+    let sources_c = &["fn c1() -> bool { true }"];
+
+    let (mst_a, store_a, _) = build_repo(sources_a);
+    let (mst_b, store_b, _) = build_repo(sources_b);
+    let (mst_c, store_c, _) = build_repo(sources_c);
+
+    let mut a = SyncSession::without_attestations(mst_a, store_a, kp(1));
+    let mut b = SyncSession::without_attestations(mst_b, store_b, kp(2));
+    run_sync(&mut a, &mut b);
+    let (mst_a, store_a, _) = a.into_parts();
+    let (mst_b, store_b, _) = b.into_parts();
+
+    let mut b = SyncSession::without_attestations(mst_b, store_b, kp(2));
+    let mut c = SyncSession::without_attestations(mst_c, store_c, kp(3));
+    run_sync(&mut b, &mut c);
+    let (mst_b, _, _) = b.into_parts();
+    let (mst_c, store_c, _) = c.into_parts();
+
+    let mut c = SyncSession::without_attestations(mst_c, store_c, kp(3));
+    let mut a = SyncSession::without_attestations(mst_a, store_a, kp(1));
+    run_sync(&mut c, &mut a);
+    let (mst_c, _, _) = c.into_parts();
+    let (mst_a, _, _) = a.into_parts();
+
+    assert_eq!(mst_a.root_hash(), mst_b.root_hash());
+    assert_eq!(mst_b.root_hash(), mst_c.root_hash());
+    assert_eq!(mst_a.len(), 4);
+}
+
+// ─── Eficiencia estructural ────────────────────────────────────────
+
+#[test]
+fn sync_subtree_short_circuit_skips_shared_branches() {
+    // Construimos dos repos que comparten muchos nodos pero difieren en
+    // uno. El short-circuit por hash de subárbol debería podar las
+    // ramas compartidas: el número de probes y delivers debe estar
+    // dominado por la divergencia, no por el tamaño total.
+    let common: Vec<String> = (0..50)
+        .map(|i| format!("fn shared_{}() -> i32 {{ {} }}", i, i))
+        .collect();
+    let common_refs: Vec<&str> = common.iter().map(|s| s.as_str()).collect();
+
+    let extra_a = "fn only_a() -> bool { true }".to_string();
+    let mut sources_a: Vec<&str> = common_refs.clone();
+    sources_a.push(&extra_a);
+
+    let extra_b = "fn only_b() -> bool { false }".to_string();
+    let mut sources_b: Vec<&str> = common_refs.clone();
+    sources_b.push(&extra_b);
+
+    let (mst_a, store_a, _) = build_repo(&sources_a);
+    let (mst_b, store_b, _) = build_repo(&sources_b);
+    let store_a_size = store_a.len();
+
+    let mut a = SyncSession::without_attestations(mst_a, store_a, kp(1));
+    let mut b = SyncSession::without_attestations(mst_b, store_b, kp(2));
+    let stats = run_sync(&mut a, &mut b);
+
+    assert_eq!(a.mst().root_hash(), b.mst().root_hash());
+
+    // Cota de eficiencia: cada peer debe pedir como máximo lo que
+    // realmente le falta. En este escenario, cada peer ignora una sola
+    // función nueva (~docena de StoredNodes). Si el short-circuit
+    // estuviera roto, transferiríamos cerca del store entero (~varios
+    // cientos). La cota es laxa pero detectaría esa regresión.
+    assert!(
+        stats.delivers < store_a_size / 2,
+        "demasiados delivers ({}); esperaba << {}",
+        stats.delivers,
+        store_a_size,
+    );
+}
+
+// ─── Seguridad: verificación criptográfica ─────────────────────────
+
+#[test]
+fn cas_hash_node_equals_hash_stored() {
+    // El invariante fundacional para verificación: hashear el árbol
+    // como `SemanticNode` y como `StoredNode` produce idéntico hash.
+    // Sin esto, el receptor no podría confiar en lo que recibe.
+    let node = parse::rust("fn add(x: i32, y: i32) -> i32 { x + y }").unwrap();
+    let direct = hash_node(&node);
+
+    let mut store = MemStore::new();
+    let via_store = store.put(&node);
+    assert_eq!(direct, via_store);
+
+    let stored = store.get(&direct).unwrap();
+    let recomputed = hash_stored(stored);
+    assert_eq!(direct, recomputed);
+}
+
+#[test]
+fn sync_rejects_tampered_deliver() {
+    // Construimos un mensaje Deliver donde `hash` y `stored` no son
+    // consistentes — simulando un peer malicioso o un bit flip en el
+    // transporte. La sesión debe rechazarlo y no contaminar su estado.
+    let (mst_a, store_a, _) = build_repo(&[]);
+    let mut a = SyncSession::without_attestations(mst_a, store_a, kp(1));
+    let initial_store_size = a.store().len();
+    let initial_mst_size = a.mst().len();
+
+    // Forjamos un StoredNode con identidad falsa: anunciamos un hash
+    // arbitrario pero adjuntamos contenido distinto.
+    let fake_stored = StoredNode {
+        kind: "function_item".to_string(),
+        field_name: None,
+        leaf_text: None,
+        children: Vec::new(),
+    };
+    // El hash real de fake_stored es x; anunciamos como otra cosa.
+    let real_hash = hash_components("function_item", None, None, &[]);
+    let bogus_hash = ContentHash([0xAB; 32]);
+    assert_ne!(real_hash, bogus_hash);
+
+    // Inyectamos como si viniera del peer (sesión recibe Hello primero
+    // para que received_hello sea true; luego le metemos el Deliver
+    // tóxico). El Hello se firma con la llave del peer simulado.
+    let peer_kp = kp(99);
+    let peer_root = minga_core::empty_subtree_hash();
+    let peer_sig = peer_kp.sign(peer_root.as_bytes());
+    a.handle(Message::Hello {
+        peer_did: peer_kp.did(),
+        root_subtree_hash: peer_root,
+        signature: peer_sig,
+    });
+    let _ = a.handle(Message::Deliver {
+        hash: bogus_hash,
+        stored: fake_stored,
+    });
+
+    // El store y MST no deben cambiar; el contador de rechazos sí.
+    assert_eq!(a.store().len(), initial_store_size);
+    assert_eq!(a.mst().len(), initial_mst_size);
+    assert_eq!(a.rejected_delivers(), 1);
+    assert!(!a.store().contains(&bogus_hash));
+}
+
+#[test]
+fn sync_accepts_well_formed_deliver() {
+    // Contraprueba del anterior: un Deliver con hash válido sí se
+    // acepta. Verifica que el rechazo es selectivo, no global.
+    let (mst_a, store_a, _) = build_repo(&[]);
+    let mut a = SyncSession::without_attestations(mst_a, store_a, kp(1));
+
+    let stored = StoredNode {
+        kind: "integer_literal".to_string(),
+        field_name: None,
+        leaf_text: Some(b"42".to_vec()),
+        children: Vec::new(),
+    };
+    let real_hash = hash_stored(&stored);
+
+    let peer_kp = kp(99);
+    let peer_root = minga_core::empty_subtree_hash();
+    let peer_sig = peer_kp.sign(peer_root.as_bytes());
+    a.handle(Message::Hello {
+        peer_did: peer_kp.did(),
+        root_subtree_hash: peer_root,
+        signature: peer_sig,
+    });
+    a.handle(Message::Deliver {
+        hash: real_hash,
+        stored,
+    });
+
+    // No estaba en awaiting_root (no llegó por probe), así que no
+    // entra al MST — pero sí al store.
+    assert!(a.store().contains(&real_hash));
+    assert_eq!(a.rejected_delivers(), 0);
+}
+
+// ─── Identidad y autenticación ─────────────────────────────────────
+
+#[test]
+fn sync_captures_peer_did_after_valid_hello() {
+    // Tras un sync exitoso, cada sesión conoce el DID del otro peer
+    // — la primera afirmación criptográficamente verificable de la
+    // identidad del interlocutor.
+    let sources = &["fn f() -> i32 { 1 }"];
+    let (mst_a, store_a, _) = build_repo(sources);
+    let (mst_b, store_b, _) = build_repo(sources);
+
+    let kp_a = kp(10);
+    let kp_b = kp(20);
+    let did_a = kp_a.did();
+    let did_b = kp_b.did();
+
+    let mut a = SyncSession::without_attestations(mst_a, store_a, kp_a);
+    let mut b = SyncSession::without_attestations(mst_b, store_b, kp_b);
+
+    assert_eq!(a.peer_did(), None);
+    assert_eq!(b.peer_did(), None);
+
+    run_sync(&mut a, &mut b);
+
+    // Cada peer ahora tiene la identidad verificada del otro.
+    assert_eq!(a.peer_did(), Some(did_b));
+    assert_eq!(b.peer_did(), Some(did_a));
+    assert_eq!(a.local_did(), did_a);
+    assert_eq!(b.local_did(), did_b);
+}
+
+#[test]
+fn sync_rejects_hello_with_tampered_signature() {
+    // Un atacante que captura un Hello legítimo pero modifica un byte
+    // de la firma debe ser rechazado. La sesión no marca
+    // received_hello, no procesa el root, no emite ProbeReq — el
+    // contador de rechazos se incrementa en su lugar.
+    let (mst_a, store_a, _) = build_repo(&[]);
+    let mut a = SyncSession::without_attestations(mst_a, store_a, kp(1));
+
+    let attacker = kp(2);
+    let root = minga_core::empty_subtree_hash();
+    let mut sig = attacker.sign(root.as_bytes());
+    sig.0[5] ^= 0xFF;
+
+    let out = a.handle(Message::Hello {
+        peer_did: attacker.did(),
+        root_subtree_hash: root,
+        signature: sig,
+    });
+
+    assert!(out.is_empty(), "Hello con firma rota no debe producir respuesta");
+    assert_eq!(a.rejected_hellos(), 1);
+    assert_eq!(a.peer_did(), None);
+}
+
+#[test]
+fn sync_rejects_hello_with_swapped_did() {
+    // Otro vector: la firma es válida bajo el DID original, pero el
+    // atacante reemplaza el campo `peer_did` por uno distinto. La
+    // verificación falla porque la firma no fue producida por la
+    // llave privada correspondiente al DID anunciado.
+    let (mst_a, store_a, _) = build_repo(&[]);
+    let mut a = SyncSession::without_attestations(mst_a, store_a, kp(1));
+
+    let real_signer = kp(50);
+    let imposter = kp(51);
+    let root = minga_core::empty_subtree_hash();
+    let sig = real_signer.sign(root.as_bytes());
+
+    a.handle(Message::Hello {
+        peer_did: imposter.did(), // dice ser imposter pero la firma es de real_signer
+        root_subtree_hash: root,
+        signature: sig,
+    });
+
+    assert_eq!(a.rejected_hellos(), 1);
+    assert_eq!(a.peer_did(), None);
+}
+
+#[test]
+fn sync_rejects_hello_signed_over_different_root() {
+    // El atacante firma un root diferente al que anuncia. La firma es
+    // válida sobre `wrong_root`, pero el mensaje dice `claimed_root`.
+    let (mst_a, store_a, _) = build_repo(&[]);
+    let mut a = SyncSession::without_attestations(mst_a, store_a, kp(1));
+
+    let signer = kp(60);
+    let claimed_root = ContentHash([0xAA; 32]);
+    let wrong_root = ContentHash([0xBB; 32]);
+    let sig_over_wrong = signer.sign(wrong_root.as_bytes());
+
+    a.handle(Message::Hello {
+        peer_did: signer.did(),
+        root_subtree_hash: claimed_root,
+        signature: sig_over_wrong,
+    });
+
+    assert_eq!(a.rejected_hellos(), 1);
+    assert_eq!(a.peer_did(), None);
+}
+
+#[test]
+fn sync_rejects_replay_of_hello_from_different_session() {
+    // El test del bloque CRÍTICO: anti-replay anti-replay.
+    //
+    // Sesión 1: el peer "alice" responde a un Challenge de A1
+    // firmando un Hello con el nonce de A1.
+    //
+    // Sesión 2: la misma A vuelve a abrir sesión (A2). A2 genera un
+    // nonce nuevo. Un atacante intenta replicar el Hello capturado de
+    // la sesión 1. Como el nonce es distinto, la firma no verifica.
+    let alice = kp(50);
+    let alice_root = ContentHash([0xAA; 32]);
+
+    // Sesión 1.
+    let mut a1 = SyncSession::without_attestations(Mst::new(), MemStore::new(), kp(1));
+    let nonce_a1 = a1.self_nonce();
+
+    // Alice firma su Hello sobre el nonce que A1 emitió.
+    let payload_1 = hello_payload(&nonce_a1, &alice.did(), &alice_root);
+    let sig_1 = alice.sign(&payload_1);
+    let captured_hello = Message::Hello {
+        peer_did: alice.did(),
+        root_subtree_hash: alice_root,
+        signature: sig_1,
+    };
+
+    // En sesión 1, el Hello se acepta limpiamente.
+    a1.handle(captured_hello.clone());
+    assert_eq!(a1.peer_did(), Some(alice.did()));
+    assert_eq!(a1.rejected_hellos(), 0);
+
+    // Sesión 2: A2 con nonce nuevo. El atacante replica `captured_hello`.
+    let mut a2 = SyncSession::without_attestations(Mst::new(), MemStore::new(), kp(2));
+    assert_ne!(a2.self_nonce(), nonce_a1, "los nonces son distintos por sesión");
+
+    a2.handle(captured_hello);
+
+    // Replay rechazado: la firma estaba sobre nonce_a1, A2 verifica
+    // contra su propio nonce, mismatch criptográfico.
+    assert_eq!(a2.rejected_hellos(), 1);
+    assert_eq!(a2.peer_did(), None);
+}
+
+#[test]
+fn sync_proceeds_after_valid_hello_following_rejection() {
+    // Si llega un Hello inválido seguido de uno válido, la sesión se
+    // recupera: acepta el válido y captura ese DID. No hay
+    // "envenenamiento" persistente del estado.
+    let (mst_a, store_a, _) = build_repo(&[]);
+    let mut a = SyncSession::without_attestations(mst_a, store_a, kp(1));
+
+    let bad_signer = kp(70);
+    let mut bad_sig = bad_signer.sign(b"otro mensaje");
+    bad_sig.0[0] ^= 0xFF;
+    let root = minga_core::empty_subtree_hash();
+    a.handle(Message::Hello {
+        peer_did: bad_signer.did(),
+        root_subtree_hash: root,
+        signature: bad_sig,
+    });
+    assert_eq!(a.rejected_hellos(), 1);
+    assert_eq!(a.peer_did(), None);
+
+    let good_signer = kp(71);
+    let nonce = a.self_nonce();
+    let good_payload = hello_payload(&nonce, &good_signer.did(), &root);
+    let good_sig = good_signer.sign(&good_payload);
+    a.handle(Message::Hello {
+        peer_did: good_signer.did(),
+        root_subtree_hash: root,
+        signature: good_sig,
+    });
+    assert_eq!(a.rejected_hellos(), 1);
+    assert_eq!(a.peer_did(), Some(good_signer.did()));
+}
+
+// Aux: dejamos `Signature` importado para que el bloque arriba siga
+// compilando en futuras refactorizaciones que lo necesiten.
+#[allow(dead_code)]
+fn _signature_marker(_: Signature) {}
+
+// ─── Propagación de atestaciones ───────────────────────────────────
+
+use minga_core::{Attestation, AttestationStore, Did};
+
+fn build_repo_with_attests(
+    sources: &[&str],
+    signers: &[&Keypair],
+) -> (Mst, MemStore, AttestationStore, Vec<ContentHash>) {
+    let mut mst = Mst::new();
+    let mut store = MemStore::new();
+    let mut attests = AttestationStore::new();
+    let mut roots = Vec::new();
+    for src in sources {
+        let n = parse::rust(src).unwrap();
+        let h = store.put(&n);
+        mst.insert(h);
+        for kp in signers {
+            attests.add(Attestation::create(kp, h)).unwrap();
+        }
+        roots.push(h);
+    }
+    (mst, store, attests, roots)
+}
+
+#[test]
+fn sync_propagates_attestations_for_owned_content() {
+    // Cada peer tiene su propio contenido y firma sus propias claves.
+    // Tras sync, ambos peers conocen ambas atestaciones.
+    let kp_a = kp(10);
+    let kp_b = kp(20);
+
+    let (mst_a, store_a, atts_a, roots_a) =
+        build_repo_with_attests(&["fn from_a() -> i32 { 1 }"], &[&kp_a]);
+    let (mst_b, store_b, atts_b, roots_b) =
+        build_repo_with_attests(&["fn from_b() -> i32 { 2 }"], &[&kp_b]);
+
+    let mut a = SyncSession::new(mst_a, store_a, atts_a, kp_a.clone());
+    let mut b = SyncSession::new(mst_b, store_b, atts_b, kp_b.clone());
+    run_sync(&mut a, &mut b);
+
+    // A debe ahora conocer la atestación de B sobre roots_b[0], y
+    // viceversa. Ambas verificables criptográficamente.
+    let h_a = roots_a[0];
+    let h_b = roots_b[0];
+
+    let a_authors_for_a: Vec<Did> = a.attestations().authors_of(&h_a);
+    let a_authors_for_b: Vec<Did> = a.attestations().authors_of(&h_b);
+    assert_eq!(a_authors_for_a, vec![kp_a.did()]);
+    assert_eq!(a_authors_for_b, vec![kp_b.did()]);
+
+    let b_authors_for_a: Vec<Did> = b.attestations().authors_of(&h_a);
+    let b_authors_for_b: Vec<Did> = b.attestations().authors_of(&h_b);
+    assert_eq!(b_authors_for_a, vec![kp_a.did()]);
+    assert_eq!(b_authors_for_b, vec![kp_b.did()]);
+}
+
+#[test]
+fn sync_merges_multiple_authors_for_shared_content() {
+    // Ambos peers tienen el MISMO contenido (mismo hash) pero
+    // atestaciones de autores DISTINTOS. Tras sync, cada peer ve el
+    // conjunto completo de autores que han respaldado ese contenido.
+    let kp_a = kp(30);
+    let kp_b = kp(40);
+    let kp_c = kp(50);
+    let kp_d = kp(60);
+
+    let src = "fn shared() -> i32 { 99 }";
+
+    // A tiene firmas de A y C sobre el contenido.
+    let (mst_a, store_a, atts_a, _) = build_repo_with_attests(&[src], &[&kp_a, &kp_c]);
+    // B tiene firmas de B y D sobre el MISMO contenido.
+    let (mst_b, store_b, atts_b, roots_b) = build_repo_with_attests(&[src], &[&kp_b, &kp_d]);
+    let h = roots_b[0];
+
+    let mut a = SyncSession::new(mst_a, store_a, atts_a, kp_a.clone());
+    let mut b = SyncSession::new(mst_b, store_b, atts_b, kp_b.clone());
+    run_sync(&mut a, &mut b);
+
+    // Ambos peers ven los cuatro autores.
+    let mut a_authors = a.attestations().authors_of(&h);
+    let mut b_authors = b.attestations().authors_of(&h);
+    a_authors.sort_by_key(|d| d.0);
+    b_authors.sort_by_key(|d| d.0);
+    assert_eq!(a_authors, b_authors);
+    assert_eq!(a_authors.len(), 4);
+    assert!(a_authors.contains(&kp_a.did()));
+    assert!(a_authors.contains(&kp_b.did()));
+    assert!(a_authors.contains(&kp_c.did()));
+    assert!(a_authors.contains(&kp_d.did()));
+}
+
+#[test]
+fn sync_attestations_are_verified_at_receiver() {
+    // Inyectamos manualmente un AttestPush con una firma corrupta
+    // entre las legítimas. La sesión solo acepta las legítimas e
+    // incrementa rejected_attests.
+    let mut a = SyncSession::without_attestations(Mst::new(), MemStore::new(), kp(1));
+
+    // Hello válido del peer simulado, para que received_hello sea true.
+    let peer_kp = kp(80);
+    let peer_root = minga_core::empty_subtree_hash();
+    let nonce = a.self_nonce();
+    let peer_payload = hello_payload(&nonce, &peer_kp.did(), &peer_root);
+    let peer_sig = peer_kp.sign(&peer_payload);
+    a.handle(Message::Hello {
+        peer_did: peer_kp.did(),
+        root_subtree_hash: peer_root,
+        signature: peer_sig,
+    });
+
+    // Tres atestaciones: dos legítimas y una con firma rota.
+    let alice = kp(81);
+    let bob = kp(82);
+    let h1 = ContentHash([1u8; 32]);
+    let h2 = ContentHash([2u8; 32]);
+    let h3 = ContentHash([3u8; 32]);
+
+    let valid1 = Attestation::create(&alice, h1);
+    let valid2 = Attestation::create(&bob, h2);
+    let mut tampered = Attestation::create(&alice, h3);
+    tampered.signature.0[10] ^= 0xFF;
+
+    a.handle(Message::AttestPush {
+        attestations: vec![valid1.clone(), tampered, valid2.clone()],
+    });
+
+    // Las dos válidas se mergean; la corrupta se rechaza.
+    assert_eq!(a.attestations().len(), 2);
+    assert_eq!(a.rejected_attests(), 1);
+    assert_eq!(a.attestations().authors_of(&h1), vec![alice.did()]);
+    assert_eq!(a.attestations().authors_of(&h2), vec![bob.did()]);
+    assert!(a.attestations().get(&h3).is_empty());
+}
+
+#[test]
+fn sync_attest_push_before_hello_is_rejected() {
+    // Una atestación que llega antes del Hello autenticado se descarta
+    // — no podemos confiar en lo que dice el remitente hasta saber
+    // quién es.
+    let mut a = SyncSession::without_attestations(Mst::new(), MemStore::new(), kp(1));
+
+    let alice = kp(90);
+    let h = ContentHash([7u8; 32]);
+    let att = Attestation::create(&alice, h);
+
+    let out = a.handle(Message::AttestPush {
+        attestations: vec![att],
+    });
+    assert!(out.is_empty());
+    assert_eq!(a.rejected_attests(), 1);
+    assert_eq!(a.attestations().len(), 0);
+}
+
+#[test]
+fn sync_attestations_are_idempotent_across_runs() {
+    // Re-correr el sync no duplica atestaciones (gracias a la
+    // idempotencia de AttestationStore::add por (autor, contenido)).
+    let kp_a = kp(100);
+    let kp_b = kp(101);
+
+    let (mst_a, store_a, atts_a, _) =
+        build_repo_with_attests(&["fn run_one() -> i32 { 1 }"], &[&kp_a]);
+    let (mst_b, store_b, atts_b, _) =
+        build_repo_with_attests(&["fn run_two() -> i32 { 2 }"], &[&kp_b]);
+
+    let mut a = SyncSession::new(mst_a, store_a, atts_a, kp_a.clone());
+    let mut b = SyncSession::new(mst_b, store_b, atts_b, kp_b.clone());
+    run_sync(&mut a, &mut b);
+    let after_first_a = a.attestations().len();
+    let after_first_b = b.attestations().len();
+    assert_eq!(after_first_a, 2);
+    assert_eq!(after_first_b, 2);
+
+    let (mst_a, store_a, atts_a) = a.into_parts();
+    let (mst_b, store_b, atts_b) = b.into_parts();
+    let mut a2 = SyncSession::new(mst_a, store_a, atts_a, kp_a);
+    let mut b2 = SyncSession::new(mst_b, store_b, atts_b, kp_b);
+    run_sync(&mut a2, &mut b2);
+
+    assert_eq!(a2.attestations().len(), after_first_a);
+    assert_eq!(b2.attestations().len(), after_first_b);
+}
+
+#[test]
+fn sync_attestations_about_remote_content() {
+    // Caso interesante: A tiene una atestación sobre contenido que
+    // **NO** posee (lo recibió por gossip de un tercero). Tras sync
+    // con B, B aprende esa atestación aunque A nunca tuvo el contenido
+    // en su store.
+    let kp_a = kp(110);
+    let kp_third_party = kp(111);
+
+    // A no tiene contenido propio pero sí una atestación de
+    // `kp_third_party` sobre un hash arbitrario.
+    let phantom_hash = ContentHash([0xCD; 32]);
+    let mut atts_a = AttestationStore::new();
+    atts_a
+        .add(Attestation::create(&kp_third_party, phantom_hash))
+        .unwrap();
+
+    let kp_b = kp(112);
+    let mut a = SyncSession::new(Mst::new(), MemStore::new(), atts_a, kp_a);
+    let mut b = SyncSession::without_attestations(Mst::new(), MemStore::new(), kp_b);
+    run_sync(&mut a, &mut b);
+
+    // B ahora conoce la atestación, aunque ni A ni B tienen el
+    // contenido en su store.
+    assert_eq!(b.attestations().len(), 1);
+    assert_eq!(b.attestations().authors_of(&phantom_hash), vec![kp_third_party.did()]);
+    assert!(!b.store().contains(&phantom_hash));
+}
+
+#[test]
+fn sync_attest_push_count_in_stats() {
+    // Cuando ambos peers tienen atestaciones, el harness registra dos
+    // AttestPushes (uno por dirección).
+    let kp_a = kp(120);
+    let kp_b = kp(121);
+    let (mst_a, store_a, atts_a, _) =
+        build_repo_with_attests(&["fn ax() -> i32 { 0 }"], &[&kp_a]);
+    let (mst_b, store_b, atts_b, _) =
+        build_repo_with_attests(&["fn bx() -> i32 { 0 }"], &[&kp_b]);
+
+    let mut a = SyncSession::new(mst_a, store_a, atts_a, kp_a);
+    let mut b = SyncSession::new(mst_b, store_b, atts_b, kp_b);
+    let stats = run_sync(&mut a, &mut b);
+
+    assert_eq!(stats.attest_pushes, 2);
+}

crates/modules/semantic_dht/minga-p2p/tests/wire_roundtrips.rs

@@ -0,0 +1,128 @@
+//! Tests de roundtrip de serialización para `Message`.
+
+use minga_core::{Attestation, ContentHash, Keypair, NodeProbe, StoredNode};
+use minga_p2p::Message;
+
+fn roundtrip(msg: &Message) {
+    let bytes = msg.encode();
+    let decoded = Message::decode(&bytes).unwrap();
+    assert_eq!(msg, &decoded);
+}
+
+fn kp(seed: u8) -> Keypair {
+    Keypair::from_seed(&[seed; 32])
+}
+
+#[test]
+fn hello_roundtrip() {
+    let k = kp(1);
+    let root = ContentHash([42; 32]);
+    let sig = k.sign(root.as_bytes());
+    let msg = Message::Hello {
+        peer_did: k.did(),
+        root_subtree_hash: root,
+        signature: sig,
+    };
+    roundtrip(&msg);
+}
+
+#[test]
+fn probe_req_roundtrip() {
+    roundtrip(&Message::ProbeReq {
+        subtree_hash: ContentHash([5; 32]),
+    });
+}
+
+#[test]
+fn probe_res_with_probe_roundtrip() {
+    let msg = Message::ProbeRes {
+        subtree_hash: ContentHash([7; 32]),
+        probe: Some(NodeProbe {
+            level: 2,
+            keys: vec![ContentHash([1; 32])],
+            child_hashes: vec![ContentHash([10; 32]), ContentHash([20; 32])],
+        }),
+    };
+    roundtrip(&msg);
+}
+
+#[test]
+fn probe_res_empty_roundtrip() {
+    roundtrip(&Message::ProbeRes {
+        subtree_hash: ContentHash([7; 32]),
+        probe: None,
+    });
+}
+
+#[test]
+fn fetch_roundtrip() {
+    roundtrip(&Message::Fetch {
+        hash: ContentHash([3; 32]),
+    });
+}
+
+#[test]
+fn deliver_roundtrip() {
+    let stored = StoredNode {
+        kind: "function_item".to_string(),
+        field_name: Some("body".to_string()),
+        leaf_text: None,
+        children: vec![ContentHash([1; 32]), ContentHash([2; 32])],
+    };
+    roundtrip(&Message::Deliver {
+        hash: ContentHash([99; 32]),
+        stored,
+    });
+}
+
+#[test]
+fn attest_push_roundtrip() {
+    let alice = kp(10);
+    let bob = kp(20);
+    let attestations = vec![
+        Attestation::create(&alice, ContentHash([1; 32])),
+        Attestation::create(&bob, ContentHash([2; 32])),
+    ];
+    roundtrip(&Message::AttestPush { attestations });
+}
+
+#[test]
+fn done_roundtrip() {
+    roundtrip(&Message::Done);
+}
+
+#[test]
+fn malformed_bytes_decode_to_error() {
+    let bogus = vec![0xFFu8; 100];
+    assert!(Message::decode(&bogus).is_err());
+}
+
+#[test]
+fn empty_bytes_decode_to_error() {
+    assert!(Message::decode(&[]).is_err());
+}
+
+#[test]
+fn message_decode_after_encode_preserves_signatures() {
+    // El roundtrip de un Hello debe preservar la firma de modo que la
+    // verificación criptográfica del receptor siga funcionando.
+    let k = kp(33);
+    let root = ContentHash([55; 32]);
+    let sig = k.sign(root.as_bytes());
+    let original = Message::Hello {
+        peer_did: k.did(),
+        root_subtree_hash: root,
+        signature: sig,
+    };
+    let bytes = original.encode();
+    let decoded = Message::decode(&bytes).unwrap();
+    let Message::Hello {
+        peer_did,
+        root_subtree_hash,
+        signature,
+    } = decoded
+    else {
+        panic!("variante incorrecta tras decode");
+    };
+    assert!(peer_did.verify(root_subtree_hash.as_bytes(), &signature));
+}