feat(service): standalone tines service with Unix-socket IPC

New `tines` crate (added to workspace members) hosting a `Node<LogStore,
BtTransport>` and serving a tiny request/response IPC over a Unix-domain
socket for robots without ROS2. Four operations: `observe`, `belief`,
`events`, and `status`, over length-prefixed JSON frames.

The `serve` loop now serves each connection request-by-request (previously
it handled one request per connection, dropping the socket after the first
response). Integration tests drive `handle`/`serve` with a mock transport
that auto-acks observed broadcasts and echoes an empty Summary, so a
confidence event fires without a real BLE radio.

Cargo.lock

@@ -1460,6 +1460,18 @@ dependencies = [
  "syn 3.0.9",
 ]
 
+[[package]]
+name = "tines"
+version = "0.3.0"
+dependencies = [
+ "heapless 0.8.0",
+ "serde",
+ "serde_json",
+ "tines-bt",
+ "tines-core",
+ "tines-store",
+]
+
 [[package]]
 name = "tines-bt"
 version = "0.3.0"

Cargo.toml

@@ -1,5 +1,5 @@
 [workspace]
-members = ["tines-core", "tines-sim", "tines-store", "tines-bt", "tines-ros2"]
+members = ["tines-core", "tines-sim", "tines-store", "tines-bt", "tines-ros2", "tines"]
 resolver = "2"
 
 [workspace.package]

tines/Cargo.toml

@@ -0,0 +1,25 @@
+[package]
+name = "tines"
+version.workspace = true
+edition.workspace = true
+license.workspace = true
+description = "Standalone tines service: hosts a tines Node over a Unix-domain-socket IPC for robots without ROS2."
+authors.workspace = true
+repository.workspace = true
+readme.workspace = true
+keywords = ["robotics", "delay-tolerant", "science::robotics", "ipc", "ble"]
+categories = ["science::robotics", "network-programming"]
+
+[dependencies]
+tines-core = { path = "../tines-core", version = "0.3.0" }
+tines-store = { path = "../tines-store", version = "0.3.0" }
+tines-bt = { path = "../tines-bt", version = "0.3.0" }
+serde = { version = "1", features = ["derive"] }
+serde_json = "1"
+
+[dev-dependencies]
+tines-core = { path = "../tines-core", version = "0.3.0" }
+tines-store = { path = "../tines-store", version = "0.3.0" }
+heapless = "0.8"
+serde = { version = "1", features = ["derive"] }
+serde_json = "1"

tines/src/ipc.rs

