feat(gui): tines-gui skeleton (ipc client, config gen, theme)

scripts/ci.sh

@@ -24,6 +24,11 @@ sh scripts/check-no-std.sh
 # compiling on the pinned stable toolchain, or the scheduled job rots unseen.
 cargo fmt --manifest-path fuzz/Cargo.toml -- --check
 cargo clippy --locked --manifest-path fuzz/Cargo.toml --all-targets -- -D warnings
+# The tines-gui crate sits outside the workspace like fuzz/ (standalone
+# manifest + lockfile, GUI release cadence), but it must keep resolving and
+# compiling on the pinned stable toolchain.
+cargo fmt --manifest-path tines-gui/Cargo.toml -- --check
+cargo clippy --locked --manifest-path tines-gui/Cargo.toml --all-targets -- -D warnings
 # Post-1.0.0 release gate: every 1.x change to tines-core must stay
 # semver-compatible with the published 1.0.0. Skipped where the tool is not
 # installed (`cargo install cargo-semver-checks --locked`); the GitHub Actions

tines-gui/Cargo.toml

@@ -0,0 +1,23 @@
+[package]
+name = "tines-gui"
+version = "0.1.0"
+edition = "2024"
+license = "Apache-2.0"
+description = "Fleet console for tines: launcher and per-node windows driving the service IPC"
+publish = false
+
+[dependencies]
+tines = { path = "../tines", version = "0.4.0" }
+eframe = { version = "0.36", default-features = false, features = ["glow", "default_fonts", "wayland"] }
+egui = "0.36"
+serde = { version = "1", features = ["derive"] }
+serde_json = "1"
+notify = "8"
+
+[dev-dependencies]
+egui_kittest = { version = "0.36", features = ["wgpu"] }
+tempfile = "3"
+
+# Like fuzz/, this crate sits outside the workspace (the GUI is a separate
+# deliverable with its own release cadence).
+[workspace]

tines-gui/src/config.rs

