feat(host): service transport choice (bt|bus) and lib-level config

Cargo.lock

@@ -1469,6 +1469,7 @@ dependencies = [
  "serde",
  "serde_json",
  "tines-bt",
+ "tines-bus",
  "tines-core",
  "tines-store",
 ]

tines-bus/src/daemon.rs

@@ -221,6 +221,15 @@ fn process_msg(idx: usize, payload: &[u8], conns: &mut [Conn], actions: &mut Vec
                         });
                     }
                 }
+                // ...and the newcomer learns everyone already here.
+                for c in before.iter().chain(after.iter()) {
+                    if let Some(existing) = c.id {
+                        actions.push(Action::Forward {
+                            to: idx,
+                            env: control(0, existing),
+                        });
+                    }
+                }
             }
         }
         Some(my_id) => {

tines-bus/tests/daemon.rs

@@ -78,6 +78,21 @@ fn expect_no_data(s: &mut UnixStream, window_ms: u64) {
     }
 }
 
+#[test]
+fn newcomer_learns_existing_peers() {
+    let dir = unique_dir("newcomer");
+    let sock = spawn_daemon(&dir);
+
+    let mut a = connect(&sock);
+    hello(&mut a, &[1; 16]);
+
+    // B joins after A: B must be told A is already there.
+    let mut b = connect(&sock);
+    hello(&mut b, &[2; 16]);
+    expect_control(&mut a, 0, &[2; 16]);
+    expect_control(&mut b, 0, &[1; 16]);
+}
+
 #[test]
 fn duplicate_hello_replaces() {
     let dir = unique_dir("dup");
@@ -113,6 +128,9 @@ fn mismatched_from_is_dropped() {
     let mut b = connect(&sock);
     hello(&mut b, &[2; 16]);
 
+    // B learns about A on join (the newcomer announcement).
+    expect_control(&mut b, 0, &[1; 16]);
+
     // A (registered as [1;16]) claims to be [2;16]: nothing may reach B.
     data(&mut a, &[2; 16], &[2; 16], &[0xAB; 10]);
     expect_no_data(&mut b, 300);
@@ -142,6 +160,8 @@ fn unknown_dest_is_dropped() {
     hello(&mut b, &[2; 16]);
     // Wait until the daemon has registered B: A sees the join control.
     expect_control(&mut a, 0, &[2; 16]);
+    // B's side of the same join: the newcomer learns A is already here.
+    expect_control(&mut b, 0, &[1; 16]);
     data(&mut a, &[1; 16], &[2; 16], &[0xCD; 10]);
     let p = read_envelope(&mut b).unwrap().expect("connection closed");
     assert_eq!(p.len(), 42);
@@ -158,8 +178,10 @@ fn oversized_frame_closes_connection() {
     let mut b = connect(&sock);
     hello(&mut b, &[2; 16]);
 
-    // Handshake: once A has seen B join, both hellos are registered.
+    // Handshake: once A has seen B join, both hellos are registered, and
+    // B has learned A on its side too.
     expect_control(&mut a, 0, &[2; 16]);
+    expect_control(&mut b, 0, &[1; 16]);
 
     // A frame one byte over MAX_FRAME: that connection is closed...
     data(&mut a, &[1; 16], &[2; 16], &vec![0u8; MAX_FRAME + 1]);

tines-bus/tests/transport.rs

@@ -102,6 +102,16 @@ fn spoofed_frame_is_dropped() {
         Duration::from_secs(3),
     );
 
+    // B's side of the same join: the newcomer learns A is there.
+    match poll_until(
+        &mut b,
+        |ev| matches!(ev, Event::Contact(_)),
+        Duration::from_secs(3),
+    ) {
+        Event::Contact(id) => assert_eq!(id, [1; 16]),
+        other => panic!("expected Contact, got {other:?}"),
+    }
+
     // A (complicit) sends a frame that claims to be from C: the bus stamps
     // A as the envelope sender, so B must never surface it.
     let spoofed = frame_from(&[9; 16]);

tines/Cargo.toml

@@ -14,6 +14,7 @@ categories = ["science::robotics", "network-programming"]
 tines-core = { path = "../tines-core", version = "1" }
 tines-store = { path = "../tines-store", version = "0.4.0" }
 tines-bt = { path = "../tines-bt", version = "0.4.0" }
+tines-bus = { path = "../tines-bus", version = "0.4.0" }
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
 

tines/src/config.rs

@@ -0,0 +1,194 @@
+//! On-disk service configuration, shared by the `tines` binary and its
+//! clients (the fleet console generates these configs and drives the service
+//! it configures).
+//!
+//! The config is JSON with `deny_unknown_fields`; every key except the core
+//! `node_id` / `log_path` / `socket_path` is optional. See [`Config`] for the
+//! per-key contract, and [`Config::validate`] for the transport constraints.
+
+use crate::trust::DEFAULT_PEER_TIME_WINDOW_MS;
+
+/// On-disk configuration for a running service. Unknown keys are rejected.
+#[derive(Debug, serde::Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct Config {
+    /// This device's advertised `NodeId`, 32 lowercase hex chars.
+    pub node_id: String,
+    /// The peer's advertised `NodeId`, 32 lowercase hex chars. Required for
+    /// the bt transport (its single peer); for the bus transport it is the
+    /// allowlist of one — `None` means any bus peer.
+    pub peer: Option<String>,
+    /// Absolute path to the LogStore append-only log.
+    pub log_path: String,
+    /// Unix-domain socket path the IPC listens on.
+    pub socket_path: String,
+    /// Exchange poll interval in milliseconds. The real default belongs to the
+    /// builder; 20 ms keeps the link responsive without burning idle cycles.
+    #[serde(default = "default_poll_interval_ms")]
+    pub poll_interval_ms: u64,
+    /// Trust the system clock for the host's wall time, instead of deriving it
+    /// from a peer's observation. Set to `"system"` when the machine's clock is
+    /// reliable — an RTC, or a GPS/NTP-disciplined clock — the operator asserts
+    /// it rather than tines detecting it. Left unset, tines falls back to a
+    /// peer observation, and, with none available, stays quarantined. Any
+    /// other value is a config error.
+    #[serde(default)]
+    pub trusted_clock: Option<TrustedClock>,
+    /// How long, from the first plausible peer estimate, to keep collecting
+    /// estimates before adopting the latest (spec decision J).
+    #[serde(default = "default_peer_time_window_ms")]
+    pub peer_time_window_ms: u64,
+    /// The `unknown_downtime_age_ms` broadcast floor passed to LogStore while
+    /// the wall clock is unknown (the fold's unknown-downtime branch). Optional;
+    /// leave the LogStore default (`24h`) when unset.
+    ///
+    /// The floor must stay below the host's trim `max_age_ms`. That bound waits
+    /// for host hardening — the trim defaults are still empty — so this check is
+    /// not validated until then.
+    #[serde(default)]
+    pub unknown_downtime_age_ms: Option<u64>,
+    /// Which transport the service runs: `bt` (default) or `bus`.
+    #[serde(default)]
+    pub transport: Option<TransportKind>,
+    /// Absolute path to the bus daemon's Unix-domain socket. Required when
+    /// `transport` is `bus`.
+    #[serde(default)]
+    pub bus_socket: Option<String>,
+}
+
+impl Config {
+    /// Check the transport constraints. Called by `main` before anything
+    /// else touches the config.
+    ///
+    /// - `bt` (the default) requires `peer`.
+    /// - `bus` requires `bus_socket`; `peer` is the allowlist of one, and
+    ///   `None` means any peer.
+    pub fn validate(&self) -> Result<(), String> {
+        match self.transport.unwrap_or(TransportKind::Bt) {
+            TransportKind::Bt if self.peer.is_none() => Err("transport bt requires peer".into()),
+            TransportKind::Bus if self.bus_socket.is_none() => {
+                Err("transport bus requires bus_socket".into())
+            }
+            _ => Ok(()),
+        }
+    }
+}
+
+/// The accepted values of `trusted_clock`.
+#[derive(Debug, PartialEq, Eq, serde::Deserialize)]
+#[serde(rename_all = "lowercase")]
+pub enum TrustedClock {
+    /// Trust the system clock (`"system"`).
+    System,
+}
+
+/// The transport the service runs over.
+#[derive(Debug, PartialEq, Eq, Clone, Copy, serde::Deserialize)]
+#[serde(rename_all = "lowercase")]
+pub enum TransportKind {
+    /// The BLE transport (the default), with the single configured peer.
+    Bt,
+    /// The host bus transport, peers = every node on the bus daemon (or the
+    /// allowlisted one).
+    Bus,
+}
+
+fn default_poll_interval_ms() -> u64 {
+    20
+}
+
+fn default_peer_time_window_ms() -> u64 {
+    DEFAULT_PEER_TIME_WINDOW_MS
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{Config, TransportKind, TrustedClock};
+
+    const BASE: &str = r#""node_id": "00000000000000000000000000000001",
+        "peer": "00000000000000000000000000000002",
+        "log_path": "/var/lib/tines/log",
+        "socket_path": "/run/tines.sock""#;
+
+    fn parse(extra: &str) -> Result<Config, serde_json::Error> {
+        serde_json::from_str(&format!("{{{BASE}{extra}}}"))
+    }
+
+    #[test]
+    fn defaults() {
+        let c = parse("").expect("minimal config");
+        assert_eq!(c.poll_interval_ms, 20);
+        assert_eq!(c.peer_time_window_ms, 10_000);
+        assert_eq!(c.trusted_clock, None);
+        assert_eq!(c.unknown_downtime_age_ms, None);
+    }
+
+    #[test]
+    fn trusted_clock_system_is_accepted() {
+        let c = parse(r#", "trusted_clock": "system""#).expect("system");
+        assert_eq!(c.trusted_clock, Some(TrustedClock::System));
+    }
+
+    #[test]
+    fn trusted_clock_unknown_value_is_rejected() {
+        assert!(parse(r#", "trusted_clock": "System""#).is_err());
+        assert!(parse(r#", "trusted_clock": "rtc""#).is_err());
+        assert!(parse(r#", "trusted_clock": "systme""#).is_err());
+    }
+
+    #[test]
+    fn unknown_field_is_rejected() {
+        assert!(parse(r#", "trusted_clok": "system""#).is_err());
+    }
+
+    #[test]
+    fn peer_time_window_is_configurable() {
+        let c = parse(r#", "peer_time_window_ms": 2500"#).expect("window");
+        assert_eq!(c.peer_time_window_ms, 2_500);
+    }
+
+    const BUS_BASE: &str = r#""node_id": "00000000000000000000000000000001",
+        "log_path": "/var/lib/tines/log",
+        "socket_path": "/run/tines.sock""#;
+
+    fn parse_bus(extra: &str) -> Result<Config, serde_json::Error> {
+        serde_json::from_str(&format!("{{{BUS_BASE}{extra}}}"))
+    }
+
+    #[test]
+    fn bus_config_parses() {
+        let c = parse_bus(r#", "transport": "bus", "bus_socket": "/run/tines-bus.sock""#)
+            .expect("bus config");
+        assert_eq!(c.transport, Some(TransportKind::Bus));
+        assert_eq!(c.bus_socket.as_deref(), Some("/run/tines-bus.sock"));
+        assert_eq!(c.peer, None);
+        assert_eq!(c.validate(), Ok(()));
+    }
+
+    #[test]
+    fn bt_without_peer_is_invalid() {
+        let c = parse_bus("").expect("bt config without peer parses");
+        assert_eq!(c.validate(), Err("transport bt requires peer".to_string()));
+    }
+
+    #[test]
+    fn bus_without_socket_is_invalid() {
+        let c = parse_bus(r#", "transport": "bus""#).expect("parses");
+        assert_eq!(
+            c.validate(),
+            Err("transport bus requires bus_socket".to_string())
+        );
+    }
+
+    #[test]
+    fn unknown_transport_is_rejected() {
+        assert!(parse_bus(r#", "transport": "wifi""#).is_err());
+    }
+
+    #[test]
+    fn transport_defaults_to_bt() {
+        let c = parse("").expect("minimal config");
+        assert_eq!(c.transport, None);
+        assert_eq!(c.validate(), Ok(()));
+    }
+}

tines/src/lib.rs

@@ -3,19 +3,24 @@
 //! 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.
+//! `Node<LogStore, HostTransport>` for real BLE or the host bus, while tests
+//! drive the same [`serve`] / [`handle`] path with a mock transport so no
+//! radio is required.
 //!
-//! The IPC protocol is documented in [`ipc`].
+//! The IPC protocol is documented in [`ipc`]; the on-disk configuration in
+//! [`config`].
 #![deny(missing_docs)]
 
 pub mod clock;
+pub mod config;
 pub mod ipc;
+pub mod transport;
 pub mod trust;
 
 pub use clock::HostClock;
+pub use config::{Config, TransportKind, TrustedClock};
 pub use ipc::{
     handle, serve, BeliefOut, ContactOut, DropOut, EventOut, ObservationOut, Request, Response,
     LIST_CAP,
 };
+pub use transport::HostTransport;

tines/src/main.rs

@@ -1,7 +1,8 @@
 //! 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.
+//! Loads a JSON config, builds a `Node<LogStore, HostTransport>` (BLE over
+//! `bluer`, or the host bus — see [`tines::config::TransportKind`]), and
+//! serves the IPC ([`tines::serve`]) over a Unix-domain socket.
 //! One instance per device — the typical deployment on robots without ROS2.
 //!
 //! The node and peer ids are 32 lowercase hex chars (the 16-byte `NodeId`).
@@ -11,14 +12,16 @@
 //! # Config
 //!
 //! A JSON object; an unknown key is an error, so a typo can't silently fall
-//! back to a default.
+//! back to a default. The full contract lives in [`tines::config::Config`].
 //!
 //! | key | required | meaning |
 //! |---|---|---|
 //! | `node_id` | yes | this device's `NodeId`, 32 hex chars |
-//! | `peer` | yes | the peer's `NodeId`, 32 hex chars |
+//! | `peer` | bt only | the peer's `NodeId`, 32 hex chars; on bus, the allowlist of one (`None` = any bus peer) |
 //! | `log_path` | yes | the LogStore log file |
 //! | `socket_path` | yes | the IPC Unix socket |
+//! | `transport` | no (`bt`) | `"bt"` or `"bus"` |
+//! | `bus_socket` | bus only | the bus daemon's Unix socket |
 //! | `poll_interval_ms` | no (20) | exchange poll cadence |
 //! | `trusted_clock` | no (unset) | `"system"` to trust the system clock; the only accepted value |
 //! | `peer_time_window_ms` | no (10000) | how long to collect peer time estimates before adopting the latest |
@@ -50,81 +53,32 @@
 use std::error::Error;
 use std::path::Path;
 
-use tines::trust::{adopt_wall_clock, system_wall_ms, DEFAULT_PEER_TIME_WINDOW_MS};
+use tines::config::{Config, TransportKind, TrustedClock};
+use tines::transport::HostTransport;
+use tines::trust::{adopt_wall_clock, system_wall_ms};
 use tines::{serve, HostClock};
 use tines_bt::{BluerRadio, BtTransport};
+use tines_bus::BusTransport;
 use tines_core::{Node, NodeId};
 use tines_store::LogStore;
 
-/// On-disk configuration for a running service. Unknown keys are rejected.
-#[derive(Debug, serde::Deserialize)]
-#[serde(deny_unknown_fields)]
-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,
-    /// Absolute path to the LogStore append-only log.
-    log_path: String,
-    /// Unix-domain socket path the IPC listens on.
-    socket_path: String,
-    /// Exchange poll interval in milliseconds. The real default belongs to the
-    /// builder; 20 ms keeps the link responsive without burning idle cycles.
-    #[serde(default = "default_poll_interval_ms")]
-    poll_interval_ms: u64,
-    /// Trust the system clock for the host's wall time, instead of deriving it
-    /// from a peer's observation. Set to `"system"` when the machine's clock is
-    /// reliable — an RTC, or a GPS/NTP-disciplined clock — the operator asserts
-    /// it rather than tines detecting it. Left unset, tines falls back to a
-    /// peer observation, and, with none available, stays quarantined. Any
-    /// other value is a config error.
-    #[serde(default)]
-    trusted_clock: Option<TrustedClock>,
-    /// How long, from the first plausible peer estimate, to keep collecting
-    /// estimates before adopting the latest (spec decision J).
-    #[serde(default = "default_peer_time_window_ms")]
-    peer_time_window_ms: u64,
-    /// The `unknown_downtime_age_ms` broadcast floor passed to LogStore while
-    /// the wall clock is unknown (the fold's unknown-downtime branch). Optional;
-    /// leave the LogStore default (`24h`) when unset.
-    ///
-    /// The floor must stay below the host's trim `max_age_ms`. That bound waits
-    /// for host hardening — the trim defaults are still empty — so this check is
-    /// not validated until then.
-    #[serde(default)]
-    unknown_downtime_age_ms: Option<u64>,
-}
-
-/// The accepted values of `trusted_clock`.
-#[derive(Debug, PartialEq, Eq, serde::Deserialize)]
-#[serde(rename_all = "lowercase")]
-enum TrustedClock {
-    /// Trust the system clock (`"system"`).
-    System,
-}
-
-fn default_poll_interval_ms() -> u64 {
-    20
-}
-
-fn default_peer_time_window_ms() -> u64 {
-    DEFAULT_PEER_TIME_WINDOW_MS
-}
-
 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)?)
         .map_err(|e| format!("invalid config: {e}"))?;
+    config
+        .validate()
+        .map_err(|e| format!("invalid config: {e}"))?;
 
     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).
-    // unknown_downtime_age_ms (the broadcast floor while the wall clock is
-    // unknown) is optional; leave the LogStore default when unset.
-    // Peer estimates outside the plausible range don't count (decision J).
+    // LogStore is the Store (S); HostTransport wraps BLE or the bus as the
+    // Transport (T). unknown_downtime_age_ms (the broadcast floor while the
+    // wall clock is unknown) is optional; leave the LogStore default when
+    // unset. Peer estimates outside the plausible range don't count
+    // (decision J).
     let builder = LogStore::builder(Path::new(&config.log_path))
         .plausible_wall_ms(tines::trust::PLAUSIBLE_WALL_MS);
     let builder = match config.unknown_downtime_age_ms {
@@ -136,8 +90,26 @@ fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
         .open(clock.now())
         .map_err(|e| format!("failed to open log store: {e:?}"))?;
 
+    // bt: the single configured peer (validation guarantees it). bus: the
+    // configured peer is the allowlist of one; None means any bus peer.
+    let transport = match config.transport.unwrap_or(TransportKind::Bt) {
+        TransportKind::Bt => {
+            let peer = parse_node_id(config.peer.as_deref().expect("validated"))?;
             let radio = BluerRadio::new(me, peer);
-    let transport = BtTransport::new(me, peer, radio);
+            HostTransport::Bt(BtTransport::new(me, peer, radio))
+        }
+        TransportKind::Bus => {
+            let allowlist = match &config.peer {
+                Some(p) => Some(parse_node_id(p)?),
+                None => None,
+            };
+            HostTransport::Bus(BusTransport::new(
+                me,
+                allowlist,
+                config.bus_socket.as_deref().expect("validated"),
+            )?)
+        }
+    };
     // `serve` runs the exchange inline (a poll loop fed by the socket read
     // timeouts) on this same thread, so observations and beliefs flow over the
     // transport, not just over IPC. The node isn't `Send` (the BLE transport
@@ -212,50 +184,3 @@ fn node_hex(id: NodeId) -> String {
     }
     s
 }
-
-#[cfg(test)]
-mod tests {
-    use super::{Config, TrustedClock};
-
-    const BASE: &str = r#""node_id": "00000000000000000000000000000001",
-        "peer": "00000000000000000000000000000002",
-        "log_path": "/var/lib/tines/log",
-        "socket_path": "/run/tines.sock""#;
-
-    fn parse(extra: &str) -> Result<Config, serde_json::Error> {
-        serde_json::from_str(&format!("{{{BASE}{extra}}}"))
-    }
-
-    #[test]
-    fn defaults() {
-        let c = parse("").expect("minimal config");
-        assert_eq!(c.poll_interval_ms, 20);
-        assert_eq!(c.peer_time_window_ms, 10_000);
-        assert_eq!(c.trusted_clock, None);
-        assert_eq!(c.unknown_downtime_age_ms, None);
-    }
-
-    #[test]
-    fn trusted_clock_system_is_accepted() {
-        let c = parse(r#", "trusted_clock": "system""#).expect("system");
-        assert_eq!(c.trusted_clock, Some(TrustedClock::System));
-    }
-
-    #[test]
-    fn trusted_clock_unknown_value_is_rejected() {
-        assert!(parse(r#", "trusted_clock": "System""#).is_err());
-        assert!(parse(r#", "trusted_clock": "rtc""#).is_err());
-        assert!(parse(r#", "trusted_clock": "systme""#).is_err());
-    }
-
-    #[test]
-    fn unknown_field_is_rejected() {
-        assert!(parse(r#", "trusted_clok": "system""#).is_err());
-    }
-
-    #[test]
-    fn peer_time_window_is_configurable() {
-        let c = parse(r#", "peer_time_window_ms": 2500"#).expect("window");
-        assert_eq!(c.peer_time_window_ms, 2_500);
-    }
-}

tines/src/transport.rs

@@ -0,0 +1,39 @@
+//! The host's chosen transport: either the BLE radio ([`BtTransport`]) or the
+//! host bus ([`BusTransport`]), per the service config.
+
+use tines_bt::BtTransport;
+use tines_bus::BusTransport;
+use tines_core::{Event, NodeId, SendError, Transport};
+
+/// The transport the service runs, selected by [`crate::config::Config`].
+pub enum HostTransport {
+    /// The BLE transport with the single configured peer.
+    Bt(BtTransport),
+    /// The host bus transport with the (optionally allowlisted) bus peers.
+    Bus(BusTransport),
+}
+
+impl Transport for HostTransport {
+    type Addr = NodeId;
+
+    fn poll(&mut self) -> Option<Event<NodeId>> {
+        match self {
+            HostTransport::Bt(t) => t.poll(),
+            HostTransport::Bus(t) => t.poll(),
+        }
+    }
+
+    fn send(&mut self, to: &NodeId, frame: &[u8]) -> Result<(), SendError> {
+        match self {
+            HostTransport::Bt(t) => t.send(to, frame),
+            HostTransport::Bus(t) => t.send(to, frame),
+        }
+    }
+
+    fn mtu(&self) -> usize {
+        match self {
+            HostTransport::Bt(t) => t.mtu(),
+            HostTransport::Bus(t) => t.mtu(),
+        }
+    }
+}

tines/tests/service_bus.rs

@@ -0,0 +1,215 @@
+//! End-to-end bus fleet test: two real `tines` service processes converge
+//! over a live `tines-busd` — the sim-fleet deliverable without docker.
+//!
+//! Node A observes a broadcast; node B learns it over the bus (belief, store
+//! list) and both see the contact change.
+
+use std::io::{Read, Write};
+use std::os::unix::net::UnixStream;
+use std::path::{Path, PathBuf};
+use std::process::{Child, Command, Stdio};
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::time::{Duration, Instant};
+
+use tines::{Request, Response};
+
+static COUNTER: AtomicUsize = AtomicUsize::new(0);
+
+fn unique_dir(tag: &str) -> PathBuf {
+    let n = COUNTER.fetch_add(1, Ordering::Relaxed);
+    let dir = std::env::temp_dir().join(format!(
+        "tines-service-bus-{tag}-{}-{n}",
+        std::process::id()
+    ));
+    std::fs::create_dir_all(&dir).unwrap();
+    dir
+}
+
+fn node_id_hex(n: u8) -> String {
+    format!("{n:02x}").repeat(16)
+}
+
+fn spawn_tines(config_path: &Path, dir: &Path) -> Child {
+    let stderr = std::fs::File::create(dir.join("stderr.log")).unwrap();
+    Command::new(env!("CARGO_BIN_EXE_tines"))
+        .arg(config_path)
+        .stderr(Stdio::from(stderr))
+        .stdout(Stdio::null())
+        .spawn()
+        .expect("spawn tines service")
+}
+
+fn connect(path: &Path, deadline: Instant) -> UnixStream {
+    loop {
+        if let Ok(s) = UnixStream::connect(path) {
+            return s;
+        }
+        assert!(
+            Instant::now() < deadline,
+            "failed to connect to {}",
+            path.display()
+        );
+        std::thread::sleep(Duration::from_millis(20));
+    }
+}
+
+fn rpc(stream: &mut UnixStream, 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(stream: &mut UnixStream) -> 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")
+}
+
+/// Poll `poll_once` until `pred` accepts a response, or fail after `deadline`.
+fn poll_until(
+    mut poll_once: impl FnMut() -> Response,
+    pred: impl Fn(&Response) -> bool,
+    deadline: Instant,
+) -> Response {
+    loop {
+        let r = poll_once();
+        if pred(&r) {
+            return r;
+        }
+        assert!(Instant::now() < deadline, "timed out polling the service");
+        std::thread::sleep(Duration::from_millis(50));
+    }
+}
+
+#[test]
+fn two_services_converge_over_bus() {
+    let dir = unique_dir("converge");
+    let bus_socket = dir.join("bus.sock");
+
+    // The bus daemon runs in-process on a thread (a thread in the test
+    // process, not an external binary, so the test has no CARGO_BIN_EXE
+    // dependency); poll until it has bound its socket.
+    let daemon_socket = bus_socket.clone();
+    std::thread::spawn(move || {
+        let _ = tines_bus::daemon::serve(&daemon_socket);
+    });
+    {
+        let mut bound = UnixStream::connect(&bus_socket);
+        let start = Instant::now();
+        while bound.is_err() && start.elapsed() < Duration::from_secs(5) {
+            std::thread::sleep(Duration::from_millis(20));
+            bound = UnixStream::connect(&bus_socket);
+        }
+        assert!(bound.is_ok(), "bus daemon did not bind its socket");
+    }
+
+    // Write both configs.
+    let mk_config = |node: &str, log: &str, sock: &str| -> String {
+        serde_json::json!({
+            "node_id": node,
+            "log_path": dir.join(log).to_str().unwrap(),
+            "socket_path": dir.join(sock).to_str().unwrap(),
+            "trusted_clock": "system",
+            "transport": "bus",
+            "bus_socket": bus_socket.to_str().unwrap(),
+        })
+        .to_string()
+    };
+    let cfg_a = dir.join("a.json");
+    let cfg_b = dir.join("b.json");
+    std::fs::write(&cfg_a, mk_config(&node_id_hex(1), "a.log", "a.sock")).unwrap();
+    std::fs::write(&cfg_b, mk_config(&node_id_hex(2), "b.log", "b.sock")).unwrap();
+
+    // The generated configs must parse as the lib-level Config type.
+    let _: tines::config::Config =
+        serde_json::from_str(&std::fs::read_to_string(&cfg_a).unwrap()).unwrap();
+
+    let mut a_proc = spawn_tines(&cfg_a, &dir);
+    let mut b_proc = spawn_tines(&cfg_b, &dir);
+
+    let deadline = Instant::now() + Duration::from_secs(10);
+    let mut a = connect(&dir.join("a.sock"), deadline);
+    let mut b = connect(&dir.join("b.sock"), deadline);
+
+    let topic = hex("status");
+    let payload = hex("ping");
+
+    // A observes a broadcast.
+    rpc(
+        &mut a,
+        &Request::Observe {
+            topic: topic.clone(),
+            payload: payload.clone(),
+            to: None,
+            lifetime_ms: None,
+        },
+    );
+    match read_response(&mut a) {
+        Response::Observe { .. } => {}
+        other => panic!("expected Observe response, got {other:?}"),
+    }
+
+    // B's belief converges over the bus.
+    poll_until(
+        || {
+            rpc(
+                &mut b,
+                &Request::Belief {
+                    topic: topic.clone(),
+                },
+            );
+            read_response(&mut b)
+        },
+        |r| matches!(r, Response::Belief { obs: Some(o) } if o.payload == payload),
+        deadline,
+    );
+
+    // B's store listing contains it too.
+    rpc(&mut b, &Request::List);
+    match read_response(&mut b) {
+        Response::List { observations, .. } => {
+            assert!(
+                observations
+                    .iter()
+                    .any(|o| o.payload == payload && o.topic == topic),
+                "observation missing from B's store list: {observations:?}"
+            );
+        }
+        other => panic!("expected List response, got {other:?}"),
+    }
+
+    // Both nodes saw the contact change.
+    for (stream, expected_peer) in [(&mut a, node_id_hex(2)), (&mut b, node_id_hex(1))] {
+        poll_until(
+            || {
+                rpc(stream, &Request::Events);
+                read_response(stream)
+            },
+            |r| {
+                matches!(r, Response::Events { contacts, .. }
+                    if contacts.iter().any(|c| c.joined && c.peer == expected_peer))
+            },
+            deadline,
+        );
+    }
+
+    drop(a);
+    drop(b);
+    let _ = a_proc.kill();
+    let _ = b_proc.kill();
+    let _ = a_proc.wait();
+    let _ = b_proc.wait();
+}
+
+fn hex(bytes: &str) -> String {
+    bytes
+        .as_bytes()
+        .iter()
+        .map(|b| format!("{b:02x}"))
+        .collect()
+}