qualia_client_core/
chat_mesh.rs

1//! Chat over the SocialWebNet mesh — bridge the chat-graph engine to peer tunnels.
2//!
3//! Chat already has a serializable message ([`RelayEnvelope`](crate::chat_relay::RelayEnvelope)), a
4//! signer, and a transport-agnostic apply path
5//! ([`apply_incoming_envelope`](crate::chat_relay::apply_incoming_envelope)). Today those ride the
6//! HTTP relay; this module routes them over the mesh instead — peer-to-peer, no relay server.
7//!
8//! It composes two lower layers:
9//! * [`crate::mesh_channel::ReliableEndpoint`] — per-peer at-least-once delivery + dedup over the
10//!   (lossy) datagram transport.
11//! * CBOR-encoded [`RelayEnvelope`] as the CHAT-port payload.
12//!
13//! Like `mesh_channel`, [`ChatMeshBridge`] is a **pure state machine**: it produces *frames to send*
14//! and *envelopes delivered*, but performs no socket I/O and touches no storage. The caller moves the
15//! frames with `SocialWebNet::send_datagram(peer, ports::CHAT, ports::CHAT, frame)` and applies each
16//! delivered envelope with [`crate::chat_relay::apply_incoming_envelope`]. This keeps the routing
17//! logic deterministically testable (two bridges exchanging a real envelope, with loss) and decoupled
18//! from the live mesh runtime, which lives in `qualia-core-db`.
19
20use std::collections::HashMap;
21
22use crate::chat_relay::RelayEnvelope;
23use crate::mesh_channel::{Inbound, ReliableEndpoint, DEFAULT_MAX_ATTEMPTS, DEFAULT_RTO_MS};
24
25/// A reliable-channel frame destined for a specific peer. The caller frames it as a mesh datagram on
26/// [`ports::CHAT`](qualia_core_db::p2p::mesh_datagram::ports) and sends it to `peer_did`.
27#[derive(Debug, Clone, PartialEq)]
28pub struct OutFrame {
29    /// The peer (by DID) to send this frame to.
30    pub peer_did: String,
31    /// The reliable-channel frame (the datagram payload).
32    pub frame: Vec<u8>,
33}
34
35/// The result of feeding one inbound frame to the bridge.
36#[derive(Debug, Default)]
37pub struct InboundChat {
38    /// Acknowledgement frames to send back to the peer.
39    pub acks: Vec<OutFrame>,
40    /// A newly-delivered chat envelope, if this frame completed one (absent for ACKs and duplicates).
41    pub delivered: Option<RelayEnvelope>,
42    /// True if a frame arrived but could not be decoded as a chat envelope (corrupt / wrong version).
43    pub decode_failed: bool,
44}
45
46/// CBOR-encode a chat envelope for the CHAT-port payload.
47pub fn encode_envelope(env: &RelayEnvelope) -> Vec<u8> {
48    let mut buf = Vec::new();
49    // ciborium only fails here if the writer errors, which a `Vec` never does.
50    let _ = ciborium::into_writer(env, &mut buf);
51    buf
52}
53
54/// Decode a CHAT-port payload back into a chat envelope. `None` if it is not a valid CBOR envelope.
55pub fn decode_envelope(bytes: &[u8]) -> Option<RelayEnvelope> {
56    ciborium::from_reader(bytes).ok()
57}
58
59/// Per-peer reliable chat routing over the mesh. One bridge per node; it holds a
60/// [`ReliableEndpoint`] per peer DID.
61pub struct ChatMeshBridge {
62    peers: HashMap<String, ReliableEndpoint>,
63    rto_ms: u64,
64    max_attempts: u32,
65}
66
67impl Default for ChatMeshBridge {
68    fn default() -> Self {
69        ChatMeshBridge {
70            peers: HashMap::new(),
71            rto_ms: DEFAULT_RTO_MS,
72            max_attempts: DEFAULT_MAX_ATTEMPTS,
73        }
74    }
75}
76
77impl ChatMeshBridge {
78    /// A bridge with explicit reliability parameters (mostly for tests).
79    pub fn new(rto_ms: u64, max_attempts: u32) -> ChatMeshBridge {
80        ChatMeshBridge {
81            peers: HashMap::new(),
82            rto_ms,
83            max_attempts,
84        }
85    }
86
87    fn endpoint(&mut self, peer_did: &str) -> &mut ReliableEndpoint {
88        let (rto, max) = (self.rto_ms, self.max_attempts);
89        self.peers
90            .entry(peer_did.to_string())
91            .or_insert_with(|| ReliableEndpoint::new(rto, max))
92    }
93
94    /// Peers this bridge currently has a channel to.
95    pub fn peers(&self) -> Vec<String> {
96        self.peers.keys().cloned().collect()
97    }
98
99    /// Reliably send `env` to every peer in `peer_dids`. The envelope is encoded once; each peer's
100    /// channel assigns its own sequence number. Returns the frames to send.
101    pub fn broadcast(
102        &mut self,
103        peer_dids: &[String],
104        env: &RelayEnvelope,
105        now_ms: u64,
106    ) -> Vec<OutFrame> {
107        let payload = encode_envelope(env);
108        peer_dids
109            .iter()
110            .map(|did| {
111                let frame = self.endpoint(did).send(&payload, now_ms);
112                OutFrame {
113                    peer_did: did.clone(),
114                    frame,
115                }
116            })
117            .collect()
118    }
119
120    /// Reliably send `env` to a single peer.
121    pub fn send_to(&mut self, peer_did: &str, env: &RelayEnvelope, now_ms: u64) -> OutFrame {
122        let payload = encode_envelope(env);
123        let frame = self.endpoint(peer_did).send(&payload, now_ms);
124        OutFrame {
125            peer_did: peer_did.to_string(),
126            frame,
127        }
128    }
129
130    /// Process one inbound reliable-channel frame from `peer_did`. Returns the ACK to send back and,
131    /// if the frame completed a new envelope, the decoded [`RelayEnvelope`] to apply.
132    pub fn on_inbound(&mut self, peer_did: &str, frame: &[u8], now_ms: u64) -> InboundChat {
133        let Inbound { delivered, to_send } = self.endpoint(peer_did).on_datagram(frame, now_ms);
134        let acks = to_send
135            .into_iter()
136            .map(|frame| OutFrame {
137                peer_did: peer_did.to_string(),
138                frame,
139            })
140            .collect();
141        let mut out = InboundChat {
142            acks,
143            delivered: None,
144            decode_failed: false,
145        };
146        if let Some(payload) = delivered {
147            match decode_envelope(&payload) {
148                Some(env) => out.delivered = Some(env),
149                None => out.decode_failed = true,
150            }
151        }
152        out
153    }
154
155    /// Retransmit any unacknowledged frames whose RTO has elapsed, across all peers.
156    pub fn on_tick(&mut self, now_ms: u64) -> Vec<OutFrame> {
157        let mut out = Vec::new();
158        for (did, ep) in self.peers.iter_mut() {
159            let (resend, _gave_up) = ep.on_tick(now_ms);
160            for frame in resend {
161                out.push(OutFrame {
162                    peer_did: did.clone(),
163                    frame,
164                });
165            }
166        }
167        out
168    }
169
170    /// Total frames still awaiting acknowledgement across all peers.
171    pub fn pending(&self) -> usize {
172        self.peers.values().map(|e| e.pending()).sum()
173    }
174}
175
176#[cfg(test)]
177mod tests {
178    use super::*;
179
180    fn envelope(session: &str, lamport: u64, content: &str) -> RelayEnvelope {
181        RelayEnvelope {
182            session_id: session.into(),
183            lamport,
184            role: "user".into(),
185            content: content.into(),
186            author_did: "did:wf:alice".into(),
187            author_name: Some("Alice".into()),
188            reply_to_fragment: None,
189            timestamp: 1_700_000_000,
190            signature_hex: "deadbeef".into(),
191            sub_agent_of: None,
192            agent_did: None,
193            model_id: None,
194            agent_backend: None,
195            outcome_sharing: None,
196        }
197    }
198
199    fn same(a: &RelayEnvelope, b: &RelayEnvelope) -> bool {
200        a.session_id == b.session_id
201            && a.lamport == b.lamport
202            && a.content == b.content
203            && a.author_did == b.author_did
204            && a.signature_hex == b.signature_hex
205    }
206
207    #[test]
208    fn envelope_cbor_round_trips() {
209        let env = envelope("s1", 7, "hello mesh");
210        let bytes = encode_envelope(&env);
211        let back = decode_envelope(&bytes).expect("decodes");
212        assert!(same(&env, &back));
213    }
214
215    #[test]
216    fn decode_rejects_garbage() {
217        assert!(decode_envelope(b"\xff\xff not cbor envelope").is_none());
218    }
219
220    #[test]
221    fn two_bridges_deliver_a_chat_envelope_reliably() {
222        // Alice's node and Bob's node, each a bridge; Alice sends a chat message to Bob.
223        let mut alice = ChatMeshBridge::default();
224        let mut bob = ChatMeshBridge::default();
225        let env = envelope("room-1", 42, "over the mesh we chat");
226
227        // Alice broadcasts to Bob → one frame for peer "bob".
228        let out = alice.broadcast(&["bob".to_string()], &env, 0);
229        assert_eq!(out.len(), 1);
230        assert_eq!(out[0].peer_did, "bob");
231        assert_eq!(alice.pending(), 1, "unacked until Bob acks");
232
233        // Bob receives it (tagging the sender as "alice"): delivers the envelope + emits an ACK.
234        let inb = bob.on_inbound("alice", &out[0].frame, 1);
235        let got = inb.delivered.expect("Bob got the envelope");
236        assert!(same(&got, &env));
237        assert_eq!(inb.acks.len(), 1);
238        assert_eq!(inb.acks[0].peer_did, "alice");
239
240        // Alice processes Bob's ACK → nothing pending.
241        let back = alice.on_inbound("bob", &inb.acks[0].frame, 2);
242        assert!(back.delivered.is_none());
243        assert_eq!(alice.pending(), 0);
244    }
245
246    #[test]
247    fn lost_envelope_is_retransmitted_and_deduplicated() {
248        let mut alice = ChatMeshBridge::new(100, 8);
249        let mut bob = ChatMeshBridge::default();
250        let env = envelope("room-1", 1, "will be lost then resent");
251
252        // Alice sends — but the frame is "lost" (never handed to Bob).
253        let _lost = alice.broadcast(&["bob".to_string()], &env, 0);
254        assert_eq!(alice.pending(), 1);
255
256        // After the RTO, Alice retransmits.
257        let resend = alice.on_tick(150);
258        assert_eq!(resend.len(), 1);
259
260        // Bob receives the retransmit and delivers once.
261        let inb = bob.on_inbound("alice", &resend[0].frame, 160);
262        assert!(same(&inb.delivered.expect("delivered on resend"), &env));
263
264        // A *second* copy of the same frame (e.g. the original arriving late) must NOT re-deliver.
265        let dup = bob.on_inbound("alice", &resend[0].frame, 170);
266        assert!(dup.delivered.is_none(), "deduplicated");
267        assert_eq!(dup.acks.len(), 1, "still acked");
268    }
269}
qualia_client_core/chat_mesh.rs

qualia_client_core/
chat_mesh.rs
