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`] / [`Subscription`] — pub/sub data transport
//! - [`ServiceTrait`] / [`ClientTrait`] — request/reply
//! - [`Rmw`] — top-level factory that creates sessions

use crate::service_inbox::ServiceInboxSpec;
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,
    /// phase-519 W3 (RFC-0074) — the rate this subscription DECLARED it can
    /// take (`arrival_curve.rate_hz`), in millihertz. `0` = undeclared. Carried to
    /// the backend in `rmw_subscription_options_t::arrival_rate_mhz`; the XRCE
    /// backend turns it into the agent's `max_bytes_per_second`, zenoh ignores
    /// it (its enforcement point 1 is the router rule). Ignored by publishers.
    pub arrival_rate_mhz: u32,
    /// phase-279 (#145) — publisher-side "express" hint: this topic's samples
    /// bypass transport tx batching (zenoh: the wire EXPRESS flag; a batching
    /// zenoh-pico session sends them immediately instead of queueing them for
    /// the next flush). For control-tier / latency-sensitive topics. Ignored by
    /// subscriptions and by backends without a batching concept.
    pub tx_express: bool,
}

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,
            arrival_rate_mhz: 0,
            tx_express: false,
        }
    }

    /// 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
    }

    /// phase-519 W3 (RFC-0074) — set the declared arrival rate, millihertz
    /// (`rate_hz x 1000`). `0` leaves it undeclared.
    pub const fn with_arrival_rate_mhz(mut self, rate_mhz: u32) -> Self {
        self.arrival_rate_mhz = rate_mhz;
        self
    }

    /// phase-279 (#145) — mark this publisher's samples express: they bypass
    /// transport tx batching (sent immediately even when `ZPICO_TX_BATCH` is
    /// on). For control-tier / latency-sensitive topics.
    pub const fn with_tx_express(mut self, express: bool) -> Self {
        self.tx_express = express;
        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,
}

/// Issue 1913 — the REP-2011 hash each of an action's five endpoints is keyed
/// on. An Iron+ `rmw_zenoh_cpp` peer keys every endpoint on its OWN type: the
/// `<Action>_SendGoal` / `<Action>_GetResult` services, the standard
/// `action_msgs/srv/CancelGoal`, `<Action>_FeedbackMessage`, and the standard
/// `action_msgs/msg/GoalStatusArray`. One hash for all five matched none of
/// them, which is what every registration path carried until 1913.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct ActionEndpointHashes<'a> {
    pub send_goal: &'a str,
    pub cancel_goal: &'a str,
    pub get_result: &'a str,
    pub feedback: &'a str,
    pub status: &'a str,
}

impl<'a> ActionEndpointHashes<'a> {
    /// Every endpoint on one value — Humble's `TypeHashNotSupported`, or a
    /// caller that has nothing better (it will not match an Iron+ peer).
    pub const fn uniform(hash: &'a str) -> Self {
        Self {
            send_goal: hash,
            cancel_goal: hash,
            get_result: hash,
            feedback: hash,
            status: hash,
        }
    }
}

impl ActionEndpointHashes<'static> {
    /// The five hashes a typed action carries.
    pub const fn of<A: nros_core::RosAction>() -> Self {
        Self {
            send_goal: A::SEND_GOAL_SERVICE_HASH,
            cancel_goal: A::CANCEL_GOAL_SERVICE_HASH,
            get_result: A::GET_RESULT_SERVICE_HASH,
            feedback: <A::FeedbackMessage as nros_core::RosMessage>::TYPE_HASH,
            status: A::STATUS_MESSAGE_HASH,
        }
    }
}

/// 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 (`rcl_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.
    ///
    /// **It NAMES the policy** — phase-417 G7. The payload is the single
    /// [`QoSPolicyMask`] bit that was refused, so
    /// `err.qos_policy_name()` (or `mask.policy_name()`) turns the refusal
    /// into `"RELIABILITY"` without an allocator. Until this carried a
    /// payload, [`QoSProfile::validate_against`] computed the offending bit
    /// and threw it away, while `rmw_vtable.h`'s `supported_qos_policies`
    /// block told the reader those creates fail "naming a policy". They did
    /// not. Issue 1329's NONE-over-union choice rests on that sentence, so the
    /// choice was right and its evidence was not yet true.
    ///
    /// [`QoSPolicyMask::NONE`] is the honest answer where the refusal crossed
    /// a boundary that cannot carry one — a C backend returns
    /// `NROS_RMW_RET_INCOMPATIBLE_QOS`, an `rmw_ret_t` with no room for a
    /// mask — and `policy_name()` then answers `None` rather than naming a
    /// policy nobody identified.
    IncompatibleQos(QoSPolicyMask),
    /// 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),
}

/// One line per variant, in the defining crate, so every consumer renders the
/// same words (issue 0783: `NodeError` used to fall back to `{:?}` because
/// this impl did not exist, and writing the prose there would have put a
/// backend's wording in the wrong crate). `no_std`: no allocator, only
/// `write_str` and the payloads the variants already carry.
impl core::fmt::Display for TransportError {
    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
        let s = match self {
            Self::ConnectionFailed => "failed to connect to the transport",
            Self::Disconnected => "the transport connection was closed",
            Self::PublisherCreationFailed => "failed to create a publisher",
            Self::SubscriberCreationFailed => "failed to create a subscriber",
            Self::ServiceServerCreationFailed => "failed to create a service server",
            Self::ServiceClientCreationFailed => "failed to create a service client",
            Self::PublishFailed => "failed to publish a message",
            Self::ServiceRequestFailed => "failed to send a service request",
            Self::ServiceReplyFailed => "failed to send a service reply",
            Self::SerializationError => "serialization failed",
            Self::DeserializationError => "deserialization failed",
            Self::BufferTooSmall => "buffer too small",
            Self::MessageTooLarge => "incoming message exceeds the static buffer",
            Self::Timeout => "timed out",
            Self::InvalidConfig => "invalid configuration",
            Self::WouldBlock => "resource momentarily unavailable; retry",
            Self::TooLarge => "requested size exceeds the backend's capacity",
            Self::TaskStartFailed => "failed to start a background task",
            Self::PollFailed => "failed to poll for incoming messages",
            Self::KeepaliveFailed => "failed to send a keepalive",
            Self::JoinFailed => "failed to send a join message",
            Self::InvalidArgument => "invalid argument",
            Self::Unsupported => "operation not supported by this backend",
            Self::BadAlloc => "memory allocation failed",
            Self::IncompatibleQos(policy) => {
                return match policy.policy_name() {
                    Some(name) => write!(f, "incompatible QoS: {name}"),
                    None => f.write_str("incompatible QoS"),
                };
            }
            Self::TopicNameInvalid => "invalid topic, service or action name",
            Self::NodeNameNonExistent => "no such node in this session",
            Self::LoanNotSupported => "message loans unavailable on this entity",
            Self::NoData => "no data available",
            Self::IncompatibleAbi => "incompatible ABI version",
            Self::Backend(msg) => return write!(f, "backend error: {msg}"),
            #[cfg(feature = "alloc")]
            Self::BackendDynamic(msg) => return write!(f, "backend error: {msg}"),
        };
        f.write_str(s)
    }
}

impl TransportError {
    /// The QoS policy this error names, or `None`.
    ///
    /// phase-417 G7 — the one accessor the three surfaces share. `Some` only
    /// for [`Self::IncompatibleQos`] carrying exactly one
    /// [`QoSPolicyMask`] bit; `None` for every other error, and for a
    /// refusal that crossed a boundary with no room for the mask.
    ///
    /// The string is borrowed from [`QoSPolicyMask::NAMED`], so this costs no
    /// allocator and no formatter and is available on a bare-metal target.
    pub fn qos_policy_name(&self) -> Option<&'static str> {
        match self {
            Self::IncompatibleQos(policy) => policy.policy_name(),
            _ => None,
        }
    }
}

/// QoS history policy
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum QoSHistoryPolicy {
    /// **The caller did not state a history policy** — the backend picks.
    /// Upstream's `RMW_QOS_POLICY_HISTORY_SYSTEM_DEFAULT`; lowers to
    /// `NROS_RMW_HISTORY_SYSTEM_DEFAULT` (0) across the C ABI.
    ///
    /// See [`QoSSystemDefaults`] for how a backend resolves it. Listed
    /// first to match rclrs's enumerator order.
    SystemDefault,
    /// Keep last N messages (where N is defined in QoSProfile)
    #[default]
    KeepLast,
    /// Keep all messages (up to resource limits)
    KeepAll,
    /// **The backend could not determine this policy.** Upstream's
    /// `RMW_QOS_POLICY_HISTORY_UNKNOWN`; lowers to `NROS_RMW_HISTORY_UNKNOWN`
    /// (3) across the C ABI.
    ///
    /// An ABSENCE, not a value: it is what a `*_get_actual_qos` read-back
    /// writes for a policy it looked at and cannot report, and it is never a
    /// request an application makes. Distinct from `SystemDefault`, which is
    /// "nobody asked, so the middleware chooses" — see
    /// [`QoSProfile::QOS_PROFILE_UNKNOWN`].
    Unknown,
}

/// QoS reliability policy
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum QoSReliabilityPolicy {
    /// **The caller did not state a reliability policy** — the backend picks.
    /// Upstream's `RMW_QOS_POLICY_RELIABILITY_SYSTEM_DEFAULT`; lowers to
    /// `NROS_RMW_RELIABILITY_SYSTEM_DEFAULT` (0) across the C ABI.
    ///
    /// See [`QoSSystemDefaults`].
    SystemDefault,
    /// Reliable delivery (retransmit if needed).
    ///
    /// Default — matches ROS 2 `rmw_qos_profile_default` and the
    /// `QoSProfile::default()` / `QOS_PROFILE_DEFAULT` aggregates.
    #[default]
    Reliable,
    /// Best-effort delivery (no retransmits)
    BestEffort,
    /// **The backend could not determine this policy.** Upstream's
    /// `RMW_QOS_POLICY_RELIABILITY_UNKNOWN`; lowers to `NROS_RMW_RELIABILITY_UNKNOWN`
    /// (3) across the C ABI.
    ///
    /// An ABSENCE, not a value: it is what a `*_get_actual_qos` read-back
    /// writes for a policy it looked at and cannot report, and it is never a
    /// request an application makes. Distinct from `SystemDefault`, which is
    /// "nobody asked, so the middleware chooses" — see
    /// [`QoSProfile::QOS_PROFILE_UNKNOWN`].
    Unknown,
    /// **Resolve at creation against the discovered graph** (issue 1973).
    /// Upstream's `RMW_QOS_POLICY_RELIABILITY_BEST_AVAILABLE` (Iron+); lowers
    /// to `NROS_RMW_RELIABILITY_BEST_AVAILABLE` (4) across the C ABI.
    ///
    /// A REQUEST, never a grant: the runtime replaces it before the backend's
    /// create slot runs ([`QoSProfile::resolve_best_available`]), so a
    /// `*_get_actual_qos` read-back reports the concrete policy chosen.
    BestAvailable,
}

/// QoS durability policy
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum QoSDurabilityPolicy {
    /// **The caller did not state a durability policy** — the backend picks.
    /// Upstream's `RMW_QOS_POLICY_DURABILITY_SYSTEM_DEFAULT`; lowers to
    /// `NROS_RMW_DURABILITY_SYSTEM_DEFAULT` (0) across the C ABI.
    ///
    /// See [`QoSSystemDefaults`].
    SystemDefault,
    /// Messages are discarded when subscriber disconnects
    #[default]
    Volatile,
    /// Messages are persisted for late-joining subscribers
    TransientLocal,
    /// **The backend could not determine this policy.** Upstream's
    /// `RMW_QOS_POLICY_DURABILITY_UNKNOWN`; lowers to `NROS_RMW_DURABILITY_UNKNOWN`
    /// (3) across the C ABI.
    ///
    /// An ABSENCE, not a value: it is what a `*_get_actual_qos` read-back
    /// writes for a policy it looked at and cannot report, and it is never a
    /// request an application makes. Distinct from `SystemDefault`, which is
    /// "nobody asked, so the middleware chooses" — see
    /// [`QoSProfile::QOS_PROFILE_UNKNOWN`].
    Unknown,
    /// **Resolve at creation against the discovered graph** (issue 1973).
    /// Upstream's `RMW_QOS_POLICY_DURABILITY_BEST_AVAILABLE` (Iron+); lowers
    /// to `NROS_RMW_DURABILITY_BEST_AVAILABLE` (4) across the C ABI. A
    /// request only — see [`QoSReliabilityPolicy::BestAvailable`].
    BestAvailable,
}

/// QoS liveliness policy. Matches DDS `LIVELINESS` semantics.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
#[repr(u8)]
pub enum QoSLivelinessPolicy {
    /// Discriminant 0 — the same value the C ABI spells
    /// `NROS_RMW_LIVELINESS_SYSTEM_DEFAULT` and upstream spells
    /// `RMW_QOS_POLICY_LIVELINESS_SYSTEM_DEFAULT`.
    ///
    /// phase-428 W10 — the NAME diverges from both and the VALUE does not, so
    /// this is what an upstream-mirroring profile carries. Read it as "the
    /// middleware's choice", not as "liveliness is off": a backend that sees 0
    /// omits the DDS call entirely and inherits the DDS default, which is
    /// AUTOMATIC. `None` is our spelling, kept because it is the C ABI
    /// discriminant three enums already agree on; renaming it is a separate
    /// question from getting the value right, and only the value crosses.
    ///
    /// No liveliness assertion or tracking of our own.
    #[default]
    None = 0,
    /// Backend's keepalive task asserts liveliness automatically.
    Automatic = 1,
    /// Application calls `assert_liveliness()` at the node level.
    ///
    /// Phase 376 W5/B2 — this was 3 and `ManualByTopic` was 2, the opposite of
    /// upstream. The discriminant IS the C ABI value (`liveliness_kind` is
    /// written with `as u8`), and the cyclonedds backend turns it into a real
    /// DDS liveliness kind that a ROS peer matches on, so the swap was visible
    /// on the wire and nowhere else.
    ManualByNode = 2,
    /// Application calls `assert_liveliness()` per topic explicitly.
    ManualByTopic = 3,
    /// **The backend could not determine this policy.** Discriminant 4, the
    /// value the C ABI spells `NROS_RMW_LIVELINESS_UNKNOWN`.
    ///
    /// An ABSENCE, not a value — see [`QoSProfile::QOS_PROFILE_UNKNOWN`]. A
    /// `*_get_actual_qos` read-back writes it for a policy it cannot report;
    /// an application never requests it.
    Unknown = 4,
    /// **Resolve at creation against the discovered graph** (issue 1973).
    /// Discriminant 5, upstream's `RMW_QOS_POLICY_LIVELINESS_BEST_AVAILABLE`
    /// (Iron+) and the C ABI's `NROS_RMW_LIVELINESS_BEST_AVAILABLE`. A
    /// request only — see [`QoSReliabilityPolicy::BestAvailable`]. Not manual:
    /// it resolves to `Automatic` or `ManualByTopic` before anything asks.
    BestAvailable = 5,
}

impl QoSLivelinessPolicy {
    /// `true` for the two kinds under which the APPLICATION owns the
    /// assertion — [`ManualByNode`](Self::ManualByNode) and
    /// [`ManualByTopic`](Self::ManualByTopic).
    ///
    /// The ONE spelling of that test (the phase-467 RMW gap-closure design
    /// study's Row 7). It decides what [`Publisher::assert_liveliness`] owes:
    /// under a manual kind the caller asked for something a backend either
    /// performs or cannot, so a backend that cannot must answer
    /// `Unsupported`; under `Automatic` / `None` there is nothing to assert
    /// and `Ok(())` is the whole truth. `Unknown` is an absence, never a
    /// request, so it is not manual.
    #[must_use]
    pub const fn is_manual(self) -> bool {
        matches!(self, Self::ManualByNode | Self::ManualByTopic)
    }
}

/// 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),
    /// Issue 0303 — `.deadline` → [`QoSProfile::deadline_ms`]. `0` /
    /// [`DURATION_INFINITE_MS`] = no deadline check.
    Deadline(u32),
    /// Issue 0303 — `.lifespan` → [`QoSProfile::lifespan_ms`].
    Lifespan(u32),
    /// Issue 0303 — `.liveliness` → [`QoSProfile::liveliness_kind`].
    Liveliness(QoSLivelinessPolicy),
    /// Issue 0303 — `.liveliness_lease_duration` →
    /// [`QoSProfile::liveliness_lease_ms`].
    LivelinessLease(u32),
}

/// Issue 0303 — the wire form a baked QoS override travels in:
/// `(topic, role, policy, value)`, all primitives.
///
/// One numbering for every language: `rclc_qos_override_t` (C),
/// `nros_cpp_qos_override_t` (C++), `RuntimeCtx::qos_overrides` (the Rust
/// entry bake) and `NodeRecord::qos_overrides` are all THIS tuple. Codes rather
/// than [`QoSOverride`] because the table crosses the C ABI and rides
/// `nros-platform`, which sits below this crate in the layer graph.
///
/// `topic` is `&'static` because every producer bakes a literal.
pub type QoSOverrideCode = (&'static str, u8, u8, u32);

/// `role` codes for [`QoSOverrideCode`].
pub mod qos_override_role {
    /// The override targets publishers on the topic.
    pub const PUBLISHER: u8 = 0;
    /// The override targets subscriptions on the topic.
    pub const SUBSCRIPTION: u8 = 1;
}

/// `policy` codes for [`QoSOverrideCode`].
///
/// Append-only: these numbers are baked into shipped images and mirrored in
/// two C headers. Never renumber — add.
pub mod qos_override_policy {
    /// value `0` = best_effort, `1` = reliable.
    pub const RELIABILITY: u8 = 0;
    /// value `0` = volatile, `1` = transient_local.
    pub const DURABILITY: u8 = 1;
    /// value `0` = keep_last, `1` = keep_all.
    pub const HISTORY: u8 = 2;
    /// value = the KeepLast depth.
    pub const DEPTH: u8 = 3;
    /// value = milliseconds (issue 0303).
    pub const DEADLINE: u8 = 4;
    /// value = milliseconds (issue 0303).
    pub const LIFESPAN: u8 = 5;
    /// value = [`super::QoSLivelinessPolicy`] discriminant (issue 0303).
    pub const LIVELINESS: u8 = 6;
    /// value = milliseconds (issue 0303).
    pub const LIVELINESS_LEASE: u8 = 7;
}

/// Decode one [`QoSOverrideCode`] into a typed [`QoSOverride`].
///
/// `None` for an unrecognised role or policy code. THE one decoder: before
/// issue 0303 this match existed four times (nros-node, nros-c, nros-cpp, and
/// the executor's node record), each with a silent catch-all, so adding a
/// policy meant finding all four — and the two FFI copies had already been
/// forgotten once.
pub fn decode_qos_override(code: &QoSOverrideCode) -> Option<QoSOverride> {
    let (topic, role, policy, value) = *code;
    decode_qos_override_parts(topic, role, policy, value)
}

/// [`decode_qos_override`] over loose parts — for the FFI paths, which read the
/// fields out of a `#[repr(C)]` struct rather than a tuple.
pub fn decode_qos_override_parts(
    topic: &'static str,
    role: u8,
    policy: u8,
    value: u32,
) -> Option<QoSOverride> {
    Some(QoSOverride {
        topic,
        role: decode_qos_override_role(role)?,
        value: decode_qos_override_value(policy, value)?,
    })
}

/// Decode a `role` code. `None` for an unrecognised one.
pub fn decode_qos_override_role(role: u8) -> Option<QoSOverrideRole> {
    match role {
        qos_override_role::PUBLISHER => Some(QoSOverrideRole::Publisher),
        qos_override_role::SUBSCRIPTION => Some(QoSOverrideRole::Subscription),
        _ => None,
    }
}

/// Decode a `(policy, value)` code pair. `None` for an unrecognised policy or
/// an out-of-range enum value.
///
/// Split out from [`decode_qos_override`] for the FFI paths: they have already
/// matched the topic against a `*const c_char`, so they need the VALUE without
/// a `&'static str` to build a whole [`QoSOverride`] around.
pub fn decode_qos_override_value(policy: u8, value: u32) -> Option<QoSOverrideValue> {
    let out = match policy {
        qos_override_policy::RELIABILITY => QoSOverrideValue::Reliability(if value == 0 {
            QoSReliabilityPolicy::BestEffort
        } else {
            QoSReliabilityPolicy::Reliable
        }),
        qos_override_policy::DURABILITY => QoSOverrideValue::Durability(if value == 1 {
            QoSDurabilityPolicy::TransientLocal
        } else {
            QoSDurabilityPolicy::Volatile
        }),
        qos_override_policy::HISTORY => QoSOverrideValue::History(if value == 1 {
            QoSHistoryPolicy::KeepAll
        } else {
            QoSHistoryPolicy::KeepLast
        }),
        qos_override_policy::DEPTH => QoSOverrideValue::Depth(value),
        qos_override_policy::DEADLINE => QoSOverrideValue::Deadline(value),
        qos_override_policy::LIFESPAN => QoSOverrideValue::Lifespan(value),
        // The encoder is `nros_orchestration_ir::qos_override`; both ends name
        // the variant rather than its number, so W5/B2's renumbering moved them
        // together instead of silently swapping two policies.
        qos_override_policy::LIVELINESS => QoSOverrideValue::Liveliness(match value {
            v if v == QoSLivelinessPolicy::None as u32 => QoSLivelinessPolicy::None,
            v if v == QoSLivelinessPolicy::Automatic as u32 => QoSLivelinessPolicy::Automatic,
            v if v == QoSLivelinessPolicy::ManualByNode as u32 => QoSLivelinessPolicy::ManualByNode,
            v if v == QoSLivelinessPolicy::ManualByTopic as u32 => {
                QoSLivelinessPolicy::ManualByTopic
            }
            _ => return None,
        }),
        qos_override_policy::LIVELINESS_LEASE => QoSOverrideValue::LivelinessLease(value),
        _ => return None,
    };
    Some(out)
}

/// 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 [`QoSProfile`] 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,
}

/// Phase-301 (issue 0241) — explicit "infinite" spelling for the u32
/// millisecond QoS duration fields (`deadline_ms`, `lifespan_ms`,
/// `liveliness_lease_ms`). Semantically identical to `0` (unset /
/// no-check) at every check site; exists so a caller can distinguish
/// "default" from "deliberately infinite". Mirrors the C header's
/// `NROS_RMW_DURATION_INFINITE_MS`.
pub const DURATION_INFINITE_MS: u32 = u32::MAX;

