motorcortex_rust/core/

request.rs

//! The async-first `Request` handle.
//!
//! Cheap to clone (carries a `mpsc::Sender`); every method is an
//! `async fn` that resolves once the driver thread serves the
//! command. See [`crate::core::driver`] for the loop on the other
//! side of the channel.

use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, RwLock};
use std::thread;

use tokio::sync::{mpsc, oneshot, watch};

use crate::client::{ParameterTree, Parameters};
use crate::connection::{ConnectionOptions, PipeEvent};
use crate::core::driver::run_request_driver;
use crate::core::state::ConnectionState;
use crate::core::util::await_reply;
use crate::error::{MotorcortexError, Result};
use crate::msg::{
    CreateGroupMsg, GetParameterListMsg, GetParameterMsg, GroupStatusMsg, ParameterListMsg,
    SetParameterListMsg, SetParameterMsg, StatusCode,
};
use crate::parameter_value::{
    GetParameterTuple, GetParameterValue, SetParameterTuple, SetParameterValue,
    decode_parameter_value, encode_parameter_value,
};

/// Commands the `Request` handle sends to its driver thread.
///
/// Each variant carries an `oneshot::Sender` for the driver to hand
/// back the result — the user `.await`s the matching `Receiver`.
pub(crate) enum Cmd {
    Connect {
        url: String,
        opts: ConnectionOptions,
        reply: oneshot::Sender<Result<()>>,
    },
    Disconnect {
        reply: oneshot::Sender<Result<()>>,
    },
    Login {
        user: String,
        pass: String,
        reply: oneshot::Sender<Result<StatusCode>>,
    },
    Logout {
        reply: oneshot::Sender<Result<StatusCode>>,
    },
    RequestParameterTree {
        reply: oneshot::Sender<Result<StatusCode>>,
    },
    GetParameter {
        path: String,
        reply: oneshot::Sender<Result<Vec<u8>>>,
    },
    SetParameter {
        path: String,
        value: Vec<u8>,
        reply: oneshot::Sender<Result<StatusCode>>,
    },
    GetParameters {
        msg: GetParameterListMsg,
        reply: oneshot::Sender<Result<ParameterListMsg>>,
    },
    SetParameters {
        msg: SetParameterListMsg,
        reply: oneshot::Sender<Result<StatusCode>>,
    },
    CreateGroup {
        msg: CreateGroupMsg,
        reply: oneshot::Sender<Result<GroupStatusMsg>>,
    },
    RemoveGroup {
        alias: String,
        reply: oneshot::Sender<Result<StatusCode>>,
    },
    GetParameterTreeHash {
        reply: oneshot::Sender<Result<u32>>,
    },
    /// Fetch a fresh session token from the server. On success,
    /// updates the driver's shared `last_token` cache (which powers
    /// the automatic restore on reconnect) *and* returns the token
    /// to the caller.
    GetSessionToken {
        reply: oneshot::Sender<Result<String>>,
    },
    /// Restore a previously-issued session without re-login. The
    /// caller supplies the token explicitly; the driver uses this
    /// path internally on reconnect too.
    RestoreSession {
        token: String,
        reply: oneshot::Sender<Result<StatusCode>>,
    },
    /// Background token-refresh tick. Does a GetSessionToken RPC,
    /// stores the result in the driver's `last_token` cache, and
    /// discards the reply. Driven by the refresh helper thread.
    RefreshTokenTick,
    /// Forwarded from the NNG pipe-notify callback via an
    /// `Arc<dyn Fn(PipeEvent)>` installed on the connection. Drains
    /// through the same command queue as user RPCs so state
    /// transitions don't race with requests in flight.
    Pipe(PipeEvent),
}

