nros_rmw/

traits.rs

//! Transport abstraction traits.
//!
//! Defines the backend-agnostic interface that transport implementations
//! (zenoh-pico, XRCE-DDS) must satisfy. The core trait hierarchy is:
//!
//! - [`Session`] — connection lifecycle and handle creation
//! - [`Publisher`] / [`Subscriber`] — pub/sub data transport
//! - [`ServiceServerTrait`] / [`ServiceClientTrait`] — request/reply
//! - [`Rmw`] — top-level factory that creates sessions

use nros_core::{Deserialize, RosMessage, RosService, Serialize};

/// Topic information for pub/sub
#[derive(Debug, Clone)]
pub struct TopicInfo<'a> {
    /// Topic name (e.g., "/chatter")
    pub name: &'a str,
    /// ROS type name (e.g., "std_msgs::msg::dds_::String_")
    pub type_name: &'a str,
    /// Type hash for compatibility checking
    pub type_hash: &'a str,
    /// Domain ID (default: 0)
    pub domain_id: u32,
    /// Node name for liveliness token generation.
    /// `None` means no node association — no liveliness token will be declared.
    pub node_name: Option<&'a str>,
    /// Node namespace for liveliness token generation (default: "/").
    /// In ROS 2, "/" is the root namespace and the standard default.
    pub namespace: &'a str,
    /// Phase 231 (RFC-0038) — receive-buffer size hint, bytes. The executor sets
    /// this from the subscription's `RX_BUF`; a backend may use it to route the
    /// subscription to a size-class receive buffer (zenoh-pico: small vs large).
    /// `0` = unset (backend picks its default). Ignored by backends that don't
    /// size-class their receive storage.
    pub rx_buffer_hint: usize,
}

impl<'a> TopicInfo<'a> {
    /// Create new topic info
    pub const fn new(name: &'a str, type_name: &'a str, type_hash: &'a str) -> Self {
        Self {
            name,
            type_name,
            type_hash,
            domain_id: 0,
            node_name: None,
            namespace: "/",
            rx_buffer_hint: 0,
        }
    }

    /// Create topic info with custom domain ID
    pub const fn with_domain(mut self, domain_id: u32) -> Self {
        self.domain_id = domain_id;
        self
    }

    /// Phase 231 (RFC-0038) — set the receive-buffer size hint (bytes) used by
    /// size-classing backends (zenoh-pico) to route the subscription's receive
    /// buffer to the small or large class.
    pub const fn with_rx_buffer_hint(mut self, hint: usize) -> Self {
        self.rx_buffer_hint = hint;
        self
    }

    /// Set the node name for liveliness token generation
    pub const fn with_node_name(mut self, node_name: &'a str) -> Self {
        self.node_name = Some(node_name);
        self
    }

    /// Set the node namespace for liveliness token generation
    pub const fn with_namespace(mut self, namespace: &'a str) -> Self {
        self.namespace = namespace;
        self
    }
}

/// Service information for service client/server
#[derive(Debug, Clone)]
pub struct ServiceInfo<'a> {
    /// Service name (e.g., "/add_two_ints")
    pub name: &'a str,
    /// ROS service type name (e.g., "example_interfaces::srv::dds_::AddTwoInts_")
    pub type_name: &'a str,
    /// Type hash for compatibility checking
    pub type_hash: &'a str,
    /// Domain ID (default: 0)
    pub domain_id: u32,
    /// Node name for liveliness token generation.
    /// `None` means no node association — no liveliness token will be declared.
    pub node_name: Option<&'a str>,
    /// Node namespace for liveliness token generation (default: "/").
    /// In ROS 2, "/" is the root namespace and the standard default.
    pub namespace: &'a str,
}

/// Action information for action client/server
///
/// Actions in ROS 2 use 5 communication channels:
/// - `send_goal` service: `<action_name>/_action/send_goal`
/// - `cancel_goal` service: `<action_name>/_action/cancel_goal`
/// - `get_result` service: `<action_name>/_action/get_result`
/// - `feedback` topic: `<action_name>/_action/feedback`
/// - `status` topic: `<action_name>/_action/status`
#[derive(Debug, Clone)]
pub struct ActionInfo<'a> {
    /// Action name (e.g., "/fibonacci")
    pub name: &'a str,
    /// ROS action type name (e.g., "example_interfaces::action::dds_::Fibonacci_")
    pub type_name: &'a str,
    /// Type hash for compatibility checking
    pub type_hash: &'a str,
    /// Domain ID (default: 0)
    pub domain_id: u32,
}

impl<'a> ActionInfo<'a> {
    /// Create new action info
    pub const fn new(name: &'a str, type_name: &'a str, type_hash: &'a str) -> Self {
        Self {
            name,
            type_name,
            type_hash,
            domain_id: 0,
        }
    }

    /// Create action info with custom domain ID
    pub const fn with_domain(mut self, domain_id: u32) -> Self {
        self.domain_id = domain_id;
        self
    }

    /// Generate the send_goal service name
    /// Returns: `<action>/_action/send_goal`
    pub fn send_goal_key<const N: usize>(&self) -> heapless::String<N> {
        self.sub_name::<N>("send_goal")
    }

    /// Generate the cancel_goal service name
    /// Returns: `<action>/_action/cancel_goal`
    pub fn cancel_goal_key<const N: usize>(&self) -> heapless::String<N> {
        self.sub_name::<N>("cancel_goal")
    }

    /// Generate the get_result service name
    /// Returns: `<action>/_action/get_result`
    pub fn get_result_key<const N: usize>(&self) -> heapless::String<N> {
        self.sub_name::<N>("get_result")
    }

    /// Generate the feedback topic name
    /// Returns: `<action>/_action/feedback`
    pub fn feedback_key<const N: usize>(&self) -> heapless::String<N> {
        self.sub_name::<N>("feedback")
    }

    /// Generate the status topic name
    /// Returns: `<action>/_action/status`
    pub fn status_key<const N: usize>(&self) -> heapless::String<N> {
        self.sub_name::<N>("status")
    }

    /// Generate a sub-entity name for an action component
    /// Returns: `<action>/_action/<suffix>` (e.g., `fibonacci/_action/send_goal`)
    ///
    /// The caller is responsible for constructing the full key expression
    /// by wrapping this name in a `ServiceInfo` or `TopicInfo` with the
    /// correct sub-service/sub-topic type name.
    fn sub_name<const N: usize>(&self, suffix: &str) -> heapless::String<N> {
        let mut name = heapless::String::new();
        let action_stripped = self.name.trim_matches('/');
        let _ = core::fmt::write(
            &mut name,
            format_args!("/{}/_action/{}", action_stripped, suffix),
        );
        name
    }
}

impl<'a> ServiceInfo<'a> {
    /// Create new service info
    pub const fn new(name: &'a str, type_name: &'a str, type_hash: &'a str) -> Self {
        Self {
            name,
            type_name,
            type_hash,
            domain_id: 0,
            node_name: None,
            namespace: "/",
        }
    }

    /// Create service info with custom domain ID
    pub const fn with_domain(mut self, domain_id: u32) -> Self {
        self.domain_id = domain_id;
        self
    }

    /// Set the node name for liveliness token generation
    pub const fn with_node_name(mut self, node_name: &'a str) -> Self {
        self.node_name = Some(node_name);
        self
    }

    /// Set the node namespace for liveliness token generation
    pub const fn with_namespace(mut self, namespace: &'a str) -> Self {
        self.namespace = namespace;
        self
    }
}

/// Transport error types.
///
/// No longer `Copy` — the `Backend` / `BackendDynamic` variants carry a
/// string diagnostic, which can't be `Copy`. Rust callers that used to
/// copy a `TransportError` value repeatedly now need `.clone()` or
/// `ref` in match arms. C/C++ callers are unaffected — both map
/// `TransportError` to integer codes (`nros_ret_t` / `ErrorCode`)
/// before crossing the FFI boundary.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum TransportError {
    /// Failed to connect to transport
    ConnectionFailed,
    /// Connection was closed
    Disconnected,
    /// Failed to create publisher
    PublisherCreationFailed,
    /// Failed to create subscriber
    SubscriberCreationFailed,
    /// Failed to create service server
    ServiceServerCreationFailed,
    /// Failed to create service client
    ServiceClientCreationFailed,
    /// Failed to publish message
    PublishFailed,
    /// Failed to send service request
    ServiceRequestFailed,
    /// Failed to send service reply
    ServiceReplyFailed,
    /// Serialization error
    SerializationError,
    /// Deserialization error
    DeserializationError,
    /// Buffer too small
    BufferTooSmall,
    /// Incoming message exceeded the static buffer capacity
    MessageTooLarge,
    /// Timeout waiting for message
    Timeout,
    /// Invalid configuration
    InvalidConfig,
    /// Resource (slot, buffer, queue) momentarily unavailable. Retry.
    /// Phase 99: returned by `try_loan` when arena slots are full and
    /// by `try_borrow` when no message is ready (alternative to
    /// `Ok(None)` for backends that prefer the error variant).
    WouldBlock,
    /// Requested allocation exceeds backend capacity. Phase 99:
    /// `try_loan(len)` returns this when `len` > arena slot size.
    TooLarge,
    /// Failed to start background tasks
    TaskStartFailed,
    /// Failed to poll for incoming messages
    PollFailed,
    /// Failed to send keepalive
    KeepaliveFailed,
    /// Failed to send join message
    JoinFailed,
    /// Caller supplied a NULL pointer, an out-of-range value, or an
    /// inconsistent argument combination. Phase 102.1.
    InvalidArgument,
    /// The backend does not implement this operation. Optional
    /// callbacks return this; the runtime then falls back to a
    /// default path. Phase 102.1.
    Unsupported,
    /// Memory allocation failed. Returned by backends on `std` /
    /// `alloc`-equipped platforms when heap allocation fails.
    /// Bare-metal backends generally do not produce this — they
    /// preallocate at session-open time. Phase 102.1.
    BadAlloc,
    /// Publisher and subscriber QoS profiles do not match in a way
    /// the backend cannot reconcile. Phase 102.1.
    IncompatibleQos,
    /// Topic, service, or action name failed validation. Phase 102.1.
    TopicNameInvalid,
    /// A request referenced a node that does not exist in this
    /// session. Phase 102.1.
    NodeNameNonExistent,
    /// The backend does not support loaned messages on this entity,
    /// or the loan slot is currently in use. Phase 102.1.
    LoanNotSupported,
    /// No data was available on a non-blocking receive. Distinct
    /// from `Timeout`: fires immediately, not after a bounded wait.
    /// Phase 102.1.
    NoData,
    /// Phase 115.A.2 — caller passed a versioned vtable struct
    /// (e.g. `NrosTransportOps`) with an `abi_version` the runtime
    /// doesn't know. Maps to `NROS_RMW_RET_INCOMPATIBLE_ABI` at
    /// the C boundary.
    IncompatibleAbi,
    /// Backend-specific error with a `'static` diagnostic string.
    ///
    /// Useful for zenoh-pico / XRCE-DDS return codes that map to a
    /// fixed set of known messages. `no_std`-compatible.
    Backend(&'static str),
    /// Backend-specific error with an owned diagnostic string.
    ///
    /// Available only with the `alloc` feature. Use this when the
    /// diagnostic is formatted at runtime (e.g. from a C error code
    /// plus a socket address).
    #[cfg(feature = "alloc")]
    BackendDynamic(alloc::string::String),
}

/// QoS history policy
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum QosHistoryPolicy {
    /// Keep last N messages (where N is defined in QosSettings)
    #[default]
    KeepLast,
    /// Keep all messages (up to resource limits)
    KeepAll,
}

/// QoS reliability policy
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum QosReliabilityPolicy {
    /// Reliable delivery (retransmit if needed).
    ///
    /// Default — matches ROS 2 `rmw_qos_profile_default` and the
    /// `QosSettings::default()` / `QOS_PROFILE_DEFAULT` aggregates.
    #[default]
    Reliable,
    /// Best-effort delivery (no retransmits)
    BestEffort,
}

/// QoS durability policy
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum QosDurabilityPolicy {
    /// Messages are discarded when subscriber disconnects
    #[default]
    Volatile,
    /// Messages are persisted for late-joining subscribers
    TransientLocal,
}