/// Issue 1973 — the "best available" spelling for `deadline_ms` and
/// `liveliness_lease_ms`: resolve at creation against the discovered graph.
///
/// Upstream spells it `RMW_QOS_DEADLINE_BEST_AVAILABLE` /
/// `RMW_QOS_LIVELINESS_LEASE_DURATION_BEST_AVAILABLE`, both
/// `RMW_DURATION_INFINITE - 1`; this is the same idea in our u32-millisecond
/// fields: [`DURATION_INFINITE_MS`] minus one. A REQUEST only — the runtime
/// replaces it before a backend sees the profile
/// ([`QoSProfile::resolve_best_available`]). Mirrors the C header's
/// `NROS_RMW_DURATION_BEST_AVAILABLE_MS`. Not meaningful for `lifespan_ms`,
/// as upstream defines no best-available lifespan.
pub const DURATION_BEST_AVAILABLE_MS: u32 = DURATION_INFINITE_MS - 1;

/// Phase-301 (issue 0241) — lower a [`core::time::Duration`] into a u32
/// millisecond QoS field. Boundary contract:
///
/// - zero stays `0` (unset / no-check);
/// - sub-millisecond remainders CEIL to the next ms (rounding down
///   would silently turn a real deadline into "no deadline");
/// - values at or past [`DURATION_INFINITE_MS`] ms are a create-time
///   error, never a clamp (infinite is requested via the sentinel or
///   `0`, not by a huge finite duration). The bound starts one below, at
///   [`DURATION_BEST_AVAILABLE_MS`] (issue 1973), so no finite duration can
///   collide with that sentinel either.
pub fn duration_to_qos_ms(d: core::time::Duration) -> Result<u32, TransportError> {
    if d.is_zero() {
        return Ok(0);
    }
    let mut ms = d.as_millis();
    if !d.subsec_nanos().is_multiple_of(1_000_000) {
        ms += 1;
    }
    if ms >= DURATION_BEST_AVAILABLE_MS as u128 {
        return Err(TransportError::InvalidArgument);
    }
    Ok(ms as u32)
}

/// 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 QoSProfile {
    /// 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,
    /// Subscription 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,
    /// Phase 282 (#145) — publisher-side "express" hint: this publisher's
    /// samples bypass transport tx batching (zenoh: the wire EXPRESS flag; a
    /// batching zenoh-pico session sends them immediately instead of queueing
    /// them for the next flush). A transport hint, not a DDS policy — no RxO
    /// matching, no backend-compat validation; ignored by subscriptions and by
    /// backends without a batching concept.
    pub tx_express: bool,
}

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

/// The depth sentinel — "the caller did not state a queue depth".
///
/// issue 0829. Upstream spells it `RMW_QOS_POLICY_DEPTH_SYSTEM_DEFAULT = 0`
/// (`rmw/include/rmw/types.h`); depth is `size_t` there and `uint16_t` in our
/// C ABI, but 0 is 0 in both. The value was already free on every backend:
/// Cyclone REJECTS `KEEP_LAST` with depth 0 outright
/// (`validate_history_qospolicy`, `ddsi_plist.c:2603-2604`), the XRCE client
/// already reads it as "unstated" and drops the field from the wire
/// (`create_entities_bin.c:148`), and both `read_entity_qos`
/// (`nros-rmw-cyclonedds/src/qos.cpp:138`) and `report_qos_downgrade`
/// (`nros-rmw-cffi/src/lib.rs:1903`) already treat a 0 read-back as "no
/// answer" rather than as an answer.
pub const DEPTH_SYSTEM_DEFAULT: u32 = 0;

/// What ONE backend resolves the `SYSTEM_DEFAULT` sentinel to.
///
/// issue 0829. `rmw_qos_profile_system_default` is an absence, and the RMW
/// fills it — which means the answer is per backend and there is no constant
/// this crate could bake. Each backend declares its own `QoSSystemDefaults`
/// and applies it with [`QoSProfile::resolve_system_default`] at its create
/// entry, **before anything is derived from the QoS**.
///
/// That ordering is load-bearing on the zenoh path: the profile is serialised
/// into the liveliness-token keyexpr that a ROS `rmw_zenoh_cpp` peer parses
/// out of the graph (`nros-rmw-zenoh/src/keyexpr.rs`). Upstream resolves in
/// best_available_qos before the entity and its token exist, so its tokens
/// never carry a sentinel; ours must not either, or we advertise `0:0:0,0` to
/// peers as though it were a policy.
///
/// The values a backend picks should mirror **the corresponding upstream
/// RMW**, not the raw middleware default — interop with a ROS peer is the
/// requirement, and the two differ. Leaving `dds_qset_reliability` unset gives
/// Cyclone's own reader default of BEST_EFFORT (`ddsi_plist.c:3470`), where
/// `rmw_cyclonedds_cpp` deliberately picks RELIABLE.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct QoSSystemDefaults {
    /// Concrete reliability. Must not itself be `SystemDefault`.
    pub reliability: QoSReliabilityPolicy,
    /// Concrete durability. Must not itself be `SystemDefault`.
    pub durability: QoSDurabilityPolicy,
    /// Concrete history. Must not itself be `SystemDefault`.
    pub history: QoSHistoryPolicy,
    /// Concrete queue depth. `0` means the backend genuinely has no answer and
    /// defers further down the stack — the XRCE case, where the client encodes
    /// exactly that (`optional_history_depth = false`) and the Agent's DDS
    /// layer resolves it.
    pub depth: u32,
}

impl QoSProfile {
    /// 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 {
                self.apply_override_value(ovr.value);
            }
        }
        self
    }

    /// Issue 0303 — apply ONE decoded override value. The single place a
    /// policy maps to the field it sets; `apply_overrides` and the FFI folds
    /// both go through it, so a new policy cannot reach some paths only.
    pub fn apply_override_value(&mut self, value: QoSOverrideValue) {
        match 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,
            QoSOverrideValue::Deadline(ms) => self.deadline_ms = ms,
            QoSOverrideValue::Lifespan(ms) => self.lifespan_ms = ms,
            QoSOverrideValue::Liveliness(k) => self.liveliness_kind = k,
            QoSOverrideValue::LivelinessLease(ms) => self.liveliness_lease_ms = ms,
        }
    }

    /// Issue 0303 — fold baked [`QoSOverrideCode`]s for one `(topic, role)`.
    /// Unrecognised codes are skipped; the producers reject them at BAKE time
    /// (`nros_orchestration_ir::qos_override`), so a code reaching here that
    /// this build does not know is an older image running newer data, not a
    /// user error to diagnose at runtime.
    pub fn apply_override_codes(
        self,
        topic: &str,
        role: QoSOverrideRole,
        codes: &[QoSOverrideCode],
    ) -> Self {
        let mut qos = self;
        for code in codes {
            if let Some(ovr) = decode_qos_override(code) {
                qos = qos.apply_overrides(topic, role, core::slice::from_ref(&ovr));
            }
        }
        qos
    }
}

/// The ONE authoring site for every named [`QoSProfile`] preset — phase-428 W10.
///
/// Upstream's profiles were hand-transcribed here and in three other places
/// (`nros::qos`, `rclcpp/qos.hpp`, `rmw_entity.h`) with nothing binding them, and
/// they drifted exactly as every other hand-mirror in this tree has: issue 0793
/// caught `PARAMETERS` claiming TRANSIENT_LOCAL, fixed that one profile, and
/// left `PARAMETER_EVENTS` sitting at `KeepAll, depth 0` under a comment saying
/// *"matches rmw_qos_profile_parameter_events"* — where upstream is
/// `KEEP_LAST, 1000`. A RELIABLE writer with KEEP_ALL **blocks** when history
/// fills where KEEP_LAST(1000) overwrites, so that was an inverting difference,
/// not a weaker one, and the test beneath it asserted the wrong value.
///
/// Three properties are what make this a table rather than ten `const`s:
///
/// 1. **Every field is stated.** The `Self::build(...)` helper this replaces
///    supplied `liveliness_kind`, the three duration windows and both flags
///    silently, so five of ten fields per profile were invisible at the
///    authoring site and no reader could compare a row against upstream.
/// 2. **Every row declares its PROVENANCE** — the upstream symbol it mirrors,
///    or `nros: <why>` for a preset we invented. A profile cannot be added
///    without answering "whose constant is this?".
/// 3. **The field ORDER is upstream's** (`rmw_qos_profile_t` in
///    `rmw/include/rmw/types.h`: history, depth, reliability, durability,
///    deadline, lifespan, liveliness, lease, avoid-ns), so a row reads against
///    the header positionally.
///
/// Expansion is exactly the `pub const` items that were here before — no table
/// is materialised, nothing is `static`, and the crate stays `no_std` and
/// zero-cost on every target it reaches.
///
/// `scripts/check-qos-profile-ssot.py` parses this invocation and asserts every
/// upstream-mirroring row field-by-field against
/// `docs/reference/rmw-qos-profiles.txt`, which is itself extracted from a real
/// ROS install. The other three sites bind to this block.
macro_rules! qos_profiles {
    ($(
        $(#[$meta:meta])*
        $name:ident : $provenance:literal = {
            history: $history:ident,
            depth: $depth:expr,
            reliability: $reliability:ident,
            durability: $durability:ident,
            deadline_ms: $deadline_ms:expr,
            lifespan_ms: $lifespan_ms:expr,
            liveliness_kind: $liveliness_kind:ident,
            liveliness_lease_ms: $liveliness_lease_ms:expr,
            avoid_ros_namespace_conventions: $avoid:expr,
            tx_express: $tx_express:expr,
        };
    )+) => {
        impl QoSProfile {
            $(
                $(#[$meta])*
                ///
                #[doc = concat!("Provenance: `", $provenance, "`.")]
                pub const $name: Self = Self {
                    history: QoSHistoryPolicy::$history,
                    depth: $depth,
                    reliability: QoSReliabilityPolicy::$reliability,
                    durability: QoSDurabilityPolicy::$durability,
                    deadline_ms: $deadline_ms,
                    lifespan_ms: $lifespan_ms,
                    liveliness_kind: QoSLivelinessPolicy::$liveliness_kind,
                    liveliness_lease_ms: $liveliness_lease_ms,
                    avoid_ros_namespace_conventions: $avoid,
                    tx_express: $tx_express,
                };
            )+
        }
    };
}

// nros-qos-table: BEGIN
// The SSoT. Everything between this marker and `nros-qos-table: END` is parsed
// by `scripts/check-qos-profile-ssot.py`; do not add a `QoSProfile` preset
// anywhere else in this crate — the gate refuses one.
qos_profiles! {
    /// Best-effort QoS (for real-time).
    BEST_EFFORT: "nros: best-effort keep-last(1); a shorthand, not an upstream constant" = {
        history: KeepLast,
        depth: 1,
        reliability: BestEffort,
        durability: Volatile,
        deadline_ms: 0,
        lifespan_ms: 0,
        liveliness_kind: Automatic,
        liveliness_lease_ms: 0,
        avoid_ros_namespace_conventions: false,
        tx_express: false,
    };

    /// Reliable QoS.
    RELIABLE: "nros: reliable keep-last(10); a shorthand, not an upstream constant" = {
        history: KeepLast,
        depth: 10,
        reliability: Reliable,
        durability: Volatile,
        deadline_ms: 0,
        lifespan_ms: 0,
        liveliness_kind: Automatic,
        liveliness_lease_ms: 0,
        avoid_ros_namespace_conventions: false,
        tx_express: false,
    };

    /// `rmw_qos_profile_system_default` — **an absence, not a profile.**
    ///
    /// issue 0829. Every field is the sentinel: upstream's constant names no
    /// concrete policy at all, and the two reference RMWs fill the absence with
    /// different numbers — `rmw_cyclonedds_cpp`'s `create_readwrite_qos` folds
    /// `RMW_QOS_POLICY_DEPTH_SYSTEM_DEFAULT` to `KEEP_LAST(1)`, while
    /// `rmw_zenoh_cpp`'s `QoS::QoS()` fills it from
    /// `RMW_ZENOH_DEFAULT_HISTORY_DEPTH`, which is 42, over a comment stating
    /// the contract outright: *"If the depth field in the qos profile is set to
    /// 0, the RMW implementation has the liberty to assign a default depth."*
    ///
    /// So no baked number can be right. This carried a concrete
    /// `Reliable / Volatile / KeepLast(1)` until 2026-09-03, and
    /// `nros::qos::SYSTEM_DEFAULT` carried a concrete depth **10**, which is how
    /// one name shipped two queue depths. Both are gone; the backend resolves
    /// this at its create entry via [`QoSProfile::resolve_system_default`].
    ///
    /// `liveliness_kind` is [`QoSLivelinessPolicy::None`], which IS the
    /// sentinel on that policy — it lowers to
    /// `NROS_RMW_LIVELINESS_SYSTEM_DEFAULT` (0), the two having collapsed onto
    /// one value in phase-376 W5/B2. This is therefore the ONE preset that
    /// matches upstream on all nine fields.
    QOS_PROFILE_SYSTEM_DEFAULT: "rmw_qos_profile_system_default" = {
        history: SystemDefault,
        depth: DEPTH_SYSTEM_DEFAULT,
        reliability: SystemDefault,
        durability: SystemDefault,
        deadline_ms: 0,
        lifespan_ms: 0,
        liveliness_kind: None,
        liveliness_lease_ms: 0,
        avoid_ros_namespace_conventions: false,
        tx_express: false,
    };

    /// `rmw_qos_profile_unknown` — **"the backend could not determine this"**,
    /// and distinct from [`QOS_PROFILE_SYSTEM_DEFAULT`](Self::QOS_PROFILE_SYSTEM_DEFAULT).
    ///
    /// `SYSTEM_DEFAULT` means "nobody asked, so the middleware chooses";
    /// this means "we looked and cannot say", which is what a
    /// `*_get_actual_qos` read-back owes its caller for a policy it cannot
    /// report. It is the starting value the read-back helper pre-loads, so a
    /// field the backend does not overwrite reads as an ABSENCE rather than as
    /// a confident grant of whatever was asked.
    ///
    /// **Never a request.** No `create_*` accepts it — a backend's
    /// `qos_admit` refuses `UNKNOWN` on every policy, because an application
    /// asking for "I cannot say" is a bug, not a preference.
    ///
    /// The C ABI twin is `NROS_RMW_QOS_PROFILE_UNKNOWN` (`rmw_entity.h`),
    /// which existed from phase-428 W10 with no Rust counterpart: three of the
    /// four policy enums had no `Unknown` variant to build it from. They do
    /// now.
    QOS_PROFILE_UNKNOWN: "rmw_qos_profile_unknown" = {
        history: Unknown,
        depth: 0,
        reliability: Unknown,
        durability: Unknown,
        deadline_ms: 0,
        lifespan_ms: 0,
        liveliness_kind: Unknown,
        liveliness_lease_ms: 0,
        avoid_ros_namespace_conventions: false,
        tx_express: false,
    };

    /// Default QoS profile — reliable, volatile, keep-last(10).
    QOS_PROFILE_DEFAULT: "rmw_qos_profile_default" = {
        history: KeepLast,
        depth: 10,
        reliability: Reliable,
        durability: Volatile,
        deadline_ms: 0,
        lifespan_ms: 0,
        liveliness_kind: None,
        liveliness_lease_ms: 0,
        avoid_ros_namespace_conventions: false,
        tx_express: false,
    };

    /// Sensor data QoS profile — best-effort, volatile, keep-last(5).
    QOS_PROFILE_SENSOR_DATA: "rmw_qos_profile_sensor_data" = {
        history: KeepLast,
        depth: 5,
        reliability: BestEffort,
        durability: Volatile,
        deadline_ms: 0,
        lifespan_ms: 0,
        liveliness_kind: None,
        liveliness_lease_ms: 0,
        avoid_ros_namespace_conventions: false,
        tx_express: false,
    };

    /// Services default QoS profile — reliable, volatile, keep-last(10).
    QOS_PROFILE_SERVICES_DEFAULT: "rmw_qos_profile_services_default" = {
        history: KeepLast,
        depth: 10,
        reliability: Reliable,
        durability: Volatile,
        deadline_ms: 0,
        lifespan_ms: 0,
        liveliness_kind: None,
        liveliness_lease_ms: 0,
        avoid_ros_namespace_conventions: false,
        tx_express: false,
    };

    /// Parameters QoS profile — KEEP_LAST(1000), RELIABLE, **VOLATILE**.
    ///
    /// issue 0793 — this said `TransientLocal` until 2026-08-25, disagreeing
    /// both with upstream (`/opt/ros/<distro>/include/rmw/rmw/qos_profiles.h`)
    /// and with our own second copy of the same profile, `nros::qos::PARAMETERS`,
    /// which was already correct. Two copies of one profile that disagree is the
    /// defect; the wrong one being the one named after the upstream constant is
    /// what made it hard to see.
    QOS_PROFILE_PARAMETERS: "rmw_qos_profile_parameters" = {
        history: KeepLast,
        depth: 1000,
        reliability: Reliable,
        durability: Volatile,
        deadline_ms: 0,
        lifespan_ms: 0,
        liveliness_kind: None,
        liveliness_lease_ms: 0,
        avoid_ros_namespace_conventions: false,
        tx_express: false,
    };

    /// Parameter events QoS profile — KEEP_LAST(1000), RELIABLE, VOLATILE.
    ///
    /// phase-428 W10 — this was `KeepAll, depth 0` until 2026-09-05, under a
    /// comment claiming it matched `rmw_qos_profile_parameter_events`. Upstream
    /// is `KEEP_LAST, 1000` (identical to `rmw_qos_profile_parameters`), and
    /// `rclcpp::ParameterEventsQoS` — which our own C++ surface mirrors — was
    /// already correct, so the two languages shipped different profiles under
    /// one name.
    ///
    /// The difference INVERTS behaviour rather than weakening it: both DDS
    /// backends map KEEP_ALL faithfully, and a RELIABLE writer with KEEP_ALL
    /// **blocks** once history fills, where KEEP_LAST(1000) overwrites. A
    /// parameter server whose events nobody is reading would stall the node
    /// instead of dropping the oldest event.
    QOS_PROFILE_PARAMETER_EVENTS: "rmw_qos_profile_parameter_events" = {
        history: KeepLast,
        depth: 1000,
        reliability: Reliable,
        durability: Volatile,
        deadline_ms: 0,
        lifespan_ms: 0,
        liveliness_kind: None,
        liveliness_lease_ms: 0,
        avoid_ros_namespace_conventions: false,
        tx_express: false,
    };

    /// Action status default QoS profile — reliable, transient-local,
    /// keep-last(1). From `rcl_action`, not `rmw`.
    QOS_PROFILE_ACTION_STATUS_DEFAULT: "rcl_action_qos_profile_status_default" = {
        history: KeepLast,
        depth: 1,
        reliability: Reliable,
        durability: TransientLocal,
        deadline_ms: 0,
        lifespan_ms: 0,
        liveliness_kind: None,
        liveliness_lease_ms: 0,
        avoid_ros_namespace_conventions: false,
        tx_express: false,
    };

    /// `rmw_qos_profile_best_available` (Iron+) — KEEP_LAST(10) with every
    /// policy that CAN be best-available set so (issue 1973).
    ///
    /// Reliability, durability, liveliness, deadline and lease are resolved at
    /// creation against the endpoints already discovered on the topic, by
    /// [`QoSProfile::resolve_best_available`]; history, depth and lifespan
    /// are stated, exactly as upstream states them. A request only: no
    /// backend is handed this profile unresolved.
    QOS_PROFILE_BEST_AVAILABLE: "rmw_qos_profile_best_available" = {
        history: KeepLast,
        depth: 10,
        reliability: BestAvailable,
        durability: BestAvailable,
        deadline_ms: DURATION_BEST_AVAILABLE_MS,
        lifespan_ms: 0,
        liveliness_kind: BestAvailable,
        liveliness_lease_ms: DURATION_BEST_AVAILABLE_MS,
        avoid_ros_namespace_conventions: false,
        tx_express: false,
    };

    /// Clock QoS profile — sensor-data shaped with depth 1.
    // phase-428 W4 — the provenance string said "/clock has no rmw constant",
    // which is true and was read as "no upstream at all". `rclrs::QOS_PROFILE_CLOCK`
    // EXISTS under this exact name (`rclrs/src/qos.rs:598`, it is rclcpp's
    // `ClockQoS`), and Rust follows rclrs. So this row mirrors something after
    // all, and the gate — which only compares upstream-MIRRORING rows — could
    // not see it.
    //
    // That is why it kept `Automatic` when W10 moved the other five to the
    // sentinel: F2's fourth instance, sitting inside the fix for F2's third.
    // The one row the instrument does not reach is the one that kept the
    // defect.
    QOS_PROFILE_CLOCK: "nros: mirrors rclrs::QOS_PROFILE_CLOCK; no rmw constant exists, so the gate cannot bind it — phase-428 W4" = {
        history: KeepLast,
        depth: 1,
        reliability: BestEffort,
        durability: Volatile,
        deadline_ms: 0,
        lifespan_ms: 0,
        liveliness_kind: None,
        liveliness_lease_ms: 0,
        avoid_ros_namespace_conventions: false,
        tx_express: false,
    };

    /// 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.
    ///
    /// Liveliness is the `None` sentinel, not `Automatic` (issue 2002): this
    /// profile exists for the XRCE backend, and `uxrQoS_t` carries no
    /// liveliness, so XRCE refuses a stated one (issue 1329). Stating it made
    /// every companion fail to advertise its first topic.
    QOS_PROFILE_PX4: "nros: PX4 uxrce_dds_client companion QoS; no upstream constant" = {
        history: KeepLast,
        depth: 1,
        reliability: BestEffort,
        durability: Volatile,
        deadline_ms: 0,
        lifespan_ms: 0,
        liveliness_kind: None,
        liveliness_lease_ms: 0,
        avoid_ros_namespace_conventions: false,
        tx_express: false,
    };
}
// nros-qos-table: END

// Where we knowingly differ from upstream, and where we knowingly have no
// counterpart. phase-428 W10.
//
// These lines are DATA, not prose: `check-qos-profile-ssot.py` requires a
// deviation to name the field, BOTH values and a tracking reference, and it
// re-measures `ours=` and `upstream=` against the table and the extracted
// record on every run. So a deviation cannot survive either side changing —
// which is the failure mode the phase-428 sweep found in `ARG_DEVIATIONS`,
// where 42 values were reason strings against a membership test and nothing
// checked that the reason was still true.
//
//   The liveliness deviation is CLOSED (issue 1329, 2026-09-12) and is
//   recorded here because the reason it survived two phases is worth keeping.
//
//   Every application-facing preset used to state AUTOMATIC where upstream
//   states the SYSTEM_DEFAULT sentinel. W10 moved this table and
//   `rmw_entity.h` onto the sentinel and could not move the remaining three —
//   `nros-c`'s `NROS_QOS_*`, ~~`rclc_c_qos_default()`~~ and `nros::QoS`'s
//   constructor — because `liveliness_kind` crosses the C ABI as its
//   discriminant, so flipping one surface alone makes a C caller's default and
//   a Rust caller's default differ in a field cyclonedds puts on the wire. All
//   three moved together instead.
//
//   MEASURED consequences of the move, 2026-09-05 and re-read 2026-09-12:
//     - cyclonedds: wire-identical. `nros_rmw_cyclonedds::make_dds_qos`
//       (`qos.cpp`; the name here read ~~`nros_rmw_cyclonedds_qos_apply`~~ until
//       phase-428 W9 went looking for it — no such symbol has ever existed)
//       calls `dds_qset_liveliness` only when the kind is
//       NOT the sentinel, and Cyclone's own reader/writer default IS
//       `DDS_LIVELINESS_AUTOMATIC` — so both spellings put AUTOMATIC on the
//       wire.
//     - zenoh: invisible. `QosKeyExpr::to_qos_string` (`keyexpr.rs`)
//       emits `"{rel}:{dur}:{hist},{depth}:,:,:,,"` (defaults elided, issue
//       1495) — the liveliness and lease
//       positions are EMPTY for every profile, so the value never reaches a
//       peer's graph parse.
//     - xrce: invisible. `uxrQoS_t` has no liveliness field at all.
//   The one live consequence was `required_policies()`, which set
//   `LIVELINESS_AUTOMATIC` for presets upstream leaves unset — an over-demand,
//   so it could only turn an acceptance into a rejection. It flipped no
//   verdict while every `supported_qos_policies()` in the tree advertised the
//   bit, and issue 1329 is exactly the change that stops them doing so: XRCE
//   honours no liveliness and now says so, which would have refused every C
//   application on that backend over a policy nobody asked for. The fix is to
//   stop asking, not to let a backend claim what it does not implement.
//
//   The `BEST_EFFORT` / `RELIABLE` shorthands still state AUTOMATIC. They
//   mirror no upstream constant — their doc strings say so — so they are
//   choices rather than deviations, and a caller picking one is asking for
//   automatic liveliness on purpose. `QOS_PROFILE_PX4` stated it too, and was
//   the counterexample to that reading (issue 2002): it is the profile FOR the
//   XRCE backend, which honours no liveliness, so nobody picking it could have
//   meant it. It carries the sentinel now.
//
// `rmw_qos_profile_unknown` HAS a counterpart now — `QOS_PROFILE_UNKNOWN` in the
// table below, added by this phase. The note that used to stand here declared it
// absent because three of our four policy enums had no `Unknown` variant to
// build it from; they have one now, which is what made the read-back able to
// say "I cannot report this policy" instead of echoing the request back as a
// grant.

/// phase-515 R8 helper -- a QoS duration as the record's whole milliseconds:
/// saturating at `u32::MAX`, and a non-zero duration under 1 ms rounded UP to
/// 1, because `0` is "infinite" in the record.
const fn qos_duration_ms(d: core::time::Duration) -> u32 {
    let ms = d.as_millis();
    if ms == 0 && !d.is_zero() {
        1
    } else if ms > u32::MAX as u128 {
        u32::MAX
    } else {
        ms as u32
    }
}

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

    // --- 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
    }

    /// The parameter services' profile — rclrs's
    /// `QoSProfile::parameter_services_default()`.
    ///
    /// `rmw_qos_profile_parameters`: KEEP_LAST(**1000**), RELIABLE, VOLATILE.
    /// The depth is the point — 1000 against the generic services profile's
    /// 10 — and it is what a tool setting many parameters at once depends on.
    ///
    /// **RENAMED in phase-417 G6, and the old spelling is GONE.** It was
    /// ~~`parameters_default`~~, an ours-only word for a name rclrs has; the
    /// ledger carried the pair as `rust:QoSProfile::parameters_default`
    /// (`rename` — ours is the one that should change) and
    /// `rust:QoSProfile::parameter_services_default` (`gap`). Adding the
    /// upstream name beside ours would have been a FOURTH spelling of one
    /// preset (`QOS_PROFILE_PARAMETERS` and `QosPresets::PARAMETERS` are the
    /// other two) and would have left the `rename` row open for good. No
    /// deprecated forwarder: a stale call should fail on the IDENTIFIER rather
    /// than on a deprecation, which is stage 6 step B's rule and what the
    /// sibling `parameter_events_default` — already at rclrs's spelling —
    /// makes the pair consistent with.
    pub const fn parameter_services_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
    }

    /// rclrs's `QoSProfile::best_available()` — `rmw_qos_profile_best_available`
    /// ([`QOS_PROFILE_BEST_AVAILABLE`](Self::QOS_PROFILE_BEST_AVAILABLE)),
    /// issue 1973.
    ///
    /// Reliability, durability, liveliness, deadline and liveliness lease are
    /// chosen when the entity is created, from the endpoints the backend has
    /// already discovered on the topic ([`QoSProfile::resolve_best_available`]
    /// states the rule); read the outcome back with the entity's
    /// `actual_qos()`. Upstream's caveat holds: the choice is not revisited
    /// when endpoints appear later.
    pub const fn best_available() -> Self {
        Self::QOS_PROFILE_BEST_AVAILABLE
    }

    // --- 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
    }

    /// rclrs's `QoSProfile::deadline_duration(deadline)` (phase-515 R8).
    ///
    /// ADOPT-BOUNDED: stored as whole MILLISECONDS in [`Self::deadline_ms`]
    /// (the repr(C) QoS record the C and C++ surfaces share), saturating at
    /// `u32::MAX`; a non-zero duration under 1 ms rounds UP to 1 ms, because
    /// `0` here means "infinite" and must not be reached by accident.
    pub const fn deadline_duration(mut self, deadline: core::time::Duration) -> Self {
        self.deadline_ms = qos_duration_ms(deadline);
        self
    }

    /// rclrs's `QoSProfile::lifespan_duration(lifespan)` (phase-515 R8),
    /// stored in [`Self::lifespan_ms`] as `deadline_duration` stores its value.
    pub const fn lifespan_duration(mut self, lifespan: core::time::Duration) -> Self {
        self.lifespan_ms = qos_duration_ms(lifespan);
        self
    }

    /// rclrs's `QoSProfile::liveliness_lease_duration(lease_duration)`
    /// (phase-515 R8), stored in [`Self::liveliness_lease_ms`] as
    /// `deadline_duration` stores its value.
    pub const fn liveliness_lease_duration(mut self, lease_duration: core::time::Duration) -> Self {
        self.liveliness_lease_ms = qos_duration_ms(lease_duration);
        self
    }

    /// Phase 282 (#145) — mark this publisher's samples "express": they
    /// bypass transport tx batching (sent immediately even when the batching
    /// knob is on). A transport hint for control-tier / latency-sensitive
    /// topics; ignored on subscriptions and by backends without batching.
    pub const fn tx_express(mut self, express: bool) -> Self {
        self.tx_express = express;
        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)],
    /// Node name for ROS 2 graph discovery liveliness token.
    ///
    /// Empty string (`""`) means no node-liveliness token is declared (preserves
    /// the pre-#104 behaviour). Non-empty causes the session to declare a
    /// `@ros2_lv/<domain>/<zid>/0/0/NN/%/<ns>/<node>` token on open.
    pub node_name: &'a str,
    /// Node namespace for the liveliness token (e.g., `""` or `"/ns1"`).
    ///
    /// Empty string is treated as root `"/"` by the keyexpr builder.
    pub namespace: &'a str,
    /// ROS 2 domain ID used in the liveliness token key expression.
    pub domain_id: u32,
}

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

