clawft_kernel/
mesh_sensor.rs

1//! Mesh sensor topics for LeWM observational frames (WEFT-526).
2//!
3//! Registers the three `mesh.sensor.v1.*` topics on the kernel mesh wire,
4//! frames payloads as [`FrameType::SensorObservation`], and indexes every
5//! publish onto an append-only ExoChain stand-in (production hosts bridge to
6//! [`crate::chain::ChainManager`]).
7//!
8//! Wire CBOR + Ed25519 signing lives in `weftos-sensor-pipeline-wire`; this
9//! module owns **mesh registration + publish/subscribe + chain index** without
10//! a hard dependency on the wire crate (payload is opaque CBOR bytes).
11
12use std::collections::HashMap;
13use std::sync::Arc;
14
15use serde::{Deserialize, Serialize};
16
17#[cfg(feature = "exochain")]
18use crate::chain::EVENT_KIND_MESH_SENSOR_FRAME;
19
20/// Same value as `chain::EVENT_KIND_MESH_SENSOR_FRAME`, which only exists
21/// with `exochain`; this keeps the mesh modules buildable without it
22/// (`check-mesh-only`). A test under `exochain` guards drift.
23#[cfg(not(feature = "exochain"))]
24const EVENT_KIND_MESH_SENSOR_FRAME: &str = "mesh.sensor.v1.frame";
25use crate::mesh_framing::{FrameType, MeshFrame};
26use crate::process::{Pid, ProcessTable};
27use crate::topic::TopicRouter;
28
29/// Primary observation stream (`mesh.sensor.v1.encoded`).
30pub const TOPIC_ENCODED: &str = "mesh.sensor.v1.encoded";
31/// Conditional consensus stream (`mesh.sensor.v1.consensus`).
32pub const TOPIC_CONSENSUS: &str = "mesh.sensor.v1.consensus";
33/// Multi-publisher control stream (`mesh.sensor.v1.control`).
34pub const TOPIC_CONTROL: &str = "mesh.sensor.v1.control";
35
36/// The three observation topics in stable order (matches wire crate).
37pub const SENSOR_TOPICS: &[&str] = &[TOPIC_ENCODED, TOPIC_CONSENSUS, TOPIC_CONTROL];
38
39/// Which of the three LeWM sensor topics a frame targets.
40#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
41#[serde(rename_all = "snake_case")]
42#[non_exhaustive]
43pub enum MeshSensorTopic {
44    /// `mesh.sensor.v1.encoded`
45    Encoded,
46    /// `mesh.sensor.v1.consensus`
47    Consensus,
48    /// `mesh.sensor.v1.control`
49    Control,
50}
51
52impl MeshSensorTopic {
53    /// Bare topic string without cluster suffix.
54    pub const fn as_str(self) -> &'static str {
55        match self {
56            Self::Encoded => TOPIC_ENCODED,
57            Self::Consensus => TOPIC_CONSENSUS,
58            Self::Control => TOPIC_CONTROL,
59        }
60    }
61
62    /// Topic with `.{cluster}` suffix for mesh routing.
63    pub fn with_cluster(self, cluster: &str) -> String {
64        format!("{}.{}", self.as_str(), cluster)
65    }
66
67    /// Parse from bare or cluster-suffixed topic string.
68    pub fn parse(s: &str) -> Option<Self> {
69        if s == TOPIC_ENCODED || s.starts_with(&format!("{TOPIC_ENCODED}.")) {
70            Some(Self::Encoded)
71        } else if s == TOPIC_CONSENSUS || s.starts_with(&format!("{TOPIC_CONSENSUS}.")) {
72            Some(Self::Consensus)
73        } else if s == TOPIC_CONTROL || s.starts_with(&format!("{TOPIC_CONTROL}.")) {
74            Some(Self::Control)
75        } else {
76            None
77        }
78    }
79
80    /// All variants in stable order.
81    pub const ALL: &'static [MeshSensorTopic] =
82        &[Self::Encoded, Self::Consensus, Self::Control];
83}
84
85/// Envelope carried on the mesh wire + local topic bus for one sensor frame.
86#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
87pub struct SensorMeshEnvelope {
88    /// Target topic kind.
89    pub topic: MeshSensorTopic,
90    /// Logical cluster name (topic suffix).
91    pub cluster: String,
92    /// Emitting node id.
93    pub node_id: u64,
94    /// Observation / emission timestamp (ms).
95    pub timestamp_ms: u64,
96    /// Opaque CBOR (typically a signed wire frame from the sensor pipeline).
97    pub payload: Vec<u8>,
98}
99
100impl SensorMeshEnvelope {
101    /// Full topic string including cluster.
102    pub fn topic_name(&self) -> String {
103        self.topic.with_cluster(&self.cluster)
104    }
105
106    /// Build a mesh frame (`FrameType::SensorObservation`) for transport.
107    pub fn to_mesh_frame(&self) -> Result<MeshFrame, String> {
108        let bytes = serde_json::to_vec(self).map_err(|e| e.to_string())?;
109        Ok(MeshFrame {
110            frame_type: FrameType::SensorObservation,
111            payload: bytes,
112        })
113    }
114
115    /// Decode from a mesh frame payload.
116    pub fn from_mesh_frame(frame: &MeshFrame) -> Result<Self, String> {
117        if frame.frame_type != FrameType::SensorObservation {
118            return Err(format!(
119                "expected SensorObservation frame, got {:?}",
120                frame.frame_type
121            ));
122        }
123        serde_json::from_slice(&frame.payload).map_err(|e| e.to_string())
124    }
125
126    /// Stable hex digest of the payload for ExoChain indexing (FNV-1a 64).
127    pub fn payload_digest_hex(&self) -> String {
128        // Lightweight non-crypto fingerprint so this path stays free of an
129        // extra hash dep. Production bridges may re-hash with blake3.
130        let mut hash: u64 = 0xcbf29ce484222325;
131        for &b in &self.payload {
132            hash ^= u64::from(b);
133            hash = hash.wrapping_mul(0x100000001b3);
134        }
135        format!("{hash:016x}")
136    }
137}
138
139/// One ExoChain index record for a published sensor frame.
140#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
141pub struct SensorFrameIndexEntry {
142    /// Monotonic sequence.
143    pub seq: u64,
144    /// Event kind (`mesh.sensor.v1.frame`).
145    pub kind: String,
146    /// Topic string (with cluster).
147    pub topic: String,
148    /// Cluster name.
149    pub cluster: String,
150    /// Emitting node.
151    pub node_id: u64,
152    /// Timestamp ms.
153    pub timestamp_ms: u64,
154    /// Payload digest hex.
155    pub digest: String,
156    /// Payload byte length.
157    pub payload_len: usize,
158}
159
160/// In-memory ExoChain indexer for sensor frames (tests + offline hosts).
161#[derive(Debug, Default, Clone)]
162pub struct SensorFrameChainIndex {
163    /// Appended index entries.
164    pub entries: Vec<SensorFrameIndexEntry>,
165    next_seq: u64,
166}
167
168impl SensorFrameChainIndex {
169    /// Empty index.
170    pub fn new() -> Self {
171        Self::default()
172    }
173
174    /// Number of indexed frames.
175    pub fn len(&self) -> usize {
176        self.entries.len()
177    }
178
179    /// True when empty.
180    pub fn is_empty(&self) -> bool {
181        self.entries.is_empty()
182    }
183
184    /// Append an envelope; returns assigned sequence.
185    pub fn index(&mut self, env: &SensorMeshEnvelope) -> u64 {
186        let seq = self.next_seq;
187        self.next_seq = self.next_seq.saturating_add(1);
188        self.entries.push(SensorFrameIndexEntry {
189            seq,
190            kind: EVENT_KIND_MESH_SENSOR_FRAME.to_string(),
191            topic: env.topic_name(),
192            cluster: env.cluster.clone(),
193            node_id: env.node_id,
194            timestamp_ms: env.timestamp_ms,
195            digest: env.payload_digest_hex(),
196            payload_len: env.payload.len(),
197        });
198        seq
199    }
200}
201
202/// Result of a sensor publish.
203#[derive(Debug, Clone, PartialEq, Eq)]
204pub struct SensorPublishResult {
205    /// Topic the frame was published on.
206    pub topic: String,
207    /// Chain index sequence.
208    pub chain_seq: u64,
209    /// Number of local in-process subscriber deliveries.
210    pub local_deliveries: usize,
211    /// Number of TopicRouter PID subscribers currently registered.
212    pub router_subscribers: usize,
213}
214
215/// Kernel-side publisher: TopicRouter registration + local fan-out + ExoChain.
216///
217/// Subscribers attach via [`TopicRouter::subscribe`] on cluster-suffixed topic
218/// names, and/or via [`SensorTopicBus::subscribe_local`] for in-process tests.
219pub struct SensorTopicBus {
220    /// Topic router for subscriber registration.
221    pub router: Arc<TopicRouter>,
222    /// ExoChain stand-in index.
223    pub chain: SensorFrameChainIndex,
224    /// Last payload published per full topic name.
225    last_by_topic: HashMap<String, Vec<u8>>,
226    /// Local in-process subscriber queues (topic → payloads).
227    local_subs: HashMap<String, Vec<Vec<u8>>>,
228}
229
230impl SensorTopicBus {
231    /// Construct a bus with a fresh process table for the topic router.
232    pub fn new() -> Self {
233        let pt = Arc::new(ProcessTable::new(64));
234        Self::with_router(Arc::new(TopicRouter::new(pt)))
235    }
236
237    /// Construct around an existing topic router (kernel boot path).
238    pub fn with_router(router: Arc<TopicRouter>) -> Self {
239        Self {
240            router,
241            chain: SensorFrameChainIndex::new(),
242            last_by_topic: HashMap::new(),
243            local_subs: HashMap::new(),
244        }
245    }
246
247    /// Register the three base topics on the router for discovery.
248    ///
249    /// TopicRouter is lazy (entries appear on subscribe). We touch each base
250    /// topic with a sentinel subscribe/unsubscribe so `list_topics` can see
251    /// them once real subscribers attach; the constants remain the source of
252    /// truth either way.
253    pub fn register_well_known_topics(&self) {
254        let sentinel: Pid = 0;
255        for t in SENSOR_TOPICS {
256            self.router.subscribe(sentinel, t);
257            self.router.unsubscribe(sentinel, t);
258        }
259    }
260
261    /// Well-known topic names (always three).
262    pub fn well_known_topics(&self) -> &'static [&'static str] {
263        SENSOR_TOPICS
264    }
265
266    /// Subscribe a process to a sensor topic (with optional cluster suffix).
267    pub fn subscribe_pid(&self, pid: Pid, topic: MeshSensorTopic, cluster: Option<&str>) {
268        let name = match cluster {
269            Some(c) if !c.is_empty() => topic.with_cluster(c),
270            _ => topic.as_str().to_string(),
271        };
272        self.router.subscribe(pid, &name);
273    }
274
275    /// Local test subscriber: buffers published payloads for a topic.
276    pub fn subscribe_local(&mut self, topic: &str) {
277        self.local_subs.entry(topic.to_string()).or_default();
278    }
279
280    /// Drain local subscriber buffer for a topic.
281    pub fn drain_local(&mut self, topic: &str) -> Vec<Vec<u8>> {
282        self.local_subs
283            .get_mut(topic)
284            .map(std::mem::take)
285            .unwrap_or_default()
286    }
287
288    /// Publish a sensor envelope: mesh frame encode check, local fan-out,
289    /// TopicRouter subscriber count, ExoChain index.
290    pub fn publish(&mut self, env: SensorMeshEnvelope) -> Result<SensorPublishResult, String> {
291        // Cross-stream mesh link: prove FrameType::SensorObservation encodes.
292        let mesh_frame = env.to_mesh_frame()?;
293        let _wire = mesh_frame
294            .encode()
295            .map_err(|e| format!("mesh frame encode: {e}"))?;
296
297        let topic = env.topic_name();
298        let payload = env.payload.clone();
299        let mut local_deliveries = 0usize;
300
301        if let Some(q) = self.local_subs.get_mut(&topic) {
302            q.push(payload.clone());
303            local_deliveries += 1;
304        }
305        let base = env.topic.as_str();
306        if base != topic
307            && let Some(q) = self.local_subs.get_mut(base)
308        {
309            q.push(payload.clone());
310            local_deliveries += 1;
311        }
312
313        self.last_by_topic.insert(topic.clone(), payload);
314        let router_subscribers = self.router.subscribers(&topic).len();
315        let chain_seq = self.chain.index(&env);
316
317        Ok(SensorPublishResult {
318            topic,
319            chain_seq,
320            local_deliveries,
321            router_subscribers,
322        })
323    }
324
325    /// Last payload for a topic (if any).
326    pub fn last_payload(&self, topic: &str) -> Option<&[u8]> {
327        self.last_by_topic.get(topic).map(|v| v.as_slice())
328    }
329}
330
331impl Default for SensorTopicBus {
332    fn default() -> Self {
333        Self::new()
334    }
335}
336
337#[cfg(test)]
338mod tests {
339    #[cfg(feature = "exochain")]
340    #[test]
341    fn frame_event_kind_matches_non_exochain_copy() {
342        assert_eq!(super::EVENT_KIND_MESH_SENSOR_FRAME, "mesh.sensor.v1.frame");
343    }
344
345    use super::*;
346
347    #[test]
348    fn three_topics_registered_and_published() {
349        let mut bus = SensorTopicBus::new();
350        bus.register_well_known_topics();
351        assert_eq!(bus.well_known_topics().len(), 3);
352        assert_eq!(MeshSensorTopic::ALL.len(), 3);
353
354        for topic in MeshSensorTopic::ALL {
355            let full = topic.with_cluster("alpha");
356            bus.subscribe_local(&full);
357        }
358
359        let mut seqs = Vec::new();
360        for (i, topic) in MeshSensorTopic::ALL.iter().enumerate() {
361            let env = SensorMeshEnvelope {
362                topic: *topic,
363                cluster: "alpha".into(),
364                node_id: 7,
365                timestamp_ms: 1000 + i as u64,
366                payload: format!("frame-{i}").into_bytes(),
367            };
368            let r = bus.publish(env).expect("publish");
369            assert_eq!(r.topic, topic.with_cluster("alpha"));
370            assert_eq!(r.local_deliveries, 1);
371            seqs.push(r.chain_seq);
372        }
373
374        assert_eq!(bus.chain.len(), 3);
375        assert_eq!(seqs, vec![0, 1, 2]);
376        for e in &bus.chain.entries {
377            assert_eq!(e.kind, EVENT_KIND_MESH_SENSOR_FRAME);
378            assert!(e.topic.starts_with("mesh.sensor.v1."));
379            assert_eq!(e.cluster, "alpha");
380            assert_eq!(e.node_id, 7);
381            assert!(!e.digest.is_empty());
382        }
383
384        // Subscriber path drained one payload per topic.
385        for topic in MeshSensorTopic::ALL {
386            let drained = bus.drain_local(&topic.with_cluster("alpha"));
387            assert_eq!(drained.len(), 1);
388        }
389    }
390
391    #[test]
392    fn mesh_frame_roundtrip_sensor_observation() {
393        let env = SensorMeshEnvelope {
394            topic: MeshSensorTopic::Encoded,
395            cluster: "c".into(),
396            node_id: 1,
397            timestamp_ms: 42,
398            payload: b"cbor-placeholder".to_vec(),
399        };
400        let frame = env.to_mesh_frame().unwrap();
401        assert_eq!(frame.frame_type, FrameType::SensorObservation);
402        let wire = frame.encode().unwrap();
403        let decoded = MeshFrame::decode(&wire[4..]).unwrap();
404        let back = SensorMeshEnvelope::from_mesh_frame(&decoded).unwrap();
405        assert_eq!(back, env);
406    }
407
408    #[test]
409    fn topic_parse_with_cluster() {
410        assert_eq!(
411            MeshSensorTopic::parse("mesh.sensor.v1.control.east"),
412            Some(MeshSensorTopic::Control)
413        );
414        assert_eq!(MeshSensorTopic::parse("other"), None);
415        assert_eq!(TOPIC_ENCODED, "mesh.sensor.v1.encoded");
416        assert_eq!(TOPIC_CONSENSUS, "mesh.sensor.v1.consensus");
417        assert_eq!(TOPIC_CONTROL, "mesh.sensor.v1.control");
418    }
419}
clawft_kernel/mesh_sensor.rs

clawft_kernel/
mesh_sensor.rs