@@ -0,0 +1,133 @@
+//! Fleet/node layout on disk and the service configs the launcher writes.
+//!
+//! The launcher owns per-fleet state under
+//! `$XDG_STATE_HOME|~/.local/state/tines-console/`, and each node's
+//! `config.json` is exactly the service's [`tines::config::Config`] schema —
+//! the service rejects unknown keys, so the generated config must match it
+//! byte-for-byte (pinned by `tests/config.rs`).
+
+use std::fs::{self, Permissions};
+use std::io::{self, Read};
+use std::os::unix::fs::PermissionsExt;
+use std::path::PathBuf;
+
+/// The per-fleet state dir, created `0700` on first use:
+/// `$XDG_STATE_HOME|~/.local/state` + `tines-console/fleets/<fleet>`.
+pub fn fleet_dir(fleet: &str) -> PathBuf {
+    let base = state_home();
+    let dir = base.join("tines-console").join("fleets").join(fleet);
+    if let Err(e) = fs::create_dir_all(&dir) {
+        eprintln!("tines-gui: cannot create fleet dir {}: {e}", dir.display());
+    }
+    // create_dir_all honors the umask; the fleets carry node sockets and
+    // logs, so they are private by design.
+    if let Err(e) = fs::set_permissions(&dir, Permissions::from_mode(0o700)) {
+        eprintln!("tines-gui: cannot chmod fleet dir {}: {e}", dir.display());
+    }
+    dir
+}
+
+/// A node's dir inside a fleet: `<fleet_dir>/<node>`. Not created here —
+/// [`NodeConfig::write`] creates it when a config lands.
+pub fn node_dir(fleet: &str, node: &str) -> PathBuf {
+    fleet_dir(fleet).join(node)
+}
+
+/// A fresh random node id: 32 lowercase hex chars from 16 bytes of
+/// `/dev/urandom`.
+pub fn generate_node_id() -> String {
+    let mut buf = [0u8; 16];
+    let mut f =
+        std::fs::File::open("/dev/urandom").expect("/dev/urandom is available on this platform");
+    f.read_exact(&mut buf)
+        .expect("reading 16 bytes from /dev/urandom");
+    buf.iter().map(|b| format!("{b:02x}")).collect()
+}
+
+/// A node's service config, serializing to exactly the service schema.
+#[derive(Debug, Clone, serde::Serialize)]
+pub struct NodeConfig {
+    /// This node's id (32 lowercase hex chars).
+    pub node_id: String,
+    /// bt: the peer; bus: the allowlist of one, `None` for any peer.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub peer: Option<String>,
+    /// Absolute path to the LogStore log inside the container.
+    pub log_path: String,
+    /// Absolute path to the IPC socket inside the container.
+    pub socket_path: String,
+    /// `"bt"` or `"bus"`.
+    pub transport: String,
+    /// bus only: the bus daemon's socket inside the container.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub bus_socket: Option<String>,
+    /// `"system"` when the host clock is trustworthy (the container's is).
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub trusted_clock: Option<String>,
+}
+
+impl NodeConfig {
+    /// A bt-mode config: one peer over the BLE transport.
+    pub fn bt(
+        node_id: String,
+        peer: Option<String>,
+        log_path: String,
+        socket_path: String,
+    ) -> Self {
+        NodeConfig {
+            node_id,
+            peer,
+            log_path,
+            socket_path,
+            transport: "bt".into(),
+            bus_socket: None,
+            trusted_clock: Some("system".into()),
+        }
+    }
+
+    /// A bus-mode config: peers are every node on the bus daemon (or the
+    /// allowlisted one).
+    pub fn bus(
+        node_id: String,
+        peer: Option<String>,
+        log_path: String,
+        socket_path: String,
+        bus_socket: String,
+    ) -> Self {
+        NodeConfig {
+            node_id,
+            peer,
+            log_path,
+            socket_path,
+            transport: "bus".into(),
+            bus_socket: Some(bus_socket),
+            trusted_clock: Some("system".into()),
+        }
+    }
+
+    /// Write this config to `<node_dir(fleet, node)>/config.json` (creating
+    /// the node dir, `0700`) and return the path.
+    pub fn write(&self, fleet: &str, node: &str) -> io::Result<PathBuf> {
+        let dir = node_dir(fleet, node);
+        fs::create_dir_all(&dir)?;
+        fs::set_permissions(&dir, Permissions::from_mode(0o700))?;
+        let path = dir.join("config.json");
+        let json = serde_json::to_string_pretty(self).map_err(io::Error::other)?;
+        fs::write(&path, json)?;
+        Ok(path)
+    }
+}
+
+/// `$XDG_STATE_HOME` when set to an absolute path, else `~/.local/state`.
+fn state_home() -> PathBuf {
+    if let Some(dir) = std::env::var_os("XDG_STATE_HOME") {
+        let p = PathBuf::from(dir);
+        if p.is_absolute() {
+            return p;
+        }
+    }
+    std::env::var_os("HOME")
+        .map(PathBuf::from)
+        .unwrap_or_else(std::env::temp_dir)
+        .join(".local/state")
+}

tines-gui/src/ipc_client.rs