/// 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/)")
        }
    }
}

/// phase-521 W3 (RFC-0113 D4) — how a session reports that data arrived.
///
/// The executor's park ends for exactly two reasons: the tier's next deadline,
/// or readiness. This is the readiness half, in one of two forms, and the
/// executor picks its wait from it rather than blocking inside the transport.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ReadinessSource {
    /// Neither form: the executor can only learn of an arrival by blocking in
    /// [`Session::drive_io`]. RFC-0113 D5 refuses this once phase-521 W6
    /// lands.
    None,
    /// The backend calls the wake callback [`Session::set_wake_callback`]
    /// installed when data arrives (a transport thread, a listener, an RX
    /// interrupt).
    Notify,
    /// An OS handle that becomes readable when data arrives — a file
    /// descriptor on POSIX and NuttX. The executor multiplexes it with its own
    /// wake object and drives the session non-blockingly after the park. It
    /// must stay valid until the session closes.
    Waitable(i32),
}

/// 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`] / [`Subscription`] 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 {
    /// RFC-0088 — the serialization format this backend speaks, as ROS 2's
    /// `rmw_get_serialization_format()` reports it ("One middleware can only
    /// have one encoding").
    ///
    /// Defaulted to CDR because every backend in tree except uORB speaks it,
    /// and a backend that speaks something else says so by overriding these
    /// two. They travel together: `SERIALIZATION_FORMAT` is the identity that
    /// crosses images (bridge config, tooling, the vtable slot) and
    /// `SERIALIZATION_FORMAT_ID` is the image-local discriminant used for the
    /// one-byte comparison a bridge makes at construction.
    const SERIALIZATION_FORMAT: &'static str =
        nros_serdes::format::SerializationFormatId::Cdr.as_str();

    /// Image-local discriminant for [`Self::SERIALIZATION_FORMAT`]. Never
    /// persisted, never compared across images — see `nros_serdes::format`.
    const SERIALIZATION_FORMAT_ID: nros_serdes::format::SerializationFormatId =
        nros_serdes::format::SerializationFormatId::Cdr;

    /// RFC-0088 — this session's serialization format, as ROS 2's
    /// `rmw_get_serialization_format()` reports it.
    ///
    /// **Per session, not per process.** ROS 2's function takes no handle
    /// because one process links one middleware; an `Executor::open_multi`
    /// image links two, so the answer must be asked of the session.
    ///
    /// **`None` means the backend has NOT SAID**, and that is not a synonym
    /// for `"cdr"`. The seam's contract is stated once, for the
    /// `get_serialization_format` slot in
    /// `packages/core/nros-rmw-abi/include/nros/rmw_vtable.h`: "NULL slot: the
    /// runtime answers NULL — it does NOT guess `cdr`." This method is the
    /// Rust face of that slot and carries the same absence. A `&'static str`
    /// could not: the only value it had to hand back was the literal `"cdr"`,
    /// indistinguishable at every call site from a backend that declared CDR.
    ///
    /// The default answers `Some(Self::SERIALIZATION_FORMAT)`, which is right
    /// for any backend whose format is a compile-time fact — it HAS said, in
    /// the const. A session that dispatches to a backend chosen at run time —
    /// the C-ABI adapter, whose vtable carries the answer — overrides this to
    /// ask the backend, and hands on the `None` a NULL slot gives it.
    fn serialization_format(&self) -> Option<&'static str> {
        Some(Self::SERIALIZATION_FORMAT)
    }

    /// The backend's name, as ROS 2's `rmw_get_implementation_identifier()`
    /// reports it — the nano-ros name (`"zenoh"`, `"cyclonedds"`, `"xrce"`,
    /// `"uorb"`), never a stock `rmw_*_cpp` name, for the reason the
    /// `get_implementation_identifier` slot in
    /// `packages/core/nros-rmw-abi/include/nros/rmw_vtable.h` states.
    ///
    /// `None` = the backend has not said. A backend whose name is a
    /// compile-time fact states it here; it must be the same string it stamps
    /// into a gid, since gids are comparable only under one identifier.
    const IMPLEMENTATION_IDENTIFIER: Option<&'static str> = None;

    /// This session's backend name — the per-session form of
    /// `rmw_get_implementation_identifier()`, for the same reason
    /// [`serialization_format`](Self::serialization_format) is per session: an
    /// `Executor::open_multi` image links two backends.
    ///
    /// The default answers [`Self::IMPLEMENTATION_IDENTIFIER`]. A session that
    /// dispatches to a backend chosen at run time (the C-ABI adapter) overrides
    /// this to ask the backend's vtable.
    fn implementation_identifier(&self) -> Option<&'static str> {
        Self::IMPLEMENTATION_IDENTIFIER
    }

    /// Error type for this session
    type Error;
    /// Publisher handle type
    type PublisherHandle;
    /// Subscription handle type
    type SubscriptionHandle;
    /// Service server handle type
    type ServiceHandle;
    /// Service client handle type
    type ClientHandle;

    /// 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: QoSProfile,
    ) -> 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_subscription(
        &mut self,
        topic: &TopicInfo,
        qos: QoSProfile,
    ) -> Result<Self::SubscriptionHandle, Self::Error>;

    /// Create a service server bound to this session. Replies are
    /// matched to requests by the sequence number returned from
    /// [`ServiceTrait::take_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 [`QoSProfile::services_default`]
    /// (RELIABLE+VOLATILE+KEEP_LAST(10)).
    ///
    /// The server receives through the backend's own inbox. A builtin family
    /// that sizes its own ring calls [`create_service_with_inbox`] instead.
    ///
    /// [`create_service_with_inbox`]: Session::create_service_with_inbox
    fn create_service(
        &mut self,
        service: &ServiceInfo,
        qos: QoSProfile,
    ) -> Result<Self::ServiceHandle, Self::Error>;

    /// phase-461 W2 -- does this backend accept a caller-owned request ring
    /// ([`ServiceInboxSpec::Caller`])?
    ///
    /// A `const`, so the family that would bring one picks at COMPILE time
    /// and a `Caller` spec never degrades silently into the backend's own
    /// geometry (RFC-0052: a knob that is dropped is a knob that lies). The
    /// caller reads this and passes [`ServiceInboxSpec::Backend`] where it is
    /// `false`, which is how a family refuses at BUILD time rather than
    /// discovering at boot that its ring went nowhere -- `nros-node`'s
    /// `PARAM_INBOX` is the worked example.
    ///
    /// `false` is the right default and every backend in tree states it
    /// explicitly with its own reason, because "this backend has no caller
    /// ring" and "nobody thought about it" are different facts.
    ///
    /// [`create_service_with_inbox`]: Session::create_service_with_inbox
    const SUPPORTS_CALLER_INBOX: bool = false;

    /// Create a service server that receives through `inbox`.
    ///
    /// phase-461 W2 / issue 1352 -- the ROS parameter services are six
    /// queryables per node and the REP-2002 lifecycle services five, and on a
    /// board they are most of the image's queryables. Their requests are
    /// bounded by the contract's declared parameters and their clients are
    /// sequential, so they want one small slot where a user service wants four
    /// large ones. The family that knows that bound brings the ring; the
    /// backend keeps the per-queryable header.
    ///
    /// The default body is [`Self::SUPPORTS_CALLER_INBOX`] `= false` spelled
    /// out: it serves [`ServiceInboxSpec::Backend`] and nothing else. It can
    /// only be reached with a `Caller` spec by a caller that ignored the
    /// const, so the `debug_assert` names that -- the REFUSAL itself is the
    /// caller's, at compile time, because a runtime error here would be a
    /// boot failure where a build failure was available (`nros-node`'s
    /// `PARAM_INBOX` is the worked example).
    ///
    /// A backend that overrides this must also state the const `true`.
    fn create_service_with_inbox(
        &mut self,
        service: &ServiceInfo,
        qos: QoSProfile,
        inbox: ServiceInboxSpec,
    ) -> Result<Self::ServiceHandle, Self::Error> {
        debug_assert!(
            inbox.caller().is_none(),
            "a caller-owned service inbox reached a backend whose \
             SUPPORTS_CALLER_INBOX is false; the caller must read the const"
        );
        self.create_service(service, qos)
    }

    /// 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 [`QoSProfile::services_default`]
    /// (RELIABLE+VOLATILE+KEEP_LAST(10)).
    fn create_client(
        &mut self,
        service: &ServiceInfo,
        qos: QoSProfile,
    ) -> Result<Self::ClientHandle, 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.**
    ///
    /// # The mask is DERIVED, not authored — phase-428 W9
    ///
    /// Every bit an implementation sets must be backed by a
    /// `nros-qos-honours:` claim sited on the code that honours the policy,
    /// and `check-qos-mask-derivation` re-measures those claims on every
    /// push. Deleting the code deletes the claim, which is the property this
    /// item exists to buy: before W9 this was a hand-written constant, so a
    /// backend could advertise a policy it had never implemented and nothing
    /// noticed. Three did — `LIVELINESS_MANUAL_BY_NODE` was advertised by
    /// every backend in the tree and implemented by none.
    ///
    /// "Honours" means the backend reads the field and either APPLIES it or
    /// REFUSES a value it cannot serve. Passing the field on to someone else
    /// (a discovery keyexpr, a lowered C struct) is not honouring it, and the
    /// gate does not count it: see `nros-qos-discovery-only`.
    ///
    /// # The default is EMPTY, deliberately
    ///
    /// It returned [`QoSPolicyMask::CORE`] until phase-428 W9 — four policies
    /// granted to any implementation that had written no code at all, which
    /// is precisely the lie above with the trait as its author. An
    /// implementation now states what it honours or is told it honours
    /// nothing; the failure is loud (`IncompatibleQos` at create) rather than
    /// a silent downgrade at runtime, which is the direction this trait has
    /// chosen everywhere else.
    ///
    /// # A C backend answers through the vtable — issue 1329
    ///
    /// `CffiSession` is the one implementation that cannot answer from
    /// compile-time knowledge: which backend registered is a run-time fact. It
    /// returned the UNION of what any C backend it routes to honours, which
    /// over-claims for each of them; it now asks the backend through
    /// `nros_rmw_vtable_t::supported_qos_policies` and returns what it said. A
    /// NULL slot there carries the same meaning as this default — the backend
    /// has not said, so it honours nothing — deliberately, so the ABI and the
    /// trait give one answer to one question.
    fn supported_qos_policies(&self) -> QoSPolicyMask {
        QoSPolicyMask::NONE
    }

    /// 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-521 W3 (RFC-0113 D4) — which form of readiness this session has.
    ///
    /// Default: [`ReadinessSource::Notify`] when the backend honours
    /// [`set_wake_callback`](Self::set_wake_callback), else
    /// [`ReadinessSource::None`] — what the executor inferred before this
    /// method existed, so a backend that already delivers a wake callback
    /// needs no change. A backend with a waitable OS handle overrides it.
    fn readiness_source(&self) -> ReadinessSource {
        if self.supports_wake_callback() {
            ReadinessSource::Notify
        } else {
            ReadinessSource::None
        }
    }

    /// 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 (XRCE: `uxr_ping_agent_session`; zenoh, which has no
    /// round-trip ping, sends one `zpico_send_keep_alive` frame) 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())
    }

    /// phase-381 W3 — enumerate the nodes this session can see.
    ///
    /// A VISITOR, not a returned collection, because upstream's
    /// `rcutils_string_array_t` allocates two levels deep and there is no
    /// allocator at this seam. A caller-provided buffer is worse than it looks:
    /// the graph has no bound the CALLER can know. So the backend streams from
    /// state it already holds, peak extra RAM is one entry, and a caller with a
    /// bound stops early by returning `false`.
    ///
    /// `namespace` and `name` are ROS names. `enclave` is `None` where the
    /// backend does not track one — which is what lets this one method answer
    /// both `rmw_get_node_names` and `rmw_get_node_names_with_enclaves`.
    ///
    /// Every string is BORROWED for the duration of the call.
    ///
    /// **Must not block on the wire, and takes no timeout.** It reports what has
    /// ALREADY arrived, so the first call after startup legitimately returns a
    /// partial graph — a backend feeds its view from `drive_io`. Letting this
    /// block was considered and rejected: it would stall the executor's only
    /// thread inside an introspection call, on a runtime whose premise is that
    /// there is no other thread to do the work.
    ///
    /// Default body: `Err(Unsupported)` — a backend with no graph (XRCE) says
    /// so, and the runtime can tell that from an empty graph.
    fn get_node_names(
        &mut self,
        visit: &mut dyn FnMut(&str, &str, Option<&str>) -> bool,
    ) -> Result<(), Self::Error>
    where
        Self::Error: From<TransportError>,
    {
        let _ = visit;
        Err(TransportError::Unsupported.into())
    }

    /// phase-381 W3 — how many publishers this session can see on `topic_name`.
    ///
    /// `topic_name` is a ROS name (`/chatter`); the backend mangles as needed.
    /// Same warm-up caveat as [`Self::get_node_names`]: a count reflects what
    /// has already been discovered, so it can be low right after startup and is
    /// never a proof of absence.
    ///
    /// Default body: `Err(Unsupported)` — distinct from `Ok(0)`, which claims
    /// there are none.
    fn count_publishers(&mut self, topic_name: &str) -> Result<usize, Self::Error>
    where
        Self::Error: From<TransportError>,
    {
        let _ = topic_name;
        Err(TransportError::Unsupported.into())
    }

    /// phase-381 W3 — how many subscribers this session can see on `topic_name`.
    /// See [`Self::count_publishers`] for the caveats.
    fn count_subscribers(&mut self, topic_name: &str) -> Result<usize, Self::Error>
    where
        Self::Error: From<TransportError>,
    {
        let _ = topic_name;
        Err(TransportError::Unsupported.into())
    }

    /// Issue 1978 — how many service CLIENTS this session can see on
    /// `service_name`, upstream (Jazzy) `rmw_count_clients`. The service twin
    /// of [`Self::count_publishers`], with the same caveats: `service_name` is
    /// a ROS name the backend mangles, and a count reflects what has already
    /// been discovered.
    ///
    /// Default body: `Err(Unsupported)` — distinct from `Ok(0)`.
    fn count_clients(&mut self, service_name: &str) -> Result<usize, Self::Error>
    where
        Self::Error: From<TransportError>,
    {
        let _ = service_name;
        Err(TransportError::Unsupported.into())
    }

    /// Issue 1978 — how many service SERVERS this session can see on
    /// `service_name`, upstream (Jazzy) `rmw_count_services`. See
    /// [`Self::count_clients`].
    fn count_services(&mut self, service_name: &str) -> Result<usize, Self::Error>
    where
        Self::Error: From<TransportError>,
    {
        let _ = service_name;
        Err(TransportError::Unsupported.into())
    }

    /// Issue 1978 — upstream (Jazzy) `rmw_event_type_is_supported`: whether
    /// ANY entity of this backend can carry status event `kind`.
    ///
    /// STATIC, like [`Self::SERIALIZATION_FORMAT`]: upstream's signature
    /// takes no entity and no session, so the C-ABI adapter answers it from
    /// the backend TYPE. Per-entity registration
    /// ([`Publisher::supports_event`], [`Subscription::supports_event`]) stays
    /// the authority for one entity; a backend's per-entity answer must be a
    /// subset of this one, and the way to keep it so is to derive both from
    /// one helper in the backend (zenoh's `shim::events`).
    ///
    /// Default: `false` — the backend states no event support, the answer a
    /// NULL vtable slot carries.
    fn event_type_is_supported(kind: crate::event::EventKind) -> bool
    where
        Self: Sized,
    {
        let _ = kind;
        false
    }

    /// phase-381 W3 — every topic, with the types published or subscribed on it.
    ///
    /// A visitor for the same reason as [`Self::get_node_names`], and one call
    /// per distinct NAME: the contract hands over a name and the types on it,
    /// so a topic carrying two types is one visit with two entries, not two
    /// visits.
    ///
    /// `types_count` may legitimately be 0 on a partially discovered graph —
    /// reporting the name without a type beats dropping it.
    ///
    /// Default body: `Err(Unsupported)`.
    fn get_topic_names_and_types(
        &mut self,
        visit: &mut dyn FnMut(&str, &[&str]) -> bool,
    ) -> Result<(), Self::Error>
    where
        Self::Error: From<TransportError>,
    {
        let _ = visit;
        Err(TransportError::Unsupported.into())
    }

    /// phase-381 W3 — every service, with its types. As
    /// [`Self::get_topic_names_and_types`], over servers and clients.
    fn get_service_names_and_types(
        &mut self,
        visit: &mut dyn FnMut(&str, &[&str]) -> bool,
    ) -> Result<(), Self::Error>
    where
        Self::Error: From<TransportError>,
    {
        let _ = visit;
        Err(TransportError::Unsupported.into())
    }

    /// phase-381 W3 — what ONE named node publishes / subscribes / serves /
    /// calls.
    ///
    /// `node_name` and `node_namespace` are ROS names; a node the graph has not
    /// discovered yields no visits, which is NOT an error — see
    /// [`Self::get_node_names`] for why an empty answer is "not seen yet".
    ///
    /// `kind` selects which of the four upstream `*_by_node` questions this
    /// answers. One method rather than four because the four differ ONLY by
    /// which entity kind they keep, and four trait methods would be four copies
    /// of one filter.
    ///
    /// The trait keeps the ABI's `subscriber` vocabulary because
    /// `rmw_get_subscriber_names_and_types_by_node` is what upstream rmw calls
    /// it and RFC-0054 makes the C headers the SSoT. The USER-facing spelling
    /// is per language and settled at that layer: rcl says `subscriber`, rclcpp
    /// and rclrs say `subscription`.
    ///
    /// Default body: `Err(Unsupported)`.
    fn get_names_and_types_by_node(
        &mut self,
        kind: GraphEntityKind,
        node_name: &str,
        node_namespace: &str,
        visit: &mut dyn FnMut(&str, &[&str]) -> bool,
    ) -> Result<(), Self::Error>
    where
        Self::Error: From<TransportError>,
    {
        let _ = (kind, node_name, node_namespace, visit);
        Err(TransportError::Unsupported.into())
    }

    /// phase-381 W3 — the endpoints on one topic, with the QoS each GRANTED.
    ///
    /// `publishers` selects `rmw_get_publishers_info_by_topic` (`true`) or
    /// `rmw_get_subscriptions_info_by_topic` (`false`).
    ///
    /// The granted profile is the whole reason a consumer asks: "why is nothing
    /// arriving" is usually a QoS incompatibility, and the REQUESTED profile
    /// cannot answer it. A backend that cannot read a remote's granted QoS says
    /// so per field rather than echoing the request back — see
    /// `rmw_topic_endpoint_info_t`.
    ///
    /// Default body: `Err(Unsupported)`.
    fn get_endpoint_info_by_topic(
        &mut self,
        publishers: bool,
        topic_name: &str,
        visit: &mut dyn FnMut(&GraphEndpointInfo<'_>) -> bool,
    ) -> Result<(), Self::Error>
    where
        Self::Error: From<TransportError>,
    {
        let _ = (publishers, topic_name, visit);
        Err(TransportError::Unsupported.into())
    }

    /// phase-467 Row 8 — install (or clear, when `callback` is `None`) the
    /// callback the backend fires when the ROS GRAPH changes.
    ///
    /// The Rust twin of `nros_rmw_vtable_t::node_get_graph_guard_condition`, which
    /// is upstream's `rmw_node_get_graph_guard_condition` under the
    /// `set_wake_callback` shape: upstream RETURNS a guard condition the caller
    /// adds to a wait set, we have no wait set, and a guard condition here is
    /// an executor concept the backend never hands out. So the edge is PUSHED,
    /// not polled — and the ABI slot has said so since phase-376 W4, while
    /// this is the trait twin it never had.
    ///
    /// The callback is an EDGE and carries no payload. `count` is how many
    /// changes the backend coalesced, never a description of WHAT changed:
    /// delivering that would mean buffering the graph, which is the cache a
    /// small target cannot afford.
    ///
    /// **A backend must not fire it for a no-op.** The contract is "the graph
    /// this session can SEE differs from what it saw", which both real
    /// backends already computed and threw away — zenoh in
    /// `zpico_graph_set_apply`'s return value, Cyclone in
    /// `on_data_available`'s first parameter.
    ///
    /// # Safety
    ///
    /// When `callback` is `Some`, `user_data` must remain valid until the
    /// callback is cleared or the session is closed; the backend may invoke it
    /// from a transport or worker thread. **The installer owes the clear** —
    /// see `Executor::clear_graph_change_signal`, which is issue 1385's lesson
    /// applied before the hazard rather than after it.
    ///
    /// Default body: `Err(Unsupported)` — a backend with no graph (XRCE, uORB)
    /// says "cannot tell you" rather than accepting a callback it will never
    /// fire. That answer is also the capability PROBE the runtime uses: a
    /// `None` clear on a graphless backend returns `Unsupported`, so nothing
    /// else has to be asked and no handle is consumed before the answer.
    unsafe fn set_graph_change_callback(
        &mut self,
        callback: Option<unsafe extern "C" fn(user_data: *const core::ffi::c_void, count: usize)>,
        user_data: *const core::ffi::c_void,
    ) -> Result<(), Self::Error>
    where
        Self::Error: From<TransportError>,
    {
        let _ = (callback, user_data);
        Err(TransportError::Unsupported.into())
    }
}