/// QoS liveliness policy. Matches DDS `LIVELINESS` semantics.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
#[repr(u8)]
pub enum QosLivelinessPolicy {
    /// No liveliness assertion or tracking. Default for entities
    /// that don't care about liveliness.
    #[default]
    None = 0,
    /// Backend's keepalive task asserts liveliness automatically.
    Automatic = 1,
    /// Application calls `assert_liveliness()` per topic explicitly.
    ManualByTopic = 2,
    /// Application calls `assert_liveliness()` at the node level.
    ManualByNode = 3,
}

/// Phase 211.H — which side of a topic a [`QosOverride`] targets.
/// Mirrors the `<role>` segment of a ROS 2
/// `qos_overrides.<topic>.<role>.<policy>` launch parameter.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum QosOverrideRole {
    /// `qos_overrides.<topic>.publisher.*`
    Publisher,
    /// `qos_overrides.<topic>.subscription.*`
    Subscription,
}

/// Phase 211.H — a single policy value a [`QosOverride`] sets. A typed enum
/// (not a string) so the codegen that bakes these from the plan catches an
/// unknown policy / mistyped value at generation time rather than silently
/// no-op-ing at runtime.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum QosOverrideValue {
    /// `.reliability` → Reliable / BestEffort.
    Reliability(QosReliabilityPolicy),
    /// `.durability` → Volatile / TransientLocal.
    Durability(QosDurabilityPolicy),
    /// `.history` → KeepLast / KeepAll.
    History(QosHistoryPolicy),
    /// `.depth` → KeepLast depth.
    Depth(u32),
}

/// Phase 211.H — one per-topic QoS override, lowered from a ROS 2
/// `qos_overrides.<topic>.<role>.<policy>` launch parameter by the planner and
/// baked into a `&'static [QosOverride]` table by the entry codegen. The node
/// folds the matching entries into the entity's [`QosSettings`] at
/// `create_publisher` / `create_subscription` time (setup-time, single
/// linear scan, no alloc), *before* the backend-compat `validate_against` —
/// so an override the active RMW can't honour still errors loudly, never a
/// silent downgrade.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct QosOverride {
    /// The resolved (remapped) topic name the override targets, e.g.
    /// `"/chatter"`. Matched exactly against the entity's topic.
    pub topic: &'static str,
    /// Publisher or subscription side.
    pub role: QosOverrideRole,
    /// The policy + value to set.
    pub value: QosOverrideValue,
}

/// Full DDS-shaped QoS profile. Matches the field set of upstream
/// `rmw_qos_profile_t`.
///
/// Backends advertise per-policy support via
/// [`Session::supported_qos_policies`]; entities created with a
/// profile the active backend can't honour return
/// [`TransportError::IncompatibleQos`] synchronously at create time
/// — no silent downgrade.
///
/// Zero-valued time-window fields ("off") mean infinite — the policy
/// is effectively disabled for the entity.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct QosSettings {
    /// History policy
    pub history: QosHistoryPolicy,
    /// Reliability policy
    pub reliability: QosReliabilityPolicy,
    /// Durability policy
    pub durability: QosDurabilityPolicy,
    /// Liveliness policy
    pub liveliness_kind: QosLivelinessPolicy,
    /// History depth (only used if history is KeepLast)
    pub depth: u32,
    /// Subscriber max-inter-arrival / publisher offered-rate, ms.
    /// `0` = infinite (no deadline check).
    pub deadline_ms: u32,
    /// Sample expiry, ms. Subscribers filter samples older than this.
    /// `0` = infinite (no expiry).
    pub lifespan_ms: u32,
    /// Liveliness lease, ms. `0` = infinite.
    pub liveliness_lease_ms: u32,
    /// If `true`, topic-name encoding skips the `/rt/` ROS prefix.
    pub avoid_ros_namespace_conventions: bool,
}

impl Default for QosSettings {
    fn default() -> Self {
        Self::QOS_PROFILE_DEFAULT
    }
}

impl QosSettings {
    /// Phase 211.H — fold the plan's `qos_overrides` matching `topic` + `role`
    /// into this profile, returning the overridden profile. Setup-time only
    /// (called from `create_publisher`/`create_subscription`): a single linear
    /// scan over the baked `&'static` table, no alloc, RT-safe. Later entries
    /// win on a duplicate `(topic, role, policy)` (last-write), matching the
    /// planner's sorted, de-conflicted emit. Non-matching entries are ignored,
    /// so passing the whole node table to every entity is cheap + correct.
    #[must_use]
    pub fn apply_overrides(
        mut self,
        topic: &str,
        role: QosOverrideRole,
        overrides: &[QosOverride],
    ) -> Self {
        for ovr in overrides {
            if ovr.topic == topic && ovr.role == role {
                match ovr.value {
                    QosOverrideValue::Reliability(r) => self.reliability = r,
                    QosOverrideValue::Durability(d) => self.durability = d,
                    QosOverrideValue::History(h) => self.history = h,
                    QosOverrideValue::Depth(d) => self.depth = d,
                }
            }
        }
        self
    }
}

impl QosSettings {
    /// Internal const builder. Extended-policy fields default to
    /// "off" (zero) and `liveliness_kind = Automatic` (the upstream
    /// `rmw_qos_profile_default` choice).
    const fn build(
        reliability: QosReliabilityPolicy,
        durability: QosDurabilityPolicy,
        history: QosHistoryPolicy,
        depth: u32,
    ) -> Self {
        Self {
            history,
            reliability,
            durability,
            liveliness_kind: QosLivelinessPolicy::Automatic,
            depth,
            deadline_ms: 0,
            lifespan_ms: 0,
            liveliness_lease_ms: 0,
            avoid_ros_namespace_conventions: false,
        }
    }

    /// Create new QoS settings with defaults (matches `QOS_PROFILE_DEFAULT`:
    /// Reliable, Volatile, KeepLast(10)).
    pub const fn new() -> Self {
        Self::QOS_PROFILE_DEFAULT
    }

    /// Best-effort QoS (for real-time)
    pub const BEST_EFFORT: Self = Self::build(
        QosReliabilityPolicy::BestEffort,
        QosDurabilityPolicy::Volatile,
        QosHistoryPolicy::KeepLast,
        1,
    );

    /// Reliable QoS
    pub const RELIABLE: Self = Self::build(
        QosReliabilityPolicy::Reliable,
        QosDurabilityPolicy::Volatile,
        QosHistoryPolicy::KeepLast,
        10,
    );

    /// System default QoS profile (matches rmw_qos_profile_system_default)
    pub const QOS_PROFILE_SYSTEM_DEFAULT: Self = Self::build(
        QosReliabilityPolicy::Reliable,
        QosDurabilityPolicy::Volatile,
        QosHistoryPolicy::KeepLast,
        1,
    );

    /// Default QoS profile (matches rmw_qos_profile_default)
    pub const QOS_PROFILE_DEFAULT: Self = Self::build(
        QosReliabilityPolicy::Reliable,
        QosDurabilityPolicy::Volatile,
        QosHistoryPolicy::KeepLast,
        10,
    );

    /// Sensor data QoS profile (matches rmw_qos_profile_sensor_data)
    pub const QOS_PROFILE_SENSOR_DATA: Self = Self::build(
        QosReliabilityPolicy::BestEffort,
        QosDurabilityPolicy::Volatile,
        QosHistoryPolicy::KeepLast,
        5,
    );

    /// Services default QoS profile (matches rmw_qos_profile_services_default)
    pub const QOS_PROFILE_SERVICES_DEFAULT: Self = Self::build(
        QosReliabilityPolicy::Reliable,
        QosDurabilityPolicy::Volatile,
        QosHistoryPolicy::KeepLast,
        10,
    );

    /// Parameters QoS profile (matches rmw_qos_profile_parameters)
    pub const QOS_PROFILE_PARAMETERS: Self = Self::build(
        QosReliabilityPolicy::Reliable,
        QosDurabilityPolicy::TransientLocal,
        QosHistoryPolicy::KeepLast,
        1000,
    );

    /// Clock QoS profile - same as sensor data but with depth 1
    pub const QOS_PROFILE_CLOCK: Self = Self::build(
        QosReliabilityPolicy::BestEffort,
        QosDurabilityPolicy::Volatile,
        QosHistoryPolicy::KeepLast,
        1,
    );

    /// Parameter events QoS profile (matches rmw_qos_profile_parameter_events)
    pub const QOS_PROFILE_PARAMETER_EVENTS: Self = Self::build(
        QosReliabilityPolicy::Reliable,
        QosDurabilityPolicy::Volatile,
        QosHistoryPolicy::KeepAll,
        0, // Not used with KeepAll
    );

    /// Action status default QoS profile (matches rcl_action_qos_profile_status_default)
    pub const QOS_PROFILE_ACTION_STATUS_DEFAULT: Self = Self::build(
        QosReliabilityPolicy::Reliable,
        QosDurabilityPolicy::TransientLocal,
        QosHistoryPolicy::KeepLast,
        1,
    );

    /// PX4 companion QoS profile (Phase 233 / RFC-0039 Track B). Matches the
    /// QoS PX4's `uxrce_dds_client` uses on `/fmu/out/*` and `/fmu/in/*` —
    /// `BEST_EFFORT` + `VOLATILE` + `KEEP_LAST(1)`. A nano-ros node talking to
    /// the same `MicroXRCEAgent` must use this (a reliable or
    /// `TRANSIENT_LOCAL` reader will not match PX4's volatile best-effort
    /// writers). Verified against real PX4 SITL (`nros-px4-sitl-test`):
    /// `TRANSIENT_LOCAL` durability silently fails to match `/fmu/out/*`.
    /// Adjust depth via `.keep_last(n)` for higher-rate streams.
    pub const QOS_PROFILE_PX4: Self = Self::build(
        QosReliabilityPolicy::BestEffort,
        QosDurabilityPolicy::Volatile,
        QosHistoryPolicy::KeepLast,
        1,
    );

    // --- Static constructor methods (matching rclrs API) ---

    /// Get the default QoS profile for ordinary topics
    pub const fn topics_default() -> Self {
        Self::QOS_PROFILE_DEFAULT
    }

    /// The PX4 companion QoS profile ([`QOS_PROFILE_PX4`](Self::QOS_PROFILE_PX4))
    /// — use for `/fmu/out/*` subscriptions and `/fmu/in/*` publications against
    /// a `MicroXRCEAgent`.
    pub const fn px4() -> Self {
        Self::QOS_PROFILE_PX4
    }

    /// Get the default QoS profile for sensor data topics
    pub const fn sensor_data_default() -> Self {
        Self::QOS_PROFILE_SENSOR_DATA
    }

    /// Get the default QoS profile for services
    pub const fn services_default() -> Self {
        Self::QOS_PROFILE_SERVICES_DEFAULT
    }

    /// Get the default QoS profile for parameter services
    pub const fn parameters_default() -> Self {
        Self::QOS_PROFILE_PARAMETERS
    }

    /// Get the default QoS profile for parameter events
    pub const fn parameter_events_default() -> Self {
        Self::QOS_PROFILE_PARAMETER_EVENTS
    }

    /// Get the system default QoS profile
    pub const fn system_default() -> Self {
        Self::QOS_PROFILE_SYSTEM_DEFAULT
    }

    /// Get the default QoS profile for action status topics
    pub const fn action_status_default() -> Self {
        Self::QOS_PROFILE_ACTION_STATUS_DEFAULT
    }

    /// Get the default QoS profile for clock topics
    pub const fn clock_default() -> Self {
        Self::QOS_PROFILE_CLOCK
    }

    // --- Builder methods ---

    /// Set history to keep last N messages
    pub const fn keep_last(mut self, depth: u32) -> Self {
        self.history = QosHistoryPolicy::KeepLast;
        self.depth = depth;
        self
    }

    /// Set history to keep all messages
    pub const fn keep_all(mut self) -> Self {
        self.history = QosHistoryPolicy::KeepAll;
        self
    }

    /// Set reliability to reliable
    pub const fn reliable(mut self) -> Self {
        self.reliability = QosReliabilityPolicy::Reliable;
        self
    }

    /// Set reliability to best-effort
    pub const fn best_effort(mut self) -> Self {
        self.reliability = QosReliabilityPolicy::BestEffort;
        self
    }

    /// Set durability to volatile
    pub const fn volatile(mut self) -> Self {
        self.durability = QosDurabilityPolicy::Volatile;
        self
    }