/// Async handle for request/reply RPCs.
///
/// Cloning a `Request` gives a second handle that multiplexes onto the
/// same driver thread — commands serialise through the driver, so the
/// NNG Req/Rep ordering invariant is enforced in the type system
/// without any user-visible `Mutex`.
pub struct Request {
    tx: mpsc::UnboundedSender<Cmd>,
    state: watch::Receiver<ConnectionState>,
    tree: Arc<RwLock<ParameterTree>>,
    /// Latest session token from the most recent `GetSessionToken`
    /// RPC (either user-invoked or via the refresh helper). Shared
    /// with the driver so both the user side (via
    /// [`session_token`](Self::session_token)) and the reconnect
    /// handler see the same value.
    last_token: Arc<RwLock<Option<String>>>,
    /// Counter the driver bumps each time the background refresh
    /// helper actually fires a `GetSessionToken` RPC (i.e. the
    /// connection was live at tick time). Lets callers — and the
    /// integration tests — observe that the refresh is paused while
    /// the pipe is down.
    refresh_count: Arc<AtomicU64>,
}

impl Request {
    /// Create a new handle and spawn its driver thread.
    ///
    /// The handle starts in [`ConnectionState::Disconnected`]; call
    /// [`connect`](Self::connect) to open the socket.
    ///
    /// ```
    /// use motorcortex_rust::core::{ConnectionState, Request};
    /// let req = Request::new();
    /// assert_eq!(*req.state().borrow(), ConnectionState::Disconnected);
    /// ```
    pub fn new() -> Self {
        let (tx, rx) = mpsc::unbounded_channel();
        let (state_tx, state_rx) = watch::channel(ConnectionState::Disconnected);
        let tree = Arc::new(RwLock::new(ParameterTree::new()));
        let last_token: Arc<RwLock<Option<String>>> = Arc::new(RwLock::new(None));
        let refresh_count = Arc::new(AtomicU64::new(0));
        let tree_for_driver = Arc::clone(&tree);
        let token_for_driver = Arc::clone(&last_token);
        let count_for_driver = Arc::clone(&refresh_count);
        let tx_for_driver = tx.clone();
        thread::Builder::new()
            .name("mcx-request-driver".into())
            .spawn(move || {
                run_request_driver(
                    tx_for_driver,
                    rx,
                    state_tx,
                    tree_for_driver,
                    token_for_driver,
                    count_for_driver,
                )
            })
            .expect("spawning the driver thread must succeed on any OS we target");
        Self {
            tx,
            state: state_rx,
            tree,
            last_token,
            refresh_count,
        }
    }

    /// Subscribe to connection-state transitions.
    ///
    /// Returns a `watch::Receiver`; consumers can `state.changed().await`
    /// or `*state.borrow()` for the current value.
    pub fn state(&self) -> watch::Receiver<ConnectionState> {
        self.state.clone()
    }

    /// Open the socket and dial `url`.
    ///
    /// ```no_run
    /// # async fn demo() -> motorcortex_rust::Result<()> {
    /// use motorcortex_rust::{ConnectionOptions, core::Request};
    /// let req = Request::new();
    /// let opts = ConnectionOptions::new("mcx.cert.crt".into(), 1000, 1000);
    /// req.connect("wss://127.0.0.1:5568", opts).await?;
    /// # Ok(()) }
    /// ```
    pub async fn connect(&self, url: &str, opts: ConnectionOptions) -> Result<()> {
        let (reply_tx, reply_rx) = oneshot::channel();
        self.send_cmd(Cmd::Connect {
            url: url.to_string(),
            opts,
            reply: reply_tx,
        })?;
        await_reply(reply_rx).await?
    }

    /// Close the socket. Subsequent RPCs will error with
    /// [`MotorcortexError::Connection`] until `connect` is called again.
    pub async fn disconnect(&self) -> Result<()> {
        let (reply_tx, reply_rx) = oneshot::channel();
        self.send_cmd(Cmd::Disconnect { reply: reply_tx })?;
        await_reply(reply_rx).await?
    }