@@ -0,0 +1,297 @@
+//! IPC over a Unix-domain socket.
+//!
+//! The [`serve`] fn hosts a tines [`Node`] and speaks a tiny
+//! request/response protocol to each connecting peer. A single instance is the
+//! control plane for its device; external tools (or another tines Node, in
+//! later tasks) drive the node through four operations: `observe`, `belief`,
+//! `events`, and `status`.
+//!
+//! # Wire format
+//!
+//! Length-prefixed JSON, one message per direction:
+//!
+//! ```text
+//!   [u32 : little-endian byte length] [u8 * length : a JSON frame]
+//! ```
+//!
+//! Each frame is one of the [`Request`] or [`Response`] enums, tagged by a
+//! `"kind"` field (serde externally-tagged). Frames are capped at 1 MiB; a
+//! longer frame is treated as a malformed request and the connection is
+//! dropped. There is no framing negotiation — the first byte after the length
+//! prefix is `"`.
+//!
+//! Observations are content-addressed, so IDs are exchanged as lowercase hex
+//! (the hex form of the 16-byte `ObservationId`). Topics and payloads are
+//! passed to `observe` the same way: as hex.
+//!
+//! # Sequence numbers
+//!
+//! `now` is the tines tick passed to `observe`/`belief`; the service reads it
+//! from its own wall clock (`now_ms()`) and ignores any caller-supplied value,
+//! so the wire request carries no timestamp.
+
+use std::fmt::Write as _;
+use std::io::{Read, Write};
+use std::os::unix::net::UnixListener;
+use std::path::{Path, PathBuf};
+use std::sync::{Arc, Mutex};
+
+use tines_core::{Node, Store, Transport};
+
+const MAX_FRAME: usize = 1 << 20; // 1 MiB cap on a single JSON frame
+
+/// A confidence event handed out by the IPC `events` request.
+#[derive(serde::Serialize, serde::Deserialize, Debug)]
+pub struct EventOut {
+    /// Hex of the `ObservationId` this confidence concerns.
+    pub id: String,
+    /// Hex of the `NodeId` of the peer that acknowledged.
+    pub peer: String,
+}
+
+/// A belief returned by the IPC `belief` request.
+#[derive(serde::Serialize, serde::Deserialize, Debug)]
+pub struct BeliefOut {
+    /// Hex of the `ObservationId`.
+    pub id: String,
+    /// Hex of the topic.
+    pub topic: String,
+    /// Hex of the payload.
+    pub payload: String,
+}
+
+/// An IPC request. See the module docs for the wire format.
+#[derive(serde::Serialize, serde::Deserialize, Debug)]
+#[serde(tag = "kind")]
+pub enum Request {
+    /// Insert an observation under `topic`.
+    Observe {
+        /// Hex of the topic bytes to observe.
+        topic: String,
+        /// Hex of the observation payload.
+        payload: String,
+    },
+    /// Read the latest belief for `topic`.
+    Belief {
+        /// Hex of the topic bytes to query.
+        topic: String,
+    },
+    /// Drain outstanding confidence events.
+    Events,
+    /// Report the node's identity and log depth.
+    Status,
+}
+
+/// An IPC response. See the module docs for the wire format.
+#[derive(serde::Serialize, serde::Deserialize, Debug)]
+#[serde(tag = "kind")]
+pub enum Response {
+    /// The inserted observation's `id` (hex).
+    Observe {
+        /// Hex of the `ObservationId`.
+        id: String,
+    },
+    /// The latest belief for the topic, if any.
+    Belief {
+        /// The stored belief, or `None` if the topic is unknown.
+        obs: Option<BeliefOut>,
+    },
+    /// Confidence events drained from the exchange.
+    Events {
+        /// The drained confidence events.
+        events: Vec<EventOut>,
+    },
+    /// The node's `id`, next tines `next_seq`, and observations `stored`.
+    Status {
+        /// Hex of the node's own `NodeId`.
+        id: String,
+        /// Next tines sequence number the node will use.
+        next_seq: u64,
+        /// Number of observations in the store.
+        stored: u32,
+    },
+    /// The request was malformed or rejected.
+    Error {
+        /// A human-readable explanation.
+        message: String,
+    },
+}
+
+/// Handle one IPC request against a node.
+///
+/// This is the single point where a request becomes a node operation;
+/// [`serve`](fn@serve) decodes frames, calls `handle`, and re-encodes the
+/// response. Kept generic over the store/transport so tests can drive it with a
+/// mock transport.
+pub fn handle<S, T>(node: &mut Node<S, T>, req: Request, now: u64) -> Response
+where
+    S: Store + 'static,
+    T: Transport,
+{
+    match req {
+        Request::Observe { topic, payload } => {
+            let topic = match decode_hex(&topic) {
+                Ok(t) => t,
+                Err(e) => return error(e),
+            };
+            let payload = match decode_hex(&payload) {
+                Ok(t) => t,
+                Err(e) => return error(e),
+            };
+            match node.observe(&topic, &payload, None, now) {
+                Ok(id) => Response::Observe {
+                    id: encode_hex(&id),
+                },
+                Err(e) => error(format!("observe failed: {e:?}")),
+            }
+        }
+        Request::Belief { topic } => {
+            let topic = match decode_hex(&topic) {
+                Ok(t) => t,
+                Err(e) => return error(e),
+            };
+            match node.belief(&topic, &tines_core::LatestWins, now) {
+                Some(obs) => Response::Belief {
+                    obs: Some(BeliefOut {
+                        id: encode_hex(&obs.id()),
+                        topic: encode_hex(&obs.topic),
+                        payload: encode_hex(&obs.payload),
+                    }),
+                },
+                None => Response::Belief { obs: None },
+            }
+        }
+        Request::Events => {
+            let events = node
+                .drain_events()
+                .into_iter()
+                .map(|ev| EventOut {
+                    id: encode_hex(&ev.id),
+                    peer: encode_hex(&ev.peer),
+                })
+                .collect();
+            Response::Events { events }
+        }
+        Request::Status => Response::Status {
+            id: encode_hex(&node.id()),
+            next_seq: node.next_seq(),
+            stored: node.store().len() as u32,
+        },
+    }
+}
+
+/// Serve IPC on `socket_path` until the listener dies or a response write
+/// fails. Blocks; each accepted connection is served request-by-request.
+///
+/// `Node<S, T>` may be any concrete store/transport, so the real binary passes
+/// `Node<LogStore, BtTransport>` and tests pass a mock.
+pub fn serve<S, T>(node: Node<S, T>, socket_path: impl AsRef<Path>) -> std::io::Result<()>
+where
+    S: Store + 'static,
+    T: Transport + 'static,
+{
+    let node = Arc::new(Mutex::new(node));
+    let path: PathBuf = socket_path.as_ref().to_path_buf();
+
+    // Clear a stale socket file left by a previous run so bind() can't fail.
+    let _ = std::fs::remove_file(&path);
+    let listener = UnixListener::bind(&path)?;
+
+    for stream in listener.incoming() {
+        let mut stream = stream?;
+        // Serve this connection request-by-request until the peer disconnects
+        // or sends a malformed frame, then move on to the next connection.
+        while let Ok(req) = read_request(&mut stream) {
+            let now = now_ms();
+            let resp = {
+                let mut guard = node.lock().unwrap();
+                handle(&mut guard, req, now)
+            };
+            // A client that vanished mid-connection shouldn't take down the
+            // whole service; just drop this one socket.
+            if write_response(&mut stream, &resp).is_err() {
+                break;
+            }
+        }
+    }
+    Ok(())
+}
+
+/// Read one length-prefixed JSON frame from a stream.
+fn read_request<S: Read>(stream: &mut S) -> std::io::Result<Request> {
+    let mut len_buf = [0u8; 4];
+    stream.read_exact(&mut len_buf)?;
+    let len = u32::from_le_bytes(len_buf) as usize;
+    if len > MAX_FRAME {
+        return Err(std::io::Error::new(
+            std::io::ErrorKind::InvalidData,
+            format!("frame {len} bytes exceeds {MAX_FRAME}"),
+        ));
+    }
+    let mut buf = vec![0u8; len];
+    stream.read_exact(&mut buf)?;
+    serde_json::from_slice(&buf)
+        .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))
+}
+
+/// Write one length-prefixed JSON frame to a stream.
+fn write_response<S: Write>(stream: &mut S, resp: &Response) -> std::io::Result<()> {
+    let body = serde_json::to_vec(resp)
+        .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?;
+    if body.len() > MAX_FRAME {
+        return Err(std::io::Error::new(
+            std::io::ErrorKind::InvalidData,
+            "response too large",
+        ));
+    }
+    let len = (body.len() as u32).to_le_bytes();
+    stream.write_all(&len)?;
+    stream.write_all(&body)?;
+    stream.flush()
+}
+
+fn error(msg: impl std::fmt::Display) -> Response {
+    Response::Error {
+        message: msg.to_string(),
+    }
+}
+
+fn now_ms() -> u64 {
+    std::time::SystemTime::now()
+        .duration_since(std::time::SystemTime::UNIX_EPOCH)
+        .map(|d| d.as_secs())
+        .unwrap_or(0)
+}
+
+fn encode_hex(bytes: &[u8]) -> String {
+    let mut s = String::with_capacity(bytes.len() * 2);
+    for b in bytes {
+        let _ = write!(s, "{b:02x}");
+    }
+    s
+}
+
+fn decode_hex(s: &str) -> Result<Vec<u8>, String> {
+    if !s.len().is_multiple_of(2) {
+        return Err(format!("odd-length hex string: {}", s.len()));
+    }
+    let bytes = s.as_bytes();
+    let mut out = Vec::with_capacity(bytes.len() / 2);
+    let mut i = 0;
+    while i < bytes.len() {
+        let hi = hex_val(bytes[i])?;
+        let lo = hex_val(bytes[i + 1])?;
+        out.push((hi << 4) | lo);
+        i += 2;
+    }
+    Ok(out)
+}
+
+fn hex_val(c: u8) -> Result<u8, String> {
+    match c {
+        b'0'..=b'9' => Ok(c - b'0'),
+        b'a'..=b'f' => Ok(c - b'a' + 10),
+        b'A'..=b'F' => Ok(c - b'A' + 10),
+        other => Err(format!("invalid hex char: {other:02x}")),
+    }
+}