    /// Set durability to transient local
    pub const fn transient_local(mut self) -> Self {
        self.durability = QosDurabilityPolicy::TransientLocal;
        self
    }

    /// Set reliability policy explicitly
    pub const fn reliability(mut self, policy: QosReliabilityPolicy) -> Self {
        self.reliability = policy;
        self
    }

    /// Set durability policy explicitly
    pub const fn durability(mut self, policy: QosDurabilityPolicy) -> Self {
        self.durability = policy;
        self
    }

    /// Set history policy explicitly
    pub const fn history(mut self, policy: QosHistoryPolicy) -> Self {
        self.history = policy;
        self
    }

    /// Set history depth explicitly
    pub const fn depth(mut self, depth: u32) -> Self {
        self.depth = depth;
        self
    }

    /// Get history depth (for backwards compatibility)
    pub const fn history_depth(&self) -> u8 {
        if self.depth > 255 {
            255
        } else {
            self.depth as u8
        }
    }
}

/// Transport session configuration
#[derive(Debug, Clone)]
pub struct TransportConfig<'a> {
    /// Peer locator (e.g., "tcp/192.168.1.1:7447" or "serial//dev/ttyUSB0#baudrate=115200")
    pub locator: Option<&'a str>,
    /// Session mode: client, peer, or router
    pub mode: SessionMode,
    /// Additional transport properties (key-value pairs)
    ///
    /// These are passed through to the underlying transport backend.
    /// For zenoh-pico, recognized keys include:
    /// - `"multicast_scouting"` - Enable/disable multicast scouting (`"true"` or `"false"`)
    /// - `"scouting_timeout_ms"` - Scouting timeout in milliseconds
    /// - `"multicast_locator"` - Multicast group address
    /// - `"listen"` - Listen endpoint (e.g., `"tcp/0.0.0.0:0"`)
    /// - `"add_timestamp"` - Add timestamps to messages (`"true"` or `"false"`)
    pub properties: &'a [(&'a str, &'a str)],
}

impl Default for TransportConfig<'_> {
    fn default() -> Self {
        Self {
            locator: None,
            mode: SessionMode::Client,
            properties: &[],
        }
    }
}

/// Middleware-agnostic session configuration.
///
/// `RmwConfig` provides a uniform interface that any RMW backend can
/// interpret. Backends map the universal fields to their own connection
/// parameters and interpret `properties` for anything backend-specific.
///
/// # Examples
///
/// ```
/// use nros_rmw::{RmwConfig, SessionMode};
///
/// let config = RmwConfig {
///     locator: "tcp/192.168.1.1:7447",
///     mode: SessionMode::Client,
///     domain_id: 0,
///     node_name: "talker",
///     namespace: "",
///     properties: &[],
/// };
/// ```
#[derive(Debug, Clone, Copy)]
pub struct RmwConfig<'a> {
    /// Middleware-specific connection string.
    ///
    /// - zenoh: `"tcp/192.168.1.1:7447"` or `"udp/224.0.0.224:7447"`
    /// - XRCE-DDS: `"udp/192.168.1.1:2019"`
    pub locator: &'a str,
    /// Session mode (zenoh: client/peer; XRCE-DDS: always client)
    pub mode: SessionMode,
    /// ROS 2 domain ID (maps to DDS domain or zenoh key prefix)
    pub domain_id: u32,
    /// Node name (e.g., `"talker"`)
    pub node_name: &'a str,
    /// Node namespace (e.g., `""` or `"/ns1"`)
    pub namespace: &'a str,
    /// Backend-specific key/value properties.
    ///
    /// Uniform escape hatch for backend-specific tuning that doesn't fit
    /// the universal fields above. Each backend documents the keys it
    /// understands; unknown keys are ignored. Passing `&[]` is always
    /// valid.
    ///
    /// Examples:
    /// - zenoh: `"tls.root_ca"`, `"scouting.multicast.enabled"`
    /// - XRCE-DDS: `"agent_port"`, `"client_key"`
    pub properties: &'a [(&'a str, &'a str)],
}

impl Default for RmwConfig<'_> {
    fn default() -> Self {
        Self {
            locator: "tcp/127.0.0.1:7447",
            mode: SessionMode::Client,
            domain_id: 0,
            node_name: "node",
            namespace: "",
            properties: &[],
        }
    }
}

/// Locator transport protocol
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum LocatorProtocol {
    /// TCP transport (e.g., "tcp/127.0.0.1:7447")
    Tcp,
    /// UDP transport (e.g., "udp/192.168.1.50:2019" — common for XRCE-DDS)
    Udp,
    /// Serial/UART transport (e.g., "serial//dev/ttyUSB0#baudrate=115200")
    Serial,
    /// Unknown protocol
    Unknown,
}

/// Parse the protocol from a locator string
pub fn locator_protocol(locator: &str) -> LocatorProtocol {
    if locator.starts_with("tcp/") {
        LocatorProtocol::Tcp
    } else if locator.starts_with("udp/") {
        LocatorProtocol::Udp
    } else if locator.starts_with("serial/") {
        LocatorProtocol::Serial
    } else {
        LocatorProtocol::Unknown
    }
}

/// Validate a locator string format.
///
/// Returns `Ok(())` if the locator is well-formed, or an error message describing
/// the problem. This provides early feedback before zenoh-pico or XRCE-DDS rejects
/// a bad locator.
///
/// Supported formats:
/// - TCP: `tcp/<host>:<port>` (e.g., `tcp/127.0.0.1:7447`)
/// - UDP: `udp/<host>:<port>` (e.g., `udp/192.168.1.50:2019`)
/// - Serial: `serial/<device>#baudrate=<rate>` (e.g., `serial//dev/ttyUSB0#baudrate=115200`)
pub fn validate_locator(locator: &str) -> Result<(), &'static str> {
    match locator_protocol(locator) {
        LocatorProtocol::Tcp => {
            let rest = &locator[4..]; // skip "tcp/"
            if !rest.contains(':') {
                return Err("TCP locator must contain host:port (e.g., tcp/127.0.0.1:7447)");
            }
            Ok(())
        }
        LocatorProtocol::Udp => {
            let rest = &locator[4..]; // skip "udp/"
            if !rest.contains(':') {
                return Err("UDP locator must contain host:port (e.g., udp/192.168.1.50:2019)");
            }
            Ok(())
        }
        LocatorProtocol::Serial => {
            let rest = &locator[7..]; // skip "serial/"
            if rest.is_empty() {
                return Err(
                    "serial locator must specify device (e.g., serial//dev/ttyUSB0#baudrate=115200)",
                );
            }
            if !rest.contains("#baudrate=") {
                return Err(
                    "serial locator must include #baudrate=RATE (e.g., serial//dev/ttyUSB0#baudrate=115200)",
                );
            }
            // Validate baudrate is numeric
            if let Some(baud_str) = rest.split("#baudrate=").nth(1) {
                let baud_str = baud_str.split('#').next().unwrap_or(baud_str);
                if baud_str.parse::<u32>().is_err() {
                    return Err("serial baudrate must be a number");
                }
            }
            Ok(())
        }
        LocatorProtocol::Unknown => {
            Err("unknown locator protocol (expected tcp/, udp/, or serial/)")
        }
    }
}

/// Session mode
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub enum SessionMode {
    /// Connect as client to a router
    #[default]
    Client,
    /// Connect as peer for peer-to-peer communication
    Peer,
}

/// Transport session trait — the per-process anchor an RMW backend
/// gives to the executor.
///
/// # Threading
///
/// `&mut self` on every method means the executor serialises all
/// session calls onto a single thread. A backend may rely on this
/// — no internal locking is required for `create_*` / `close` /
/// `drive_io`. **Publisher / subscriber / service handles created
/// from the session, however, are typically used from worker
/// threads** and must carry their own synchronisation (see the
/// [`Publisher`] / [`Subscriber`] trait docs).
///
/// # Calling pattern
///
/// 1. Open the session (backend-specific factory; not on this trait).
/// 2. `create_*` for every entity at startup. Creating entities mid-
///    flight after `drive_io` has run is allowed but not common.
/// 3. The executor calls `drive_io` periodically. Worker threads
///    publish / receive in parallel.
/// 4. `close` once at shutdown. Entities must be dropped first.
pub trait Session {
    /// Error type for this session
    type Error;
    /// Publisher handle type
    type PublisherHandle;
    /// Subscriber handle type
    type SubscriberHandle;
    /// Service server handle type
    type ServiceServerHandle;
    /// Service client handle type
    type ServiceClientHandle;

    /// Create a publisher bound to this session.
    ///
    /// May allocate transport resources (zenoh declarations, DDS
    /// writers). Returns a handle that outlives the call but not the
    /// session — drop the handle before `close()`.
    fn create_publisher(
        &mut self,
        topic: &TopicInfo,
        qos: QosSettings,
    ) -> Result<Self::PublisherHandle, Self::Error>;

    /// Create a subscriber bound to this session.
    ///
    /// Subscribers may start receiving immediately after creation if
    /// the transport supports late-joining publishers. Late messages
    /// are buffered up to the QoS depth.
    fn create_subscriber(
        &mut self,
        topic: &TopicInfo,
        qos: QosSettings,
    ) -> Result<Self::SubscriberHandle, Self::Error>;

    /// Create a service server bound to this session. Replies are
    /// matched to requests by the sequence number returned from
    /// [`ServiceServerTrait::try_recv_request`].
    ///
    /// `qos` is applied to both the request and reply endpoints (a
    /// service is two DDS topics; rmw uses one profile for both). The
    /// default is [`QosSettings::services_default`]
    /// (RELIABLE+VOLATILE+KEEP_LAST(10)).
    fn create_service_server(
        &mut self,
        service: &ServiceInfo,
        qos: QosSettings,
    ) -> Result<Self::ServiceServerHandle, Self::Error>;

    /// Create a service client bound to this session.
    ///
    /// `qos` is applied to both the request and reply endpoints (a
    /// service is two DDS topics; rmw uses one profile for both). The
    /// default is [`QosSettings::services_default`]
    /// (RELIABLE+VOLATILE+KEEP_LAST(10)).
    fn create_service_client(
        &mut self,
        service: &ServiceInfo,
        qos: QosSettings,
    ) -> Result<Self::ServiceClientHandle, Self::Error>;

    /// Close the session, releasing transport resources. All entity
    /// handles created from this session must already be dropped.
    fn close(&mut self) -> Result<(), Self::Error>;

    /// Drive transport I/O (poll network, dispatch callbacks).
    ///
    /// Both zenoh-pico and XRCE-DDS are pull-based: they require the
    /// application to periodically call this method to read from the
    /// network socket and dispatch incoming messages to subscriber
    /// buffers.
    ///
    /// `timeout_ms` is the maximum time to wait for data (0 = non-blocking;
    /// negative values mean "block indefinitely" — see Phase 84.D7 for the
    /// planned migration to `core::time::Duration`).
    ///
    /// **Required**. There is no default body — both shipped backends
    /// (zenoh and XRCE) must drive I/O, and a silent no-op default was a
    /// trap for third-party implementers. If your backend genuinely
    /// receives data via OS callbacks (push-based) and has nothing to do
    /// here, return `Ok(())` explicitly.
    fn drive_io(&mut self, timeout_ms: i32) -> Result<(), Self::Error>;

    /// Phase 109 — report which QoS policies the active backend
    /// honours. The runtime validates requested QoS against this mask
    /// at entity-create time and returns
    /// [`TransportError::IncompatibleQos`] if the requested profile
    /// includes a policy the backend can't enforce. **No silent
    /// downgrade.**
    ///
    /// Default returns [`QosPolicyMask::CORE`] — reliability +
    /// durability VOLATILE + history + depth. Backends override per
    /// supported policy.
    fn supported_qos_policies(&self) -> QosPolicyMask {
        QosPolicyMask::CORE
    }

    /// Phase 110.0 — backend's next internal-event deadline in
    /// milliseconds from now (lease keepalive, heartbeat, reader
    /// ACK-NACK timeout, etc.).
    ///
    /// The executor caps its `drive_io` timeout against
    /// `min(user_timeout, timer_deadline, this)` so quiet links don't
    /// wake early, see no user-visible work, and round-trip back into
    /// `drive_io`. Returns `None` when the backend has no internal
    /// deadlines or chooses not to expose them.
    ///
    /// Default `None` keeps existing backends working unchanged; opt-in
    /// per backend.
    fn next_deadline_ms(&self) -> Option<u32> {
        None
    }