/// Which entity kind a `*_by_node` graph query keeps — phase-381 W3.
///
/// Named for upstream rmw's four `*_by_node` slots. `Subscriber` carries rmw's
/// word, not rclcpp's `subscription`; the user-facing spelling is chosen per
/// language one layer up.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum GraphEntityKind {
    Publisher,
    Subscriber,
    Service,
    Client,
}

/// One discovered endpoint on a topic — phase-381 W3, the Rust view of
/// `rmw_topic_endpoint_info_t`.
///
/// Every string BORROWS from the backend's own state for the duration of the
/// visit. A caller that needs one afterwards copies it; that is what lets the
/// graph stream without an allocator.
#[derive(Debug, Clone)]
pub struct GraphEndpointInfo<'a> {
    /// Node that owns the endpoint.
    pub node_name: &'a str,
    /// That node's namespace.
    pub node_namespace: &'a str,
    /// Fully-qualified type on the wire, e.g. `"std_msgs/msg/Int32"`.
    pub topic_type: &'a str,
    /// `true` for a publisher, `false` for a subscription.
    pub is_publisher: bool,
    /// The endpoint's 24-byte identity; all-zero when the backend has none.
    pub endpoint_gid: [u8; 24],
    /// The profile the endpoint is known by — what the backend MATCHED on.
    ///
    /// [`QoSProfile::QOS_PROFILE_UNKNOWN`] (every policy `Unknown`, windows 0)
    /// when the backend cannot read it, which is the C seam's contract on
    /// `rmw_topic_endpoint_info_t::qos_profile` too.
    pub qos: QoSProfile,
}

// The `qos` field arrived with issue 1973, because resolving a best-available
// profile needs every peer's profile and this is the seam that walks them.
//
// It was left out before on the reading that zenoh's liveliness token carries
// the DECLARING side's profile, not a negotiated grant. Measured against what
// the token actually holds, that distinction does not exist on zenoh: there is
// no negotiation, our token is written from `shim/qos.rs`'s GRANT
// (`keyexpr.rs`), and a stock `rmw_zenoh_cpp` peer writes its own resolved
// profile and reports the same string from `rmw_get_publishers_info_by_topic`.
// So the token is what the endpoint does, which is what the C seam's
// `qos_profile` promises ("reports what it MATCHED on"). Cyclone fills it from
// the discovered endpoint's DDS QoS (`graph_query.cpp`). A backend that cannot
// read one writes `QOS_PROFILE_UNKNOWN`, never the local request.

/// Bitmask of QoS policies a backend can honour. See
/// [`Session::supported_qos_policies`].
///
/// One bit per value `QoSProfile::required_policies` can request, and that
/// function is the definition of what each bit MEANS — `check-qos-mask-
/// derivation` reads the bit->field mapping out of its body rather than
/// restating it, so a policy that gains a bit here and no arm there is a gate
/// failure instead of a bit nothing can ever ask for.
#[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);

    /// Honours nothing. The default answer for an implementation that has not
    /// said otherwise — phase-428 W9, see [`Session::supported_qos_policies`].
    pub const NONE: Self = Self(0);

    /// The four policies every `QoSProfile` states a value for.
    ///
    /// NOT "policies every nano-ros backend implements", which is what this
    /// said until phase-428 W9 measured it: the zenoh shim honoured exactly
    /// none of the four, serialising all four into a discovery keyexpr and
    /// enforcing a build-time ring depth. The name describes the SHAPE of a
    /// profile — a caller who states nothing else still states these — and a
    /// backend still has to earn each bit.
    pub const CORE: Self =
        Self(Self::RELIABILITY.0 | Self::DURABILITY_VOLATILE.0 | Self::HISTORY.0 | Self::DEPTH.0);

    /// Every policy bit paired with the name it PRINTS — the vocabulary the
    /// three languages name a refused policy in (phase-417 G7, ledger row
    /// `cpp:qos_policy_kind_to_cstr`).
    ///
    /// The name of a bit is the CONSTANT'S OWN IDENTIFIER, never an
    /// independently chosen word. That is what keeps this from becoming the
    /// parallel table this campaign has paid for twice — the RMW parity map's
    /// 28 stale slots and the layout gate's three authored type names — and it
    /// is what `check-qos-mask-derivation`'s R8 measures: the entries here must
    /// be exactly the `pub const … = Self(1 << n)` bits above, each paired with
    /// its own spelling. A bit added above and not here is a gate failure, not
    /// a policy that silently prints as nothing.
    ///
    /// Uppercase because that is upstream's convention for the same strings:
    /// `rclcpp::qos_policy_kind_to_cstr` returns `"DURABILITY"`, `"DEADLINE"`,
    /// `"LIVELINESS"` and the rest.
    ///
    /// `NONE` and `CORE` are deliberately absent: they are not policies, they
    /// are the empty mask and a named union, and neither can be the answer to
    /// "which policy was refused".
    ///
    /// The name is stored as a [`core::ffi::CStr`] and not as a `&str`,
    /// because the C entry point (`rclc_qos_policy_kind_to_cstr`) needs the
    /// trailing NUL and a `&str` does not carry one. Storing the `&str` form
    /// and NUL-terminating it in `nros-c` was the first shape here, and it put
    /// the twelve identifiers in a second file — which is the drift this row
    /// exists to avoid. One table, two readings: `policy_name` slices the NUL
    /// off for Rust, the C entry point hands the pointer straight over.
    pub const NAMED: &'static [(Self, &'static core::ffi::CStr)] = &[
        (Self::RELIABILITY, c"RELIABILITY"),
        (Self::DURABILITY_VOLATILE, c"DURABILITY_VOLATILE"),
        (
            Self::DURABILITY_TRANSIENT_LOCAL,
            c"DURABILITY_TRANSIENT_LOCAL",
        ),
        (Self::HISTORY, c"HISTORY"),
        (Self::DEPTH, c"DEPTH"),
        (Self::DEADLINE, c"DEADLINE"),
        (Self::LIFESPAN, c"LIFESPAN"),
        (Self::LIVELINESS_AUTOMATIC, c"LIVELINESS_AUTOMATIC"),
        (
            Self::LIVELINESS_MANUAL_BY_TOPIC,
            c"LIVELINESS_MANUAL_BY_TOPIC",
        ),
        (
            Self::LIVELINESS_MANUAL_BY_NODE,
            c"LIVELINESS_MANUAL_BY_NODE",
        ),
        (Self::LIVELINESS_LEASE, c"LIVELINESS_LEASE"),
        (
            Self::AVOID_ROS_NAMESPACE_CONVENTIONS,
            c"AVOID_ROS_NAMESPACE_CONVENTIONS",
        ),
    ];

    /// The name of the ONE policy this mask is, as a C string.
    ///
    /// `None` for the empty mask and for a mask carrying more than one bit:
    /// both are real answers a diagnostic must be able to give. "The backend
    /// did not say which" is not a policy name, and printing a neighbouring
    /// one would be the plausible wrong answer this whole family exists to
    /// refuse.
    ///
    /// Borrowed from a table in flash: no allocator, no formatting, reaches a
    /// bare-metal target, and these are the exact bytes the C and C++ entry
    /// points hand back.
    pub const fn policy_name_cstr(self) -> Option<&'static core::ffi::CStr> {
        let mut i = 0;
        while i < Self::NAMED.len() {
            if Self::NAMED[i].0.0 == self.0 {
                return Some(Self::NAMED[i].1);
            }
            i += 1;
        }
        None
    }

    /// [`Self::policy_name_cstr`] as a Rust string — the same bytes without
    /// the terminator.
    pub fn policy_name(self) -> Option<&'static str> {
        // Every entry is an ASCII literal, so this cannot fail; `ok()` rather
        // than `expect()` because a panic in a diagnostic path is worse than
        // the diagnostic being absent.
        self.policy_name_cstr().and_then(|c| c.to_str().ok())
    }

    /// The policies in `self` that `supported` does not carry.
    pub const fn missing_from(self, supported: Self) -> Self {
        Self(self.0 & !supported.0)
    }

    /// The lowest-numbered policy bit set here, or [`Self::NONE`].
    ///
    /// Which single policy a multi-policy refusal REPORTS. Lowest bit rather
    /// than "any": a stable choice means the same profile against the same
    /// backend names the same policy on every run and in every language, so a
    /// test can assert it and two surfaces cannot disagree about which one to
    /// mention first.
    pub const fn lowest_policy(self) -> Self {
        Self(self.0 & self.0.wrapping_neg())
    }

    /// `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 QoSProfile {
    /// Does any policy in this profile carry its `Unknown` sentinel?
    ///
    /// True only for a profile that came out of a `*_get_actual_qos`
    /// read-back on a backend that could not report every policy — never for
    /// one an application built. [`validate_against`](Self::validate_against)
    /// refuses such a profile at create time, so "I cannot say" cannot be
    /// laundered into a request.
    pub const fn has_unknown(&self) -> bool {
        matches!(self.reliability, QoSReliabilityPolicy::Unknown)
            || matches!(self.durability, QoSDurabilityPolicy::Unknown)
            || matches!(self.history, QoSHistoryPolicy::Unknown)
            || matches!(self.liveliness_kind, QoSLivelinessPolicy::Unknown)
    }

    /// Compute the set of QoS policies actually requested by this profile.
    ///
    /// Zero-valued time fields and `QoSLivelinessPolicy::None` count as "not
    /// requesting" the corresponding policy — the cheap default.
    ///
    /// issue 0829 — the four `CORE` bits (reliability, durability, history,
    /// depth) used to be added UNCONDITIONALLY, on the reasoning that every
    /// nano-ros backend honours them. That is true and still not the right
    /// test: this function answers *what did the caller ASK FOR*, and a
    /// `SYSTEM_DEFAULT` policy asks for nothing. Starting from `CORE` made
    /// `QOS_PROFILE_SYSTEM_DEFAULT` demand all four, so a backend that could
    /// not honour one would reject the profile with `IncompatibleQos` — for
    /// requesting nothing. The pattern was already here for the extended
    /// policies three lines down (a zero `deadline_ms` declines its bit); it
    /// just never reached the four CORE ones.
    ///
    /// This RELAXES the mask, so it can only turn a rejection into an
    /// acceptance, never the reverse.
    ///
    /// It used to say it flipped no verdict "because every
    /// `supported_qos_policies` impl in the tree returns at least `CORE`".
    /// That stopped being true in phase-428 W9: the trait default is
    /// [`QoSPolicyMask::NONE`] now, so an implementation that claims nothing
    /// is refused on the first CORE bit a caller states — which is the whole
    /// point, and which makes THIS function's sentinel handling the thing
    /// standing between `QOS_PROFILE_SYSTEM_DEFAULT` and a hard refusal.
    pub fn required_policies(&self) -> QoSPolicyMask {
        let mut mask = QoSPolicyMask(0);
        if self.reliability != QoSReliabilityPolicy::SystemDefault
            && self.reliability != QoSReliabilityPolicy::BestAvailable
        {
            mask |= QoSPolicyMask::RELIABILITY;
        }
        if self.history != QoSHistoryPolicy::SystemDefault {
            mask |= QoSPolicyMask::HISTORY;
        }
        // Depth's sentinel is a VALUE, not a variant: `DEPTH_SYSTEM_DEFAULT`
        // is 0, the same 0 that a `KEEP_ALL` profile carries to mean "depth is
        // not used here" (`QoSProfile::keep_all()` leaves whatever depth the
        // profile had, and a caller building one from scratch leaves it 0).
        // Both mean the caller did not ask for a depth, so both decline the
        // bit. No NAMED preset is `KEEP_ALL` any more — phase-428 W10;
        // `QOS_PROFILE_PARAMETER_EVENTS` used to be the exemplar here and was
        // the defect, not the example.
        if self.depth != DEPTH_SYSTEM_DEFAULT {
            mask |= QoSPolicyMask::DEPTH;
        }
        match self.durability {
            QoSDurabilityPolicy::SystemDefault => {}
            QoSDurabilityPolicy::Volatile => mask |= QoSPolicyMask::DURABILITY_VOLATILE,
            QoSDurabilityPolicy::TransientLocal => {
                mask |= QoSPolicyMask::DURABILITY_TRANSIENT_LOCAL
            }
            // `Unknown` is a READ-BACK sentinel, never a request, so there is
            // no capability bit to demand for it — there is nothing a backend
            // could support. Refusing it belongs one level up, in
            // `validate_against`, where it is refused for every policy at once
            // rather than smuggled in as an unsatisfiable mask bit.
            QoSDurabilityPolicy::Unknown => {}
            // issue 1973 — a best-available field demands NOTHING: the
            // resolver clamps its choice to the backend's own mask
            // (`resolve_best_available`), so whatever it picks is honoured.
            QoSDurabilityPolicy::BestAvailable => {}
        }
        // Phase-301 (issue 0241): DURATION_INFINITE_MS reads the same as 0
        // (infinite = no check) at every duration check site. The
        // best-available sentinel demands nothing either (see above).
        if self.deadline_ms != 0
            && self.deadline_ms != DURATION_INFINITE_MS
            && self.deadline_ms != DURATION_BEST_AVAILABLE_MS
        {
            mask |= QoSPolicyMask::DEADLINE;
        }
        if self.lifespan_ms != 0 && self.lifespan_ms != DURATION_INFINITE_MS {
            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,
            // See the durability arm.
            QoSLivelinessPolicy::Unknown => {}
            QoSLivelinessPolicy::BestAvailable => {}
        }
        if self.liveliness_lease_ms != 0
            && self.liveliness_lease_ms != DURATION_INFINITE_MS
            && self.liveliness_lease_ms != DURATION_BEST_AVAILABLE_MS
        {
            mask |= QoSPolicyMask::LIVELINESS_LEASE;
        }
        if self.avoid_ros_namespace_conventions {
            mask |= QoSPolicyMask::AVOID_ROS_NAMESPACE_CONVENTIONS;
        }
        mask
    }

    /// Returns `Err(TransportError::IncompatibleQos(policy))` 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.
    ///
    /// **The error NAMES the policy** (phase-417 G7). This computed the
    /// offending bit and discarded it until then, returning a bare
    /// `IncompatibleQos` — so every one of the six `create_*` paths in
    /// `nros-node` refused with the one fact the caller needed missing, while
    /// `rmw_vtable.h` told the reader the opposite. Which bit is reported when
    /// several are missing is [`QoSPolicyMask::lowest_policy`]'s choice, and it
    /// is stable so a test can assert it.
    pub fn validate_against(&self, supported: QoSPolicyMask) -> Result<(), TransportError> {
        // issue 1437 — `UNKNOWN` is what a `*_get_actual_qos` read-back writes
        // for a policy it cannot report. It is an ABSENCE, so it is meaningless
        // as a REQUEST, and the way it reaches a create call is a caller
        // feeding a read-back profile straight back in — the exact round trip
        // that would otherwise turn "I cannot say" into a silently different
        // entity. Refused here, once, for all four policies.
        //
        // The refusal names NO policy, and that is not laziness: the mask's
        // bits are CAPABILITIES a backend declares, and no backend could ever
        // declare one that makes "I cannot say" servable — `required_policies`
        // has no bit to set, so there is none to report. `policy_name` already
        // answers `None` for the empty mask, which the diagnostic renders as
        // the honest "the profile is not a request", rather than naming
        // `DURABILITY_VOLATILE` for a durability that is not volatile.
        if self.has_unknown() {
            return Err(TransportError::IncompatibleQos(QoSPolicyMask::NONE));
        }
        let missing = self.required_policies().missing_from(supported);
        if missing.0 == 0 {
            Ok(())
        } else {
            Err(TransportError::IncompatibleQos(missing.lowest_policy()))
        }
    }

    /// Replace every `SYSTEM_DEFAULT` field with the backend's own answer.
    ///
    /// issue 0829 — a backend calls this at its create entry, **before**
    /// anything is derived from the profile (see [`QoSSystemDefaults`] for why
    /// the ordering matters on the zenoh path). Fields the caller DID state
    /// are left exactly as they are: this resolves an absence, it never
    /// overrides a request.
    ///
    /// Idempotent, and safe to call on a fully concrete profile — a profile
    /// with no sentinel in it is returned unchanged.
    #[must_use]
    pub const fn resolve_system_default(mut self, defaults: &QoSSystemDefaults) -> Self {
        if matches!(self.reliability, QoSReliabilityPolicy::SystemDefault) {
            self.reliability = defaults.reliability;
        }
        if matches!(self.durability, QoSDurabilityPolicy::SystemDefault) {
            self.durability = defaults.durability;
        }
        if matches!(self.history, QoSHistoryPolicy::SystemDefault) {
            self.history = defaults.history;
        }
        if self.depth == DEPTH_SYSTEM_DEFAULT {
            self.depth = defaults.depth;
        }
        self
    }

    /// `true` if any field is still the `SYSTEM_DEFAULT` sentinel.
    ///
    /// Depth is deliberately NOT part of this test: a `KEEP_ALL` profile
    /// legitimately carries depth 0 forever (any profile a caller builds with
    /// [`QoSProfile::keep_all`] and never gave a depth to — no NAMED preset is
    /// `KEEP_ALL` since phase-428 W10), and a backend that resolves the
    /// sentinel depth to 0 — XRCE does, on purpose — leaves a resolved profile
    /// reading 0 here. This asks about the
    /// three POLICY fields, where the sentinel is a distinct variant and so
    /// cannot be confused with a stated value.
    #[must_use]
    pub const fn has_unresolved_system_default(&self) -> bool {
        matches!(self.reliability, QoSReliabilityPolicy::SystemDefault)
            || matches!(self.durability, QoSDurabilityPolicy::SystemDefault)
            || matches!(self.history, QoSHistoryPolicy::SystemDefault)
    }

    /// `true` if any field still asks to be resolved best-available (issue
    /// 1973): one of the three policy variants, or the
    /// [`DURATION_BEST_AVAILABLE_MS`] sentinel in `deadline_ms` /
    /// `liveliness_lease_ms`.
    #[must_use]
    pub const fn has_best_available(&self) -> bool {
        matches!(self.reliability, QoSReliabilityPolicy::BestAvailable)
            || matches!(self.durability, QoSDurabilityPolicy::BestAvailable)
            || matches!(self.liveliness_kind, QoSLivelinessPolicy::BestAvailable)
            || self.deadline_ms == DURATION_BEST_AVAILABLE_MS
            || self.liveliness_lease_ms == DURATION_BEST_AVAILABLE_MS
    }

    /// Replace every best-available field with a concrete value — issue 1973,
    /// the ONE resolution rule every backend reaches.
    ///
    /// The rule is upstream's, transcribed from `rmw_dds_common`'s
    /// `qos_profile_get_best_available_for_{subscription,publisher}` and
    /// `qos_profile_update_best_available_for_services` (jazzy):
    ///
    /// | policy      | subscription (peers = publishers)                  | publisher (peers = subscriptions)              |
    /// |-------------|----------------------------------------------------|------------------------------------------------|
    /// | reliability | RELIABLE iff ≥1 peer and every peer is RELIABLE, else BEST_EFFORT | RELIABLE                           |
    /// | durability  | TRANSIENT_LOCAL iff ≥1 peer and all are, else VOLATILE | TRANSIENT_LOCAL                            |
    /// | liveliness  | MANUAL_BY_TOPIC iff ≥1 peer and all are, else AUTOMATIC | MANUAL_BY_TOPIC iff any peer is, else AUTOMATIC |
    /// | deadline    | largest finite peer deadline, else default (0)     | smallest finite peer deadline, else default (0) |
    /// | lease       | largest finite peer lease, else default (0)        | smallest finite peer lease, else default (0)    |
    ///
    /// A service or client ([`BestAvailableRole::Service`]) takes each field
    /// from `rmw_qos_profile_services_default`, as upstream does.
    ///
    /// **No peers is a real answer, not a failure.** It is what a backend with
    /// no graph (XRCE, uORB) always reports, and it yields upstream's own
    /// "nothing discovered yet" profile: BEST_EFFORT / VOLATILE / AUTOMATIC for
    /// a subscription, RELIABLE / TRANSIENT_LOCAL / AUTOMATIC for a publisher.
    ///
    /// **A peer whose profile the backend cannot read** (its fields are the
    /// `Unknown` sentinels, durations 0) contributes its LEAST demanding value:
    /// not reliable, not transient-local, not manual, no deadline. For a
    /// subscription that keeps the choice compatible with the peer whatever it
    /// really offers; a publisher cannot be protected the same way and offers
    /// what the readable peers ask for.
    ///
    /// **Last, the choice is clamped to what the backend honours**
    /// (`supported`, its [`Session::supported_qos_policies`]): a resolved value
    /// the backend could not grant steps down to the next one it can —
    /// TRANSIENT_LOCAL → VOLATILE, MANUAL_BY_TOPIC → AUTOMATIC, a finite
    /// deadline or lease → default — and to the `SYSTEM_DEFAULT` sentinel when
    /// the backend honours the policy not at all. "Best available" is then
    /// literally the best this backend has; the alternative, choosing a value
    /// the next line refuses with `IncompatibleQos`, would make the request
    /// fail for asking to be accommodated. Fields the caller stated are never
    /// touched.
    ///
    /// Idempotent: a profile with nothing best-available is returned as is.
    #[must_use]
    pub fn resolve_best_available(
        mut self,
        role: BestAvailableRole,
        peers: &BestAvailablePeers,
        supported: QoSPolicyMask,
    ) -> Self {
        let rel = matches!(self.reliability, QoSReliabilityPolicy::BestAvailable);
        let dur = matches!(self.durability, QoSDurabilityPolicy::BestAvailable);
        let liv = matches!(self.liveliness_kind, QoSLivelinessPolicy::BestAvailable);
        let dl = self.deadline_ms == DURATION_BEST_AVAILABLE_MS;
        let lease = self.liveliness_lease_ms == DURATION_BEST_AVAILABLE_MS;
        let any = peers.count > 0;
        match role {
            BestAvailableRole::Subscription => {
                if rel {
                    self.reliability = if any && peers.reliable == peers.count {
                        QoSReliabilityPolicy::Reliable
                    } else {
                        QoSReliabilityPolicy::BestEffort
                    };
                }
                if dur {
                    self.durability = if any && peers.transient_local == peers.count {
                        QoSDurabilityPolicy::TransientLocal
                    } else {
                        QoSDurabilityPolicy::Volatile
                    };
                }
                if liv {
                    self.liveliness_kind = if any && peers.manual_by_topic == peers.count {
                        QoSLivelinessPolicy::ManualByTopic
                    } else {
                        QoSLivelinessPolicy::Automatic
                    };
                }
                if dl {
                    self.deadline_ms = peers.deadline_max;
                }
                if lease {
                    self.liveliness_lease_ms = peers.lease_max;
                }
            }
            BestAvailableRole::Publisher => {
                if rel {
                    self.reliability = QoSReliabilityPolicy::Reliable;
                }
                if dur {
                    self.durability = QoSDurabilityPolicy::TransientLocal;
                }
                if liv {
                    self.liveliness_kind = if peers.manual_by_topic > 0 {
                        QoSLivelinessPolicy::ManualByTopic
                    } else {
                        QoSLivelinessPolicy::Automatic
                    };
                }
                if dl {
                    self.deadline_ms = peers.deadline_min;
                }
                if lease {
                    self.liveliness_lease_ms = peers.lease_min;
                }
            }
            BestAvailableRole::Service => {
                let svc = Self::QOS_PROFILE_SERVICES_DEFAULT;
                if rel {
                    self.reliability = svc.reliability;
                }
                if dur {
                    self.durability = svc.durability;
                }
                if liv {
                    self.liveliness_kind = svc.liveliness_kind;
                }
                if dl {
                    self.deadline_ms = svc.deadline_ms;
                }
                if lease {
                    self.liveliness_lease_ms = svc.liveliness_lease_ms;
                }
            }
        }

        // The clamp — only on fields that WERE best-available.
        if rel && !supported.contains(QoSPolicyMask::RELIABILITY) {
            self.reliability = QoSReliabilityPolicy::SystemDefault;
        }
        if dur {
            if matches!(self.durability, QoSDurabilityPolicy::TransientLocal)
                && !supported.contains(QoSPolicyMask::DURABILITY_TRANSIENT_LOCAL)
            {
                self.durability = QoSDurabilityPolicy::Volatile;
            }
            if matches!(self.durability, QoSDurabilityPolicy::Volatile)
                && !supported.contains(QoSPolicyMask::DURABILITY_VOLATILE)
            {
                self.durability = QoSDurabilityPolicy::SystemDefault;
            }
        }
        if liv {
            if matches!(self.liveliness_kind, QoSLivelinessPolicy::ManualByTopic)
                && !supported.contains(QoSPolicyMask::LIVELINESS_MANUAL_BY_TOPIC)
            {
                self.liveliness_kind = QoSLivelinessPolicy::Automatic;
            }
            if matches!(self.liveliness_kind, QoSLivelinessPolicy::Automatic)
                && !supported.contains(QoSPolicyMask::LIVELINESS_AUTOMATIC)
            {
                self.liveliness_kind = QoSLivelinessPolicy::None;
            }
        }
        if dl && self.deadline_ms != 0 && !supported.contains(QoSPolicyMask::DEADLINE) {
            self.deadline_ms = 0;
        }
        if lease
            && self.liveliness_lease_ms != 0
            && !supported.contains(QoSPolicyMask::LIVELINESS_LEASE)
        {
            self.liveliness_lease_ms = 0;
        }
        self
    }
}