@@ -0,0 +1,198 @@
+//! An async, queue-based client for the `tines` service IPC.
+//!
+//! The wire protocol is the service's own: `[u32 LE len][JSON]`. The client
+//! tags frames with a monotonically increasing message id **on the
+//! client-side queue only** — `request` enqueues `[id, JSON body]` and the
+//! worker thread writes `[len][JSON]` to the socket, so the service-side
+//! protocol is untouched. The service answers one request per connection in
+//! order, so the worker pairs responses with the outstanding ids FIFO; the
+//! GUI polls [`IpcClient::try_recv`] for `(id, response)` pairs.
+//!
+//! The worker owns the socket and re-dials forever: a dead service never
+//! blocks `request` (requests queue), and [`IpcClient::reconnect`] forces a
+//! re-dial without dropping queued requests. If the connection dies between
+//! a write and its response, the request is replayed after the re-dial —
+//! observe is the dominant verb here, and re-observing beats losing data.
+
+use std::collections::VecDeque;
+use std::fmt;
+use std::io::{self, Read, Write};
+use std::os::unix::net::UnixStream;
+use std::path::PathBuf;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
+use std::sync::mpsc::{self, Receiver, Sender};
+use std::thread;
+use std::time::Duration;
+
+use tines::{Request, Response};
+
+/// Errors the client can report.
+#[derive(Debug)]
+pub enum ClientError {
+    /// A socket-level failure (including a dead worker thread).
+    Io(io::Error),
+    /// A request failed to serialize.
+    Json(serde_json::Error),
+}
+
+impl fmt::Display for ClientError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            ClientError::Io(e) => write!(f, "ipc client io: {e}"),
+            ClientError::Json(e) => write!(f, "ipc client json: {e}"),
+        }
+    }
+}
+
+impl std::error::Error for ClientError {}
+
+/// Shared state between the handle and the worker thread.
+struct Shared {
+    /// Set by [`IpcClient::reconnect`]; the worker redials and clears it.
+    force_reconnect: AtomicBool,
+}
+
+/// A queue-based client for one service socket.
+pub struct IpcClient {
+    out: Sender<(u64, Vec<u8>)>,
+    rx: Receiver<(u64, Response)>,
+    next_id: Arc<AtomicU64>,
+    shared: Arc<Shared>,
+    _socket: PathBuf,
+}
+
+impl IpcClient {
+    /// Spawn the worker thread and return the client. The socket is dialed
+    /// on the worker thread, so this succeeds even while the service is
+    /// down.
+    pub fn spawn(socket: PathBuf) -> io::Result<IpcClient> {
+        let (out_tx, out_rx) = mpsc::channel();
+        let (resp_tx, resp_rx) = mpsc::channel();
+        let next_id = Arc::new(AtomicU64::new(0));
+        let shared = Arc::new(Shared {
+            force_reconnect: AtomicBool::new(false),
+        });
+        let worker_shared = Arc::clone(&shared);
+        let worker_socket = socket.clone();
+        thread::Builder::new()
+            .name("tines-gui-ipc".into())
+            .spawn(move || worker(worker_socket, out_rx, resp_tx, worker_shared))?;
+        Ok(IpcClient {
+            out: out_tx,
+            rx: resp_rx,
+            next_id,
+            shared,
+            _socket: socket,
+        })
+    }
+
+    /// Encode and queue a request; returns the message id it will be
+    /// answered under.
+    pub fn request(&self, req: Request) -> Result<u64, ClientError> {
+        let body = serde_json::to_vec(&req).map_err(ClientError::Json)?;
+        let id = self.next_id.fetch_add(1, Ordering::Relaxed);
+        self.out
+            .send((id, body))
+            .map_err(|e| ClientError::Io(io::Error::new(io::ErrorKind::BrokenPipe, e)))?;
+        Ok(id)
+    }
+
+    /// Nonblocking: the next completed `(message id, response)`, if any.
+    pub fn try_recv(&self) -> Option<(u64, Response)> {
+        self.rx.try_recv().ok()
+    }
+
+    /// Force the worker to drop its current connection and re-dial. Queued
+    /// requests are kept and replayed after the new dial.
+    pub fn reconnect(&self) {
+        self.shared.force_reconnect.store(true, Ordering::Relaxed);
+    }
+}
+
+/// Write one `[u32 LE len][body]` frame to the stream.
+fn write_frame(stream: &mut UnixStream, body: &[u8]) -> io::Result<()> {
+    let len = (body.len() as u32).to_le_bytes();
+    stream.write_all(&len)?;
+    stream.write_all(body)?;
+    stream.flush()
+}
+
+/// Read one `[u32 LE len][body]` frame from the stream.
+fn read_frame(stream: &mut UnixStream) -> io::Result<Vec<u8>> {
+    let mut len_buf = [0u8; 4];
+    stream.read_exact(&mut len_buf)?;
+    let len = u32::from_le_bytes(len_buf) as usize;
+    let mut buf = vec![0u8; len];
+    stream.read_exact(&mut buf)?;
+    Ok(buf)
+}
+
+/// The worker loop: dial, serve requests in order, redial forever.
+fn worker(
+    socket: PathBuf,
+    out: Receiver<(u64, Vec<u8>)>,
+    resp: Sender<(u64, Response)>,
+    shared: Arc<Shared>,
+) {
+    let mut backlog: VecDeque<(u64, Vec<u8>)> = VecDeque::new();
+    loop {
+        // Dial (keeping the queue drained while we wait for the service).
+        let mut stream = loop {
+            match UnixStream::connect(&socket) {
+                Ok(s) => break s,
+                Err(_) => {
+                    while let Ok(x) = out.try_recv() {
+                        backlog.push_back(x);
+                    }
+                    thread::sleep(Duration::from_millis(50));
+                }
+            }
+        };
+
+        // Serve this connection until it dies or reconnect() asks.
+        loop {
+            if shared.force_reconnect.swap(false, Ordering::Relaxed) {
+                break;
+            }
+            let (id, body) = if let Some(x) = backlog.pop_front() {
+                x
+            } else {
+                match out.recv_timeout(Duration::from_millis(50)) {
+                    Ok(x) => x,
+                    Err(mpsc::RecvTimeoutError::Timeout) => continue,
+                    Err(mpsc::RecvTimeoutError::Disconnected) => return,
+                }
+            };
+
+            // A reconnect() may have landed while this request waited in the
+            // channel: redial first rather than write on a connection that
+            // may already be dead.
+            if shared.force_reconnect.swap(false, Ordering::Relaxed) {
+                backlog.push_front((id, body));
+                break;
+            }
+
+            if write_frame(&mut stream, &body).is_err() {
+                backlog.push_front((id, body));
+                break;
+            }
+            match read_frame(&mut stream) {
+                Ok(buf) => {
+                    let response = serde_json::from_slice::<Response>(&buf).unwrap_or_else(|e| {
+                        Response::Error {
+                            message: format!("unparseable service response: {e}"),
+                        }
+                    });
+                    let _ = resp.send((id, response));
+                }
+                Err(_) => {
+                    // The connection died mid-exchange; replay the request
+                    // on the next dial rather than dropping it.
+                    backlog.push_front((id, body));
+                    break;
+                }
+            }
+        }
+    }
+}