    /// Authenticate with the server.
    ///
    /// Returns the server's [`StatusCode`] — `Ok` on success,
    /// `WrongPassword` / other variants on rejection. Transport-level
    /// failures surface as [`MotorcortexError`].
    pub async fn login(&self, user: &str, pass: &str) -> Result<StatusCode> {
        let (reply_tx, reply_rx) = oneshot::channel();
        self.send_cmd(Cmd::Login {
            user: user.to_string(),
            pass: pass.to_string(),
            reply: reply_tx,
        })?;
        await_reply(reply_rx).await?
    }

    /// Drop the current session on the server.
    pub async fn logout(&self) -> Result<StatusCode> {
        let (reply_tx, reply_rx) = oneshot::channel();
        self.send_cmd(Cmd::Logout { reply: reply_tx })?;
        await_reply(reply_rx).await?
    }

    /// Fetch the parameter tree from the server and store it in the
    /// shared cache accessible via [`parameter_tree`](Self::parameter_tree).
    ///
    /// Returns the server's [`StatusCode`]. The cache is only updated
    /// on `StatusCode::Ok`; a non-OK reply leaves the previous cache
    /// intact.
    pub async fn request_parameter_tree(&self) -> Result<StatusCode> {
        let (reply_tx, reply_rx) = oneshot::channel();
        self.send_cmd(Cmd::RequestParameterTree { reply: reply_tx })?;
        await_reply(reply_rx).await?
    }

    /// Shared read handle to the local parameter-tree cache. Every
    /// cloned [`Request`] sees the same tree; reads are `RwLock::read`
    /// (cheap, no channel round-trip). Populated by
    /// [`request_parameter_tree`](Self::request_parameter_tree).
    pub fn parameter_tree(&self) -> Arc<RwLock<ParameterTree>> {
        Arc::clone(&self.tree)
    }

    /// Read a single parameter. The caller-specified `V` is the Rust
    /// type the server value should be converted to (see
    /// [`GetParameterValue`]).
    ///
    /// Returns [`MotorcortexError::ParameterNotFound`] if `path` is
    /// unknown locally — call [`request_parameter_tree`](Self::request_parameter_tree)
    /// first.
    ///
    /// ```no_run
    /// # async fn demo(req: motorcortex_rust::core::Request) -> motorcortex_rust::Result<()> {
    /// // Same path, three different Rust types — the server value is
    /// // converted per call (lossy casts allowed).
    /// let as_double: f64    = req.get_parameter("root/Control/dummyDouble").await?;
    /// let as_int:    i64    = req.get_parameter("root/Control/dummyDouble").await?;
    /// let as_text:   String = req.get_parameter("root/Control/dummyDouble").await?;
    /// # Ok(()) }
    /// ```
    pub async fn get_parameter<V>(&self, path: &str) -> Result<V>
    where
        V: GetParameterValue + Default,
    {
        let data_type = self.data_type_of(path)?;

        let (reply_tx, reply_rx) = oneshot::channel();
        self.send_cmd(Cmd::GetParameter {
            path: path.to_string(),
            reply: reply_tx,
        })?;
        let value_bytes = await_reply(reply_rx).await??;
        Ok(decode_parameter_value::<V>(data_type, &value_bytes))
    }

    /// Write a single parameter. Returns the server's [`StatusCode`].
    ///
    /// ```no_run
    /// # async fn demo(req: motorcortex_rust::core::Request) -> motorcortex_rust::Result<()> {
    /// // Scalar.
    /// req.set_parameter("root/Control/dummyDouble", 2.345_f64).await?;
    /// // Fixed-size array.
    /// req.set_parameter("root/Control/dummyDoubleVec", [1.0, 2.0, 3.0]).await?;
    /// // Dynamic Vec.
    /// req.set_parameter("root/Control/dummyDoubleVec", vec![1.0, 2.0]).await?;
    /// # Ok(()) }
    /// ```
    pub async fn set_parameter<V>(&self, path: &str, value: V) -> Result<StatusCode>
    where
        V: SetParameterValue,
    {
        let data_type = self.data_type_of(path)?;
        let value_bytes = encode_parameter_value(data_type, &value);

        let (reply_tx, reply_rx) = oneshot::channel();
        self.send_cmd(Cmd::SetParameter {
            path: path.to_string(),
            value: value_bytes,
            reply: reply_tx,
        })?;
        await_reply(reply_rx).await?
    }