/// Which end of the graph a best-available profile is resolved for — issue
/// 1973. See [`QoSProfile::resolve_best_available`].
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum BestAvailableRole {
    /// A publisher; resolved against the topic's discovered SUBSCRIPTIONS.
    Publisher,
    /// A subscription; resolved against the topic's discovered PUBLISHERS.
    Subscription,
    /// A service server or client; resolved to
    /// `rmw_qos_profile_services_default`, never against the graph.
    Service,
}

/// The discovered peers of one topic, folded down to what
/// [`QoSProfile::resolve_best_available`] reads — issue 1973.
///
/// A fold rather than a list so a graph walk can feed it one endpoint at a
/// time, with no allocator and no bound on how many endpoints there are.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct BestAvailablePeers {
    /// Peers observed.
    pub count: usize,
    /// …of which RELIABLE.
    pub reliable: usize,
    /// …of which TRANSIENT_LOCAL.
    pub transient_local: usize,
    /// …of which MANUAL_BY_TOPIC.
    pub manual_by_topic: usize,
    /// Smallest finite deadline seen, ms; `0` when none was finite.
    pub deadline_min: u32,
    /// Largest finite deadline seen, ms; `0` when none was finite.
    pub deadline_max: u32,
    /// Smallest finite liveliness lease seen, ms; `0` when none was finite.
    pub lease_min: u32,
    /// Largest finite liveliness lease seen, ms; `0` when none was finite.
    pub lease_max: u32,
}

impl BestAvailablePeers {
    /// No peers — what a backend with no graph reports.
    pub const NONE: Self = Self {
        count: 0,
        reliable: 0,
        transient_local: 0,
        manual_by_topic: 0,
        deadline_min: 0,
        deadline_max: 0,
        lease_min: 0,
        lease_max: 0,
    };

    /// Fold one discovered endpoint's profile in.
    ///
    /// `0`, [`DURATION_INFINITE_MS`] and the best-available sentinel all read
    /// as "no finite window" — upstream's `RMW_QOS_DEADLINE_DEFAULT`
    /// comparison, widened to our two spellings of infinite.
    pub fn observe(&mut self, peer: &QoSProfile) {
        self.count += 1;
        if matches!(peer.reliability, QoSReliabilityPolicy::Reliable) {
            self.reliable += 1;
        }
        if matches!(peer.durability, QoSDurabilityPolicy::TransientLocal) {
            self.transient_local += 1;
        }
        if matches!(peer.liveliness_kind, QoSLivelinessPolicy::ManualByTopic) {
            self.manual_by_topic += 1;
        }
        fn finite(ms: u32) -> Option<u32> {
            (ms != 0 && ms < DURATION_BEST_AVAILABLE_MS).then_some(ms)
        }
        fn fold(min: &mut u32, max: &mut u32, v: Option<u32>) {
            if let Some(v) = v {
                *min = if *min == 0 { v } else { (*min).min(v) };
                *max = (*max).max(v);
            }
        }
        fold(
            &mut self.deadline_min,
            &mut self.deadline_max,
            finite(peer.deadline_ms),
        );
        fold(
            &mut self.lease_min,
            &mut self.lease_max,
            finite(peer.liveliness_lease_ms),
        );
    }
}

/// Resolve `qos`'s best-available fields against what `session` has
/// discovered on `topic_name` — issue 1973. The graph half of
/// [`QoSProfile::resolve_best_available`], shared by every create path.
///
/// Queries [`Session::get_endpoint_info_by_topic`] for the OTHER side of the
/// topic (subscriptions for a publisher, publishers for a subscription) and
/// folds each endpoint's [`GraphEndpointInfo::qos`]. A backend that cannot
/// answer — `Unsupported`, or any error from the walk — is treated as having
/// discovered nothing, which resolves to upstream's documented
/// nothing-discovered profile rather than to a guess. A
/// [`BestAvailableRole::Service`] never walks the graph.
///
/// Returns `qos` unchanged when nothing in it is best-available, so it is
/// free to call on every create.
pub fn resolve_best_available_on<S: Session + ?Sized>(
    session: &mut S,
    role: BestAvailableRole,
    topic_name: &str,
    qos: QoSProfile,
) -> QoSProfile
where
    S::Error: From<TransportError>,
{
    if !qos.has_best_available() {
        return qos;
    }
    let mut peers = BestAvailablePeers::NONE;
    let publishers = match role {
        BestAvailableRole::Publisher => Some(false),
        BestAvailableRole::Subscription => Some(true),
        BestAvailableRole::Service => None,
    };
    if let Some(publishers) = publishers {
        let mut walked = BestAvailablePeers::NONE;
        let ok = session
            .get_endpoint_info_by_topic(publishers, topic_name, &mut |info| {
                walked.observe(&info.qos);
                true
            })
            .is_ok();
        if ok {
            peers = walked;
        }
    }
    let supported = session.supported_qos_policies();
    qos.resolve_best_available(role, &peers, supported)
}

/// Publisher trait for sending messages.
///
/// The name is rclrs's, the role is the backend seam: rclrs's `Publisher<T>` is a
/// typed, shared handle (`Arc<PublisherState<T>>`); this trait is what an RMW
/// backend implements, raw bytes in. The typed handle a node hands an application
/// is one type with no `Arc` (`nros::EmbeddedPublisher<M>`). Loaning exists:
/// `EmbeddedRawPublisher::try_loan` / `loan`, zero-copy through `SlotLending`
/// when the `lending` feature is on and the backend supports it.
///
/// # 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
    /// [`Subscription::take_serialized_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. What it saves is the CALLER-side
    /// buffer holding a whole serialised message — which is STACK,
    /// not `.bss`: `EmbeddedPublisher::publish` serialises into a
    /// per-CALL array of `DEFAULT_TX_BUF`, and this trait's own
    /// fallback below stages on the caller's stack too. There is no
    /// per-publisher staging buffer and none of it is in `.bss`;
    /// issue 0782 corrected that fiction on the C slot and left this
    /// copy of it standing (phase-428 W12's F2, one layer over).
    ///
    /// Default body — the **staging-buffer fallback** (124.E.2).
    /// Asks `size_cb` for the total length, fills a stack-allocated
    /// `[u8; STAGE_CAP]` via `chunk_cb`, then forwards to
    /// `publish_raw`. `STAGE_CAP` is the 4096 constant declared in
    /// the body, not a knob: this said `NROS_MAX_STREAM_CHUNK` until
    /// 2026-09-12 and no such setting has ever existed in any form,
    /// so a reader looking for it found nothing and had no way to
    /// learn the limit. Returns `Err(BufferTooSmall)` if the total
    /// exceeds that 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`] / [`ManualByNode`](QoSLivelinessPolicy::ManualByNode);
    /// nothing to do for the other kinds.
    ///
    /// **`Ok(())` MEANS THE ASSERTION REACHED THE WIRE, or that there was
    /// nothing to assert** (the phase-467 RMW gap-closure design study's
    /// Row 7). Under a kind for which [`QoSLivelinessPolicy::is_manual`]
    /// holds, a backend that cannot renew a lease a PEER can observe must
    /// answer `Unsupported` — an `Ok(())` from a call that sent nothing is
    /// the silently-wrong shape RFC-0089 Part I refuses, and it is what the
    /// zenoh shim answered until this study. A backend is free to keep a
    /// LOCAL record of the call (zenoh drives its own `LivelinessLost`
    /// watchdog off one) and still report `Unsupported`; the return value is
    /// about what a peer can see, not about what we noted down.
    ///
    /// The default body is `Ok(())`, which is correct only for a publisher
    /// whose kind is not manual — a backend that can be created with a manual
    /// kind must override.
    fn assert_liveliness(&self) -> Result<(), Self::Error> {
        Ok(())
    }

    /// This publisher's own global identifier — upstream
    /// `rmw_get_gid_for_publisher`, the vtable's `get_gid_for_publisher`.
    ///
    /// The phase-467 RMW gap-closure design study's Q1. **Not the same
    /// question as [`MessageInfo::publisher_gid`](nros_core::MessageInfo::publisher_gid)**,
    /// which is the identity of whoever sent a message we RECEIVED. Since
    /// that study's Q1(a) the two are one TYPE — both
    /// [`PUBLISHER_GID_SIZE`](nros_core::PUBLISHER_GID_SIZE) wide, nothing
    /// truncated, no mapping — and since issue 1495 they are one VALUE on
    /// zenoh and Cyclone: a take reports the gid this publisher's own
    /// `get_gid` reports, so the two compare. (XRCE answers `get_gid` — the
    /// DDS GID its graph derives, issue 2110 — but reports no take-side gid;
    /// uORB answers neither.)
    ///
    /// **How many of the 24 bytes MEAN anything is a backend property.** A
    /// backend whose identity is narrower zero-extends through
    /// [`pad_publisher_gid`](nros_core::pad_publisher_gid) rather than
    /// leaving the tail undefined, so two gids naming one entity compare
    /// equal. Gids from DIFFERENT backends are never comparable.
    ///
    /// The default is `Err(Unsupported)`, never a fabricated or all-zero id:
    /// an all-zero gid is what an unwritten buffer holds, so handing one back
    /// would make "this backend has no identity for this publisher"
    /// indistinguishable from an answer. That is the same NULL-slot
    /// discipline the ABI header states for `get_serialization_format`.
    fn get_gid(&self) -> Result<[u8; nros_core::PUBLISHER_GID_SIZE], Self::Error>
    where
        Self::Error: From<TransportError>,
    {
        Err(TransportError::Unsupported.into())
    }

    /// How many subscriptions are MATCHED to this publisher — upstream
    /// `rmw_publisher_count_matched_subscriptions`, the vtable's
    /// `publisher_count_matched_subscriptions`. Issue 1643.
    ///
    /// Matched means QoS-compatible with THIS publisher, which is upstream's
    /// question and a stronger one than "how many subscriptions are on the
    /// topic" (the graph's `count_subscribers`): an incompatible peer is on
    /// the topic and not matched.
    ///
    /// The default is `Err(Unsupported)`, never `Ok(0)` — `0` means "nobody
    /// is listening", and a backend that cannot tell must not say that. A
    /// caller with a weaker fallback (the node API's topic-wide graph count)
    /// takes it on `Unsupported`.
    fn count_matched_subscriptions(&self) -> Result<usize, Self::Error>
    where
        Self::Error: From<TransportError>,
    {
        Err(TransportError::Unsupported.into())
    }

    /// The profile this entity is ACTUALLY running — what the backend GRANTED,
    /// not what the call site requested.
    ///
    /// Upstream's `rmw_publisher_get_actual_qos`. The two differ whenever a
    /// backend serves something other than what it was asked for, and the
    /// difference is exactly what answers "why is nothing arriving": a
    /// RELIABLE reader does not match a BEST_EFFORT writer, and a clamped
    /// history depth drops samples the caller believed were kept.
    ///
    /// **Per-field, and an unreportable field is an ABSENCE.** A backend that
    /// determines four policies and not the fifth returns the four and leaves
    /// the fifth at its `Unknown` variant. Never echo the request into a field
    /// you did not read back — that is the inverse of this method's meaning,
    /// and it is what the six C-ABI slots did until this phase.
    ///
    /// The default is [`QoSProfile::QOS_PROFILE_UNKNOWN`] — every field an
    /// absence — which is the honest answer for a backend with no read-back at
    /// all (XRCE has none; zenoh-pico's QoS is per-message flags). It is NOT
    /// the request, and it is NOT `QOS_PROFILE_DEFAULT`.
    fn actual_qos(&self) -> QoSProfile {
        QoSProfile::QOS_PROFILE_UNKNOWN
    }
}

/// Subscription trait for receiving messages.
///
/// # Threading
///
/// `&mut self` on `take_serialized` — 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
///
/// `take_serialized` 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 `take_serialized`.
pub trait Subscription {
    /// 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 take_serialized(&mut self, buf: &mut [u8]) -> Result<Option<usize>, Self::Error>;

    /// Issue 1612 — how many bytes the sample the most recent take REFUSED
    /// as too big for the caller's buffer needed.
    ///
    /// Read it right after a take returns `BufferTooSmall`; a later take may
    /// overwrite it. `None` means the backend does not know the size (the
    /// default, and the honest answer for one that never measured it), never
    /// "zero bytes" — a zero-byte sample fits every buffer, so no refusal can
    /// need zero.
    ///
    /// A query beside the error rather than a payload on
    /// `TransportError::BufferTooSmall`: that variant is the same word a dozen
    /// unrelated paths return (arena exhaustion, a short reply buffer), so
    /// giving it a field would make every one of them invent a size. Only a
    /// take ever has one, and only a take is asked.
    fn refused_sample_len(&self) -> Option<usize> {
        None
    }