tines-gui/src/lib.rs

@@ -0,0 +1,12 @@
+//! The tines fleet console.
+//!
+//! A launcher plus one window per container: each window drives a running
+//! `tines` service over its Unix-socket IPC. This crate holds the pieces
+//! shared by those windows — the IPC client ([`ipc_client`]), the
+//! config-generation half of the launcher ([`config`]), and the omarchy
+//! theme mechanism ([`theme`]).
+#![deny(missing_docs)]
+
+pub mod config;
+pub mod ipc_client;
+pub mod theme;

tines-gui/src/main.rs

@@ -0,0 +1,9 @@
+//! The console binary. Task 8 wires the real UI; for now the launcher
+//! sentence is the whole program.
+
+fn main() {
+    let args: Vec<String> = std::env::args().collect();
+    if args.len() < 2 {
+        println!("tines-gui: launcher");
+    }
+}

tines-gui/src/theme.rs

@@ -0,0 +1,110 @@
+//! The omarchy theme: the live palette plus a watch that repaints on change.
+//!
+//! Reads `~/.local/state/omarchy/current/theme/colors.toml` for the three
+//! keys the console uses — `background`, `foreground`, `accent`. Omarchy
+//! rewrites the `current` directory in place on theme change, so the watch
+//! is on the directory (recursive), not the file.
+
+use std::path::PathBuf;
+
+use egui::Color32;
+use notify::Watcher;
+
+/// The three palette colors the console paints with.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct OmarchyTheme {
+    /// The window background.
+    pub background: Color32,
+    /// The window foreground (text).
+    pub foreground: Color32,
+    /// The accent (highlights, focus).
+    pub accent: Color32,
+}
+
+/// The app's own named fallback — used only when the live theme is missing
+/// or unreadable. Deliberately not the omarchy palette: the console must be
+/// visibly on fallback, not approximating the real theme.
+pub const FALLBACK: OmarchyTheme = OmarchyTheme {
+    background: Color32::from_rgb(0x16, 0x18, 0x1d),
+    foreground: Color32::from_rgb(0xd7, 0xdb, 0xe2),
+    accent: Color32::from_rgb(0x7a, 0xa2, 0xf7),
+};
+
+impl OmarchyTheme {
+    /// Load the live theme; [`FALLBACK`] when it is missing or malformed.
+    pub fn load() -> OmarchyTheme {
+        match read_theme_file() {
+            Some(t) => t,
+            None => FALLBACK,
+        }
+    }
+
+    /// Watch the omarchy `current` theme dir (recursively) and call
+    /// `on_change` whenever it changes — the caller repaints from there.
+    /// Returns an error only if the watcher cannot be installed.
+    pub fn watch<F>(on_change: F) -> notify::Result<()>
+    where
+        F: FnMut() + Send + 'static,
+    {
+        // notify's callback is Fn, ours is FnMut: share it through a mutex.
+        let on_change = std::sync::Mutex::new(on_change);
+        let mut watcher =
+            notify::recommended_watcher(move |res: notify::Result<notify::Event>| {
+                if res.is_ok()
+                    && let Ok(mut f) = on_change.lock()
+                {
+                    f();
+                }
+            })?;
+        watcher.watch(&current_dir(), notify::RecursiveMode::Recursive)?;
+        // The watcher must outlive the call: park it on a thread.
+        std::thread::spawn(move || {
+            std::thread::park();
+            let _ = watcher;
+        });
+        Ok(())
+    }
+}
+
+/// `~/.local/state/omarchy/current` — the live-theme indirection.
+fn current_dir() -> PathBuf {
+    let home = std::env::var_os("HOME")
+        .map(PathBuf::from)
+        .unwrap_or_else(std::env::temp_dir);
+    home.join(".local/state/omarchy/current")
+}
+
+fn read_theme_file() -> Option<OmarchyTheme> {
+    let text = std::fs::read_to_string(current_dir().join("theme/colors.toml")).ok()?;
+    let mut background = None;
+    let mut foreground = None;
+    let mut accent = None;
+    for line in text.lines() {
+        let Some((key, value)) = line.split_once('=') else {
+            continue; // blank lines and comments carry no key
+        };
+        let parsed = parse_hex(value);
+        match key.trim() {
+            "background" => background = parsed,
+            "foreground" => foreground = parsed,
+            "accent" => accent = parsed,
+            _ => {}
+        }
+    }
+    Some(OmarchyTheme {
+        background: background?,
+        foreground: foreground?,
+        accent: accent?,
+    })
+}
+
+fn parse_hex(s: &str) -> Option<Color32> {
+    let s = s.trim().trim_matches('"').strip_prefix('#')?;
+    if s.len() != 6 {
+        return None;
+    }
+    let r = u8::from_str_radix(&s[0..2], 16).ok()?;
+    let g = u8::from_str_radix(&s[2..4], 16).ok()?;
+    let b = u8::from_str_radix(&s[4..6], 16).ok()?;
+    Some(Color32::from_rgb(r, g, b))
+}

