Bus input: serve envelopes from yukart/vehicle/+ and yukart/host/+

Newest per signal, stale after 5 s or when only the retained copy is known.
Bus or mock, never both; hello says which (source bus|mock|none).
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01DDwfd6Ki8UfG4hK6W5YpJz

src/bus.rs

@@ -0,0 +1,88 @@
+//! The real input: envelopes from the bus, latest per signal.
+//!
+//! Subscribes to the envelope topics and keeps the newest envelope of each signal,
+//! keyed by the topic's last segment (the signal name, contracts guideline 1).
+//! Staleness is judged here, by arrival: a signal not refreshed within `STALE`,
+//! or known only from the broker's retained copy (it may predate a reboot), is
+//! sent with `quality: "stale"`. Signals are never dropped once seen; an old value
+//! shown stale is the honest picture.
+
+use std::collections::HashMap;
+use std::sync::{Arc, Mutex};
+use std::thread;
+use std::time::{Duration, Instant};
+
+use rumqttc::{Client, Event, MqttOptions, Packet, QoS};
+use serde_json::{json, Map, Value};
+
+const TOPICS: [&str; 2] = ["yukart/vehicle/+", "yukart/host/+"];
+const STALE: Duration = Duration::from_secs(5);
+
+struct Entry {
+    envelope: Value,
+    at: Instant,
+    retained: bool,
+}
+
+#[derive(Clone)]
+pub struct Bus {
+    latest: Arc<Mutex<HashMap<String, Entry>>>,
+}
+
+impl Bus {
+    /// Connects in the background and reconnects forever; never blocks the caller.
+    pub fn start(host: &str, port: u16) -> Self {
+        let bus = Bus { latest: Arc::default() };
+        let mut opts = MqttOptions::new("svc-gw", host, port);
+        opts.set_keep_alive(Duration::from_secs(5));
+        let (client, mut connection) = Client::new(opts, 64);
+        let latest = bus.latest.clone();
+        thread::spawn(move || {
+            let mut up = false;
+            for ev in connection.iter() {
+                match ev {
+                    Ok(Event::Incoming(Packet::ConnAck(_))) => {
+                        up = true;
+                        eprintln!("bus: connected; subscribing to {}", TOPICS.join(", "));
+                        // A clean session forgets subscriptions: renew them on every connect.
+                        for t in TOPICS {
+                            let _ = client.try_subscribe(t, QoS::AtMostOnce);
+                        }
+                    }
+                    Ok(Event::Incoming(Packet::Publish(p))) => {
+                        let Some(name) = p.topic.rsplit('/').next() else { continue };
+                        let Ok(envelope) = serde_json::from_slice::<Value>(&p.payload) else { continue };
+                        if !envelope.is_object() {
+                            continue;
+                        }
+                        let entry = Entry { envelope, at: Instant::now(), retained: p.retain };
+                        latest.lock().unwrap().insert(name.to_string(), entry);
+                    }
+                    Ok(_) => {}
+                    Err(e) => {
+                        if std::mem::take(&mut up) {
+                            eprintln!("bus: {e}; retrying");
+                        }
+                        thread::sleep(Duration::from_secs(1));
+                    }
+                }
+            }
+        });
+        bus
+    }
+
+    /// Every signal seen, newest envelope each, marked stale where it is.
+    pub fn snapshot(&self) -> Map<String, Value> {
+        let latest = self.latest.lock().unwrap();
+        latest
+            .iter()
+            .map(|(name, e)| {
+                let mut env = e.envelope.clone();
+                if (e.retained || e.at.elapsed() > STALE) && env.get("quality").is_none() {
+                    env["quality"] = json!("stale");
+                }
+                (name.clone(), env)
+            })
+            .collect()
+    }
+}

src/main.rs

@@ -2,8 +2,10 @@
 //! hello, heartbeats, state snapshots down and acknowledged commands up.
 //! Defined in contracts `docs/BOUNDARY.md`.
 //!
-//! v0 has no bus input yet: without `--mock` it sends hello and heartbeats only.
+//! State comes from the bus (`bus.rs`), or from the mock with `--mock`: one input
+//! or the other, never both, so mock data cannot mix into real data.
 