    /// Read a batch of parameters in one RPC. The generic `T` is a
    /// tuple type like `(bool, f64, i32)` whose arity matches
    /// `paths.len()`. Each element is decoded into its position using
    /// the dtype recorded in the local tree cache.
    ///
    /// ```no_run
    /// # async fn demo(req: motorcortex_rust::core::Request) -> motorcortex_rust::Result<()> {
    /// let (b, d, i): (bool, f64, i32) = req.get_parameters(&[
    ///     "root/Control/dummyBool",
    ///     "root/Control/dummyDouble",
    ///     "root/Control/dummyInt32",
    /// ]).await?;
    /// # Ok(()) }
    /// ```
    pub async fn get_parameters<T>(&self, paths: &[&str]) -> Result<T>
    where
        T: GetParameterTuple,
    {
        // Resolve every dtype up-front so the first missing path is
        // reported before we even build the request message.
        let dtypes: Vec<u32> = paths
            .iter()
            .map(|p| self.data_type_of(p))
            .collect::<Result<_>>()?;

        let msg = GetParameterListMsg {
            header: None,
            params: paths
                .iter()
                .map(|p| GetParameterMsg {
                    header: None,
                    path: p.to_string(),
                })
                .collect(),
        };

        let (reply_tx, reply_rx) = oneshot::channel();
        self.send_cmd(Cmd::GetParameters {
            msg,
            reply: reply_tx,
        })?;
        let reply = await_reply(reply_rx).await??;

        // Use the dtypes the client looked up locally, not whatever
        // the server echoed back. Matches legacy semantics (dtype is
        // authoritative client-side — it's what the caller used when
        // picking `T`'s element types).
        let iter = reply
            .params
            .iter()
            .zip(dtypes.iter())
            .map(|(param, dt)| (dt, param.value.as_slice()));
        T::get_parameters(iter).map_err(MotorcortexError::Decode)
    }

    /// Write a batch of parameters in one RPC. `values` is a tuple
    /// (or single-element tuple) whose arity matches `paths.len()`.
    /// Each element is encoded against the dtype recorded in the
    /// local tree cache.
    pub async fn set_parameters<T>(&self, paths: &[&str], values: T) -> Result<StatusCode>
    where
        T: SetParameterTuple,
    {
        let mut params = Vec::with_capacity(paths.len());
        for (i, path) in paths.iter().enumerate() {
            let data_type = self.data_type_of(path)?;
            let value = values
                .get_tuple_element(i, data_type)
                .map_err(MotorcortexError::Encode)?;
            params.push(SetParameterMsg {
                header: None,
                offset: None,
                path: path.to_string(),
                value,
            });
        }

        let msg = SetParameterListMsg {
            header: None,
            params,
        };

        let (reply_tx, reply_rx) = oneshot::channel();
        self.send_cmd(Cmd::SetParameters {
            msg,
            reply: reply_tx,
        })?;
        await_reply(reply_rx).await?
    }

    /// Create a server-side subscription group. Returns the
    /// [`GroupStatusMsg`] the subscribe-side code uses as the
    /// group descriptor.
    ///
    /// `paths` accepts anything implementing [`Parameters`] — a
    /// string literal, a `Vec<String>`, an array of `&str`, etc.
    pub async fn create_group<I>(
        &self,
        paths: I,
        alias: &str,
        frequency_divider: u32,
    ) -> Result<GroupStatusMsg>
    where
        I: Parameters,
    {
        let msg = CreateGroupMsg {
            header: None,
            frq_divider: frequency_divider,
            alias: alias.to_string(),
            paths: paths.into_vec(),
        };
        let (reply_tx, reply_rx) = oneshot::channel();
        self.send_cmd(Cmd::CreateGroup {
            msg,
            reply: reply_tx,
        })?;
        await_reply(reply_rx).await?
    }