tines-gui/tests/config.rs

@@ -0,0 +1,75 @@
+//! The GUI's generated configs must deserialize as the service's `Config`
+//! byte-for-byte: unknown keys are fatal on the service side.
+
+use tines_gui::config::{NodeConfig, generate_node_id};
+
+#[test]
+fn generated_config_parses_as_service_config() {
+    // bt: transport bt, a peer, no bus socket.
+    let bt = NodeConfig::bt(
+        generate_node_id(),
+        Some(generate_node_id()),
+        "/tmp/fleet/log".into(),
+        "/tmp/fleet/node.sock".into(),
+    );
+    let json = serde_json::to_string(&bt).unwrap();
+    let parsed: tines::config::Config =
+        serde_json::from_str(&json).expect("bt config must parse as the service config");
+    assert_eq!(parsed.node_id, bt.node_id);
+    assert_eq!(parsed.peer, bt.peer);
+    assert_eq!(parsed.bus_socket, None);
+    assert_eq!(parsed.validate(), Ok(()));
+
+    // bus: transport bus, no peer, a bus socket.
+    let bus = NodeConfig::bus(
+        generate_node_id(),
+        None,
+        "/tmp/fleet/log".into(),
+        "/tmp/fleet/node.sock".into(),
+        "/tmp/fleet/bus.sock".into(),
+    );
+    let json = serde_json::to_string(&bus).unwrap();
+    let parsed: tines::config::Config =
+        serde_json::from_str(&json).expect("bus config must parse as the service config");
+    assert_eq!(parsed.node_id, bus.node_id);
+    assert_eq!(parsed.peer, None);
+    assert_eq!(parsed.bus_socket, Some("/tmp/fleet/bus.sock".to_string()));
+    assert_eq!(parsed.validate(), Ok(()));
+}
+
+#[test]
+fn node_id_is_32_hex_chars() {
+    let id = generate_node_id();
+    assert_eq!(id.len(), 32, "node id must be 32 chars, got {id:?}");
+    assert!(
+        id.chars()
+            .all(|c| c.is_ascii_hexdigit() && !c.is_ascii_uppercase()),
+        "node id must be lowercase hex, got {id:?}"
+    );
+}
+
+#[test]
+fn fleet_dir_is_0700() {
+    use std::os::unix::fs::PermissionsExt;
+
+    // Point XDG_STATE_HOME at a scratch dir so the test never touches the
+    // real ~/.local/state, and fleet_dir's creation is what we inspect.
+    let scratch = std::env::temp_dir().join(format!("tines-gui-state-{}", std::process::id()));
+    std::fs::remove_dir_all(&scratch).ok();
+    std::fs::create_dir_all(&scratch).unwrap();
+    // SAFETY: this is the only test in this binary touching fleet_dir, and
+    // the test binary dies right after; the env change cannot leak to other
+    // tests in other binaries.
+    unsafe { std::env::set_var("XDG_STATE_HOME", &scratch) };
+
+    let dir = tines_gui::config::fleet_dir("fleet-a");
+    let meta = std::fs::metadata(&dir).expect("fleet dir must be created");
+    assert!(meta.is_dir());
+    assert_eq!(
+        meta.permissions().mode() & 0o777,
+        0o700,
+        "fleet dir must be 0700"
+    );
+
+    std::fs::remove_dir_all(&scratch).ok();
+}