    /// Phase 124.B.1 — install (or clear, when `cb.is_none()`) the
    /// executor wake callback. The runtime calls this once per
    /// session after `open` with `cb` pointing at a runtime-owned
    /// function and `ctx` pointing at the executor's wake state.
    /// The backend stores `(cb, ctx)` in its per-session state and
    /// calls `cb(ctx)` whenever its transport notification path
    /// fires (datagram arrival, condvar wake, etc.) — the runtime
    /// cb does flag-write + condvar-signal atomically, so a
    /// `spin_once` blocked on the wake condvar resumes immediately
    /// instead of waiting for the next poll iteration.
    ///
    /// # Safety
    ///
    /// When `cb` is `Some`, `ctx` must remain valid until the callback is
    /// cleared or the session is closed. The backend may invoke `cb(ctx)` from
    /// its transport notification path.
    ///
    /// Default body: ignore the call. Poll-only backends (XRCE,
    /// bare-metal) leave the default in place; the executor still
    /// drains them on its deadline-bound cv-wait boundary.
    unsafe fn set_wake_callback(
        &mut self,
        cb: Option<unsafe extern "C" fn(ctx: *mut core::ffi::c_void)>,
        ctx: *mut core::ffi::c_void,
    ) {
        let _ = (cb, ctx);
    }

    /// Phase 130.4 — does this backend actually honour
    /// [`set_wake_callback`]?
    ///
    /// `true` means the backend installs the callback and will
    /// fire it from its async notify path (worker thread, ISR,
    /// signalfd, …). `false` (the default) means
    /// `set_wake_callback` was a no-op — the executor must drive
    /// I/O for the caller's full timeout because no async wake
    /// will pre-empt it.
    ///
    /// The executor uses this to choose between the wake-primitive
    /// wait (`NodeWake::wait_ms` / `std::Condvar::wait_timeout_while`)
    /// and a direct `drive_io(timeout_ms)`. Poll-only backends
    /// (XRCE-DDS-Client, bare-metal smoltcp) return `false`;
    /// event-driven backends (zenoh-pico with an RX task that
    /// invokes the callback on packet arrival) return `true`.
    ///
    /// [`set_wake_callback`]: Self::set_wake_callback
    fn supports_wake_callback(&self) -> bool {
        false
    }

    /// Phase 124.F.1 — session-level connectivity probe.
    ///
    /// Sends a wire-level round-trip probe and waits up to
    /// `timeout_ms`. `Ok(())` on reply, `Err(TransportError::Timeout)`
    /// on no reply, `Err(TransportError::Unsupported)` when the
    /// backend can't probe (DDS without participant introspection).
    /// Lesson from micro-ROS's `rmw_uros_ping_agent`.
    ///
    /// Default body: `Err(Unsupported)`. Backends with a native
    /// ping API (zenoh: `z_send_ping`; XRCE:
    /// `uxr_ping_agent_session_until_timeout`) opt in by overriding.
    fn ping_session(&mut self, timeout_ms: i32) -> Result<(), Self::Error>
    where
        Self::Error: From<TransportError>,
    {
        let _ = timeout_ms;
        Err(TransportError::Unsupported.into())
    }
}

/// Bitmask of QoS policies a backend can honour. See
/// [`Session::supported_qos_policies`].
///
/// `CORE` covers the policies every nano-ros backend implements:
/// reliability, durability=VOLATILE, history, depth. Backends opt
/// into additional policies by OR-ing the relevant flags.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct QosPolicyMask(pub u32);

impl QosPolicyMask {
    pub const RELIABILITY: Self = Self(1 << 0);
    pub const DURABILITY_VOLATILE: Self = Self(1 << 1);
    pub const DURABILITY_TRANSIENT_LOCAL: Self = Self(1 << 2);
    pub const HISTORY: Self = Self(1 << 3);
    pub const DEPTH: Self = Self(1 << 4);
    pub const DEADLINE: Self = Self(1 << 5);
    pub const LIFESPAN: Self = Self(1 << 6);
    pub const LIVELINESS_AUTOMATIC: Self = Self(1 << 7);
    pub const LIVELINESS_MANUAL_BY_TOPIC: Self = Self(1 << 8);
    pub const LIVELINESS_MANUAL_BY_NODE: Self = Self(1 << 9);
    pub const LIVELINESS_LEASE: Self = Self(1 << 10);
    pub const AVOID_ROS_NAMESPACE_CONVENTIONS: Self = Self(1 << 11);

    /// Policies every nano-ros backend implements.
    pub const CORE: Self =
        Self(Self::RELIABILITY.0 | Self::DURABILITY_VOLATILE.0 | Self::HISTORY.0 | Self::DEPTH.0);

    /// `true` if `self` contains every policy in `other`.
    pub const fn contains(self, other: Self) -> bool {
        self.0 & other.0 == other.0
    }

    /// Bitwise OR of two masks.
    pub const fn union(self, other: Self) -> Self {
        Self(self.0 | other.0)
    }
}

impl core::ops::BitOr for QosPolicyMask {
    type Output = Self;
    fn bitor(self, rhs: Self) -> Self {
        self.union(rhs)
    }
}

impl core::ops::BitOrAssign for QosPolicyMask {
    fn bitor_assign(&mut self, rhs: Self) {
        self.0 |= rhs.0;
    }
}

impl QosSettings {
    /// Compute the set of QoS policies actually requested by this profile.
    ///
    /// Zero-valued time fields and `LivelinessKind::None` count as "not
    /// requesting" the corresponding policy — the cheap default. The
    /// `CORE` bits (reliability, durability=VOLATILE, history, depth)
    /// are always present because every nano-ros backend honours them.
    pub fn required_policies(&self) -> QosPolicyMask {
        let mut mask = QosPolicyMask::CORE;
        if self.durability == QosDurabilityPolicy::TransientLocal {
            mask = QosPolicyMask(
                (mask.0 & !QosPolicyMask::DURABILITY_VOLATILE.0)
                    | QosPolicyMask::DURABILITY_TRANSIENT_LOCAL.0,
            );
        }
        if self.deadline_ms != 0 {
            mask |= QosPolicyMask::DEADLINE;
        }
        if self.lifespan_ms != 0 {
            mask |= QosPolicyMask::LIFESPAN;
        }
        match self.liveliness_kind {
            QosLivelinessPolicy::None => {}
            QosLivelinessPolicy::Automatic => mask |= QosPolicyMask::LIVELINESS_AUTOMATIC,
            QosLivelinessPolicy::ManualByTopic => mask |= QosPolicyMask::LIVELINESS_MANUAL_BY_TOPIC,
            QosLivelinessPolicy::ManualByNode => mask |= QosPolicyMask::LIVELINESS_MANUAL_BY_NODE,
        }
        if self.liveliness_lease_ms != 0 {
            mask |= QosPolicyMask::LIVELINESS_LEASE;
        }
        if self.avoid_ros_namespace_conventions {
            mask |= QosPolicyMask::AVOID_ROS_NAMESPACE_CONVENTIONS;
        }
        mask
    }

    /// Returns `Err(TransportError::IncompatibleQos)` if any policy this
    /// profile requires is missing from the backend's `supported` mask.
    /// Used at entity-create time to enforce the **no silent
    /// degradation** contract.
    pub fn validate_against(&self, supported: QosPolicyMask) -> Result<(), TransportError> {
        if supported.contains(self.required_policies()) {
            Ok(())
        } else {
            Err(TransportError::IncompatibleQos)
        }
    }
}

/// Publisher trait for sending messages.
///
/// # Threading
///
/// `&self` on `publish_raw` — implementors must allow concurrent
/// publishes from multiple threads. Internal locking (or lock-free
/// queues) is the backend's responsibility.
///
/// # Buffer ownership
///
/// `data` in `publish_raw` is borrowed for the duration of the call.
/// The backend must either send it inline or copy into its own
/// buffer before returning — the slice is invalid after the call.
///
/// # Blocking
///
/// `publish_raw` is expected to be non-blocking on best-effort QoS
/// and bounded-blocking on reliable QoS (waiting for outbound queue
/// space). Backends should *not* block waiting for ack from a
/// matched subscriber.
pub trait Publisher {
    /// Error type for publish operations
    type Error;

    /// Publish a CDR-serialised message.
    ///
    /// Returns once the message has been handed to the transport
    /// (queued or fired-and-forgotten depending on QoS). Does **not**
    /// wait for delivery.
    fn publish_raw(&self, data: &[u8]) -> Result<(), Self::Error>;

    /// Phase 128.F.4 — publish with an opaque attachment block.
    ///
    /// `attachment` rides alongside the payload at the wire layer.
    /// Receivers can read it back via
    /// [`Subscriber::try_recv_raw_with_attachment`].
    ///
    /// Primary use case: cross-RMW bridges stamp a `bridge_origin`
    /// tag (the source backend's RMW name) so a paired return
    /// bridge can deterministically drop echoed frames.
    ///
    /// Default body delegates to [`publish_raw`](Self::publish_raw)
    /// and discards the attachment — backends that do not natively
    /// carry attachments (XRCE today, DDS without a user-data hook)
    /// see no change. Backends with native attachment support
    /// (zenoh-pico's `z_publisher_put_options::attachment`,
    /// Cyclone DDS user-data) override to write the bytes onto the
    /// wire.
    fn publish_raw_with_attachment(
        &self,
        data: &[u8],
        _attachment: &[u8],
    ) -> Result<(), Self::Error> {
        self.publish_raw(data)
    }

    /// Phase 124.E.1 — streamed publish.
    ///
    /// `size_cb` reports the total payload length once; `chunk_cb`
    /// fills the slot in chunks. Saves the per-publisher staging
    /// buffer when the message is large enough to dominate the
    /// device's `.bss`.
    ///
    /// Default body — the **staging-buffer fallback** (124.E.2).
    /// Asks `size_cb` for the total length, fills a stack-allocated
    /// `[u8; NROS_MAX_STREAM_CHUNK]` via `chunk_cb`, then forwards
    /// to `publish_raw`. Returns `Err(BufferTooSmall)` if the total
    /// exceeds the stack cap so the caller can drop back to a
    /// regular `publish_raw` with a heap-sized buffer.
    ///
    /// Concrete backends opt in by overriding to stream straight
    /// into the network buffer (zenoh: write into the zenoh-pico
    /// outbound buffer; XRCE: micro-CDR streaming APIs).
    ///
    /// # Safety
    ///
    /// The caller must ensure `user_ctx` is valid for every invocation of
    /// `size_cb` and `chunk_cb` during this call, and that both callbacks obey
    /// their out-pointer contracts.
    ///
    /// `size_cb` and `chunk_cb` may be called from the same thread
    /// that called `publish_streamed`. Backends MUST NOT defer the
    /// calls past the function return; the caller's `user_ctx`
    /// pointer is only guaranteed valid for the duration of the
    /// call.
    unsafe fn publish_streamed(
        &self,
        size_cb: unsafe extern "C" fn(out_total_len: *mut usize, user_ctx: *mut core::ffi::c_void),
        chunk_cb: unsafe extern "C" fn(
            out_buf: *mut u8,
            cap: usize,
            out_written: *mut usize,
            user_ctx: *mut core::ffi::c_void,
        ),
        user_ctx: *mut core::ffi::c_void,
    ) -> Result<(), Self::Error>
    where
        Self::Error: From<TransportError>,
    {
        /// Default staging-buffer cap. Stack-allocated, so embedded
        /// callers don't pay for it unless they actually invoke this
        /// fallback. 4 KiB matches typical RTPS frag-size +
        /// micro-XRCE message ceilings.
        const STAGE_CAP: usize = 4096;

        let mut total = 0usize;
        // SAFETY: caller's contract on `size_cb` matches our trait
        // doc — fire once with a writable `*mut usize` slot.
        unsafe { size_cb(&mut total as *mut usize, user_ctx) };
        if total > STAGE_CAP {
            return Err(TransportError::BufferTooSmall.into());
        }
        let mut stage = [0u8; STAGE_CAP];
        let mut written_so_far = 0usize;
        while written_so_far < total {
            let mut chunk_written = 0usize;
            let remaining = total - written_so_far;
            // SAFETY: `chunk_cb` writes ≤ `cap` bytes to
            // `out_buf` and reports the count via `out_written`.
            unsafe {
                chunk_cb(
                    stage.as_mut_ptr().add(written_so_far),
                    remaining,
                    &mut chunk_written as *mut usize,
                    user_ctx,
                );
            }
            if chunk_written == 0 {
                // Caller signalled EOF early — treat the partial
                // write as a malformed sequence; reporting it as
                // BufferTooSmall keeps the surface tight without
                // adding a new variant.
                return Err(TransportError::BufferTooSmall.into());
            }
            written_so_far += chunk_written;
        }
        self.publish_raw(&stage[..total])
    }