tines/src/lib.rs

@@ -0,0 +1,15 @@
+//! The standalone `tines` service.
+//!
+//! Hosts a `Node` over a Unix-domain socket and serves a
+//! small request/response IPC for `observe` / `belief` / `events` / `status`.
+//! One instance runs per device — the binary (`src/main.rs`) wires a
+//! `Node<LogStore, BtTransport>` for real BLE, while tests drive the same
+//! [`serve`] / [`handle`] path with a mock transport so no radio
+//! is required.
+//!
+//! The IPC protocol is documented in [`ipc`].
+#![deny(missing_docs)]
+
+pub mod ipc;
+
+pub use ipc::{handle, serve, BeliefOut, EventOut, Request, Response};

tines/src/main.rs

@@ -0,0 +1,88 @@
+//! Standalone `tines` service binary.
+//!
+//! Loads a JSON config, builds a `Node<LogStore, BtTransport>` (BLE over
+//! `bluer`), and serves the IPC ([`tines::serve`]) over a Unix-domain socket.
+//! One instance per device — the typical deployment on robots without ROS2.
+//!
+//! The node/peer ids are 32 lowercase hex chars (the 16-byte `NodeId`), and
+//! `first_seq` is the next tines sequence number to use (persisted across
+//! reboots, starting from `tines`'s 1-based numbering).
+
+use std::error::Error;
+use std::path::Path;
+
+use tines::serve;
+use tines_bt::{BluerRadio, BtTransport};
+use tines_core::{Node, NodeId};
+use tines_store::LogStore;
+
+/// On-disk configuration for a running service.
+#[derive(serde::Deserialize)]
+struct Config {
+    /// This device's advertised `NodeId`, 32 lowercase hex chars.
+    node_id: String,
+    /// The peer's advertised `NodeId`, 32 lowercase hex chars.
+    peer: String,
+    /// Next tines sequence number to use (persisted across reboots).
+    first_seq: u64,
+    /// Absolute path to the LogStore append-only log.
+    log_path: String,
+    /// Unix-domain socket path the IPC listens on.
+    socket_path: String,
+}
+
+fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
+    let cfg_path = std::env::args()
+        .nth(1)
+        .ok_or("usage: tines <config.json>")?;
+    let config: Config = serde_json::from_str(&std::fs::read_to_string(cfg_path)?)?;
+
+    let me = parse_node_id(&config.node_id)?;
+    let peer = parse_node_id(&config.peer)?;
+
+    // LogStore is the Store (S); BtTransport wraps it as the Transport (T).
+    let store = LogStore::new(Path::new(&config.log_path), 0, config.first_seq)
+        .map_err(|e| format!("failed to open log store: {e:?}"))?;
+    let radio = BluerRadio::new(me, peer);
+    let transport = BtTransport::new(me, peer, radio);
+    let node = Node::new(me, config.first_seq, store, transport);
+
+    println!(
+        "tines: listening on {} (node {})",
+        config.socket_path,
+        node_hex(me)
+    );
+    serve(node, Path::new(&config.socket_path))?;
+    Ok(())
+}
+
+fn parse_node_id(s: &str) -> Result<NodeId, Box<dyn Error + Send + Sync>> {
+    if s.len() != 32 {
+        return Err(format!("node id must be 32 hex chars, got {}", s.len()).into());
+    }
+    let mut id = [0u8; 16];
+    for (i, chunk) in s.as_bytes().chunks(2).enumerate() {
+        let hi = hex_val(chunk[0])?;
+        let lo = hex_val(chunk[1])?;
+        id[i] = (hi << 4) | lo;
+    }
+    Ok(id)
+}
+
+fn hex_val(c: u8) -> Result<u8, Box<dyn Error + Send + Sync>> {
+    match c {
+        b'0'..=b'9' => Ok(c - b'0'),
+        b'a'..=b'f' => Ok(c - b'a' + 10),
+        b'A'..=b'F' => Ok(c - b'A' + 10),
+        other => Err(format!("invalid hex char: {other:02x}").into()),
+    }
+}
+
+fn node_hex(id: NodeId) -> String {
+    use std::fmt::Write as _;
+    let mut s = String::with_capacity(32);
+    for b in id {
+        let _ = write!(s, "{b:02x}");
+    }
+    s
+}