    /// Remove a previously-created subscription group by alias.
    /// Returns the server's [`StatusCode`] unchanged — `Ok` on
    /// success, `Failed` (or similar) if the group wasn't there.
    /// Transport / decode failures surface as
    /// [`MotorcortexError`].
    ///
    /// The crate-wide rule: RPCs that return `Result<StatusCode>`
    /// never promote a non-OK server reply into `Err`. Branch on
    /// the returned code if you care which way the request went.
    pub async fn remove_group(&self, alias: &str) -> Result<StatusCode> {
        let (reply_tx, reply_rx) = oneshot::channel();
        self.send_cmd(Cmd::RemoveGroup {
            alias: alias.to_string(),
            reply: reply_tx,
        })?;
        await_reply(reply_rx).await?
    }

    /// Fetch a fresh session token from the server.
    ///
    /// On success the token is cached in the driver and returned to
    /// the caller. Callers that want to persist a session across
    /// process restarts can stash the returned string and hand it
    /// back to a fresh `Request` via [`restore_session`](Self::restore_session).
    ///
    /// While a connection is live, the driver also refreshes this
    /// token periodically in the background (see
    /// [`ConnectionOptions::token_refresh_interval`]) so the cache
    /// stays warm for the automatic reconnect path.
    pub async fn get_session_token(&self) -> Result<String> {
        let (reply_tx, reply_rx) = oneshot::channel();
        self.send_cmd(Cmd::GetSessionToken { reply: reply_tx })?;
        await_reply(reply_rx).await?
    }

    /// Restore a previously-issued session by supplying the token.
    /// Returns the server's [`StatusCode`]:
    ///
    /// - `Ok` / `ReadOnlyMode` — the session was accepted, subsequent
    ///   RPCs run under that identity.
    /// - `PermissionDenied` / `Failed` — the token is stale or the
    ///   server has lost its state.
    ///
    /// The driver calls this internally on reconnect using the token
    /// stashed by the refresh loop; callers generally don't need to
    /// invoke it explicitly unless they're recovering from a process
    /// restart.
    pub async fn restore_session(&self, token: &str) -> Result<StatusCode> {
        let (reply_tx, reply_rx) = oneshot::channel();
        self.send_cmd(Cmd::RestoreSession {
            token: token.to_string(),
            reply: reply_tx,
        })?;
        await_reply(reply_rx).await?
    }

    /// Snapshot of the most recently cached session token, if any.
    ///
    /// Populated by [`get_session_token`](Self::get_session_token)
    /// and by the periodic refresh loop. Returns `None` before the
    /// first successful fetch.
    pub fn session_token(&self) -> Option<String> {
        self.last_token.read().ok().and_then(|g| g.clone())
    }

    /// Count of `GetSessionToken` RPCs the background refresh helper
    /// has fired since this handle was created.
    ///
    /// Bumped by the driver only when the refresh tick runs against
    /// a live pipe — ticks that fire while the state is
    /// `ConnectionLost` / `SessionExpired` / `Disconnected` are
    /// skipped and don't count. Useful for observing that the
    /// refresh loop is paused while the transport is down, or as a
    /// lightweight liveness metric.
    pub fn session_refresh_count(&self) -> u64 {
        self.refresh_count.load(Ordering::Relaxed)
    }

    /// Fetch the server's parameter-tree hash — useful for cheap
    /// change detection. A non-zero return on a populated server
    /// signals the caller can skip a full [`request_parameter_tree`]
    /// if the hash matches what they cached previously.
    ///
    /// [`request_parameter_tree`]: Self::request_parameter_tree
    pub async fn get_parameter_tree_hash(&self) -> Result<u32> {
        let (reply_tx, reply_rx) = oneshot::channel();
        self.send_cmd(Cmd::GetParameterTreeHash { reply: reply_tx })?;
        await_reply(reply_rx).await?
    }