    /// Publish a typed message (serializes automatically)
    fn publish<M: RosMessage>(&self, msg: &M, buf: &mut [u8]) -> Result<(), Self::Error> {
        use nros_core::CdrWriter;

        let mut writer = CdrWriter::new_with_header(buf).map_err(|_| self.buffer_error())?;
        msg.serialize(&mut writer)
            .map_err(|_| self.serialization_error())?;
        let len = writer.position();
        self.publish_raw(&buf[..len])
    }

    /// Return a buffer-too-small error (implementation specific)
    fn buffer_error(&self) -> Self::Error;

    /// Return a serialization error (implementation specific)
    fn serialization_error(&self) -> Self::Error;

    /// Phase 108 — `true` if the backend can generate this event for
    /// this publisher. Default returns `false`; backends override per
    /// supported event kind.
    ///
    /// Only [`EventKind::LivelinessLost`](crate::event::EventKind::LivelinessLost) and
    /// [`EventKind::OfferedDeadlineMissed`](crate::event::EventKind::OfferedDeadlineMissed) are publisher-side events;
    /// other kinds always return `false` here.
    fn supports_event(&self, _kind: crate::event::EventKind) -> bool {
        false
    }

    /// Phase 108 — register a callback fired when the named status
    /// event occurs. `deadline_ms` applies to
    /// [`EventKind::OfferedDeadlineMissed`](crate::event::EventKind::OfferedDeadlineMissed) only; ignored otherwise.
    /// Default impl returns the backend's "unsupported"-shaped error.
    ///
    /// # Safety
    ///
    /// `cb` and `user_ctx` must remain valid for the entity's
    /// lifetime. Caller (typically `nros-node`'s typed wrapper) is
    /// responsible for keeping the closure / context arena alive.
    unsafe fn register_event_callback(
        &mut self,
        _kind: crate::event::EventKind,
        _deadline_ms: u32,
        _cb: crate::event::EventCallback,
        _user_ctx: *mut core::ffi::c_void,
    ) -> Result<(), Self::Error> {
        Err(self.unsupported_event_error())
    }

    /// Phase 108 — backend's error variant for "this event kind is
    /// not supported." Default impl reuses `serialization_error()`
    /// since most backends share an `Unsupported` variant; backends
    /// override if they have a distinct `Unsupported` mapping.
    fn unsupported_event_error(&self) -> Self::Error {
        self.serialization_error()
    }

    /// Phase 109 — assert this publisher's liveliness manually.
    /// Required for publishers configured with
    /// `QosLivelinessPolicy::ManualByTopic`. No-op for other
    /// liveliness kinds. Default impl returns `Ok(())` (no-op);
    /// backends override when they implement manual liveliness.
    fn assert_liveliness(&self) -> Result<(), Self::Error> {
        Ok(())
    }
}

/// Subscriber trait for receiving messages.
///
/// # Threading
///
/// `&mut self` on `try_recv_raw` — the executor takes exclusive
/// ownership of the subscriber for the duration of a receive. A
/// backend that wants to allow concurrent receives must split into
/// per-thread sub-handles internally.
///
/// # Buffer ownership
///
/// `buf` is caller-owned. The implementation copies the next ready
/// message into `buf` and returns the byte count. The caller may
/// re-use or drop `buf` immediately after the call.
///
/// # Blocking
///
/// `try_recv_raw` is **non-blocking**: returns `Ok(None)` (or
/// equivalent for backends that map empty into a zero-length read)
/// when no message is ready. Use [`Session::drive_io`] to wait for
/// data; never sleep inside `try_recv_raw`.
pub trait Subscriber {
    /// Error type for receive operations
    type Error;

    /// Check if data is available without consuming it.
    ///
    /// Non-destructive — does not advance the receive cursor.
    /// Conservative default returns `true` (always assume data may
    /// be available); backends should override with a real check
    /// to avoid spurious receive attempts.
    fn has_data(&self) -> bool {
        true
    }

    /// Try to receive one message into `buf`.
    ///
    /// Non-blocking. On success returns `Ok(Some(len))` where `len`
    /// is the byte count written into `buf[..len]`. Returns
    /// `Ok(None)` if no message is ready. If `buf` is too small the
    /// backend may either truncate (and document it) or return an
    /// error (preferred).
    fn try_recv_raw(&mut self, buf: &mut [u8]) -> Result<Option<usize>, Self::Error>;

    /// Phase 128.F.4 — receive with attachment bytes alongside the
    /// payload.
    ///
    /// On success returns `Ok(Some((payload_len, attachment_len)))`
    /// with the payload written into `buf[..payload_len]` and the
    /// attachment (if any) written into
    /// `att_buf[..attachment_len]`. `attachment_len == 0` means the
    /// incoming sample carried no attachment.
    ///
    /// Default body falls back to [`try_recv_raw`](Self::try_recv_raw)
    /// and reports a 0-length attachment. Backends with native
    /// attachment support override to populate `att_buf`. Cross-RMW
    /// bridges use the attachment to read the `bridge_origin` tag
    /// stamped by the sending side.
    fn try_recv_raw_with_attachment(
        &mut self,
        buf: &mut [u8],
        _att_buf: &mut [u8],
    ) -> Result<Option<(usize, usize)>, Self::Error> {
        match self.try_recv_raw(buf)? {
            Some(len) => Ok(Some((len, 0))),
            None => Ok(None),
        }
    }

    /// Phase 124.D.1 — burst-take.
    ///
    /// Drain up to `max_msgs` queued samples into the contiguous
    /// `buf` block in one call, with the i-th sample at
    /// `buf[i * per_msg_cap .. i * per_msg_cap + out_lens[i]]`.
    /// Returns the number of messages actually delivered. Partial
    /// drains MUST report the count, not error out.
    ///
    /// Default body loop-drives `try_recv_raw` so callers can
    /// commit to the batched API regardless of backend support.
    /// Concrete backends opt in by overriding with a native batch
    /// take (zenoh queue drain, `dds_take(max_samples)`).
    fn try_recv_sequence(
        &mut self,
        buf: &mut [u8],
        per_msg_cap: usize,
        max_msgs: usize,
        out_lens: &mut [usize],
    ) -> Result<usize, Self::Error> {
        if per_msg_cap == 0 || max_msgs == 0 {
            return Ok(0);
        }
        let limit = max_msgs.min(out_lens.len());
        let mut count = 0;
        for i in 0..limit {
            let slot = &mut buf[i * per_msg_cap..(i + 1) * per_msg_cap];
            match self.try_recv_raw(slot)? {
                Some(len) => {
                    out_lens[i] = len;
                    count += 1;
                }
                None => break,
            }
        }
        Ok(count)
    }

    /// Try to receive a typed message (non-blocking)
    fn try_recv<M: RosMessage>(&mut self, buf: &mut [u8]) -> Result<Option<M>, Self::Error> {
        use nros_core::CdrReader;

        match self.try_recv_raw(buf)? {
            Some(len) => {
                let mut reader = CdrReader::new_with_header(&buf[..len])
                    .map_err(|_| self.deserialization_error())?;
                let msg = M::deserialize(&mut reader).map_err(|_| self.deserialization_error())?;
                Ok(Some(msg))
            }
            None => Ok(None),
        }
    }

    /// Process the received message in-place without copying.
    ///
    /// Calls `f` with a reference to the raw CDR bytes in the subscriber's
    /// internal receive buffer, avoiding a copy into a caller-provided buffer.
    /// While `f` executes the buffer is exclusively borrowed — any messages
    /// arriving from the transport during that time are dropped to prevent
    /// data races.
    ///
    /// Returns `Ok(true)` if a message was available and `f` was called,
    /// `Ok(false)` if no message was available.
    ///
    /// **Default body**: returns `Err(MessageTooLarge)` — the old default
    /// silently truncated anything larger than 1 KB into a stack buffer,
    /// which broke large messages with no diagnostic. Backends must
    /// override this with a real zero-copy path if they advertise support
    /// for `process_raw_in_place`; callers that hit the default should
    /// use `try_recv_raw` with a caller-sized buffer instead.
    fn process_raw_in_place(&mut self, f: impl FnOnce(&[u8])) -> Result<bool, Self::Error>
    where
        Self::Error: From<TransportError>,
    {
        let _ = f;
        Err(TransportError::MessageTooLarge.into())
    }

    /// Whether this backend implements the in-place dispatch methods
    /// ([`process_raw_in_place`](Subscriber::process_raw_in_place) /
    /// [`process_raw_in_place_with_info`](Subscriber::process_raw_in_place_with_info))
    /// with a real zero-copy borrow.
    ///
    /// The executor consults this at subscription registration to choose the
    /// **in-place** arena dispatch (borrow + deserialize from the backend slot, no
    /// arena buffer) over the **buffered** dispatch (copy into an arena buffer
    /// first). Backends that leave the in-place methods at their unsupported
    /// default return `false` (the default) and keep the buffered path. (RFC-0038,
    /// Phase 231 Wave 0.2.)
    fn supports_process_in_place(&self) -> bool {
        false
    }

    /// In-place processing variant that also surfaces publisher metadata.
    ///
    /// Same borrow contract as
    /// [`process_raw_in_place`](Subscriber::process_raw_in_place): `f` receives
    /// the raw CDR bytes plus the parsed [`MessageInfo`](nros_core::MessageInfo)
    /// — the co-located attachment (publisher GID / sequence / source timestamp),
    /// or `None` when no attachment was present — for the duration of the call;
    /// the slot is released after `f` returns. `Ok(true)` = a message was
    /// available and `f` was called; `Ok(false)` = none ready.
    ///
    /// **Default body**: returns the unsupported error (mirrors
    /// `process_raw_in_place`). Backends that advertise in-place support override
    /// this with a real zero-copy path; callers that hit the default should use
    /// the buffered [`try_recv_raw_with_info`](Subscriber::try_recv_raw_with_info)
    /// path instead. (RFC-0038, Phase 231 Wave 0.1.)
    fn process_raw_in_place_with_info(
        &mut self,
        f: impl FnOnce(&[u8], Option<nros_core::MessageInfo>),
    ) -> Result<bool, Self::Error>
    where
        Self::Error: From<TransportError>,
    {
        let _ = f;
        Err(TransportError::MessageTooLarge.into())
    }

    /// Try to receive raw data along with publisher metadata.
    ///
    /// When available, [`MessageInfo`](nros_core::MessageInfo) contains
    /// the publisher's GID (Global Identifier) and source timestamp,
    /// extracted from a transport-level attachment on the incoming message.
    ///
    /// Returns `Ok(Some((len, info)))` if data is available, where:
    /// - `len` is the number of bytes written to the buffer
    /// - `info` is the parsed publisher metadata (if attachment was present)
    ///
    /// Default: delegates to [`try_recv_raw`](Subscriber::try_recv_raw) with no info.
    fn try_recv_raw_with_info(
        &mut self,
        buf: &mut [u8],
    ) -> Result<Option<(usize, Option<nros_core::MessageInfo>)>, Self::Error> {
        self.try_recv_raw(buf).map(|opt| opt.map(|len| (len, None)))
    }

    /// Try to receive raw data with E2E safety validation (CRC + sequence tracking).
    ///
    /// Returns `Ok(Some((len, status)))` if data is available, where:
    /// - `len` is the number of bytes written to the buffer
    /// - `status` is the integrity validation result
    ///
    /// Default: delegates to `try_recv_raw` with no CRC info.
    #[cfg(feature = "safety-e2e")]
    fn try_recv_validated(
        &mut self,
        buf: &mut [u8],
    ) -> Result<Option<(usize, crate::IntegrityStatus)>, Self::Error> {
        self.try_recv_raw(buf).map(|opt| {
            opt.map(|len| {
                (
                    len,
                    crate::IntegrityStatus {
                        gap: 0,
                        duplicate: false,
                        crc_valid: None,
                    },
                )
            })
        })
    }