+mod bus;
 mod client;
 mod hub;
 mod mock;
@@ -20,7 +22,8 @@ use serde_json::json;
 const USAGE: &str = "\
 usage: svc-gw [options]
   --listen ADDR       WebSocket listen address              (0.0.0.0:7700)
-  --mock              synthesise vehicle state, flagged as mock in every message
+  --mock              synthesise vehicle state, flagged as mock in every message (no bus)
+  --broker HOST:PORT  MQTT broker; '' for none                (127.0.0.1:1883)
   --recorder-ctl PATH recorder control socket, for 'mark'   (/run/yukart-recorder/ctl.sock)";
 
 pub fn now_us() -> i64 {
@@ -50,12 +53,14 @@ fn main() -> ExitCode {
     let mut listen = "0.0.0.0:7700".to_string();
     let mut mock = false;
     let mut recorder_ctl = "/run/yukart-recorder/ctl.sock".to_string();
+    let mut broker = "127.0.0.1:1883".to_string();
     let mut args = std::env::args().skip(1);
     while let Some(a) = args.next() {
         match a.as_str() {
             "--mock" => mock = true,
             "--listen" => listen = args.next().unwrap_or_default(),
             "--recorder-ctl" => recorder_ctl = args.next().unwrap_or_default(),
+            "--broker" => broker = args.next().unwrap_or_default(),
             _ => {
                 eprintln!("svc-gw: unknown option {a}\n{USAGE}");
                 return ExitCode::from(2);
@@ -64,6 +69,18 @@ fn main() -> ExitCode {
     }
 
     let registry = signals::Registry::load();
+    let bus = if mock || broker.is_empty() {
+        None
+    } else {
+        match broker.rsplit_once(':').and_then(|(h, p)| Some((h, p.parse::<u16>().ok()?))) {
+            Some((host, port)) => Some(bus::Bus::start(host, port)),
+            None => {
+                eprintln!("svc-gw: --broker must be HOST:PORT\n{USAGE}");
+                return ExitCode::from(2);
+            }
+        }
+    };
+    let source = if mock { "mock" } else if bus.is_some() { "bus" } else { "none" };
     let hub = hub::Hub::default();
     let ctx = Arc::new(client::Ctx {
         hub: hub.clone(),
@@ -72,7 +89,7 @@ fn main() -> ExitCode {
             "proto": 0,
             "server": concat!("svc-gw ", env!("CARGO_PKG_VERSION")),
             "host": std::fs::read_to_string("/proc/sys/kernel/hostname").unwrap_or_default().trim(),
-            "source": if mock { "mock" } else { "none" },
+            "source": source,
             "mock": mock,
             "hb_ms": client::HEARTBEAT.as_millis() as u64,
             "timeout_ms": client::TIMEOUT.as_millis() as u64,
@@ -88,7 +105,7 @@ fn main() -> ExitCode {
             return ExitCode::from(1);
         }
     };
-    eprintln!("svc-gw {} on ws://{listen}{}", env!("CARGO_PKG_VERSION"), if mock { " — MOCK DATA" } else { "" });
+    eprintln!("svc-gw {} on ws://{listen}, source {source}{}", env!("CARGO_PKG_VERSION"), if mock { " — MOCK DATA" } else { "" });
 
     // Steady state is reached with zero clients: the producer runs regardless.
     thread::spawn(move || {
@@ -100,9 +117,14 @@ fn main() -> ExitCode {
             thread::sleep(Duration::from_millis(200));
             let dt = last.elapsed().as_secs_f64();
             last = Instant::now();
-            if mock {
+            let signals = match &bus {
+                Some(b) => Some(b.snapshot()).filter(|s| !s.is_empty()),
+                None if mock => Some(m.step(&registry, dt)),
+                None => None,
+            };
+            if let Some(signals) = signals {
                 seq += 1;
-                let state = json!({ "t": "state", "seq": seq, "mock": true, "signals": m.step(&registry, dt) });
+                let state = json!({ "t": "state", "seq": seq, "mock": mock, "signals": signals });
                 hub.broadcast(&state.to_string());
             }
             if last_hb.elapsed() >= client::HEARTBEAT {