    /// Convenience: `Request::new()` + [`connect`](Self::connect) in
    /// one call. Useful for the "I just want a connected client"
    /// entry path.
    ///
    /// ```no_run
    /// # async fn demo() -> motorcortex_rust::Result<()> {
    /// use motorcortex_rust::{ConnectionOptions, core::Request};
    /// let opts = ConnectionOptions::new("mcx.cert.crt".into(), 1000, 1000);
    /// let req = Request::connect_to("wss://127.0.0.1:5568", opts).await?;
    /// req.request_parameter_tree().await?;
    /// # Ok(()) }
    /// ```
    pub async fn connect_to(url: &str, opts: ConnectionOptions) -> Result<Self> {
        let req = Self::new();
        req.connect(url, opts).await?;
        Ok(req)
    }

    /// Shared helper: local tree lookup with a clear
    /// `ParameterNotFound` error when the path isn't cached.
    fn data_type_of(&self, path: &str) -> Result<u32> {
        self.tree
            .read()
            .map_err(|_| MotorcortexError::Decode("parameter tree lock poisoned".into()))?
            .get_parameter_data_type(path)
            .ok_or_else(|| MotorcortexError::ParameterNotFound(path.to_string()))
    }

    fn send_cmd(&self, cmd: Cmd) -> Result<()> {
        self.tx
            .send(cmd)
            .map_err(|_| MotorcortexError::Connection("driver thread is gone".into()))
    }
}

impl Default for Request {
    fn default() -> Self {
        Self::new()
    }
}