    /// Register an async waker to be notified when data arrives.
    ///
    /// Called from `Future::poll()` implementations to store the waker.
    /// The transport backend calls `waker.wake()` from its receive callback
    /// when new data is available, enabling event-driven async without
    /// busy-polling.
    ///
    /// Default: no-op (backends that don't support waking simply ignore this).
    fn register_waker(&self, _waker: &core::task::Waker) {}

    /// Return a deserialization error (implementation specific)
    fn deserialization_error(&self) -> Self::Error;

    /// Phase 108 — `true` if the backend can generate this event for
    /// this subscriber. Default returns `false`; backends override per
    /// supported event kind.
    ///
    /// Subscriber-side event kinds:
    /// [`EventKind::LivelinessChanged`](crate::event::EventKind::LivelinessChanged),
    /// [`EventKind::RequestedDeadlineMissed`](crate::event::EventKind::RequestedDeadlineMissed),
    /// [`EventKind::MessageLost`](crate::event::EventKind::MessageLost).
    /// Publisher kinds always return `false` here.
    fn supports_event(&self, _kind: crate::event::EventKind) -> bool {
        false
    }

    /// Phase 108 — register a callback fired when the named status
    /// event occurs. `deadline_ms` applies to
    /// [`EventKind::RequestedDeadlineMissed`](crate::event::EventKind::RequestedDeadlineMissed) only; ignored otherwise.
    /// Default impl returns the backend's "unsupported"-shaped error.
    ///
    /// # Safety
    ///
    /// `cb` and `user_ctx` must remain valid for the entity's
    /// lifetime. Caller (typically `nros-node`'s typed wrapper) is
    /// responsible for keeping the closure / context arena alive.
    unsafe fn register_event_callback(
        &mut self,
        _kind: crate::event::EventKind,
        _deadline_ms: u32,
        _cb: crate::event::EventCallback,
        _user_ctx: *mut core::ffi::c_void,
    ) -> Result<(), Self::Error> {
        Err(self.unsupported_event_error())
    }

    /// Phase 108 — backend's error variant for "this event kind is
    /// not supported." Default reuses `deserialization_error()` for
    /// backends that don't have a distinct `Unsupported` mapping.
    fn unsupported_event_error(&self) -> Self::Error {
        self.deserialization_error()
    }
}

/// Service request from a client
pub struct ServiceRequest<'a> {
    /// Raw request data (CDR encoded)
    pub data: &'a [u8],
    /// Sequence number for request/response matching
    pub sequence_number: i64,
}

// ============================================================================
// Phase 99 — zero-copy raw API: SlotLending / SlotBorrowing
// ============================================================================
//
// Backends that can lend a slot directly into their outbound buffer
// (zenoh-pico w/ unstable-zenoh-api, XRCE-DDS via uxr_prepare_output_stream,
// full DDS w/ SHM transport) implement these traits. Backends that cannot
// (uORB, default zenoh-pico) do NOT impl them — `EmbeddedRawPublisher` then
// falls back to its per-publisher arena and memcpys at commit time. Both
// paths land at `Publisher::publish_raw` for the actual wire write; only
// the user-side copy is eliminated when lending is available.
//
// Selection is **compile-time** via the `lending` Cargo feature. Each
// backend crate forwards its own `lending` feature to `nros-rmw/lending`
// when it can satisfy the trait. nros-node's `rmw-lending` aggregates.
// User opting `nros/rmw-lending` w/ a non-lending backend (e.g. uORB)
// gets a clear compile error from the unsatisfied trait bound on the
// concrete `RmwPublisher`.

/// Backend can lend a writable slot into its outbound buffer.
///
/// The returned slot's lifetime is tied to `&self`; user fills it in
/// place, then calls [`commit_slot`](Self::commit_slot) to publish.
/// Dropping the slot without commit is a no-op (slot returned to free
/// pool); concurrent loan attempts that would exceed backend capacity
/// return [`TransportError::WouldBlock`] — never block.
#[cfg(feature = "lending")]
pub trait SlotLending: Publisher {
    /// Backend-owned writable slot. Holds a `&'a mut [u8]` and any
    /// state needed for commit_slot.
    type Slot<'a>: AsMut<[u8]> + 'a
    where
        Self: 'a;

    /// Reserve a writable slot of `len` bytes from the backend's
    /// outbound buffer. Returns `Ok(None)` if the backend has no slot
    /// available (full); never blocks.
    fn try_lend_slot(&self, len: usize) -> Result<Option<Self::Slot<'_>>, Self::Error>;

    /// Commit a previously-lent slot. Consumes the slot and triggers
    /// the actual wire write. Returns `Err` on backend send failure;
    /// the slot's bytes are lost in that case (caller must re-lend +
    /// re-fill to retry).
    fn commit_slot(&self, slot: Self::Slot<'_>) -> Result<(), Self::Error>;
}

/// Backend can lend a read-only view into its receive buffer.
///
/// The returned view's lifetime is tied to `&mut self` (subscriber-
/// exclusive); dropping the view releases any backend lock and lets
/// the next message advance into the buffer.
#[cfg(feature = "lending")]
pub trait SlotBorrowing: Subscriber {
    /// Backend-owned read-only view. Holds a `&'a [u8]` and any state
    /// needed to release the borrow on Drop.
    type View<'a>: AsRef<[u8]> + 'a
    where
        Self: 'a;

    /// Try to borrow the next available message in place. Returns
    /// `Ok(None)` if no message is ready; never blocks.
    fn try_borrow(&mut self) -> Result<Option<Self::View<'_>>, Self::Error>;
}

/// Service server trait for handling requests.
///
/// # Threading
///
/// `&mut self` on `try_recv_request` and `send_reply` — the executor
/// owns the server while a request is being handled. Handler bodies
/// run synchronously on the executor thread; long handlers should
/// dispatch work to a worker queue and reply later via the recorded
/// `sequence_number`.
///
/// # Calling pattern
///
/// 1. Executor calls `try_recv_request(buf)`.
/// 2. If `Some(req)` returned, decode, run handler, encode reply.
/// 3. `send_reply(req.sequence_number, &reply_buf)`.
///
/// `sequence_number` is the canonical request → reply correlation
/// token; backends derive it from the wire-level metadata (zenoh
/// query id, DDS sample identity).
pub trait ServiceServerTrait {
    /// Error type for service operations
    type Error;

    /// Check if a request is available without consuming it.
    ///
    /// Non-destructive. Default returns `true` (always assume one
    /// may be available); backends should override with a real
    /// check.
    fn has_request(&self) -> bool {
        true
    }

    /// Phase 122.3.c.6.e — register a `Waker` for event-driven
    /// service servers. Mirrors the matching method on
    /// `SubscriberTrait` / `ServiceClientTrait`. Backends that
    /// surface incoming-request notifications wake `waker` when
    /// `has_request()` flips true. Default: no-op (backends without
    /// wake support ignore — caller falls back to polling).
    fn register_waker(&self, _waker: &core::task::Waker) {}

    /// Try to receive a service request into `buf` (non-blocking).
    ///
    /// On success returns a `ServiceRequest` that borrows from
    /// `buf`. The borrow is released when the returned struct is
    /// dropped — typically before `send_reply` is called, since
    /// `send_reply` takes `&mut self`.
    fn try_recv_request<'a>(
        &mut self,
        buf: &'a mut [u8],
    ) -> Result<Option<ServiceRequest<'a>>, Self::Error>;

    /// Send a reply for the given sequence number. Non-blocking
    /// from the application's perspective; the backend may queue
    /// the reply for transport-level transmission.
    fn send_reply(&mut self, sequence_number: i64, data: &[u8]) -> Result<(), Self::Error>;

    /// Handle a service request with typed messages
    fn handle_request<S: RosService>(
        &mut self,
        req_buf: &mut [u8],
        reply_buf: &mut [u8],
        handler: impl FnOnce(&S::Request) -> S::Reply,
    ) -> Result<bool, Self::Error>
    where
        Self::Error: From<TransportError>,
    {
        use nros_core::{CdrReader, CdrWriter};

        // First, try to receive a request and extract necessary data.
        // Capture the data slice's offset within `req_buf` so we can
        // re-borrow it after the `ServiceRequest` (which holds a
        // borrow into `req_buf`) is dropped. Some backends prepend a
        // header (DDS: 8-byte sequence number) and place the CDR
        // payload at a non-zero offset in the buffer; others (zenoh)
        // put it at offset 0. Reading from offset 0 unconditionally
        // would feed the prefix bytes to the CDR deserializer and
        // silently corrupt the request.
        let buf_start = req_buf.as_ptr() as usize;
        let (data_offset, data_len, sequence_number) = match self.try_recv_request(req_buf)? {
            Some(request) => {
                let offset = (request.data.as_ptr() as usize).saturating_sub(buf_start);
                (offset, request.data.len(), request.sequence_number)
            }
            None => return Ok(false),
        };

        // Deserialize request from the captured offset.
        let mut reader = CdrReader::new_with_header(&req_buf[data_offset..data_offset + data_len])
            .map_err(|_| TransportError::DeserializationError)?;
        let req = S::Request::deserialize(&mut reader)
            .map_err(|_| TransportError::DeserializationError)?;

        // Call handler
        let reply = handler(&req);

        // Serialize reply
        let mut writer =
            CdrWriter::new_with_header(reply_buf).map_err(|_| TransportError::BufferTooSmall)?;
        reply
            .serialize(&mut writer)
            .map_err(|_| TransportError::SerializationError)?;
        let len = writer.position();

        // Send reply (now we can borrow self mutably again)
        self.send_reply(sequence_number, &reply_buf[..len])?;
        Ok(true)
    }

    /// Handle a service request where the handler returns `Box<S::Reply>`
    ///
    /// Identical to `handle_request` but the handler returns a heap-allocated reply.
    /// This is needed for services with large response types (e.g., parameter services
    /// where `Vec<ParameterValue, 64>` is ~1MB+) that would overflow the stack.
    #[cfg(feature = "alloc")]
    fn handle_request_boxed<S: RosService>(
        &mut self,
        req_buf: &mut [u8],
        reply_buf: &mut [u8],
        handler: impl FnOnce(&S::Request) -> alloc::boxed::Box<S::Reply>,
    ) -> Result<bool, Self::Error>
    where
        Self::Error: From<TransportError>,
    {
        use nros_core::{CdrReader, CdrWriter};

        let buf_start = req_buf.as_ptr() as usize;
        let (data_offset, data_len, sequence_number) = match self.try_recv_request(req_buf)? {
            Some(request) => {
                let offset = (request.data.as_ptr() as usize).saturating_sub(buf_start);
                (offset, request.data.len(), request.sequence_number)
            }
            None => return Ok(false),
        };

        let mut reader = CdrReader::new_with_header(&req_buf[data_offset..data_offset + data_len])
            .map_err(|_| TransportError::DeserializationError)?;
        let req = S::Request::deserialize(&mut reader)
            .map_err(|_| TransportError::DeserializationError)?;

        let reply = handler(&req);

        let mut writer =
            CdrWriter::new_with_header(reply_buf).map_err(|_| TransportError::BufferTooSmall)?;
        reply
            .serialize(&mut writer)
            .map_err(|_| TransportError::SerializationError)?;
        let len = writer.position();

        self.send_reply(sequence_number, &reply_buf[..len])?;
        Ok(true)
    }
}

/// Service client trait for sending requests.
///
/// # Threading
///
/// `&mut self` on every method — the client is single-owner. For
/// fan-out request patterns, create one client per worker thread.
///
/// # Calling pattern
///
/// All in-tree backends route blocking waits through the executor:
///
/// 1. `send_request_raw(buf)` — non-blocking; returns once the
///    request is queued for transmission.
/// 2. The executor's `drive_io` runs.
/// 3. `try_recv_reply_raw(buf)` — non-blocking; returns
///    `Ok(Some(len))` when the reply is back.
///
/// The deprecated [`call_raw`](Self::call_raw) blocking path is
/// kept for backwards compatibility but should not be called.
pub trait ServiceClientTrait {
    /// Error type for service operations
    type Error;