    /// 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 [`take_serialized`](Self::take_serialized)
    /// 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 take_serialized_with_attachment(
        &mut self,
        buf: &mut [u8],
        _att_buf: &mut [u8],
    ) -> Result<Option<(usize, usize)>, Self::Error> {
        match self.take_serialized(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 `take_serialized` 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 take_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];
            // Issue 0971 — NOT `take_serialized(slot)?`. The `?` propagates the
            // error and DISCARDS `count`, which is precisely what the doc
            // comment above forbids ("Partial drains MUST report the count, not
            // error out") and what `c3af8c1d1` removed from the two concrete
            // implementations. A backend that does not override this body got
            // the original defect back through the default.
            //
            // The shape is the one that fix established: a drain that has
            // already taken messages reports the COUNT, and the error is
            // delivered by the NEXT call — the caller sees every message it was
            // handed, then the reason the drain stopped. With nothing taken
            // there is no count to protect, so the error goes out immediately.
            //
            // Parking is per-implementation state, which a default body has no
            // place to keep. So it does the half it can do correctly: it
            // returns the partial count and lets the error surface on the
            // caller's next `take_serialized`, which is where an unconsumed
            // backend error still sits.
            match self.take_serialized(slot) {
                Ok(Some(len)) => {
                    out_lens[i] = len;
                    count += 1;
                }
                Ok(None) => break,
                Err(e) => {
                    if count == 0 {
                        return Err(e);
                    }
                    break;
                }
            }
        }
        Ok(count)
    }

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

        match self.take_serialized(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 `take_serialized` 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`](Subscription::process_raw_in_place) /
    /// [`process_raw_in_place_with_info`](Subscription::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`](Subscription::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 [`take_serialized_with_info`](Subscription::take_serialized_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 [`take_serialized`](Subscription::take_serialized) with no info.
    fn take_serialized_with_info(
        &mut self,
        buf: &mut [u8],
    ) -> Result<Option<(usize, Option<nros_core::MessageInfo>)>, Self::Error> {
        self.take_serialized(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 `take_serialized` with no CRC info.
    #[cfg(feature = "safety-e2e")]
    fn take_validated(
        &mut self,
        buf: &mut [u8],
    ) -> Result<Option<(usize, crate::IntegrityStatus)>, Self::Error> {
        self.take_serialized(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.
    ///
    /// Returns `true` when the backend will wake `waker` on an arrival, and
    /// `false` when it cannot (issue 1940): the caller must then poll, or
    /// rely on the session-wide wake, and a C caller reports
    /// `RCL_RET_UNSUPPORTED` rather than `OK`.
    ///
    /// Default: `false` — a backend that writes no code cannot wake anyone.
    fn register_waker(&self, _waker: &core::task::Waker) -> bool {
        false
    }

    /// 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.
    ///
    /// Subscription-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()
    }
    /// The profile this entity is ACTUALLY running — what the backend GRANTED,
    /// not what the call site requested.
    ///
    /// Upstream's `rmw_subscription_get_actual_qos`. The two differ whenever a
    /// backend serves something other than what it was asked for, and the
    /// difference is exactly what answers "why is nothing arriving": a
    /// RELIABLE reader does not match a BEST_EFFORT writer, and a clamped
    /// history depth drops samples the caller believed were kept.
    ///
    /// **Per-field, and an unreportable field is an ABSENCE.** A backend that
    /// determines four policies and not the fifth returns the four and leaves
    /// the fifth at its `Unknown` variant. Never echo the request into a field
    /// you did not read back — that is the inverse of this method's meaning,
    /// and it is what the six C-ABI slots did until this phase.
    ///
    /// The default is [`QoSProfile::QOS_PROFILE_UNKNOWN`] — every field an
    /// absence — which is the honest answer for a backend with no read-back at
    /// all (XRCE has none; zenoh-pico's QoS is per-message flags). It is NOT
    /// the request, and it is NOT `QOS_PROFILE_DEFAULT`.
    fn actual_qos(&self) -> QoSProfile {
        QoSProfile::QOS_PROFILE_UNKNOWN
    }

    /// How many publishers are MATCHED to this subscription — upstream
    /// `rmw_subscription_count_matched_publishers`, the vtable's
    /// `subscription_count_matched_publishers`. Issue 1643.
    ///
    /// The subscription half of
    /// [`Publisher::count_matched_subscriptions`], which states the contract:
    /// matched, not merely discovered; `Err(Unsupported)` by default, never a
    /// fabricated `0`.
    fn count_matched_publishers(&self) -> Result<usize, Self::Error>
    where
        Self::Error: From<TransportError>,
    {
        Err(TransportError::Unsupported.into())
    }
}

/// 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: Subscription {
    /// 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>;
}

/// Issues 1612 / 1632 — the size a refused take needed, as a drop log prints
/// it: the number, or `?` when unknown. No allocator.
///
/// [`RefusedLen::for_error`] is the one rule for WHICH refusal may carry a
/// size: only `BufferTooSmall`, the caller's-buffer refusal a take measures.
/// `MessageTooLarge` is the backend's own staging buffer refusing before any
/// take saw a length, so a size recorded by an earlier `BufferTooSmall` must
/// not be printed beside it. The C++ subscription drop log and `nros-node`'s
/// arena drop log both read it from here, so the two cannot disagree.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct RefusedLen(pub Option<usize>);

impl RefusedLen {
    /// The size `recorded` (a `refused_*_len()` reading) for `err`, or
    /// unknown when `err` is not a refusal a take measured.
    pub fn for_error(err: &TransportError, recorded: Option<usize>) -> Self {
        Self(match err {
            TransportError::BufferTooSmall => recorded,
            _ => None,
        })
    }
}

impl core::fmt::Display for RefusedLen {
    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
        match self.0 {
            Some(n) => write!(f, "{n}"),
            None => f.write_str("?"),
        }
    }
}

/// Service server trait for handling requests.
///
/// # Threading
///
/// `&mut self` on `take_request` and `send_response` — 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 `take_request(buf)`.
/// 2. If `Some(req)` returned, decode, run handler, encode reply.
/// 3. `send_response(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 ServiceTrait {
    /// 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` / `ClientTrait`. Backends that
    /// surface incoming-request notifications wake `waker` when
    /// `has_request()` flips true. Returns `false` when the backend cannot
    /// (issue 1940); the caller then falls back to polling. Default: `false`.
    fn register_waker(&self, _waker: &core::task::Waker) -> bool {
        false
    }

    /// 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_response` is called, since
    /// `send_response` takes `&mut self`.
    fn take_request<'a>(
        &mut self,
        buf: &'a mut [u8],
    ) -> Result<Option<ServiceRequest<'a>>, Self::Error>;

    /// Issue 1632 — how many bytes the request the most recent
    /// [`take_request`](Self::take_request) REFUSED as too big for the
    /// caller's buffer needed.
    ///
    /// [`Subscription::refused_sample_len`]'s contract, for the service
    /// server: read it right after a take returns `BufferTooSmall`; `None`
    /// means the backend does not know (the default), never "zero bytes".
    fn refused_request_len(&self) -> Option<usize> {
        None
    }

    /// 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_response(&mut self, sequence_number: i64, data: &[u8]) -> Result<(), Self::Error>;

    /// Issue 1942 — the per-request metadata of the request the most recent
    /// successful [`take_request`](Self::take_request) returned: client GID +
    /// client sequence number, and the timestamps the backend can supply.
    ///
    /// rclrs's `ServiceInfo`. Read it after a take and before the next one.
    /// **A field the backend cannot report is `None`**, and the default — for
    /// a backend with no per-request metadata at all — is
    /// [`ServiceInfo::NOT_REPORTED`](nros_core::ServiceInfo::NOT_REPORTED),
    /// never a zeroed or invented value (the `MessageInfo` precedent).
    ///
    /// Not [`ServiceRequest::sequence_number`]: that is the token this backend
    /// replies against, which on zenoh is a local reply-slot index, while
    /// `request_id.sequence_number` is the number the CLIENT assigned.
    fn request_info(&self) -> nros_core::ServiceInfo {
        nros_core::ServiceInfo::NOT_REPORTED
    }

    /// 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>,
    {
        self.handle_request_with_info::<S>(req_buf, reply_buf, |req, _info| handler(req))
    }

    /// Issue 1942 — [`handle_request`](Self::handle_request), with the
    /// request's [`ServiceInfo`](nros_core::ServiceInfo) (read through
    /// [`request_info`](Self::request_info) right after the take) handed to
    /// the handler beside it. rclrs's `(request, ServiceInfo)` callback.
    fn handle_request_with_info<S: RosService>(
        &mut self,
        req_buf: &mut [u8],
        reply_buf: &mut [u8],
        handler: impl FnOnce(&S::Request, &nros_core::ServiceInfo) -> 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.take_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),
        };

        // Issue 1942 — the take that just succeeded is the one this describes.
        let info = self.request_info();

        // 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, &info);

        // 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_response(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.take_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_response(sequence_number, &reply_buf[..len])?;
        Ok(true)
    }

    /// Handle one request by STREAMING: the handler reads fields off the wire
    /// and writes the reply's fields straight back, so neither the request nor
    /// the reply is ever materialised as a value.
    ///
    /// phase-382 W1'. `handle_request_boxed` above boxes the REPLY because the
    /// parameter responses are enormous — `GetParametersResponse` measures
    /// 1,176,072 bytes. It does not box the REQUEST, which is deserialized by
    /// value into a stack local one line above the handler, and
    /// `SetParametersRequest` measures **1,192,968 bytes**. So every
    /// `ros2 param set` against a node put a 1.19 MB local on the calling
    /// task's stack — larger than the reply the boxing exists for, on every
    /// platform, with `param-services` live on Zephyr.
    ///
    /// Streaming removes both, and removes the `alloc` requirement with them:
    /// the whole value is never needed, because serialisation happens on the
    /// line after construction. Three things make it safe rather than clever:
    ///
    /// * **No `rcl_interfaces` message uses a DHEADER** — every `Serialize` impl
    ///   is plain sequential CDR, so hand-written field writes are byte-identical
    ///   to the generated ones. (If a future message gains XCDR2 extensibility
    ///   this stops being true for THAT message; see RFC-0055.)
    /// * `req_buf` and `reply_buf` are disjoint, so a handler can hold the
    ///   reader and the writer at once.
    /// * `CdrReader::read_string` borrows out of `req_buf` rather than copying,
    ///   so a handler can look a name up without a buffer of its own.
    ///
    /// The cost is that the hand-written writes can drift from the generated
    /// `Serialize`. Guard it with a round-trip test that deserialises the
    /// streamed bytes back into the generated type — the by-value handler makes
    /// a good test-only oracle.
    ///
    /// No `alloc`, deliberately: this is the seam that lets `param-services` and
    /// `lifecycle-services` build without it.
    fn handle_request_raw(
        &mut self,
        req_buf: &mut [u8],
        reply_buf: &mut [u8],
        handler: impl FnOnce(
            &mut nros_core::CdrReader<'_>,
            &mut nros_core::CdrWriter<'_>,
        ) -> Result<(), TransportError>,
    ) -> 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.take_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),
        };

        // Split the borrow so the reader (over `req_buf`) and the writer (over
        // `reply_buf`) can be live at the same time. They are separate
        // parameters, so this is disjoint by construction.
        let mut reader = CdrReader::new_with_header(&req_buf[data_offset..data_offset + data_len])
            .map_err(|_| TransportError::DeserializationError)?;
        let mut writer =
            CdrWriter::new_with_header(reply_buf).map_err(|_| TransportError::BufferTooSmall)?;

        handler(&mut reader, &mut writer)?;
        let len = writer.position();

        self.send_response(sequence_number, &reply_buf[..len])?;
        Ok(true)
    }
    /// The profile this entity is ACTUALLY running — what the backend GRANTED,
    /// not what the call site requested.
    ///
    /// Upstream's `rmw_service_request_subscription_get_actual_qos`. The two differ whenever a
    /// backend serves something other than what it was asked for, and the
    /// difference is exactly what answers "why is nothing arriving": a
    /// RELIABLE reader does not match a BEST_EFFORT writer, and a clamped
    /// history depth drops samples the caller believed were kept.
    ///
    /// **Per-field, and an unreportable field is an ABSENCE.** A backend that
    /// determines four policies and not the fifth returns the four and leaves
    /// the fifth at its `Unknown` variant. Never echo the request into a field
    /// you did not read back — that is the inverse of this method's meaning,
    /// and it is what the six C-ABI slots did until this phase.
    ///
    /// The default is [`QoSProfile::QOS_PROFILE_UNKNOWN`] — every field an
    /// absence — which is the honest answer for a backend with no read-back at
    /// all (XRCE has none; zenoh-pico's QoS is per-message flags). It is NOT
    /// the request, and it is NOT `QOS_PROFILE_DEFAULT`.
    fn request_subscription_actual_qos(&self) -> QoSProfile {
        QoSProfile::QOS_PROFILE_UNKNOWN
    }

    /// The profile this entity is ACTUALLY running — what the backend GRANTED,
    /// not what the call site requested.
    ///
    /// Upstream's `rmw_service_response_publisher_get_actual_qos`. The two differ whenever a
    /// backend serves something other than what it was asked for, and the
    /// difference is exactly what answers "why is nothing arriving": a
    /// RELIABLE reader does not match a BEST_EFFORT writer, and a clamped
    /// history depth drops samples the caller believed were kept.
    ///
    /// **Per-field, and an unreportable field is an ABSENCE.** A backend that
    /// determines four policies and not the fifth returns the four and leaves
    /// the fifth at its `Unknown` variant. Never echo the request into a field
    /// you did not read back — that is the inverse of this method's meaning,
    /// and it is what the six C-ABI slots did until this phase.
    ///
    /// The default is [`QoSProfile::QOS_PROFILE_UNKNOWN`] — every field an
    /// absence — which is the honest answer for a backend with no read-back at
    /// all (XRCE has none; zenoh-pico's QoS is per-message flags). It is NOT
    /// the request, and it is NOT `QOS_PROFILE_DEFAULT`.
    fn response_publisher_actual_qos(&self) -> QoSProfile {
        QoSProfile::QOS_PROFILE_UNKNOWN
    }
}

/// Decode a CDR service reply (encapsulation header included) — the typed
/// half [`ClientTrait::take_response`] and [`ClientTrait::take_response_for`]
/// share.
fn decode_reply<S: RosService>(bytes: &[u8]) -> Result<S::Reply, TransportError> {
    use nros_core::CdrReader;

    let mut reader =
        CdrReader::new_with_header(bytes).map_err(|_| TransportError::DeserializationError)?;
    S::Reply::deserialize(&mut reader).map_err(|_| TransportError::DeserializationError)
}

/// 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. `take_response_raw(buf)` — non-blocking; returns
///    `Ok(Some(len))` when the reply is back.
///
/// Phase-301 (issue 0240): the deprecated blocking `call_raw` path is
/// DELETED — `send_request_raw` + `take_response_raw` is the one
/// request/reply path, and both are required for service-capable
/// backends.
pub trait ClientTrait {
    /// Error type for service operations
    type Error;

    /// Send a service request without waiting for a reply (non-blocking).
    ///
    /// Returns the SEQUENCE ID the backend assigned. The caller must
    /// subsequently poll [`take_response_raw`](Self::take_response_raw) and
    /// match that id against the one the reply carries.
    ///
    /// Issue 0778 — this returned `()` until 2026-08-25, and every backend
    /// computed an id and discarded it. With nothing to correlate by, a client
    /// with two calls outstanding could not tell the replies apart, so each
    /// backend picked a policy: cyclonedds abandoned the older request, zenoh
    /// took the first reply. Both are wrong for `send_goal` and
    /// `SetParameters`, which travel this path and are not idempotent.
    fn send_request_raw(&mut self, request: &[u8]) -> Result<i64, Self::Error>;

    /// Poll for a reply (non-blocking).
    ///
    /// Returns `Ok(Some((len, sequence_id)))` when a reply has arrived,
    /// `Ok(None)` if not yet available, or `Err` on failure. The
    /// `sequence_id` is the one [`send_request_raw`](Self::send_request_raw)
    /// returned for the request this answers.
    ///
    /// It used to say "a reply to the MOST RECENTLY sent request", which was
    /// the single-outstanding-call assumption written into the contract.
    fn take_response_raw(
        &mut self,
        reply_buf: &mut [u8],
    ) -> Result<Option<(usize, i64)>, Self::Error>;

    /// Issue 1632 — how many bytes the reply the most recent
    /// [`take_response_raw`](Self::take_response_raw) REFUSED as too big for
    /// the caller's buffer needed. [`Subscription::refused_sample_len`]'s
    /// contract, for the client: `None` is "the backend does not know".
    fn refused_response_len(&self) -> Option<usize> {
        None
    }

    /// Issue 2136 — the metadata of the reply the most recent successful
    /// [`take_response_raw`](Self::take_response_raw) returned: upstream
    /// `rmw_take_response`'s `rmw_service_info_t`, the client twin of
    /// [`ServiceTrait::request_info`].
    ///
    /// - `request_id` — the request this reply answers: this client's own GID
    ///   and the sequence number of its request;
    /// - `source_timestamp` — when the SERVER sent the reply, on its clock;
    /// - `received_timestamp` — when this client received it, on its clock.
    ///
    /// Read it after a take and before the next one. **A field the backend
    /// cannot report is `None`**, and the default — for a backend with no
    /// per-reply metadata at all — is
    /// [`ServiceInfo::NOT_REPORTED`](nros_core::ServiceInfo::NOT_REPORTED).
    fn response_info(&self) -> nros_core::ServiceInfo {
        nros_core::ServiceInfo::NOT_REPORTED
    }

    /// 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), returning the SEQUENCE ID
    /// the backend assigned — [`crate::SEQUENCE_ID_UNKNOWN`] on a backend that
    /// cannot assign one. Hand it to
    /// [`take_response_for`](Self::take_response_for) so this call takes only
    /// its own reply.
    ///
    /// Issue 2033 — this returned `()` after issue 2009, discarding the id the
    /// raw path had carried since issue 0778, so a typed caller had nothing to
    /// correlate by.
    fn send_request<S: RosService>(
        &mut self,
        request: &S::Request,
        req_buf: &mut [u8],
    ) -> Result<i64, 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 the next typed reply (non-blocking), WHICHEVER request it
    /// answers.
    ///
    /// Returns `Ok(Some((reply, sequence_id)))`: `sequence_id` is the id
    /// [`send_request`](Self::send_request) returned for the request this
    /// reply answers, and the caller decides whom it belongs to — upstream's
    /// `rcl_take_response` request header. A caller with ONE request in mind
    /// wants [`take_response_for`](Self::take_response_for), which drops the
    /// replies of requests it abandoned.
    ///
    /// Issue 2033 — this used to return the reply alone and was documented as
    /// "the reply to the most recently sent request", which it never checked:
    /// after a resend or a timeout, the abandoned request's late reply was
    /// handed to the new one.
    fn take_response<S: RosService>(
        &mut self,
        reply_buf: &mut [u8],
    ) -> Result<Option<(S::Reply, i64)>, Self::Error>
    where
        Self::Error: From<TransportError>,
    {
        match self.take_response_raw(reply_buf)? {
            Some((len, seq)) => Ok(Some((decode_reply::<S>(&reply_buf[..len])?, seq))),
            None => Ok(None),
        }
    }