tines/tests/ipc.rs

@@ -0,0 +1,315 @@
+//! Integration tests for the `tines` IPC.
+//!
+//! These drive `handle`/`serve` with a mock transport, so no real BLE radio is
+//! needed: the mock surfaces a single `Contact`, records outbound frames, and
+//! auto-acks any observation broadcast to it (simulating a well-behaved peer).
+
+use std::collections::{HashSet, VecDeque};
+use std::io::{Read, Write};
+use std::os::unix::net::UnixStream;
+use std::path::{Path, PathBuf};
+use std::process;
+use std::thread;
+use std::time::Duration;
+
+use heapless::Vec as HeaplessVec;
+use std::fmt::Write as _;
+use tines::{handle, serve, Request, Response};
+use tines_core::{
+    consts, Event, Frame, FrameBuf, Node, NodeId, ObservationId, SendError, Transport,
+};
+use tines_store::LogStore;
+
+/// A no-op [`Transport`] for tests: it never surfaces inbound frames except
+/// the ones injected by the test, and it records outbound frames.
+struct MockTransport {
+    peer: NodeId,
+    contacted: bool,
+    pending: VecDeque<Event<NodeId>>,
+    sent: Vec<Vec<u8>>,
+    acked: HashSet<ObservationId>,
+}
+
+/// A well-behaved peer: auto-acks each distinct observation it is sent.
+impl MockTransport {
+    fn new(peer: NodeId) -> Self {
+        Self {
+            peer,
+            contacted: false,
+            pending: VecDeque::new(),
+            sent: Vec::new(),
+            acked: HashSet::new(),
+        }
+    }
+}
+
+impl Transport for MockTransport {
+    type Addr = NodeId;
+
+    fn poll(&mut self) -> Option<Event<NodeId>> {
+        if !self.contacted {
+            self.contacted = true;
+            return Some(Event::Contact(self.peer));
+        }
+        self.pending.pop_front()
+    }
+
+    fn send(&mut self, to: &NodeId, frame: &[u8]) -> Result<(), SendError> {
+        self.sent.push(frame.to_vec());
+        match Frame::decode(frame) {
+            Ok(Frame::Obs { obs, .. }) => {
+                // A well-behaved peer acks each distinct observed broadcast.
+                if self.acked.insert(obs.id()) {
+                    let ack = ack_frame(*to, obs.id());
+                    self.pending.push_back(frame_event(*to, &ack));
+                }
+            }
+            Ok(Frame::Summary { salt, .. }) => {
+                // Echo back an empty, full-size Summary so the exchange learns
+                // the peer holds nothing and broadcasts its diff (which we then
+                // ack, firing a ConfidenceEvent).
+                let raw: [u8; consts::SUMMARY_BYTES] = [0u8; consts::SUMMARY_BYTES];
+                let bits: HeaplessVec<u8, { consts::SUMMARY_BYTES }> =
+                    HeaplessVec::from_slice(&raw).expect("512 fits SUMMARY_BYTES");
+                let reply = Frame::Summary {
+                    from: *to,
+                    salt,
+                    bits,
+                };
+                self.pending.push_back(frame_event(*to, &reply));
+            }
+            _ => {}
+        }
+        Ok(())
+    }
+
+    fn mtu(&self) -> usize {
+        consts::MAX_FRAME
+    }
+}
+
+fn ack_frame(from: NodeId, id: ObservationId) -> Frame {
+    let ids: HeaplessVec<ObservationId, { consts::ACK_IDS_CAP }> =
+        HeaplessVec::from_slice(&[id]).expect("one id fits ACK_IDS_CAP");
+    Frame::Ack { from, ids }
+}
+
+fn frame_event(from: NodeId, frame: &Frame) -> Event<NodeId> {
+    let mut buf = [0u8; consts::MAX_FRAME];
+    let n = frame.encode(&mut buf).expect("frame fits MAX_FRAME");
+    Event::Frame(
+        from,
+        FrameBuf::from_slice(&buf[..n]).expect("frame fits FrameBuf"),
+    )
+}
+
+fn topic() -> Vec<u8> {
+    b"tines/test".to_vec()
+}
+
+fn payload() -> Vec<u8> {
+    b"hello".to_vec()
+}
+
+fn hex_encode(bytes: &[u8]) -> String {
+    let mut s = String::with_capacity(bytes.len() * 2);
+    for b in bytes {
+        let _ = write!(s, "{b:02x}");
+    }
+    s
+}
+
+fn unique_dir(tag: &str) -> PathBuf {
+    let dir = std::env::temp_dir().join(format!("tines-ipc-{}-{}", tag, process::id()));
+    std::fs::create_dir_all(&dir).unwrap();
+    dir
+}
+
+fn build_node(peer: NodeId, dir: &Path) -> Node<LogStore, MockTransport> {
+    let store_path = dir.join("store.log");
+    let store = LogStore::new(&store_path, 0, 1).expect("store open");
+    Node::new(peer, 1, store, MockTransport::new(peer))
+}
+
+fn connect(path: &Path) -> UnixStream {
+    let mut delay = 5;
+    for _ in 0..500 {
+        if let Ok(s) = UnixStream::connect(path) {
+            return s;
+        }
+        thread::sleep(Duration::from_millis(delay));
+        delay = delay.min(50);
+    }
+    panic!("failed to connect to {}", path.display());
+}
+
+fn rpc<S: Read + Write>(stream: &mut S, req: &Request) {
+    let body = serde_json::to_vec(req).expect("serialize request");
+    let len = (body.len() as u32).to_le_bytes();
+    stream.write_all(&len).unwrap();
+    stream.write_all(&body).unwrap();
+    stream.flush().unwrap();
+}
+
+fn read_response<S: Read>(stream: &mut S) -> Response {
+    let mut len_buf = [0u8; 4];
+    stream.read_exact(&mut len_buf).unwrap();
+    let len = u32::from_le_bytes(len_buf) as usize;
+    let mut buf = vec![0u8; len];
+    stream.read_exact(&mut buf).unwrap();
+    serde_json::from_slice(&buf).expect("valid response")
+}
+
+#[test]
+fn handle_status_and_belief_direct() {
+    let dir = unique_dir("direct");
+    let mut node = build_node([3; 16], &dir);
+    let x = node.observe(&topic(), &payload(), None, 0).unwrap();
+
+    let status = handle(&mut node, Request::Status, 0);
+    match status {
+        Response::Status {
+            id,
+            next_seq,
+            stored,
+        } => {
+            assert_eq!(id, hex_encode(&node.id()));
+            assert_eq!(next_seq, 2);
+            assert_eq!(stored, 1);
+        }
+        other => panic!("expected status, got {other:?}"),
+    }
+
+    let belief = handle(
+        &mut node,
+        Request::Belief {
+            topic: hex_encode(&topic()),
+        },
+        0,
+    );
+    match belief {
+        Response::Belief { obs: Some(b) } => {
+            assert_eq!(b.id, hex_encode(&x));
+            assert_eq!(b.topic, hex_encode(&topic()));
+            assert_eq!(b.payload, hex_encode(&payload()));
+        }
+        other => panic!("expected belief, got {other:?}"),
+    }
+}
+
+#[test]
+fn handle_unknown_obs_is_belief_none() {
+    let dir = unique_dir("direct");
+    let mut node = build_node([3; 16], &dir);
+    let resp = handle(
+        &mut node,
+        Request::Belief {
+            topic: hex_encode(&topic()),
+        },
+        0,
+    );
+    match resp {
+        Response::Belief { obs: None } => {}
+        other => panic!("expected empty belief, got {other:?}"),
+    }
+}
+
+#[test]
+fn observe_and_belief_over_socket() {
+    let dir = unique_dir("roundtrip");
+    let socket = dir.join("ipc.sock");
+    let node = build_node([9; 16], &dir);
+    let serve_path = socket.clone();
+    let _handle = thread::spawn(move || serve(node, serve_path).unwrap());
+
+    let mut stream = connect(&socket);
+
+    rpc(
+        &mut stream,
+        &Request::Observe {
+            topic: hex_encode(&topic()),
+            payload: hex_encode(&payload()),
+        },
+    );
+    let id = match read_response(&mut stream) {
+        Response::Observe { id } => id,
+        other => panic!("expected Observe, got {other:?}"),
+    };
+    assert_eq!(id.len(), 32);
+
+    rpc(
+        &mut stream,
+        &Request::Belief {
+            topic: hex_encode(&topic()),
+        },
+    );
+    match read_response(&mut stream) {
+        Response::Belief { obs: Some(b) } => {
+            assert_eq!(b.id, id);
+            assert_eq!(b.topic, hex_encode(&topic()));
+            assert_eq!(b.payload, hex_encode(&payload()));
+        }
+        other => panic!("expected Belief, got {other:?}"),
+    }
+}
+
+#[test]
+fn events_over_socket() {
+    let dir = unique_dir("events");
+    let socket = dir.join("ipc.sock");
+    let peer = [7; 16];
+    let mut node = build_node(peer, &dir);
+
+    // Drive a confidence event into the node directly: observe (broadcast),
+    // let the mock auto-ack it, and poll it out of the exchange.
+    let x = node.observe(&topic(), &payload(), None, 0).unwrap();
+    node.poll(0); // contact -> node emits its summary
+    node.poll(0); // node broadcasts the observed topic; mock injects the ack
+    node.poll(0); // ack processed -> confidence event fires
+
+    let serve_path = socket.clone();
+    let _handle = thread::spawn(move || serve(node, serve_path).unwrap());
+    let mut stream = connect(&socket);
+
+    rpc(&mut stream, &Request::Events);
+    match read_response(&mut stream) {
+        Response::Events { events } => {
+            assert_eq!(events.len(), 1, "expected one confidence event");
+            assert_eq!(events[0].id, hex_encode(&x), "event id matches observed id");
+            assert_eq!(events[0].peer, hex_encode(&peer), "event came from peer");
+        }
+        other => panic!("expected Events, got {other:?}"),
+    }
+}
+
+#[test]
+fn handle_bad_hex_is_error() {
+    let dir = unique_dir("hex");
+    let mut node = build_node([3; 16], &dir);
+    let resp = handle(
+        &mut node,
+        Request::Observe {
+            topic: "z".into(),
+            payload: "ff".into(),
+        },
+        0,
+    );
+    match resp {
+        Response::Error { .. } => {}
+        other => panic!("expected error, got {other:?}"),
+    }
+}
+
+#[test]
+fn confidence_event_direct_probe() {
+    let dir = unique_dir("ev");
+    let mut node = build_node([7; 16], &dir);
+    let x = node.observe(&topic(), &payload(), None, 0).unwrap();
+    node.poll(0);
+    node.poll(0);
+    node.poll(0);
+    let evs = node.drain_events();
+    assert_eq!(evs.len(), 1, "one confidence event after observe+ack");
+    assert_eq!(evs[0].id, x);
+    assert_eq!(evs[0].peer, [7; 16]);
+}