    /// Send a service request and wait for reply (blocking).
    ///
    /// **Deprecated — do not call.** The default body returns `Timeout`
    /// immediately without polling. Use `Client::call` →
    /// `Promise::wait(executor, timeout_ms)` which lets the executor
    /// drive I/O while waiting instead of busy-looping on
    /// `try_recv_reply_raw` with no sleep (which starves the transport
    /// on FreeRTOS / Zephyr single-threaded schedulers).
    ///
    /// Backends that still need an internal blocking path should
    /// override this with a real sleep-between-polls implementation,
    /// but all in-tree backends (zenoh, XRCE) route blocking waits
    /// through the executor.
    #[deprecated(note = "use Client::call → Promise::wait with an executor instead")]
    fn call_raw(&mut self, request: &[u8], _reply_buf: &mut [u8]) -> Result<usize, Self::Error>
    where
        Self::Error: From<TransportError>,
    {
        let _ = request;
        Err(TransportError::Timeout.into())
    }

    /// Send a service request without waiting for a reply (non-blocking).
    ///
    /// The caller must subsequently poll [`try_recv_reply_raw`](Self::try_recv_reply_raw)
    /// to retrieve the reply.
    fn send_request_raw(&mut self, request: &[u8]) -> Result<(), Self::Error>;

    /// Poll for a reply to the most recently sent request (non-blocking).
    ///
    /// Returns `Ok(Some(len))` when a reply has arrived, `Ok(None)` if not yet
    /// available, or `Err` on failure.
    fn try_recv_reply_raw(&mut self, reply_buf: &mut [u8]) -> Result<Option<usize>, Self::Error>;

    /// Send a typed service request without waiting for a reply (non-blocking).
    ///
    /// Serializes the request into `req_buf` and calls [`send_request_raw`](Self::send_request_raw).
    fn send_request<S: RosService>(
        &mut self,
        request: &S::Request,
        req_buf: &mut [u8],
    ) -> Result<(), Self::Error>
    where
        Self::Error: From<TransportError>,
    {
        use nros_core::CdrWriter;

        let mut writer =
            CdrWriter::new_with_header(req_buf).map_err(|_| TransportError::BufferTooSmall)?;
        request
            .serialize(&mut writer)
            .map_err(|_| TransportError::SerializationError)?;
        let req_len = writer.position();

        self.send_request_raw(&req_buf[..req_len])
    }

    /// Poll for a typed reply to the most recently sent request (non-blocking).
    ///
    /// Calls [`try_recv_reply_raw`](Self::try_recv_reply_raw) and deserializes if available.
    fn try_recv_reply<S: RosService>(
        &mut self,
        reply_buf: &mut [u8],
    ) -> Result<Option<S::Reply>, Self::Error>
    where
        Self::Error: From<TransportError>,
    {
        use nros_core::CdrReader;

        match self.try_recv_reply_raw(reply_buf)? {
            Some(len) => {
                let mut reader = CdrReader::new_with_header(&reply_buf[..len])
                    .map_err(|_| TransportError::DeserializationError)?;
                let reply = S::Reply::deserialize(&mut reader)
                    .map_err(|_| TransportError::DeserializationError)?;
                Ok(Some(reply))
            }
            None => Ok(None),
        }
    }

    /// Register an async waker to be notified when a reply arrives.
    ///
    /// Called from `Future::poll()` implementations to store the waker.
    /// The transport backend calls `waker.wake()` from its reply callback
    /// when a response is available, enabling event-driven async without
    /// busy-polling.
    ///
    /// Default: no-op (backends that don't support waking simply ignore this).
    fn register_waker(&self, _waker: &core::task::Waker) {}

    /// Begin a server-discovery query on this client (non-blocking).
    ///
    /// Models `rclcpp::ClientBase::wait_for_service` machinery: the backend
    /// fires off a discovery probe (typically a Zenoh liveliness query
    /// against the matching server's wildcarded liveliness keyexpr) and
    /// the caller polls [`poll_server_discovery`](Self::poll_server_discovery)
    /// to collect the result.
    ///
    /// Default impl: no-op success. Backends without a discovery channel
    /// (or those that always assume the server is reachable) can leave
    /// this default and have `poll_server_discovery` return
    /// `Ok(Some(true))` immediately.
    fn start_server_discovery(&mut self, _timeout_ms: u32) -> Result<(), Self::Error> {
        Ok(())
    }

    /// Poll an in-flight server-discovery query.
    ///
    /// - `Ok(Some(true))` — at least one matching server has reported
    ///   back; safe to send the first request.
    /// - `Ok(Some(false))` — discovery query finished without finding
    ///   any matching server (timeout / no-replies).
    /// - `Ok(None)` — query still in flight.
    /// - `Err(_)` — transport-level failure unrelated to server presence.
    ///
    /// Default impl: returns `Ok(Some(true))` (i.e., "server is always
    /// assumed reachable"). The Zenoh backend overrides this with a
    /// liveliness-token check.
    fn poll_server_discovery(&mut self) -> Result<Option<bool>, Self::Error> {
        Ok(Some(true))
    }

    /// Synchronous, non-blocking check of whether a matching server is
    /// currently visible.
    ///
    /// Mirrors `rclcpp::ClientBase::service_is_ready`. Backends that lack
    /// discovery should keep the default `true` so existing call sites
    /// don't regress.
    ///
    /// Default impl: always `true`.
    fn is_server_ready(&self) -> bool {
        true
    }

    /// Phase 124.C.1 — graph-aware server-availability probe.
    ///
    /// Returns `Ok(true)` if at least one matching server has been
    /// discovered, `Ok(false)` if none yet, or `Err(_)` if the
    /// backend cannot answer (e.g. XRCE — micro-XRCE-DDS-Client has
    /// no participant enumeration). Distinct from
    /// [`is_server_ready`](Self::is_server_ready), which collapses
    /// "don't know" and "no server" into the same `false` answer.
    ///
    /// User-facing surface: `Client<S>::server_available()` in Rust,
    /// `nros_client_server_available()` in C/C++. Clients use this
    /// to gate the first `call_raw` so a startup-ordering race
    /// (client opens before server's discovery announcement lands)
    /// doesn't surface as a request-side timeout.
    ///
    /// Default impl: `Err(TransportError::Unsupported)` — backends
    /// that support graph introspection (zenoh queryable interest,
    /// DDS built-in topic readers) opt in by overriding.
    fn server_available(&self) -> Result<bool, Self::Error>
    where
        Self::Error: From<TransportError>,
    {
        Err(TransportError::Unsupported.into())
    }

    /// Call a service with typed messages (blocking).
    ///
    /// **Deprecated — do not call.** The default body returns `Timeout`
    /// immediately without polling. Use `Client::call` on the executor
    /// instead, which drives I/O while waiting. See
    /// [`call_raw`](Self::call_raw) for the same reasoning.
    #[deprecated(note = "use Client::call → Promise::wait with an executor instead")]
    fn call<S: RosService>(
        &mut self,
        request: &S::Request,
        req_buf: &mut [u8],
        _reply_buf: &mut [u8],
    ) -> Result<S::Reply, Self::Error>
    where
        Self::Error: From<TransportError>,
    {
        use nros_core::CdrWriter;

        // Serialize request so the error surface matches the old impl
        // for the "bad request" path; but skip the receive busy-loop.
        let mut writer =
            CdrWriter::new_with_header(req_buf).map_err(|_| TransportError::BufferTooSmall)?;
        request
            .serialize(&mut writer)
            .map_err(|_| TransportError::SerializationError)?;
        Err(TransportError::Timeout.into())
    }
}

/// Transport backend trait (legacy).
///
/// Use [`Rmw`] for new code. This trait is retained for backward compatibility
/// with existing code that uses [`TransportConfig`] directly.
pub trait Transport {
    /// Error type for this transport
    type Error;
    /// Session type for this transport
    type Session: Session;

    /// Open a new session with the given configuration
    fn open(config: &TransportConfig) -> Result<Self::Session, Self::Error>;
}

/// Factory trait for compile-time middleware selection.
///
/// Embedded crates select a backend via feature flag:
/// ```rust,ignore
/// #[cfg(feature = "rmw-cffi")]
/// type DefaultRmw = nros_rmw_cffi::CffiRmw;
/// ```
///
/// Each backend provides its own `Rmw` implementation that bridges
/// from the middleware-agnostic [`RmwConfig`] to backend-specific
/// initialization.
///
/// Phase 84.E2: `open` consumes `self`. Backends carry their own
/// configuration (agent addresses, serial ports, TLS CA slots)
/// inside the factory value and hand that over to the session at
/// `open` time. All in-repo backends also implement
/// [`Default`]; most callers spell this as
/// `BackendRmw::default().open(&config)`.
pub trait Rmw {
    /// Session type returned by [`open`](Rmw::open)
    type Session: Session;
    /// Error type for session creation
    type Error: core::fmt::Debug;