    /// Poll for the typed reply to the request `sequence_id` names
    /// (non-blocking), dropping any reply that answers a different one.
    ///
    /// `sequence_id` is what [`send_request`](Self::send_request) returned.
    /// The rule is [`crate::correlation`]'s, the one every other client
    /// surface applies (issue 2009): a reply is delivered only to the request
    /// whose id it carries, a stale reply is DROPPED (at most
    /// [`crate::MAX_STALE_DRAIN_PER_POLL`] per call), and an id-less reply or
    /// request ([`crate::SEQUENCE_ID_UNKNOWN`]) matches the sole request in
    /// mind. `Ok(None)` — nothing for this request yet.
    fn take_response_for<S: RosService>(
        &mut self,
        sequence_id: i64,
        reply_buf: &mut [u8],
    ) -> Result<Option<S::Reply>, Self::Error>
    where
        Self::Error: From<TransportError>,
    {
        let mut pending = crate::RequestCorrelator::<(), 1>::new();
        // A one-entry table cannot be full on its first insert.
        let _ = pending.insert(sequence_id, ());
        match crate::take_correlated(self, reply_buf, &mut pending)? {
            Some((len, _seq, ())) => Ok(Some(decode_reply::<S>(&reply_buf[..len])?)),
            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.
    ///
    /// Returns `false` when the backend cannot wake `waker` (issue 1940);
    /// the caller then polls. Default: `false`.
    fn register_waker(&self, _waker: &core::task::Waker) -> bool {
        false
    }

    // phase-428 W13 — the ~~`start_server_discovery`~~ / ~~`poll_server_discovery`~~
    // pair was DELETED here. It existed because the only zenoh backend could
    // answer "is a server up?" only by issuing a `z_liveliness_get`, which is
    // start-then-poll, so the synchronous `service_is_ready` below had nothing
    // to return but a latch (issue 1087). The pair had no vtable slot, so every
    // C-ABI backend inherited its default and `wait_for_service` answered
    // without asking. The zenoh backend now keeps a matched-server SET fed by
    // its liveliness subscriber — the state upstream's DDS discovery cache
    // keeps — and `service_is_ready` reads it synchronously; the wait loops
    // spin on that one query, and nothing needs the pair.

    /// Whether a matching service server is currently discoverable.
    ///
    /// Mirrors `rclcpp::ClientBase::service_is_ready` — the NAME is upstream's.
    /// The SHAPE is `rcl`'s: `rcl_service_server_is_available(node, client,
    /// bool *is_available)` returns `RCL_RET_OK` "if the check was made
    /// successfully (regardless of the service readiness)", i.e. the return
    /// code says whether the CHECK worked and the out-param carries the ANSWER.
    /// rclcpp collapses that to a bare `bool` and moves the error to
    /// exceptions; RFC-0018 forbids exceptions, so `Result<bool, _>` is how the
    /// same contract is expressed here (phase-379 W6, RFC-0036).
    ///
    /// Returns `Ok(true)` if at least one matching server is CURRENTLY
    /// discoverable, `Ok(false)` if none is, or `Err(_)` if the backend
    /// cannot answer (e.g. XRCE — micro-XRCE-DDS-Client has no participant
    /// enumeration).
    ///
    /// CURRENT, not latched: upstream `rmw.h` says the outcome reflects a
    /// change in either direction, and `rclcpp::ClientBase::service_is_ready`
    /// calls `rcl_service_server_is_available` on every invocation. A backend
    /// answers from the discovery state it maintains — the DDS built-in
    /// topics, or the zenoh matched-server set fed by liveliness PUT/DELETE
    /// (phase-428 W13) — so a server that goes away reads unavailable again.
    /// Every wait loop in the tree (`wait_for_service`,
    /// `wait_for_action_server`, their C and C++ bindings) is a spin on this
    /// one method; there is no separate asynchronous probe (issue 1087).
    ///
    /// Those are three answers, not two, and keeping them apart is the whole
    /// point: `is_server_ready` used to sit beside this method collapsing
    /// "don't know" and "no server" into one `false`, its trait default was
    /// `true`, and only zenoh overrode it — so on every other image
    /// `wait_for_service` returned success without probing anything. Issue
    /// 1008 deleted it rather than fixing it, because a caller cannot recover
    /// from an answer that does not distinguish the two.
    ///
    /// User-facing surface: `Client<S>::server_available()` in Rust,
    /// `rclc_client_server_available()` in C/C++. Clients use this
    /// to gate the first request 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 service_is_ready(&self) -> Result<bool, Self::Error>
    where
        Self::Error: From<TransportError>,
    {
        Err(TransportError::Unsupported.into())
    }

    /// This client's own global identifier — upstream (Jazzy)
    /// `rmw_get_gid_for_client`, the vtable's `get_gid_for_client`. Issue
    /// 1978.
    ///
    /// The client twin of [`Publisher::get_gid`]: the identity a server keys
    /// a request id on (zenoh: the gid stamped into every request
    /// attachment; Cyclone: the request writer's GUID), zero-extended to
    /// [`PUBLISHER_GID_SIZE`](nros_core::PUBLISHER_GID_SIZE) by
    /// [`pad_publisher_gid`](nros_core::pad_publisher_gid).
    ///
    /// The default is `Err(Unsupported)`, never an all-zero id — the same
    /// discipline, for the same reason, as `Publisher::get_gid`.
    fn get_gid(&self) -> Result<[u8; nros_core::PUBLISHER_GID_SIZE], Self::Error>
    where
        Self::Error: From<TransportError>,
    {
        Err(TransportError::Unsupported.into())
    }

    /// The profile this entity is ACTUALLY running — what the backend GRANTED,
    /// not what the call site requested.
    ///
    /// Upstream's `rmw_client_request_publisher_get_actual_qos`. The two differ whenever a
    /// backend serves something other than what it was asked for, and the
    /// difference is exactly what answers "why is nothing arriving": a
    /// RELIABLE reader does not match a BEST_EFFORT writer, and a clamped
    /// history depth drops samples the caller believed were kept.
    ///
    /// **Per-field, and an unreportable field is an ABSENCE.** A backend that
    /// determines four policies and not the fifth returns the four and leaves
    /// the fifth at its `Unknown` variant. Never echo the request into a field
    /// you did not read back — that is the inverse of this method's meaning,
    /// and it is what the six C-ABI slots did until this phase.
    ///
    /// The default is [`QoSProfile::QOS_PROFILE_UNKNOWN`] — every field an
    /// absence — which is the honest answer for a backend with no read-back at
    /// all (XRCE has none; zenoh-pico's QoS is per-message flags). It is NOT
    /// the request, and it is NOT `QOS_PROFILE_DEFAULT`.
    fn request_publisher_actual_qos(&self) -> QoSProfile {
        QoSProfile::QOS_PROFILE_UNKNOWN
    }

    /// The profile this entity is ACTUALLY running — what the backend GRANTED,
    /// not what the call site requested.
    ///
    /// Upstream's `rmw_client_response_subscription_get_actual_qos`. The two differ whenever a
    /// backend serves something other than what it was asked for, and the
    /// difference is exactly what answers "why is nothing arriving": a
    /// RELIABLE reader does not match a BEST_EFFORT writer, and a clamped
    /// history depth drops samples the caller believed were kept.
    ///
    /// **Per-field, and an unreportable field is an ABSENCE.** A backend that
    /// determines four policies and not the fifth returns the four and leaves
    /// the fifth at its `Unknown` variant. Never echo the request into a field
    /// you did not read back — that is the inverse of this method's meaning,
    /// and it is what the six C-ABI slots did until this phase.
    ///
    /// The default is [`QoSProfile::QOS_PROFILE_UNKNOWN`] — every field an
    /// absence — which is the honest answer for a backend with no read-back at
    /// all (XRCE has none; zenoh-pico's QoS is per-message flags). It is NOT
    /// the request, and it is NOT `QOS_PROFILE_DEFAULT`.
    fn response_subscription_actual_qos(&self) -> QoSProfile {
        QoSProfile::QOS_PROFILE_UNKNOWN
    }
}

/// 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. In-tree, every backend sits behind the C vtable, so the one
/// production impl is `CffiRmw`, spelled `CffiRmw::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::*;

    /// phase-515 R8 -- rclrs's `*_duration` QoS builders, in the record's ms.
    #[test]
    fn qos_duration_builders_store_whole_milliseconds() {
        use core::time::Duration;
        let q = QoSProfile::new()
            .deadline_duration(Duration::from_millis(250))
            .lifespan_duration(Duration::from_secs(2))
            .liveliness_lease_duration(Duration::from_micros(1500));
        assert_eq!(q.deadline_ms, 250);
        assert_eq!(q.lifespan_ms, 2000);
        assert_eq!(q.liveliness_lease_ms, 1, "1.5 ms truncates to whole ms");
        // Sub-millisecond is never "infinite" (0) by accident.
        assert_eq!(
            QoSProfile::new()
                .deadline_duration(Duration::from_nanos(1))
                .deadline_ms,
            1
        );
        assert_eq!(
            QoSProfile::new()
                .deadline_duration(Duration::ZERO)
                .deadline_ms,
            0
        );
        assert_eq!(
            QoSProfile::new()
                .lifespan_duration(Duration::from_secs(u64::MAX))
                .lifespan_ms,
            u32::MAX,
            "saturates"
        );
    }

    /// issue 0783 -- the variants with a payload render it; the rest render
    /// their fixed line. `heapless`, so the check runs without `alloc`.
    #[test]
    fn transport_error_display_renders_payloads() {
        use core::fmt::Write;
        fn show(e: &TransportError) -> heapless::String<64> {
            let mut s = heapless::String::new();
            write!(s, "{e}").unwrap();
            s
        }
        assert_eq!(show(&TransportError::Timeout).as_str(), "timed out");
        assert_eq!(
            show(&TransportError::IncompatibleQos(QoSPolicyMask::RELIABILITY)).as_str(),
            "incompatible QoS: RELIABILITY"
        );
        assert_eq!(
            show(&TransportError::IncompatibleQos(QoSPolicyMask::NONE)).as_str(),
            "incompatible QoS"
        );
        assert_eq!(
            show(&TransportError::Backend("z_open refused")).as_str(),
            "backend error: z_open refused"
        );
    }

    /// Issue 0971 — the DEFAULT `take_sequence` body must report a partial
    /// count rather than discard it, which is what its own doc comment says and
    /// what `?` did not do.
    ///
    /// This is the default's negative control: a backend that takes two
    /// messages and then errors must hand the caller those two. Before the fix
    /// the `?` threw them away and the caller could not tell two messages had
    /// been consumed — they were gone from the queue and absent from the
    /// result.
    mod take_sequence_default {
        use super::*;

        #[derive(Debug, PartialEq)]
        struct Boom;

        /// Yields `n` one-byte messages, then errors forever.
        struct ThenErrors {
            left: usize,
        }

        impl Subscription for ThenErrors {
            type Error = Boom;

            fn take_serialized(&mut self, buf: &mut [u8]) -> Result<Option<usize>, Self::Error> {
                if self.left == 0 {
                    return Err(Boom);
                }
                self.left -= 1;
                buf[0] = 0xAB;
                Ok(Some(1))
            }

            fn deserialization_error(&self) -> Self::Error {
                Boom
            }
        }

        #[test]
        fn a_partial_drain_reports_its_count_instead_of_the_error() {
            let mut sub = ThenErrors { left: 2 };
            let mut buf = [0u8; 4 * 8];
            let mut lens = [0usize; 4];

            // Two messages are available, the third call errors.
            let got = sub.take_sequence(&mut buf, 8, 4, &mut lens);

            assert_eq!(
                got,
                Ok(2),
                "a drain that took messages must report them; `?` discarded the \
                 count and the caller lost two consumed messages"
            );
            assert_eq!(&lens[..2], &[1, 1]);
        }

        #[test]
        fn an_error_with_nothing_taken_is_returned_immediately() {
            let mut sub = ThenErrors { left: 0 };
            let mut buf = [0u8; 4 * 8];
            let mut lens = [0usize; 4];

            // No count to protect, so the caller should see the error itself
            // rather than an ambiguous `Ok(0)`.
            assert_eq!(sub.take_sequence(&mut buf, 8, 4, &mut lens), Err(Boom));
        }
    }

    #[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 = QoSProfile::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 = QoSProfile::default().apply_overrides(
            "/chatter",
            QoSOverrideRole::Subscription,
            OVERRIDES,
        );
        assert_eq!(sub_qos, QoSProfile::default());

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

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

    #[test]
    fn test_qos_defaults() {
        let qos = QoSProfile::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 = QoSProfile::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 = QoSProfile::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 = QoSProfile::QOS_PROFILE_SERVICES_DEFAULT;
        assert_eq!(qos.reliability, QoSReliabilityPolicy::Reliable);
        assert_eq!(qos.durability, QoSDurabilityPolicy::Volatile);
    }

    /// issue 0829 — `SYSTEM_DEFAULT` asks for NOTHING, so it demands no policy.
    ///
    /// `required_policies` started from `QoSPolicyMask::CORE` unconditionally,
    /// which made the sentinel profile demand reliability, durability, history
    /// AND depth — so a backend that could not honour one would answer
    /// `IncompatibleQos` to a profile that requested none of them.
    #[test]
    fn system_default_profile_requires_no_policy() {
        let required = QoSProfile::QOS_PROFILE_SYSTEM_DEFAULT.required_policies();
        assert_eq!(required.0, 0, "SYSTEM_DEFAULT demanded {required:?}");
        // Therefore it is admissible against a backend that advertises nothing.
        assert!(
            QoSProfile::QOS_PROFILE_SYSTEM_DEFAULT
                .validate_against(QoSPolicyMask(0))
                .is_ok()
        );
    }

    /// phase-428 W9 — the roadmap predicted that deriving zenoh's mask would
    /// drop `DEPTH` and that "nothing that works today breaks". The second
    /// half is FALSE, and this is the measurement that says so.
    ///
    /// `QOS_PROFILE_DEFAULT` states `depth: 10`, so it REQUIRES `DEPTH`; a
    /// mask without that bit refuses it, and that profile is what every
    /// `create_publisher` / `create_subscription` with no explicit QoS passes
    /// (`nros-node/src/executor/node.rs:238,372`). Withdrawing the bit would
    /// therefore refuse every default entity on the backend, not just the ones
    /// asking for more depth than the ring holds.
    ///
    /// So the zenoh shim EARNS the bit instead of withdrawing it (phase-428
    /// W9: `shim/qos.rs` reads the depth, grants what the ring can hold, and
    /// publishes the GRANTED value to the graph). Which is the honest reading
    /// of the mask anyway: the bit says "I will not silently ignore this
    /// policy", and a depth lie lives in the VALUE, not in the bit.
    #[test]
    fn withdrawing_depth_would_refuse_the_default_profile() {
        let without_depth = QoSPolicyMask(!QoSPolicyMask::DEPTH.0);
        assert!(
            QoSProfile::QOS_PROFILE_DEFAULT
                .validate_against(without_depth)
                .is_err(),
            "QOS_PROFILE_DEFAULT states depth 10; a mask without DEPTH must refuse it"
        );
        // And the same for the two other presets a default caller reaches.
        for qos in [
            QoSProfile::QOS_PROFILE_SERVICES_DEFAULT,
            QoSProfile::QOS_PROFILE_SENSOR_DATA,
        ] {
            assert!(qos.validate_against(without_depth).is_err());
        }
        // The sentinel profile is the one that survives, because it states no
        // depth at all.
        assert!(
            QoSProfile::QOS_PROFILE_SYSTEM_DEFAULT
                .validate_against(without_depth)
                .is_ok()
        );
    }

    /// phase-428 W9 — the trait default advertises NOTHING, so an
    /// implementation that has written no QoS code refuses any stated policy.
    ///
    /// The default was `CORE` until W9: four policies granted to a backend
    /// that had implemented none of them, by the trait itself.
    #[test]
    fn the_empty_mask_refuses_every_stated_policy_and_nothing_else() {
        assert_eq!(QoSPolicyMask::NONE.0, 0);
        assert!(
            QoSProfile::QOS_PROFILE_DEFAULT
                .validate_against(QoSPolicyMask::NONE)
                .is_err()
        );
        assert!(
            QoSProfile::QOS_PROFILE_SYSTEM_DEFAULT
                .validate_against(QoSPolicyMask::NONE)
                .is_ok(),
            "an absence demands nothing, so it is admissible anywhere"
        );
    }

    // ------------------------------------------------------------------
    // phase-417 G7 — the refusal NAMES the policy, and the vocabulary it
    // names it in is `QoSPolicyMask`'s own constants.
    // ------------------------------------------------------------------

    /// The table is the twelve bits and nothing else, and a name is never a
    /// second authoring of the constant it belongs to.
    ///
    /// The buildless half of this lives in `check-qos-mask-derivation` (R8),
    /// which reads the `pub const` block and compares it against `NAMED`
    /// entry for entry. This is the half a `no_std` unit run can assert: every
    /// entry is one bit, the bits are distinct, and every name is distinct.
    /// A bit added above with no entry here would leave a policy that refuses
    /// a create and prints nothing.
    #[test]
    fn every_named_policy_is_one_distinct_bit_with_one_distinct_name() {
        assert_eq!(QoSPolicyMask::NAMED.len(), 12);
        let mut seen_bits = 0u32;
        for (bit, cstr) in QoSPolicyMask::NAMED {
            let name = cstr.to_str().expect("every entry is an ASCII literal");
            assert_eq!(
                bit.0.count_ones(),
                1,
                "{name} is not a single policy bit ({:#x})",
                bit.0
            );
            assert_eq!(seen_bits & bit.0, 0, "{name} repeats a bit already named");
            seen_bits |= bit.0;
            assert_eq!(
                bit.policy_name(),
                Some(name),
                "{name} does not name itself back"
            );
            // The C entry points hand this pointer straight over, so the NUL
            // is load-bearing and the byte before it must be the last of the
            // name — a `&str` view that silently dropped a byte would be a
            // truncated C string.
            assert_eq!(cstr.to_bytes().len(), name.len());
        }
        // The twelve bits are contiguous from 0, which is what
        // `rmw_entity.h`'s `NROS_RMW_QOS_POLICY_*` mirrors.
        assert_eq!(seen_bits, (1 << 12) - 1);
    }

    /// `None` is a real answer in both directions — neither the empty mask nor
    /// a union is a policy, so neither may borrow a neighbour's name.
    #[test]
    fn neither_the_empty_mask_nor_a_union_names_a_policy() {
        assert_eq!(QoSPolicyMask::NONE.policy_name(), None);
        assert_eq!(QoSPolicyMask::CORE.policy_name(), None);
        assert_eq!(
            (QoSPolicyMask::DEADLINE | QoSPolicyMask::LIFESPAN).policy_name(),
            None
        );
        // And an error that is not a QoS refusal names no policy either.
        assert_eq!(TransportError::Timeout.qos_policy_name(), None);
        assert_eq!(
            TransportError::IncompatibleQos(QoSPolicyMask::NONE).qos_policy_name(),
            None,
            "a refusal relayed across `rmw_ret_t` identified no policy"
        );
    }

    /// Several missing policies still produce ONE name, the same one every
    /// time — otherwise two surfaces could report different policies for the
    /// same refusal and a test could not assert either.
    #[test]
    fn a_refusal_missing_several_policies_names_the_lowest_one() {
        // DEFAULT states reliability, durability(volatile), history and depth.
        let qos = QoSProfile::QOS_PROFILE_DEFAULT;
        assert_eq!(
            qos.validate_against(QoSPolicyMask::NONE),
            Err(TransportError::IncompatibleQos(QoSPolicyMask::RELIABILITY)),
            "RELIABILITY is bit 0, so it is the reported one"
        );
        // Grant reliability and the next-lowest missing policy is reported.
        assert_eq!(
            qos.validate_against(QoSPolicyMask::RELIABILITY)
                .unwrap_err()
                .qos_policy_name(),
            Some("DURABILITY_VOLATILE")
        );
        // Grant everything the profile states and it is admitted.
        assert!(qos.validate_against(qos.required_policies()).is_ok());
    }

    /// The policy a caller stated is the policy the refusal names — not a
    /// policy the profile never asked for. `DEADLINE` is the interesting one
    /// because it is stated by a VALUE rather than by a variant.
    #[test]
    fn the_refusal_names_the_policy_the_caller_actually_stated() {
        let qos = QoSProfile {
            deadline_ms: 250,
            ..QoSProfile::QOS_PROFILE_SYSTEM_DEFAULT
        };
        assert_eq!(qos.required_policies(), QoSPolicyMask::DEADLINE);
        assert_eq!(
            qos.validate_against(QoSPolicyMask::NONE)
                .unwrap_err()
                .qos_policy_name(),
            Some("DEADLINE")
        );
    }

    /// The relaxation is per FIELD, not all-or-nothing: a profile that states
    /// SOME policies still demands exactly those.
    #[test]
    fn a_partly_stated_profile_demands_only_what_it_states() {
        let qos = QoSProfile {
            reliability: QoSReliabilityPolicy::BestEffort,
            ..QoSProfile::QOS_PROFILE_SYSTEM_DEFAULT
        };
        let required = qos.required_policies();
        assert!(required.contains(QoSPolicyMask::RELIABILITY));
        assert!(!required.contains(QoSPolicyMask::HISTORY));
        assert!(!required.contains(QoSPolicyMask::DEPTH));
        assert!(!required.contains(QoSPolicyMask::DURABILITY_VOLATILE));
        assert!(!required.contains(QoSPolicyMask::DURABILITY_TRANSIENT_LOCAL));
    }

    /// A fully concrete profile is unaffected by the relaxation — this is the
    /// guard on "do not silently stop demanding what the caller asked for".
    #[test]
    fn concrete_profiles_still_demand_their_core_policies() {
        let required = QoSProfile::QOS_PROFILE_DEFAULT.required_policies();
        assert!(required.contains(QoSPolicyMask::RELIABILITY));
        assert!(required.contains(QoSPolicyMask::DURABILITY_VOLATILE));
        assert!(required.contains(QoSPolicyMask::HISTORY));
        assert!(required.contains(QoSPolicyMask::DEPTH));
        // And a backend missing one still rejects it.
        let missing = QoSPolicyMask(required.0 & !QoSPolicyMask::DEPTH.0);
        // phase-417 G7 — and it says WHICH policy. `DEPTH` is the only bit
        // dropped, so there is exactly one right answer here.
        assert_eq!(
            QoSProfile::QOS_PROFILE_DEFAULT.validate_against(missing),
            Err(TransportError::IncompatibleQos(QoSPolicyMask::DEPTH))
        );
    }

    /// issue 0829 — resolution replaces ONLY the sentinel fields, and the
    /// answer is the backend's, not a constant this crate bakes.
    #[test]
    fn resolve_system_default_fills_absences_and_touches_nothing_else() {
        // Two backends, two different answers to the SAME sentinel — the whole
        // reason no concrete `QOS_PROFILE_SYSTEM_DEFAULT` could be right.
        const CYCLONE: QoSSystemDefaults = QoSSystemDefaults {
            reliability: QoSReliabilityPolicy::Reliable,
            durability: QoSDurabilityPolicy::Volatile,
            history: QoSHistoryPolicy::KeepLast,
            depth: 1,
        };
        const ZENOH: QoSSystemDefaults = QoSSystemDefaults {
            depth: 4,
            ..CYCLONE
        };

        let a = QoSProfile::QOS_PROFILE_SYSTEM_DEFAULT.resolve_system_default(&CYCLONE);
        let b = QoSProfile::QOS_PROFILE_SYSTEM_DEFAULT.resolve_system_default(&ZENOH);
        assert_eq!(a.reliability, QoSReliabilityPolicy::Reliable);
        assert_eq!(a.history, QoSHistoryPolicy::KeepLast);
        assert_eq!(a.durability, QoSDurabilityPolicy::Volatile);
        assert_eq!(a.depth, 1);
        assert_eq!(b.depth, 4);
        assert!(!a.has_unresolved_system_default());
        assert!(!b.has_unresolved_system_default());

        // A STATED policy is never overridden — resolution fills an absence.
        let stated = QoSProfile {
            reliability: QoSReliabilityPolicy::BestEffort,
            history: QoSHistoryPolicy::KeepAll,
            durability: QoSDurabilityPolicy::TransientLocal,
            depth: 7,
            ..QoSProfile::QOS_PROFILE_SYSTEM_DEFAULT
        };
        let resolved = stated.resolve_system_default(&CYCLONE);
        assert_eq!(resolved, stated, "resolution overrode a stated policy");

        // Idempotent.
        assert_eq!(a.resolve_system_default(&ZENOH), a);
    }