tines-gui/tests/ipc_client.rs

@@ -0,0 +1,133 @@
+//! The IPC client against a fake service speaking the real wire framing
+//! (`[u32 LE len][JSON]`), from a test-local UnixListener.
+
+use std::io::{Read, Write};
+use std::os::unix::net::{UnixListener, UnixStream};
+use std::path::{Path, PathBuf};
+use std::thread;
+use std::time::{Duration, Instant};
+
+use tines::{Request, Response};
+use tines_gui::ipc_client::IpcClient;
+
+fn scratch_dir(tag: &str) -> PathBuf {
+    let dir = std::env::temp_dir().join(format!("tines-gui-ipc-{tag}-{}", std::process::id()));
+    std::fs::create_dir_all(&dir).unwrap();
+    dir
+}
+
+fn read_frame(stream: &mut UnixStream) -> Vec<u8> {
+    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();
+    buf
+}
+
+fn write_frame(stream: &mut UnixStream, body: &[u8]) {
+    let len = (body.len() as u32).to_le_bytes();
+    stream.write_all(&len).unwrap();
+    stream.write_all(body).unwrap();
+    stream.flush().unwrap();
+}
+
+/// Answer one Observe request with an Observe response, then end the
+/// connection (EOF).
+fn answer_once(socket: &Path) {
+    // A restarted server finds the previous server's socket path still on
+    // disk (Linux leaves the file behind when a bound listener drops), so
+    // it removes the stale path first — exactly what a real service does.
+    let _ = std::fs::remove_file(socket);
+    let (mut stream, _) = UnixListener::bind(socket)
+        .unwrap()
+        .accept()
+        .expect("server accepts");
+    let body = read_frame(&mut stream);
+    // Request-agnostic: the client-side tests only assert the id pairing,
+    // so any well-formed request gets an Observe-shaped response.
+    let _: Request = serde_json::from_slice(&body).expect("well-formed request");
+    write_frame(
+        &mut stream,
+        &serde_json::to_vec(&Response::Observe { id: "00".into() }).unwrap(),
+    );
+    // Dropping the stream here closes the connection: the client sees EOF.
+}
+
+#[test]
+fn client_roundtrips_observe() {
+    let dir = scratch_dir("rt");
+    let socket = dir.join("s.sock");
+
+    let server_socket = socket.clone();
+    let server = thread::spawn(move || answer_once(&server_socket));
+
+    let client = IpcClient::spawn(socket.clone()).expect("client spawns");
+    let id = client
+        .request(Request::Observe {
+            topic: "ab".into(),
+            payload: "cd".into(),
+            to: None,
+            lifetime_ms: None,
+        })
+        .expect("request queues");
+
+    let deadline = Instant::now() + Duration::from_secs(5);
+    loop {
+        if let Some((got_id, resp)) = client.try_recv() {
+            assert_eq!(got_id, id, "response must carry the request id");
+            assert!(
+                matches!(resp, Response::Observe { .. }),
+                "expected Observe response, got {resp:?}"
+            );
+            break;
+        }
+        assert!(Instant::now() < deadline, "no response arrived");
+        thread::sleep(Duration::from_millis(20));
+    }
+    server.join().unwrap();
+}
+
+#[test]
+fn client_survives_server_restart() {
+    let dir = scratch_dir("restart");
+    let socket = dir.join("s.sock");
+
+    // First server: one round trip, then it goes away.
+    let server_socket = socket.clone();
+    let first = thread::spawn(move || answer_once(&server_socket));
+
+    let client = IpcClient::spawn(socket.clone()).expect("client spawns");
+    let first_id = client
+        .request(Request::Events)
+        .expect("first request queues");
+    let deadline = Instant::now() + Duration::from_secs(5);
+    loop {
+        if let Some((got, _)) = client.try_recv() {
+            assert_eq!(got, first_id);
+            break;
+        }
+        assert!(Instant::now() < deadline, "no response to first request");
+        thread::sleep(Duration::from_millis(20));
+    }
+    first.join().unwrap();
+
+    // Restart: a fresh listener on the same path answers the next request.
+    client.reconnect();
+    let server_socket = socket.clone();
+    let second = thread::spawn(move || answer_once(&server_socket));
+
+    let second_id = client
+        .request(Request::Events)
+        .expect("second request queues");
+    let deadline = Instant::now() + Duration::from_secs(5);
+    loop {
+        if let Some((got, _)) = client.try_recv() {
+            assert_eq!(got, second_id, "ids keep increasing across restarts");
+            break;
+        }
+        assert!(Instant::now() < deadline, "no response after restart");
+        thread::sleep(Duration::from_millis(20));
+    }
+    second.join().unwrap();
+}