    /// Open a new middleware session with the given configuration.
    ///
    /// The backend maps [`RmwConfig`] fields to its own connection
    /// parameters (e.g., zenoh locator and session mode, XRCE-DDS
    /// agent address). Any backend-specific pre-open state stored
    /// on `self` (e.g. configured agent IP / port) is moved into the
    /// returned `Session`.
    fn open(self, config: &RmwConfig) -> Result<Self::Session, Self::Error>;
}

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

    #[test]
    fn test_topic_info() {
        let topic = TopicInfo::new("/chatter", "std_msgs::msg::dds_::String_", "abc123");
        assert_eq!(topic.name, "/chatter");
        assert_eq!(topic.domain_id, 0);
    }

    #[test]
    fn qos_apply_overrides_matches_topic_and_role() {
        // Default is Reliable / Volatile / KeepLast(10).
        static OVERRIDES: &[QosOverride] = &[
            QosOverride {
                topic: "/chatter",
                role: QosOverrideRole::Publisher,
                value: QosOverrideValue::Reliability(QosReliabilityPolicy::BestEffort),
            },
            QosOverride {
                topic: "/chatter",
                role: QosOverrideRole::Publisher,
                value: QosOverrideValue::Depth(5),
            },
            QosOverride {
                topic: "/scan",
                role: QosOverrideRole::Subscription,
                value: QosOverrideValue::Durability(QosDurabilityPolicy::TransientLocal),
            },
        ];

        // Matching topic + publisher role → reliability + depth applied.
        let pub_qos = QosSettings::default().apply_overrides(
            "/chatter",
            QosOverrideRole::Publisher,
            OVERRIDES,
        );
        assert_eq!(pub_qos.reliability, QosReliabilityPolicy::BestEffort);
        assert_eq!(pub_qos.depth, 5);
        assert_eq!(pub_qos.durability, QosDurabilityPolicy::Volatile); // untouched

        // Same topic but subscription role → publisher overrides DON'T apply;
        // the /scan override is for a different topic → also no change.
        let sub_qos = QosSettings::default().apply_overrides(
            "/chatter",
            QosOverrideRole::Subscription,
            OVERRIDES,
        );
        assert_eq!(sub_qos, QosSettings::default());

        // The /scan subscription override applies only to /scan+subscription.
        let scan_qos = QosSettings::default().apply_overrides(
            "/scan",
            QosOverrideRole::Subscription,
            OVERRIDES,
        );
        assert_eq!(scan_qos.durability, QosDurabilityPolicy::TransientLocal);

        // Empty table → identity (the zero-override fast path).
        assert_eq!(
            QosSettings::default().apply_overrides("/x", QosOverrideRole::Publisher, &[]),
            QosSettings::default()
        );
    }

    #[test]
    fn test_qos_defaults() {
        let qos = QosSettings::default();
        assert_eq!(qos.reliability, QosReliabilityPolicy::Reliable);
    }

    #[test]
    fn test_action_info() {
        let action = ActionInfo::new(
            "/fibonacci",
            "example_interfaces::action::dds_::Fibonacci_",
            "abc123",
        );
        assert_eq!(action.name, "/fibonacci");
        assert_eq!(action.domain_id, 0);
    }

    #[test]
    fn test_action_info_with_domain() {
        let action = ActionInfo::new(
            "/fibonacci",
            "example_interfaces::action::dds_::Fibonacci_",
            "abc123",
        )
        .with_domain(42);
        assert_eq!(action.domain_id, 42);
    }

    #[test]
    fn test_action_send_goal_key() {
        let action = ActionInfo::new(
            "/fibonacci",
            "example_interfaces::action::dds_::Fibonacci_",
            "abc123",
        )
        .with_domain(0);

        let key: heapless::String<256> = action.send_goal_key();
        // ActionInfo returns the sub-entity name with leading slash for ROS 2 compatibility
        assert_eq!(key.as_str(), "/fibonacci/_action/send_goal");
    }

    #[test]
    fn test_action_feedback_key() {
        let action = ActionInfo::new(
            "/fibonacci",
            "example_interfaces::action::dds_::Fibonacci_",
            "abc123",
        )
        .with_domain(0);

        let key: heapless::String<256> = action.feedback_key();
        assert_eq!(key.as_str(), "/fibonacci/_action/feedback");
    }

    #[test]
    fn test_action_all_sub_names() {
        let action = ActionInfo::new(
            "/fibonacci",
            "example_interfaces::action::dds_::Fibonacci_",
            "abc123",
        )
        .with_domain(0);

        let cancel: heapless::String<256> = action.cancel_goal_key();
        assert_eq!(cancel.as_str(), "/fibonacci/_action/cancel_goal");

        let result: heapless::String<256> = action.get_result_key();
        assert_eq!(result.as_str(), "/fibonacci/_action/get_result");

        let status: heapless::String<256> = action.status_key();
        assert_eq!(status.as_str(), "/fibonacci/_action/status");
    }

    // --- QoS Profile Tests ---

    #[test]
    fn test_qos_profile_sensor_data() {
        let qos = QosSettings::QOS_PROFILE_SENSOR_DATA;
        assert_eq!(qos.reliability, QosReliabilityPolicy::BestEffort);
        assert_eq!(qos.durability, QosDurabilityPolicy::Volatile);
        assert_eq!(qos.history, QosHistoryPolicy::KeepLast);
        assert_eq!(qos.depth, 5);
    }

    #[test]
    fn test_qos_profile_default() {
        let qos = QosSettings::QOS_PROFILE_DEFAULT;
        assert_eq!(qos.reliability, QosReliabilityPolicy::Reliable);
        assert_eq!(qos.durability, QosDurabilityPolicy::Volatile);
        assert_eq!(qos.depth, 10);
    }

    #[test]
    fn test_qos_profile_services_default() {
        let qos = QosSettings::QOS_PROFILE_SERVICES_DEFAULT;
        assert_eq!(qos.reliability, QosReliabilityPolicy::Reliable);
        assert_eq!(qos.durability, QosDurabilityPolicy::Volatile);
    }

    #[test]
    fn services_default_validates_and_rejects_missing_policy() {
        // Phase 193.5 — the service-create path (node `create_service*_sized` +
        // the typed-arena `register_service*_on`) runs `validate_against` on the
        // caller's profile, exactly like pub/sub. A backend advertising the
        // profile's required policies admits it; dropping any required bit (here
        // RELIABILITY) rejects it with `IncompatibleQos` — no silent downgrade.
        let qos = QosSettings::services_default();
        let required = qos.required_policies();
        assert!(qos.validate_against(required).is_ok());
        assert!(qos.validate_against(QosPolicyMask(u32::MAX)).is_ok());
        let missing = QosPolicyMask(required.0 & !QosPolicyMask::RELIABILITY.0);
        assert_eq!(
            qos.validate_against(missing),
            Err(TransportError::IncompatibleQos)
        );
    }

    #[test]
    fn test_qos_profile_parameters() {
        let qos = QosSettings::QOS_PROFILE_PARAMETERS;
        assert_eq!(qos.reliability, QosReliabilityPolicy::Reliable);
        assert_eq!(qos.durability, QosDurabilityPolicy::TransientLocal);
        assert_eq!(qos.depth, 1000);
    }

    #[test]
    fn test_qos_profile_clock() {
        let qos = QosSettings::QOS_PROFILE_CLOCK;
        assert_eq!(qos.reliability, QosReliabilityPolicy::BestEffort);
        assert_eq!(qos.depth, 1);
    }

    #[test]
    fn test_qos_profile_parameter_events() {
        let qos = QosSettings::QOS_PROFILE_PARAMETER_EVENTS;
        assert_eq!(qos.reliability, QosReliabilityPolicy::Reliable);
        assert_eq!(qos.history, QosHistoryPolicy::KeepAll);
    }

    #[test]
    fn test_qos_profile_action_status() {
        let qos = QosSettings::QOS_PROFILE_ACTION_STATUS_DEFAULT;
        assert_eq!(qos.reliability, QosReliabilityPolicy::Reliable);
        assert_eq!(qos.durability, QosDurabilityPolicy::TransientLocal);
        assert_eq!(qos.depth, 1);
    }

    #[test]
    fn test_qos_static_constructors() {
        assert_eq!(
            QosSettings::topics_default(),
            QosSettings::QOS_PROFILE_DEFAULT
        );
        assert_eq!(
            QosSettings::sensor_data_default(),
            QosSettings::QOS_PROFILE_SENSOR_DATA
        );
        assert_eq!(
            QosSettings::services_default(),
            QosSettings::QOS_PROFILE_SERVICES_DEFAULT
        );
        assert_eq!(
            QosSettings::parameters_default(),
            QosSettings::QOS_PROFILE_PARAMETERS
        );
        assert_eq!(
            QosSettings::action_status_default(),
            QosSettings::QOS_PROFILE_ACTION_STATUS_DEFAULT
        );
    }

    #[test]
    fn test_qos_builder_explicit_setters() {
        let qos = QosSettings::new()
            .reliability(QosReliabilityPolicy::Reliable)
            .durability(QosDurabilityPolicy::TransientLocal)
            .history(QosHistoryPolicy::KeepAll)
            .depth(100);

        assert_eq!(qos.reliability, QosReliabilityPolicy::Reliable);
        assert_eq!(qos.durability, QosDurabilityPolicy::TransientLocal);
        assert_eq!(qos.history, QosHistoryPolicy::KeepAll);
        assert_eq!(qos.depth, 100);
    }

    #[test]
    fn test_qos_builder_chaining() {
        // Test that builder methods can be chained in any order
        let qos = QosSettings::sensor_data_default()
            .reliable()
            .transient_local()
            .keep_last(20);

        assert_eq!(qos.reliability, QosReliabilityPolicy::Reliable);
        assert_eq!(qos.durability, QosDurabilityPolicy::TransientLocal);
        assert_eq!(qos.history, QosHistoryPolicy::KeepLast);
        assert_eq!(qos.depth, 20);
    }

    #[test]
    fn test_qos_eq_impl() {
        // Verify that PartialEq works correctly via derive on QosSettings
        let qos1 = QosSettings::QOS_PROFILE_DEFAULT;
        let qos2 = QosSettings::topics_default();
        // Both should have same values - verify field by field
        assert_eq!(qos1.reliability, qos2.reliability);
        assert_eq!(qos1.durability, qos2.durability);
        assert_eq!(qos1.history, qos2.history);
        assert_eq!(qos1.depth, qos2.depth);
    }

    // --- Locator validation tests ---

    #[test]
    fn test_locator_protocol_tcp() {
        assert_eq!(locator_protocol("tcp/127.0.0.1:7447"), LocatorProtocol::Tcp);
    }

    #[test]
    fn test_locator_protocol_serial() {
        assert_eq!(
            locator_protocol("serial//dev/ttyUSB0#baudrate=115200"),
            LocatorProtocol::Serial
        );
    }

    #[test]
    fn test_locator_protocol_unknown() {
        assert_eq!(locator_protocol(""), LocatorProtocol::Unknown);
        assert_eq!(locator_protocol("http://foo"), LocatorProtocol::Unknown);
        assert_eq!(locator_protocol("tls/host:port"), LocatorProtocol::Unknown);
    }

    #[test]
    fn test_locator_protocol_udp() {
        assert_eq!(locator_protocol("udp/127.0.0.1:7447"), LocatorProtocol::Udp);
        assert_eq!(
            locator_protocol("udp/192.168.1.50:2019"),
            LocatorProtocol::Udp
        );
    }

    #[test]
    fn test_validate_tcp_locator_ok() {
        assert!(validate_locator("tcp/127.0.0.1:7447").is_ok());
        assert!(validate_locator("tcp/192.168.1.1:7447").is_ok());
    }

    #[test]
    fn test_validate_tcp_locator_missing_port() {
        assert!(validate_locator("tcp/127.0.0.1").is_err());
    }

    #[test]
    fn test_validate_serial_locator_ok() {
        assert!(validate_locator("serial//dev/ttyUSB0#baudrate=115200").is_ok());
        assert!(validate_locator("serial//dev/ttyACM0#baudrate=9600").is_ok());
        assert!(validate_locator("serial/uart1#baudrate=921600").is_ok());
    }

    #[test]
    fn test_validate_serial_locator_empty_device() {
        assert!(validate_locator("serial/").is_err());
    }

    #[test]
    fn test_validate_serial_locator_missing_baudrate() {
        assert!(validate_locator("serial//dev/ttyUSB0").is_err());
    }

    #[test]
    fn test_validate_serial_locator_invalid_baudrate() {
        assert!(validate_locator("serial//dev/ttyUSB0#baudrate=abc").is_err());
    }

    #[test]
    fn test_validate_unknown_protocol() {
        assert!(validate_locator("http://foo").is_err());
        assert!(validate_locator("tls/host:port").is_err());
    }

    #[test]
    fn test_validate_udp_locator_ok() {
        assert!(validate_locator("udp/127.0.0.1:7447").is_ok());
        assert!(validate_locator("udp/192.168.1.50:2019").is_ok());
    }

    #[test]
    fn test_validate_udp_locator_missing_port() {
        assert!(validate_locator("udp/127.0.0.1").is_err());
    }

    // Phase 233.2 — the PX4 companion QoS profile must be BEST_EFFORT +
    // TRANSIENT_LOCAL + KEEP_LAST so it matches PX4's uxrce_dds_client endpoints.
    #[test]
    fn px4_qos_profile_matches_uxrce_dds_client() {
        let q = QosSettings::px4();
        assert_eq!(q.reliability, QosReliabilityPolicy::BestEffort);
        // VOLATILE — PX4's /fmu/out writers are volatile; a TRANSIENT_LOCAL
        // reader silently fails to match (verified against real PX4 SITL).
        assert_eq!(q.durability, QosDurabilityPolicy::Volatile);
        assert_eq!(q.history, QosHistoryPolicy::KeepLast);
        assert_eq!(q, QosSettings::QOS_PROFILE_PX4);
        // Depth is tunable via the builder without losing the PX4 policies.
        let deep = QosSettings::px4().keep_last(5);
        assert_eq!(deep.depth, 5);
        assert_eq!(deep.reliability, QosReliabilityPolicy::BestEffort);
        assert_eq!(deep.durability, QosDurabilityPolicy::Volatile);
    }

    // --- RmwConfig Tests ---

    #[test]
    fn test_rmw_config_default() {
        let config = RmwConfig::default();
        assert_eq!(config.locator, "tcp/127.0.0.1:7447");
        assert_eq!(config.mode, SessionMode::Client);
        assert_eq!(config.domain_id, 0);
        assert_eq!(config.node_name, "node");
        assert_eq!(config.namespace, "");
    }

    #[test]
    fn test_rmw_config_custom() {
        let config = RmwConfig {
            locator: "tcp/192.168.1.1:7447",
            mode: SessionMode::Peer,
            domain_id: 42,
            node_name: "talker",
            namespace: "/ns1",
            properties: &[("agent_port", "2019")],
        };
        assert_eq!(config.locator, "tcp/192.168.1.1:7447");
        assert_eq!(config.mode, SessionMode::Peer);
        assert_eq!(config.domain_id, 42);
        assert_eq!(config.node_name, "talker");
        assert_eq!(config.namespace, "/ns1");
        assert_eq!(config.properties.len(), 1);
        assert_eq!(config.properties[0].0, "agent_port");
    }

    #[test]
    fn test_rmw_config_is_copy() {
        let config = RmwConfig::default();
        let config2 = config; // Copy
        assert_eq!(config.locator, config2.locator);
        assert_eq!(config.domain_id, config2.domain_id);
    }

    #[test]
    fn test_rmw_config_clone() {
        let config = RmwConfig::default();
        let cloned = RmwConfig { ..config };
        assert_eq!(cloned.locator, config.locator);
        assert_eq!(cloned.node_name, config.node_name);
    }
}