    #[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 = QoSProfile::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);
        // phase-417 G7 — RELIABILITY is the bit dropped and RELIABILITY is
        // what the refusal has to name.
        assert_eq!(
            qos.validate_against(missing),
            Err(TransportError::IncompatibleQos(QoSPolicyMask::RELIABILITY))
        );
        assert_eq!(
            qos.validate_against(missing).unwrap_err().qos_policy_name(),
            Some("RELIABILITY")
        );
    }

    /// issue 0793 — this asserted `TransientLocal`, which PINNED THE DEFECT:
    /// upstream `rmw_qos_profile_parameters` is KEEP_LAST(1000) + RELIABLE +
    /// **VOLATILE** (`/opt/ros/<distro>/include/rmw/rmw/qos_profiles.h`), and
    /// our own second copy of the profile, `nros::qos::PARAMETERS`, was already
    /// correct. The test agreed with the wrong copy, so the disagreement between
    /// the two survived every run.
    #[test]
    fn test_qos_profile_parameters() {
        let qos = QoSProfile::QOS_PROFILE_PARAMETERS;
        assert_eq!(qos.reliability, QoSReliabilityPolicy::Reliable);
        assert_eq!(
            qos.durability,
            QoSDurabilityPolicy::Volatile,
            "rmw_qos_profile_parameters is VOLATILE; transient-local would make \
             every parameter server retain for late joiners, which ROS 2 does not do"
        );
        assert_eq!(qos.depth, 1000);
    }

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

    /// phase-428 W10 — this asserted `KeepAll`, which PINNED THE DEFECT, in
    /// exactly the shape issue 0793 had already been caught in one profile
    /// over: upstream `rmw_qos_profile_parameter_events` is KEEP_LAST(1000) +
    /// RELIABLE + VOLATILE, byte-identical to `rmw_qos_profile_parameters`, and
    /// our C++ surface (`rclcpp::ParameterEventsQoS`) was already correct — so
    /// the two languages shipped different profiles under one name while this
    /// test reported green.
    ///
    /// The old value inverted behaviour rather than weakening it: a RELIABLE
    /// writer with KEEP_ALL blocks once history fills, where KEEP_LAST(1000)
    /// overwrites.
    #[test]
    fn test_qos_profile_parameter_events() {
        let qos = QoSProfile::QOS_PROFILE_PARAMETER_EVENTS;
        assert_eq!(qos.reliability, QoSReliabilityPolicy::Reliable);
        assert_eq!(qos.durability, QoSDurabilityPolicy::Volatile);
        assert_eq!(
            qos.history,
            QoSHistoryPolicy::KeepLast,
            "rmw_qos_profile_parameter_events is KEEP_LAST; KEEP_ALL makes a \
             reliable parameter-event writer BLOCK where upstream overwrites"
        );
        assert_eq!(qos.depth, 1000);

        // Upstream defines the two constants with identical field values.
        assert_eq!(qos, QoSProfile::QOS_PROFILE_PARAMETERS);
    }

    /// phase-428 W10 — the class behind the `PARAMETER_EVENTS` defect, not just
    /// the instance. `KEEP_ALL` under RELIABLE is a blocking configuration, and
    /// no upstream `rmw_qos_profile_*` nor `rcl_action_qos_profile_*` uses it;
    /// every one of them is `KEEP_LAST` or the SYSTEM_DEFAULT sentinel. A NAMED
    /// preset that is `KeepAll` is therefore a transcription error by
    /// construction — a caller who genuinely wants it says
    /// [`QoSProfile::keep_all`].
    #[test]
    fn no_named_preset_is_keep_all() {
        for (name, qos) in [
            ("BEST_EFFORT", QoSProfile::BEST_EFFORT),
            ("RELIABLE", QoSProfile::RELIABLE),
            ("SYSTEM_DEFAULT", QoSProfile::QOS_PROFILE_SYSTEM_DEFAULT),
            ("DEFAULT", QoSProfile::QOS_PROFILE_DEFAULT),
            ("SENSOR_DATA", QoSProfile::QOS_PROFILE_SENSOR_DATA),
            ("SERVICES_DEFAULT", QoSProfile::QOS_PROFILE_SERVICES_DEFAULT),
            ("PARAMETERS", QoSProfile::QOS_PROFILE_PARAMETERS),
            ("PARAMETER_EVENTS", QoSProfile::QOS_PROFILE_PARAMETER_EVENTS),
            (
                "ACTION_STATUS_DEFAULT",
                QoSProfile::QOS_PROFILE_ACTION_STATUS_DEFAULT,
            ),
            ("CLOCK", QoSProfile::QOS_PROFILE_CLOCK),
            ("PX4", QoSProfile::QOS_PROFILE_PX4),
        ] {
            assert_ne!(
                qos.history,
                QoSHistoryPolicy::KeepAll,
                "preset {name} is KEEP_ALL; no upstream profile is"
            );
        }
    }

    #[test]
    fn test_qos_profile_action_status() {
        let qos = QoSProfile::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!(
            QoSProfile::topics_default(),
            QoSProfile::QOS_PROFILE_DEFAULT
        );
        assert_eq!(
            QoSProfile::sensor_data_default(),
            QoSProfile::QOS_PROFILE_SENSOR_DATA
        );
        assert_eq!(
            QoSProfile::services_default(),
            QoSProfile::QOS_PROFILE_SERVICES_DEFAULT
        );
        assert_eq!(
            QoSProfile::parameter_services_default(),
            QoSProfile::QOS_PROFILE_PARAMETERS
        );
        assert_eq!(
            QoSProfile::action_status_default(),
            QoSProfile::QOS_PROFILE_ACTION_STATUS_DEFAULT
        );
    }

    #[test]
    fn test_qos_builder_explicit_setters() {
        let qos = QoSProfile::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 = QoSProfile::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 QoSProfile
        let qos1 = QoSProfile::QOS_PROFILE_DEFAULT;
        let qos2 = QoSProfile::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 = QoSProfile::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, QoSProfile::QOS_PROFILE_PX4);
        // Depth is tunable via the builder without losing the PX4 policies.
        let deep = QoSProfile::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);
    }
}

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

    /// issue 0896 / phase-402 — the hint must SURVIVE being set on `TopicInfo`.
    ///
    /// This is the seam every path funnels through: the C register writes it
    /// here, `nros-rmw-cffi` reads `topic.rx_buffer_hint` back out when
    /// building `rmw_subscription_options_t`, and a size-classing backend
    /// routes on it. A default that silently stayed 0 is the original defect,
    /// so the round trip is worth an assertion rather than an assumption.
    #[test]
    fn a_hint_set_on_topic_info_is_readable_back() {
        let t = TopicInfo::new("/chatter", "std_msgs/msg/Int32", "").with_rx_buffer_hint(4096);
        assert_eq!(t.rx_buffer_hint, 4096);
    }

    /// Zero is "no opinion", not "zero bytes". The C options struct uses 0 as
    /// its unset sentinel and the executor only calls the setter when the
    /// caller stated something, so the default must stay 0 for the
    /// backend-default path to remain reachable.
    #[test]
    fn the_default_hint_is_zero_meaning_no_opinion() {
        let t = TopicInfo::new("/chatter", "std_msgs/msg/Int32", "");
        assert_eq!(t.rx_buffer_hint, 0);
    }

    /// The builder must not disturb the rest of the descriptor — it is applied
    /// AFTER domain/namespace/node in the C path, so a setter that reset a
    /// field would drop identity that discovery depends on.
    #[test]
    fn setting_the_hint_preserves_the_rest_of_the_descriptor() {
        let make = || {
            TopicInfo::new("/chatter", "std_msgs/msg/Int32", "hash")
                .with_domain(7)
                .with_namespace("/ns")
        };
        let base = make();
        let hinted = make().with_rx_buffer_hint(1234);
        assert_eq!(hinted.name, base.name);
        assert_eq!(hinted.type_name, base.type_name);
        assert_eq!(hinted.type_hash, base.type_hash);
        assert_eq!(hinted.domain_id, base.domain_id);
        assert_eq!(hinted.namespace, base.namespace);
        assert_eq!(hinted.rx_buffer_hint, 1234);
    }
}

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

    /// A backend with no read-back answers an ABSENCE, and the trait default
    /// is where that is decided for every backend that does not override it.
    ///
    /// The tempting defaults are both wrong and both silent: the REQUEST
    /// ("you got what you asked for", which is the claim this accessor exists
    /// to stop being made for free) and `QOS_PROFILE_DEFAULT` ("RELIABLE
    /// KEEP_LAST(10)", a profile nobody chose).
    struct Mute;

    impl Publisher for Mute {
        type Error = TransportError;
        fn publish_raw(&self, _data: &[u8]) -> Result<(), TransportError> {
            Ok(())
        }
        fn buffer_error(&self) -> TransportError {
            TransportError::BufferTooSmall
        }
        fn serialization_error(&self) -> TransportError {
            TransportError::SerializationError
        }
    }

    #[test]
    fn the_default_read_back_is_an_absence_not_the_request() {
        let answered = Publisher::actual_qos(&Mute);
        assert_eq!(answered, QoSProfile::QOS_PROFILE_UNKNOWN);
        assert!(answered.has_unknown());
        assert_ne!(
            answered,
            QoSProfile::QOS_PROFILE_DEFAULT,
            "a backend that cannot say must not answer a profile nobody chose"
        );
    }

    /// The phase-467 RMW gap-closure design study's Q1 — the trait default is
    /// a REFUSAL, never a gid.
    ///
    /// `Mute` overrides nothing, so it stands for every backend that has not
    /// implemented the accessor. The tempting default is `Ok([0u8; 24])`: it
    /// compiles, it needs no error plumbing, and it is indistinguishable from
    /// a real gid a backend zero-padded — which is the RFC-0089 Part I shape
    /// this method exists on the loud side of.
    #[test]
    fn the_default_gid_is_a_refusal_not_zeros() {
        let answered = Publisher::get_gid(&Mute);
        assert_eq!(answered, Err(TransportError::Unsupported));
        assert_ne!(answered, Ok([0u8; nros_core::PUBLISHER_GID_SIZE]));
    }

    /// Row 7 of the same study — the ONE spelling of "the application owns
    /// the assertion", which two backends had written out by hand.
    ///
    /// `Unknown` is deliberately not manual: it is what a `*_get_actual_qos`
    /// read-back writes for a policy the backend could not report, and an
    /// absence is not a request. Reading it as manual would make a backend
    /// with no read-back refuse an assertion nobody asked for.
    #[test]
    fn only_the_two_manual_kinds_are_manual() {
        assert!(QoSLivelinessPolicy::ManualByNode.is_manual());
        assert!(QoSLivelinessPolicy::ManualByTopic.is_manual());
        assert!(!QoSLivelinessPolicy::Automatic.is_manual());
        assert!(!QoSLivelinessPolicy::None.is_manual());
        assert!(!QoSLivelinessPolicy::Unknown.is_manual());
    }

    /// `UNKNOWN` and `SYSTEM_DEFAULT` are DIFFERENT absences and the ABI keeps
    /// them apart: "nobody asked" is resolvable by the backend, "I cannot say"
    /// is not. A profile that confused them would let a read-back round-trip
    /// into a create call and silently pick policies.
    #[test]
    fn unknown_is_not_the_system_default_sentinel() {
        assert_ne!(
            QoSProfile::QOS_PROFILE_UNKNOWN,
            QoSProfile::QOS_PROFILE_SYSTEM_DEFAULT
        );
        assert!(QoSProfile::QOS_PROFILE_UNKNOWN.has_unknown());
        assert!(!QoSProfile::QOS_PROFILE_SYSTEM_DEFAULT.has_unknown());
    }

    /// The round trip is REFUSED. Feeding a read-back straight back into a
    /// create is the one way `UNKNOWN` reaches a request, and it must fail
    /// loudly rather than resolve to whatever the backend picks.
    #[test]
    fn a_read_back_profile_cannot_be_used_as_a_request() {
        // A mask that supports everything, so the refusal cannot be an
        // accident of a narrow backend.
        let everything = QoSPolicyMask(u32::MAX);
        assert_eq!(
            QoSProfile::QOS_PROFILE_UNKNOWN.validate_against(everything),
            Err(TransportError::IncompatibleQos(QoSPolicyMask::NONE)),
            "UNKNOWN is an absence; nothing can grant it, so no policy is named"
        );
        // And exactly one unknown policy is enough — a partial read-back is
        // the common case, not the exotic one.
        let mut partial = QoSProfile::QOS_PROFILE_DEFAULT;
        partial.durability = QoSDurabilityPolicy::Unknown;
        assert_eq!(
            partial.validate_against(everything),
            Err(TransportError::IncompatibleQos(QoSPolicyMask::NONE))
        );
        // The same profile without the sentinel is admitted, so the refusal
        // above is about `UNKNOWN` and not about the mask.
        assert_eq!(
            QoSProfile::QOS_PROFILE_DEFAULT.validate_against(everything),
            Ok(())
        );
    }

    /// `required_policies` cannot express `UNKNOWN`, which is why the refusal
    /// lives one level up. Asserted so that a future attempt to smuggle it in
    /// as an unsatisfiable mask bit has to change a test that says why not.
    #[test]
    fn unknown_demands_no_capability_bit() {
        let mut p = QoSProfile::QOS_PROFILE_SYSTEM_DEFAULT;
        p.durability = QoSDurabilityPolicy::Unknown;
        p.liveliness_kind = QoSLivelinessPolicy::Unknown;
        assert_eq!(p.required_policies().0, QoSPolicyMask(0).0);
    }
}

#[cfg(test)]
mod best_available_tests {
    //! issue 1973 — the ONE best-available rule, asserted against upstream's
    //! `rmw_dds_common` behaviour case by case.
    use super::*;

    const ALL: QoSPolicyMask = QoSPolicyMask(u32::MAX);

    fn peer(
        rel: QoSReliabilityPolicy,
        dur: QoSDurabilityPolicy,
        liv: QoSLivelinessPolicy,
        deadline_ms: u32,
        lease_ms: u32,
    ) -> QoSProfile {
        let mut q = QoSProfile::QOS_PROFILE_DEFAULT;
        q.reliability = rel;
        q.durability = dur;
        q.liveliness_kind = liv;
        q.deadline_ms = deadline_ms;
        q.liveliness_lease_ms = lease_ms;
        q
    }

    fn fold(peers: &[QoSProfile]) -> BestAvailablePeers {
        let mut f = BestAvailablePeers::NONE;
        for p in peers {
            f.observe(p);
        }
        f
    }

    #[test]
    fn the_profile_matches_upstream_and_asks_for_resolution() {
        let q = QoSProfile::best_available();
        assert_eq!(q, QoSProfile::QOS_PROFILE_BEST_AVAILABLE);
        assert!(q.has_best_available());
        assert_eq!((q.history, q.depth), (QoSHistoryPolicy::KeepLast, 10));
        assert_eq!(q.lifespan_ms, 0, "lifespan has no best-available spelling");
        assert!(!QoSProfile::QOS_PROFILE_DEFAULT.has_best_available());
    }

    /// A best-available field demands nothing of the backend: the resolver
    /// clamps to the backend's mask, so validation of the REQUEST cannot fail
    /// on it.
    #[test]
    fn a_best_available_request_requires_no_policy_bit() {
        let q = QoSProfile::best_available();
        assert_eq!(
            q.required_policies(),
            QoSPolicyMask::HISTORY | QoSPolicyMask::DEPTH
        );
        assert!(
            q.validate_against(QoSPolicyMask::HISTORY | QoSPolicyMask::DEPTH)
                .is_ok()
        );
    }

    #[test]
    fn subscription_with_no_publishers_gets_upstreams_nothing_discovered_answer() {
        let q = QoSProfile::best_available().resolve_best_available(
            BestAvailableRole::Subscription,
            &BestAvailablePeers::NONE,
            ALL,
        );
        assert_eq!(q.reliability, QoSReliabilityPolicy::BestEffort);
        assert_eq!(q.durability, QoSDurabilityPolicy::Volatile);
        assert_eq!(q.liveliness_kind, QoSLivelinessPolicy::Automatic);
        assert_eq!((q.deadline_ms, q.liveliness_lease_ms), (0, 0));
        assert!(!q.has_best_available());
    }

    #[test]
    fn publisher_with_no_subscriptions_gets_the_highest_service() {
        let q = QoSProfile::best_available().resolve_best_available(
            BestAvailableRole::Publisher,
            &BestAvailablePeers::NONE,
            ALL,
        );
        assert_eq!(q.reliability, QoSReliabilityPolicy::Reliable);
        assert_eq!(q.durability, QoSDurabilityPolicy::TransientLocal);
        assert_eq!(q.liveliness_kind, QoSLivelinessPolicy::Automatic);
        assert_eq!((q.deadline_ms, q.liveliness_lease_ms), (0, 0));
    }

    #[test]
    fn subscription_takes_the_strictest_value_every_publisher_offers() {
        use QoSDurabilityPolicy as D;
        use QoSLivelinessPolicy as L;
        use QoSReliabilityPolicy as R;
        // All reliable + transient-local + manual: the subscription may ask
        // for all three; largest finite deadline / lease.
        let all = fold(&[
            peer(R::Reliable, D::TransientLocal, L::ManualByTopic, 100, 10),
            peer(R::Reliable, D::TransientLocal, L::ManualByTopic, 300, 30),
        ]);
        let q = QoSProfile::best_available().resolve_best_available(
            BestAvailableRole::Subscription,
            &all,
            ALL,
        );
        assert_eq!(q.reliability, R::Reliable);
        assert_eq!(q.durability, D::TransientLocal);
        assert_eq!(q.liveliness_kind, L::ManualByTopic);
        assert_eq!((q.deadline_ms, q.liveliness_lease_ms), (300, 30));

        // One best-effort volatile automatic publisher with no deadline drags
        // each policy down — and an infinite window makes the deadline default.
        let mixed = fold(&[
            peer(R::Reliable, D::TransientLocal, L::ManualByTopic, 100, 0),
            peer(
                R::BestEffort,
                D::Volatile,
                L::Automatic,
                DURATION_INFINITE_MS,
                0,
            ),
        ]);
        let q = QoSProfile::best_available().resolve_best_available(
            BestAvailableRole::Subscription,
            &mixed,
            ALL,
        );
        assert_eq!(q.reliability, R::BestEffort);
        assert_eq!(q.durability, D::Volatile);
        assert_eq!(q.liveliness_kind, L::Automatic);
        assert_eq!(
            q.deadline_ms, 100,
            "infinite reads as no finite window, like upstream's DEFAULT"
        );
    }

    #[test]
    fn publisher_follows_any_manual_subscription_and_the_smallest_window() {
        use QoSDurabilityPolicy as D;
        use QoSLivelinessPolicy as L;
        use QoSReliabilityPolicy as R;
        let subs = fold(&[
            peer(R::BestEffort, D::Volatile, L::Automatic, 300, 0),
            peer(R::BestEffort, D::Volatile, L::ManualByTopic, 100, 40),
        ]);
        let q = QoSProfile::best_available().resolve_best_available(
            BestAvailableRole::Publisher,
            &subs,
            ALL,
        );
        assert_eq!(
            q.reliability,
            R::Reliable,
            "reliable serves every subscription"
        );
        assert_eq!(
            q.durability,
            D::TransientLocal,
            "transient-local serves every subscription"
        );
        assert_eq!(q.liveliness_kind, L::ManualByTopic);
        assert_eq!((q.deadline_ms, q.liveliness_lease_ms), (100, 40));
    }

    /// A peer the backend could not read contributes its least demanding
    /// value — never counted reliable — so a subscription stays compatible.
    #[test]
    fn an_unreadable_peer_counts_as_least_demanding() {
        let unknown = fold(&[QoSProfile::QOS_PROFILE_UNKNOWN]);
        let q = QoSProfile::best_available().resolve_best_available(
            BestAvailableRole::Subscription,
            &unknown,
            ALL,
        );
        assert_eq!(q.reliability, QoSReliabilityPolicy::BestEffort);
        assert_eq!(q.durability, QoSDurabilityPolicy::Volatile);
    }

    #[test]
    fn services_take_the_services_default_profile() {
        let q = QoSProfile::best_available().resolve_best_available(
            BestAvailableRole::Service,
            &BestAvailablePeers::NONE,
            ALL,
        );
        let svc = QoSProfile::QOS_PROFILE_SERVICES_DEFAULT;
        assert_eq!(q.reliability, svc.reliability);
        assert_eq!(q.durability, svc.durability);
        assert_eq!(q.liveliness_kind, svc.liveliness_kind);
        assert_eq!((q.deadline_ms, q.liveliness_lease_ms), (0, 0));
    }

    /// The choice never lands on a value the backend would refuse: it steps
    /// down to the next one the mask honours, and to the sentinel when the
    /// policy is not honoured at all.
    #[test]
    fn the_choice_is_clamped_to_what_the_backend_honours() {
        let mask = QoSPolicyMask::RELIABILITY
            | QoSPolicyMask::DURABILITY_VOLATILE
            | QoSPolicyMask::LIVELINESS_AUTOMATIC;
        let subs = fold(&[peer(
            QoSReliabilityPolicy::Reliable,
            QoSDurabilityPolicy::TransientLocal,
            QoSLivelinessPolicy::ManualByTopic,
            100,
            40,
        )]);
        let q = QoSProfile::best_available().resolve_best_available(
            BestAvailableRole::Publisher,
            &subs,
            mask,
        );
        assert_eq!(q.durability, QoSDurabilityPolicy::Volatile);
        assert_eq!(q.liveliness_kind, QoSLivelinessPolicy::Automatic);
        assert_eq!((q.deadline_ms, q.liveliness_lease_ms), (0, 0));
        assert!(
            q.validate_against(mask | QoSPolicyMask::HISTORY | QoSPolicyMask::DEPTH)
                .is_ok()
        );

        let q = QoSProfile::best_available().resolve_best_available(
            BestAvailableRole::Publisher,
            &subs,
            QoSPolicyMask::NONE,
        );
        assert_eq!(q.reliability, QoSReliabilityPolicy::SystemDefault);
        assert_eq!(q.durability, QoSDurabilityPolicy::SystemDefault);
        assert_eq!(q.liveliness_kind, QoSLivelinessPolicy::None);
    }

    /// Fields the caller STATED are never touched — only the best-available
    /// ones resolve.
    #[test]
    fn stated_fields_are_left_alone() {
        let q = QoSProfile::best_available().best_effort();
        let r =
            q.resolve_best_available(BestAvailableRole::Publisher, &BestAvailablePeers::NONE, ALL);
        assert_eq!(r.reliability, QoSReliabilityPolicy::BestEffort);
        assert_eq!(r.durability, QoSDurabilityPolicy::TransientLocal);
        let plain = QoSProfile::QOS_PROFILE_SENSOR_DATA;
        assert_eq!(
            plain.resolve_best_available(
                BestAvailableRole::Subscription,
                &BestAvailablePeers::NONE,
                ALL
            ),
            plain
        );
    }

    /// No finite duration lowers onto the best-available sentinel.
    #[test]
    fn a_finite_duration_cannot_collide_with_the_sentinel() {
        use core::time::Duration;
        assert_eq!(
            duration_to_qos_ms(Duration::from_millis(DURATION_BEST_AVAILABLE_MS as u64)),
            Err(TransportError::InvalidArgument)
        );
        assert_eq!(
            duration_to_qos_ms(Duration::from_millis(DURATION_BEST_AVAILABLE_MS as u64 - 1)),
            Ok(DURATION_BEST_AVAILABLE_MS - 1)
        );
    }
}