impl Clone for Request {
    fn clone(&self) -> Self {
        Self {
            tx: self.tx.clone(),
            state: self.state.clone(),
            tree: Arc::clone(&self.tree),
            last_token: Arc::clone(&self.last_token),
            refresh_count: Arc::clone(&self.refresh_count),
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn new_starts_disconnected() {
        let req = Request::new();
        assert_eq!(*req.state().borrow(), ConnectionState::Disconnected);
    }

    #[test]
    fn clone_shares_the_same_state_watch() {
        let a = Request::new();
        let b = a.clone();
        // Both handles should observe the same initial value, because
        // the watch channel is shared (clone clones a Receiver).
        assert_eq!(*a.state().borrow(), *b.state().borrow());
    }

    #[test]
    fn default_is_equivalent_to_new() {
        let req = Request::default();
        assert_eq!(*req.state().borrow(), ConnectionState::Disconnected);
    }

    #[test]
    fn dropping_handle_does_not_panic() {
        // When the last handle drops, the mpsc channel closes and the
        // driver exits cleanly via blocking_recv → None. We can't
        // directly observe the thread join without tracking the
        // JoinHandle, but we can at least confirm dropping is safe.
        let req = Request::new();
        drop(req);
    }

    #[tokio::test]
    async fn disconnect_without_connect_is_ok() {
        // ConnectionManager::disconnect is idempotent — both sock and
        // tls_cfg are None, so nothing happens at the NNG layer.
        let req = Request::new();
        req.disconnect().await.expect("no-op disconnect must succeed");
    }

    #[tokio::test]
    async fn login_without_connect_errors_with_connection_variant() {
        let req = Request::new();
        let err = req
            .login("u", "p")
            .await
            .expect_err("no socket → Connection error");
        assert!(matches!(err, MotorcortexError::Connection(_)));
    }

    #[tokio::test]
    async fn logout_without_connect_errors_with_connection_variant() {
        let req = Request::new();
        let err = req
            .logout()
            .await
            .expect_err("no socket → Connection error");
        assert!(matches!(err, MotorcortexError::Connection(_)));
    }

    #[tokio::test]
    async fn get_session_token_without_connect_errors() {
        let req = Request::new();
        let err = req
            .get_session_token()
            .await
            .expect_err("no socket → Connection error");
        assert!(matches!(err, MotorcortexError::Connection(_)));
    }

    #[tokio::test]
    async fn restore_session_without_connect_errors() {
        let req = Request::new();
        let err = req
            .restore_session("dummy-token")
            .await
            .expect_err("no socket → Connection error");
        assert!(matches!(err, MotorcortexError::Connection(_)));
    }

    #[test]
    fn session_token_is_none_on_fresh_handle() {
        let req = Request::new();
        assert!(req.session_token().is_none());
    }

    #[tokio::test]
    async fn get_parameter_tree_hash_without_connect_errors() {
        let req = Request::new();
        let err = req
            .get_parameter_tree_hash()
            .await
            .expect_err("no socket → Connection error");
        assert!(matches!(err, MotorcortexError::Connection(_)));
    }

    #[tokio::test]
    async fn create_group_without_connect_errors() {
        let req = Request::new();
        let err = req
            .create_group("root/x", "g", 1)
            .await
            .expect_err("no socket → Connection error");
        assert!(matches!(err, MotorcortexError::Connection(_)));
    }

    #[tokio::test]
    async fn remove_group_without_connect_errors() {
        let req = Request::new();
        let err = req
            .remove_group("g")
            .await
            .expect_err("no socket → Connection error");
        assert!(matches!(err, MotorcortexError::Connection(_)));
    }

    #[tokio::test]
    async fn set_parameters_on_empty_tree_returns_parameter_not_found() {
        let req = Request::new();
        let err = req
            .set_parameters(&["root/missing"], (1.0f64,))
            .await
            .expect_err("empty tree → ParameterNotFound");
        assert!(matches!(err, MotorcortexError::ParameterNotFound(ref p) if p == "root/missing"));
    }

    #[tokio::test]
    async fn get_parameters_on_empty_tree_returns_parameter_not_found() {
        let req = Request::new();
        let err = req
            .get_parameters::<(f64,)>(&["root/missing"])
            .await
            .expect_err("empty tree → ParameterNotFound");
        assert!(matches!(err, MotorcortexError::ParameterNotFound(ref p) if p == "root/missing"));
    }

    #[tokio::test]
    async fn set_parameter_on_empty_tree_returns_parameter_not_found() {
        let req = Request::new();
        let err = req
            .set_parameter("root/missing", 1.0f64)
            .await
            .expect_err("empty tree → ParameterNotFound");
        assert!(matches!(err, MotorcortexError::ParameterNotFound(ref p) if p == "root/missing"));
    }

    #[tokio::test]
    async fn get_parameter_on_empty_tree_returns_parameter_not_found() {
        let req = Request::new();
        let err = req
            .get_parameter::<f64>("root/missing")
            .await
            .expect_err("empty tree → ParameterNotFound");
        assert!(matches!(err, MotorcortexError::ParameterNotFound(ref p) if p == "root/missing"));
    }

    #[tokio::test]
    async fn request_parameter_tree_without_connect_errors() {
        let req = Request::new();
        let err = req
            .request_parameter_tree()
            .await
            .expect_err("no socket → Connection error");
        assert!(matches!(err, MotorcortexError::Connection(_)));
    }

    #[test]
    fn parameter_tree_is_empty_on_fresh_handle() {
        let req = Request::new();
        let tree = req.parameter_tree();
        let guard = tree.read().unwrap();
        assert!(guard.get_parameter_info("anything").is_none());
    }

    #[test]
    fn parameter_tree_is_shared_across_clones() {
        // Both handles hold the same Arc<RwLock<ParameterTree>>.
        let a = Request::new();
        let b = a.clone();
        assert!(Arc::ptr_eq(&a.parameter_tree(), &b.parameter_tree()));
    }

    #[tokio::test]
    async fn state_observer_sees_disconnect() {
        // After a disconnect the watch should still read Disconnected.
        // `state.changed().await` fires only on a *transition*, and our
        // initial value is already Disconnected — but calling
        // disconnect() flushes a fresh publish through the driver, so
        // the value is still correct.
        let req = Request::new();
        let mut state = req.state();
        req.disconnect().await.unwrap();
        assert_eq!(*state.borrow_and_update(), ConnectionState::Disconnected);
    }
}