nros_node/executor/

node.rs

//! Node — borrows the session to create typed entities.

use core::marker::PhantomData;

use nros_core::{RosAction, RosMessage, RosService};
use nros_rmw::{ActionInfo, QoSProfile, ServiceInfo, Session as _, TopicInfo, TransportError};

use crate::{
    rmw_type_registry::{MessageForRmw, register_type},
    session,
};

use super::{
    handles::{
        ActionClient, ActionClientCallback, ActionServer, EmbeddedPublisher, EmbeddedServiceClient,
        EmbeddedServiceServer, ServiceClientCallback, Subscription,
    },
    types::NodeError,
};

/// Validate a profile against what the backend honours, and SAY WHICH POLICY
/// when it does not — phase-417 G7.
///
/// Six `create_*` paths ran `qos.validate_against(…).map_err(NodeError::Transport)`
/// and that was the whole diagnostic: the C caller saw `RCLC_RET_REJECTED`, the
/// C++ caller `NROS_CPP_RET_NOT_ALLOWED`, and neither could learn WHICH policy
/// the backend declined — while `rmw_vtable.h`'s `supported_qos_policies` block
/// told the reader those creates fail "naming a policy". This is the one place
/// that sentence becomes true for a caller who only gets a return code.
///
/// ONE helper rather than a log line per site, for the reason CLAUDE.md's "fix
/// the CLASS" rule gives: six copies of a diagnostic are six places for the
/// wording, the logger name and the fallback to drift, and this family has
/// already paid that twice (the sizes-header mirror, the Zephyr unset-variable
/// guard).
///
/// The policy NAME comes from [`nros_rmw::QoSPolicyMask::NAMED`] by way of
/// `TransportError::qos_policy_name`, so this crate states no policy vocabulary
/// of its own, and the token a C++ caller reads out of
/// `nros::qos_policy_kind_to_cstr` is the same byte string that appears here.
pub(crate) fn validate_qos_or_report(
    qos: &QoSProfile,
    supported: nros_rmw::QoSPolicyMask,
    kind: &str,
    name: &str,
) -> Result<(), NodeError> {
    match qos.validate_against(supported) {
        Ok(()) => Ok(()),
        Err(err) => {
            nros_log::log_error!(
                nros_log::get_logger("nros_node"),
                "qos: {} '{}' refused — the backend does not honour {}",
                kind,
                name,
                // `None` is a real answer, not a missing one: a refusal
                // relayed from a C backend crosses `rmw_ret_t`, which has no
                // room for a mask, so no policy was identified. Naming a
                // neighbouring one would be the plausible wrong answer.
                err.qos_policy_name()
                    .unwrap_or("a QoS policy it did not name")
            );
            Err(NodeError::Transport(err))
        }
    }
}

// ============================================================================
// Node
// ============================================================================

/// Backend-agnostic node — borrows the session to create typed entities.
pub struct NodeHandle<'a> {
    name: heapless::String<64>,
    namespace: heapless::String<64>,
    session: &'a mut session::ConcreteSession,
    /// The executor's remap rules, both tiers (issue 1971: the launch table in
    /// flash plus the runtime table). Consulted only to SUBSTITUTE a name some
    /// rule matches — see `NodeHandle::remapped`.
    remaps: super::spin::RemapView<'a>,
    domain_id: u32,
    /// Phase 211.H — per-node QoS overrides lowered from the launch
    /// `qos_overrides.<topic>.<role>.<policy>` params and baked into a
    /// `&'static` table by the entry codegen. Folded into each entity's
    /// `QoSProfile` at `create_publisher`/`create_subscription` time
    /// (setup-time, no alloc). Empty (`&[]`) by default → zero cost for
    /// systems without overrides.
    qos_overrides: &'static [nros_rmw::QoSOverride],
    /// RFC-0052 W3b.4 — baked contract-monitor table (mirror of
    /// `qos_overrides`): `create_publisher` attaches the matching
    /// endpoint's counter cell so `publish` can bump it lock-free.
    monitors: &'static [crate::executor::monitor::MonitorSpec],
    /// W3b.5 — baked subscriber age-contract table + epoch clock;
    /// `create_subscription` attaches the matching endpoint's age cell.
    age_monitors: &'static [crate::executor::monitor::AgeMonitorSpec],
    epoch_us_fn: Option<fn() -> u64>,
}

impl<'a> NodeHandle<'a> {
    /// Create a new node (called by Executor::create_node).
    pub(crate) fn new(
        name: heapless::String<64>,
        namespace: heapless::String<64>,
        session: &'a mut session::ConcreteSession,
        remaps: super::spin::RemapView<'a>,
        domain_id: u32,
    ) -> Self {
        Self {
            name,
            namespace,
            session,
            remaps,
            domain_id,
            qos_overrides: &[],
            monitors: &[],
            age_monitors: &[],
            epoch_us_fn: None,
        }
    }

    /// Phase 211.H — install the plan's QoS-override table on this node. Called
    /// by the generated entry BEFORE the component constructs its entities, so
    /// `create_publisher`/`create_subscription` fold the matching overrides in.
    /// The table is `&'static` (codegen bakes it as a `static`), so there is no
    /// lifetime to thread and no runtime allocation. Plan = authority: an
    /// override for a topic the entity creates is applied transparently (the
    /// user's `create_publisher(topic)` call is unchanged, matching rclcpp).
    pub fn set_qos_overrides(&mut self, overrides: &'static [nros_rmw::QoSOverride]) {
        self.qos_overrides = overrides;
    }

    /// RFC-0052 W3b.4 — install the executor's monitor table on this node
    /// (called by the entry glue / fixture alongside `set_qos_overrides`).
    pub fn set_monitors(&mut self, monitors: &'static [crate::executor::monitor::MonitorSpec]) {
        self.monitors = monitors;
    }

    /// W3b.5 — install the subscriber age-contract table + epoch clock
    /// (auto-seeded from the executor's `set_age_table` / config epoch).
    pub fn set_age_monitors(
        &mut self,
        table: &'static [crate::executor::monitor::AgeMonitorSpec],
        epoch_us: Option<fn() -> u64>,
    ) {
        self.age_monitors = table;
        self.epoch_us_fn = epoch_us;
    }

    /// The installed QoS-override table (empty unless the entry set one).
    #[must_use]
    pub fn qos_overrides(&self) -> &'static [nros_rmw::QoSOverride] {
        self.qos_overrides
    }

    /// Get the node name.
    pub fn name(&self) -> &str {
        &self.name
    }

    /// RFC-0088 — the serialization format this node's entities speak.
    ///
    /// The counterpart to ROS 2's `rmw_get_serialization_format()`, asked of
    /// the node rather than the process: an `Executor::open_multi` image holds
    /// two sessions and has no single answer, which is the case the reserved
    /// vtable slot was cut for.
    ///
    /// A single-backend image already knows this at compile time — see
    /// [`crate::session::IMAGE_SERIALIZATION_FORMAT`], which is what the
    /// entity-creation assertions compare against. Use this accessor when the
    /// answer must be a value: a bridge, a diagnostic, a tool.
    ///
    /// **`None` means the backend has not said** — the same discipline the C
    /// entry point `rclc_node_get_serialization_format` has always kept by
    /// returning NULL, and the one `rmw_vtable.h` states for the slot
    /// underneath. It is not a synonym for `"cdr"`.
    #[must_use]
    pub fn serialization_format(&self) -> Option<&'static str> {
        nros_rmw::Session::serialization_format(&*self.session)
    }

    /// The RMW backend behind this node, by name — the counterpart to ROS 2's
    /// `rmw_get_implementation_identifier()` (rclpy's
    /// `get_rmw_implementation_identifier()`), asked of the node because an
    /// `Executor::open_multi` image holds two sessions.
    ///
    /// The nano-ros backend name (`"zenoh"`, `"cyclonedds"`, `"xrce"`,
    /// `"uorb"`), never a stock `rmw_*_cpp` name: a zenoh image speaks
    /// `rmw_zenoh_cpp`'s wire but is not that implementation, and code that
    /// branches on the identifier is asking about the implementation. A
    /// backend reached through the C ABI that states no name answers the name
    /// it was registered under. `None` = the backend has not said.
    #[must_use]
    pub fn rmw_implementation_identifier(&self) -> Option<&'static str> {
        nros_rmw::Session::implementation_identifier(&*self.session)
    }

    /// Phase 88.12 — return the [`nros_log::Logger`] NAMED FOR THIS NODE.
    ///
    /// Loggers are interned in nros-log's bounded global table
    /// ([`nros_log::MAX_LOGGERS`] slots). If the caller has
    /// pre-registered a `'static Logger` whose name matches this
    /// node's name (via [`nros_log::register_logger`]), this method
    /// returns that exact reference — so subsequent `nros_*!` calls
    /// share per-logger runtime threshold state with any other call
    /// site that resolves the same name.
    ///
    /// phase-427 W5 — otherwise one is CREATED under the node's name, in
    /// nros-log's bounded arena. It used to answer
    /// [`nros_log::DEFAULT_LOGGER`] instead, which kept the API total by
    /// making every node in an image share the name `"nros"`: no record could
    /// say which node emitted it, and `set_level` on one node's handle moved
    /// every other node's bar with it. [`nros_log::resolve_logger`] is the one
    /// spelling of this resolution; the catch-all remains the fallback for an
    /// exhausted arena, and it reports itself when it fires.
    ///
    /// ```ignore
    /// // Pre-register if you want a `'static` logger you can also name
    /// // from elsewhere in the image:
    /// static MY_NODE_LOGGER: nros_log::Logger =
    ///     nros_log::Logger::new("my_node");
    /// nros_log::register_logger(&MY_NODE_LOGGER);
    ///
    /// // Inside any node-creating code:
    /// let logger = node.logger();
    /// nros_log::log_info!(logger, "started; domain = {}", node.domain_id());
    /// ```
    #[must_use]
    pub fn logger(&self) -> &'static nros_log::Logger {
        nros_log::resolve_node_logger(self.name())
    }

    /// Get the domain ID.
    pub fn domain_id(&self) -> u32 {
        self.domain_id
    }

    /// Set the domain ID.
    pub fn set_domain_id(&mut self, domain_id: u32) {
        self.domain_id = domain_id;
    }

    /// Get a mutable reference to the underlying session.
    pub fn session_mut(&mut self) -> &mut session::ConcreteSession {
        self.session
    }

    // ------------------------------------------------------------------
    // Routing-info builders (Phase 91.F)
    //
    // Every `create_*` below threads the same node identity (domain_id +
    // name + namespace) into a TopicInfo / ServiceInfo / ActionInfo. The
    // shape repeats verbatim ~12 times across this file. Centralised
    // here so a future change to the routing-info shape (e.g. adding a
    // security context) updates one site instead of twelve, and
    // so the per-`create_*` function bodies focus on the parts that
    // actually differ between them.
    // ------------------------------------------------------------------

    // Associated fns (NOT `&self` methods) so the returned `*Info`
    // value's borrow tracks only the explicit `&str` arguments, not
    // the whole `Node`. A `&self` form would block the immediately-
    // following `self.session.create_*(&info, …)` mut borrow on the
    // `name` / `namespace` reborrow held inside the returned `*Info`,
    // because going through a method call hides the field-disjoint
    // path that lets `&self.name` + `&mut self.session` coexist.
    fn topic_info<'b>(
        domain_id: u32,
        node_name: &'b str,
        namespace: &'b str,
        topic_name: &'b str,
        type_name: &'b str,
        type_hash: &'b str,
    ) -> TopicInfo<'b> {
        TopicInfo::new(topic_name, type_name, type_hash)
            .with_domain(domain_id)
            .with_node_name(node_name)
            .with_namespace(namespace)
    }

    fn service_info<'b>(
        domain_id: u32,
        node_name: &'b str,
        namespace: &'b str,
        service_name: &'b str,
        type_name: &'b str,
        type_hash: &'b str,
    ) -> ServiceInfo<'b> {
        ServiceInfo::new(service_name, type_name, type_hash)
            .with_domain(domain_id)
            .with_node_name(node_name)
            .with_namespace(namespace)
    }

    fn action_info<'b>(
        domain_id: u32,
        action_name: &'b str,
        type_name: &'b str,
        type_hash: &'b str,
    ) -> ActionInfo<'b> {
        // Action root only needs the domain — per-channel ServiceInfo /
        // TopicInfo derived from action_info.{send_goal,cancel_goal,...}_key()
        // carry the full node identity via service_info() / topic_info().
        ActionInfo::new(action_name, type_name, type_hash).with_domain(domain_id)
    }

    /// The name a rule in either tier sends `source` to, or `None` when no
    /// rule matches — in which case the caller passes `source` through
    /// UNCHANGED, exactly as this handle always has.
    ///
    /// Launch rules (`Executor::declare_remap`) are authoritative and
    /// `--ros-args` rules are the fallback, through
    /// [`crate::names::remap_layered`] — the same two tiers
    /// `Executor::resolve_entity_name_for` gives the C and C++ roads and
    /// `ExecutorSink` gives Rust components.
    ///
    /// Substitution-only, deliberately. This handle has never expanded a name
    /// (`~`/relative) before handing it to the backend, and doing so now for
    /// every entity would move wire names nobody asked to move. A matched rule
    /// yields a fully-qualified name, which is what every other road hands the
    /// backend too.
    ///
    /// Before this the handle consulted no remap at all: a rule declared on
    /// the executor reached C, C++ and component entities and silently missed
    /// every entity created through `Executor::create_node`'s handle.
    fn remapped(&self, source: &str) -> Option<crate::names::ResolvedName> {
        crate::names::remap_layered(
            source,
            &self.name,
            &self.namespace,
            self.remaps.launch_rules(&self.name, &self.namespace),
            self.remaps.argv_rules(&self.name, &self.namespace),
        )
    }

    // -- Publishers --

    /// Create a publisher for the given topic.
    pub fn create_publisher<M: MessageForRmw>(
        &mut self,
        topic_name: &str,
    ) -> Result<EmbeddedPublisher<M>, NodeError> {
        self.create_publisher_with_qos::<M>(topic_name, QoSProfile::default())
    }

    /// Create a publisher with custom QoS settings.
    pub fn create_publisher_with_qos<M: MessageForRmw>(
        &mut self,
        topic_name: &str,
        qos: QoSProfile,
    ) -> Result<EmbeddedPublisher<M>, NodeError> {
        // RFC-0088 / phase-421 W1 — the message's declared format must be the
        // one the linked backend speaks. Universal: the const lives on
        // `RosMessage`, which `MessageForRmw` requires under every backend.
        crate::format_check::assert_message_format::<M>();
        // Phase 212.K.7.6.b — under `rmw-cyclonedds`, ensure the runtime
        // type-descriptor exists before the cffi vtable creates the
        // entity. No-op for other RMWs.
        register_type::<M>()?;
        // Phase 211.H — fold any plan qos_overrides for this topic+publisher
        // into the profile (setup-time, no alloc) BEFORE validation, so an
        // override the backend can't honour still errors loudly below.
        let qos = qos.apply_overrides(
            topic_name,
            nros_rmw::QoSOverrideRole::Publisher,
            self.qos_overrides,
        );
        // Phase 108.B — synchronous QoS validation against backend's
        // `supported_qos_policies()` mask. No silent downgrade.
        let remapped = self.remapped(topic_name);
        let topic = Self::topic_info(
            self.domain_id,
            &self.name,
            &self.namespace,
            remapped.as_deref().unwrap_or(topic_name),
            <M as RosMessage>::TYPE_NAME,
            <M as RosMessage>::TYPE_HASH,
        );
        // issue 1608 -- the publisher half of the declared-QoS check. BEFORE
        // validation, so a policy the contract raises this publisher to is
        // one the backend is asked whether it can honour.
        let qos = crate::declared_qos::honour_publisher(topic.type_name, topic.name, qos)?;
        validate_qos_or_report(
            &qos,
            nros_rmw::Session::supported_qos_policies(self.session),
            "publisher",
            topic_name,
        )?;
        let handle = self
            .session
            .create_publisher(&topic, qos)
            .map_err(|_| NodeError::Transport(TransportError::PublisherCreationFailed))?;
        // RFC-0052 W3b.4 — attach the contracted endpoint's counter cell
        // (exact topic-name match against the baked table; None = free).
        let monitor = self
            .monitors
            .iter()
            .find(|m| m.topic == topic_name)
            .map(|m| m.cell);
        Ok(EmbeddedPublisher {
            handle,
            event_regs: crate::executor::handles::empty_event_regs(),
            monitor,
            epoch: self.epoch_us_fn,
            _phantom: PhantomData,
        })
    }

    /// Create a typeless publisher for non-ROS wire formats (e.g. PX4 uORB
    /// raw POD bytes, custom binary protocols). The caller supplies the
    /// `type_name` and `type_hash` strings used by backends that need them
    /// for liveliness/discovery; backends that don't (uORB) can pass any
    /// stable string.
    pub fn create_publisher_raw(
        &mut self,
        topic_name: &str,
        type_name: &str,
        type_hash: &str,
    ) -> Result<crate::executor::handles::EmbeddedRawPublisher, NodeError> {
        self.create_publisher_raw_with_qos(topic_name, type_name, type_hash, QoSProfile::default())
    }

    /// Typeless publisher with custom QoS.
    pub fn create_publisher_raw_with_qos(
        &mut self,
        topic_name: &str,
        type_name: &str,
        type_hash: &str,
        qos: QoSProfile,
    ) -> Result<crate::executor::handles::EmbeddedRawPublisher, NodeError> {
        // Phase 211.H — apply plan qos_overrides (publisher side) before validate.
        let qos = qos.apply_overrides(
            topic_name,
            nros_rmw::QoSOverrideRole::Publisher,
            self.qos_overrides,
        );
        let remapped = self.remapped(topic_name);
        let topic = Self::topic_info(
            self.domain_id,
            &self.name,
            &self.namespace,
            remapped.as_deref().unwrap_or(topic_name),
            type_name,
            type_hash,
        );
        // issue 1608 -- see `create_publisher_with_qos`.
        let qos = crate::declared_qos::honour_publisher(topic.type_name, topic.name, qos)?;
        validate_qos_or_report(
            &qos,
            nros_rmw::Session::supported_qos_policies(self.session),
            "publisher",
            topic_name,
        )?;
        let handle = self
            .session
            .create_publisher(&topic, qos)
            .map_err(|_| NodeError::Transport(TransportError::PublisherCreationFailed))?;
        Ok(crate::executor::handles::EmbeddedRawPublisher {
            handle,
            arena: crate::executor::handles::TxArena::new(),
            event_regs: crate::executor::handles::empty_event_regs(),
            // Issue 1676 — the typed path's rule, verbatim.
            monitor: self
                .monitors
                .iter()
                .find(|m| m.topic == topic_name)
                .map(|m| m.cell),
        })
    }

    /// Phase 189.M1 — the customizable publisher **builder** (the `clone` tier;
    /// see `docs/design/0022-entity-api-tiers.md`). Pick a mode with `.typed::<M>()`
    /// or `.generic(type, hash)`, set knobs (`.qos`), then `.build()`. The
    /// convenient `create_publisher` / `create_publisher_raw` are the `fork`
    /// tier — sugar over this with defaults.
    pub fn publisher<'t>(&mut self, topic: &'t str) -> PublisherBuilder<'_, 'a, 't> {
        PublisherBuilder {
            node: self,
            topic,
            qos: QoSProfile::default(),
        }
    }

    // -- Subscriptions --

    /// Create a subscription for the given topic.
    pub fn create_subscription<M: MessageForRmw>(
        &mut self,
        topic_name: &str,
    ) -> Result<Subscription<M>, NodeError> {
        self.create_subscription_sized::<M, { crate::config::DEFAULT_RX_BUF_SIZE }>(topic_name)
    }

    /// Create a subscription with custom buffer size.
    pub fn create_subscription_sized<M: MessageForRmw, const RX_BUF: usize>(
        &mut self,
        topic_name: &str,
    ) -> Result<Subscription<M, RX_BUF>, NodeError> {
        self.create_subscription_with_qos::<M, RX_BUF>(topic_name, QoSProfile::default())
    }

    /// Create a subscription with custom QoS and buffer size.
    pub fn create_subscription_with_qos<M: MessageForRmw, const RX_BUF: usize>(
        &mut self,
        topic_name: &str,
        qos: QoSProfile,
    ) -> Result<Subscription<M, RX_BUF>, NodeError> {
        // RFC-0088 / phase-421 W1 — the message's declared format must be the
        // one the linked backend speaks. Universal: the const lives on
        // `RosMessage`, which `MessageForRmw` requires under every backend.
        crate::format_check::assert_message_format::<M>();
        // Phase 212.K.7.6.b — see `create_publisher_with_qos`.
        register_type::<M>()?;
        // Phase 211.H — apply plan qos_overrides (subscription side) before validate.
        let qos = qos.apply_overrides(
            topic_name,
            nros_rmw::QoSOverrideRole::Subscription,
            self.qos_overrides,
        );
        validate_qos_or_report(
            &qos,
            nros_rmw::Session::supported_qos_policies(self.session),
            "subscription",
            topic_name,
        )?;
        let remapped = self.remapped(topic_name);
        let topic = Self::topic_info(
            self.domain_id,
            &self.name,
            &self.namespace,
            remapped.as_deref().unwrap_or(topic_name),
            <M as RosMessage>::TYPE_NAME,
            <M as RosMessage>::TYPE_HASH,
        );
        // phase-454 W10/W13 — see the same line in `executor/spin.rs`.
        let qos = crate::declared_qos::honour(topic.type_name, topic.name, qos)?;
        // phase-519 W3b — and the declared arrival rate, likewise.
        let topic = crate::declared_qos::with_declared_arrival_rate(topic);
        let handle = self
            .session
            .create_subscription(&topic, qos)
            .map_err(NodeError::Transport)?;
        // W3b.5 — attach the contracted endpoint's age cell (stamped
        // types only). Issue 2215: with no epoch source the take is still
        // recorded, unaged.
        let age_mon = match <M as RosMessage>::STAMP_OFFSET {
            Some(_) => self
                .age_monitors
                .iter()
                .find(|a| a.topic == topic_name)
                .map(|a| (a.cell, self.epoch_us_fn)),
            None => None,
        };
        Ok(Subscription {
            handle,
            buffer: [0u8; RX_BUF],
            event_regs: crate::executor::handles::empty_event_regs(),
            age_mon,
            _phantom: PhantomData,
        })
    }

    /// Create a typeless subscription. Caller decodes raw bytes themselves.
    pub fn create_subscription_raw(
        &mut self,
        topic_name: &str,
        type_name: &str,
        type_hash: &str,
    ) -> Result<crate::executor::handles::RawSubscription, NodeError> {
        self.create_subscription_raw_sized::<{ crate::config::DEFAULT_RX_BUF_SIZE }>(
            topic_name, type_name, type_hash,
        )
    }

    /// Typeless subscription with custom buffer size.
    pub fn create_subscription_raw_sized<const RX_BUF: usize>(
        &mut self,
        topic_name: &str,
        type_name: &str,
        type_hash: &str,
    ) -> Result<crate::executor::handles::RawSubscription<RX_BUF>, NodeError> {
        // Phase 211.H — apply plan qos_overrides (subscription side) before
        // validate, mirroring `create_publisher_raw_with_qos`. The raw entity
        // paths honour node overrides exactly like the typed ones — an
        // override the active RMW can't meet errors loudly, never silently.
        let qos = QoSProfile::default().apply_overrides(
            topic_name,
            nros_rmw::QoSOverrideRole::Subscription,
            self.qos_overrides,
        );
        validate_qos_or_report(
            &qos,
            nros_rmw::Session::supported_qos_policies(self.session),
            "subscription",
            topic_name,
        )?;
        let remapped = self.remapped(topic_name);
        let topic = Self::topic_info(
            self.domain_id,
            &self.name,
            &self.namespace,
            remapped.as_deref().unwrap_or(topic_name),
            type_name,
            type_hash,
        );
        // phase-454 W10/W13 — the raw path costs the arena exactly what
        // the typed one does, so leaving it out would make the declared depth
        // enforceable everywhere except in the images that decode bytes
        // themselves.
        let qos = crate::declared_qos::honour(topic.type_name, topic.name, qos)?;
        // phase-519 W3b — and the declared arrival rate, likewise.
        let topic = crate::declared_qos::with_declared_arrival_rate(topic);
        let handle = self
            .session
            .create_subscription(&topic, qos)
            .map_err(NodeError::Transport)?;
        Ok(crate::executor::handles::RawSubscription {
            handle,
            buffer: [0u8; RX_BUF],
            event_regs: crate::executor::handles::empty_event_regs(),
        })
    }

    // -- Services --

    /// Create a service server.
    pub fn create_service<Svc: RosService>(
        &mut self,
        service_name: &str,
    ) -> Result<EmbeddedServiceServer<Svc>, NodeError>
    where
        Svc::Request: MessageForRmw,
        Svc::Reply: MessageForRmw,
    {
        self.create_service_sized::<Svc, { crate::config::DEFAULT_RX_BUF_SIZE }, { crate::config::DEFAULT_RX_BUF_SIZE }>(service_name, QoSProfile::services_default())
    }

    /// Phase 193.2b — service server with an explicit QoS profile (applied to
    /// both the request + reply endpoints; rclcpp's `create_service(name, qos)`).
    pub fn create_service_with_qos<Svc: RosService>(
        &mut self,
        service_name: &str,
        qos: QoSProfile,
    ) -> Result<EmbeddedServiceServer<Svc>, NodeError>
    where
        Svc::Request: MessageForRmw,
        Svc::Reply: MessageForRmw,
    {
        self.create_service_sized::<Svc, { crate::config::DEFAULT_RX_BUF_SIZE }, { crate::config::DEFAULT_RX_BUF_SIZE }>(service_name, qos)
    }

    /// Create a service server with custom buffer sizes + QoS.
    pub fn create_service_sized<Svc: RosService, const REQ_BUF: usize, const REPLY_BUF: usize>(
        &mut self,
        service_name: &str,
        qos: QoSProfile,
    ) -> Result<EmbeddedServiceServer<Svc, REQ_BUF, REPLY_BUF>, NodeError>
    where
        Svc::Request: MessageForRmw,
        Svc::Reply: MessageForRmw,
    {
        // Phase 212.K.7.6.b — register both halves of the service round-trip
        // under cyclonedds. No-op for other RMWs.
        register_type::<Svc::Request>()?;
        register_type::<Svc::Reply>()?;
        // Phase 193.5 — validate the service profile against the backend's
        // supported policies (mirrors pub/sub); no silent downgrade. RELIABLE is
        // effectively required for request/reply, so a backend that only honours
        // a fixed profile rejects an incompatible request here.
        validate_qos_or_report(
            &qos,
            nros_rmw::Session::supported_qos_policies(self.session),
            "service",
            service_name,
        )?;
        let remapped = self.remapped(service_name);
        let info = Self::service_info(
            self.domain_id,
            &self.name,
            &self.namespace,
            remapped.as_deref().unwrap_or(service_name),
            Svc::SERVICE_NAME,
            Svc::SERVICE_HASH,
        );
        let handle = self
            .session
            .create_service(&info, qos)
            .map_err(NodeError::Transport)?;
        Ok(EmbeddedServiceServer {
            handle,
            req_buffer: [0u8; REQ_BUF],
            reply_buffer: [0u8; REPLY_BUF],
            _phantom: PhantomData,
        })
    }

    /// Create a service client.
    pub fn create_client<Svc: RosService>(
        &mut self,
        service_name: &str,
    ) -> Result<EmbeddedServiceClient<Svc>, NodeError>
    where
        Svc::Request: MessageForRmw,
        Svc::Reply: MessageForRmw,
    {
        self.create_client_sized::<Svc, { crate::config::DEFAULT_RX_BUF_SIZE }, { crate::config::DEFAULT_RX_BUF_SIZE }>(service_name, QoSProfile::services_default())
    }

    /// Phase 193.2b — service client with an explicit QoS profile.
    pub fn create_client_with_qos<Svc: RosService>(
        &mut self,
        service_name: &str,
        qos: QoSProfile,
    ) -> Result<EmbeddedServiceClient<Svc>, NodeError>
    where
        Svc::Request: MessageForRmw,
        Svc::Reply: MessageForRmw,
    {
        self.create_client_sized::<Svc, { crate::config::DEFAULT_RX_BUF_SIZE }, { crate::config::DEFAULT_RX_BUF_SIZE }>(service_name, qos)
    }

    /// Create a service client with custom buffer sizes + QoS.
    pub fn create_client_sized<Svc: RosService, const REQ_BUF: usize, const REPLY_BUF: usize>(
        &mut self,
        service_name: &str,
        qos: QoSProfile,
    ) -> Result<EmbeddedServiceClient<Svc, REQ_BUF, REPLY_BUF>, NodeError>
    where
        Svc::Request: MessageForRmw,
        Svc::Reply: MessageForRmw,
    {
        // Phase 212.K.7.6.b — see `create_service_sized`.
        register_type::<Svc::Request>()?;
        register_type::<Svc::Reply>()?;
        // Phase 193.5 — validate against the backend's supported policies (no
        // silent downgrade); request/reply effectively requires RELIABLE.
        validate_qos_or_report(
            &qos,
            nros_rmw::Session::supported_qos_policies(self.session),
            "client",
            service_name,
        )?;
        let remapped = self.remapped(service_name);
        let info = Self::service_info(
            self.domain_id,
            &self.name,
            &self.namespace,
            remapped.as_deref().unwrap_or(service_name),
            Svc::SERVICE_NAME,
            Svc::SERVICE_HASH,
        );
        let handle = self
            .session
            .create_client(&info, qos)
            .map_err(|_| NodeError::Transport(TransportError::ServiceClientCreationFailed))?;
        Ok(EmbeddedServiceClient {
            handle,
            req_buffer: [0u8; REQ_BUF],
            reply_buffer: [0u8; REPLY_BUF],
            in_flight: false,
            pending: nros_rmw::RequestCorrelator::new(),
            _phantom: PhantomData,
        })
    }

    /// Typeless service server. L1 counterpart of [`Self::create_service`]
    /// for the C / C++ FFI shims and callers that own their own
    /// scheduler. Returns a [`crate::executor::handles::RawServiceServer`]
    /// which polls request bytes directly.
    pub fn create_service_raw(
        &mut self,
        service_name: &str,
        type_name: &str,
        type_hash: &str,
    ) -> Result<crate::executor::handles::RawServiceServer, NodeError> {
        self.create_service_raw_sized::<
            { crate::config::DEFAULT_RX_BUF_SIZE },
            { crate::config::DEFAULT_RX_BUF_SIZE },
        >(service_name, type_name, type_hash)
    }

    /// Typeless service server with custom buffer sizes.
    pub fn create_service_raw_sized<const REQ_BUF: usize, const RESP_BUF: usize>(
        &mut self,
        service_name: &str,
        type_name: &str,
        type_hash: &str,
    ) -> Result<crate::executor::handles::RawServiceServer<REQ_BUF, RESP_BUF>, NodeError> {
        let remapped = self.remapped(service_name);
        let info = Self::service_info(
            self.domain_id,
            &self.name,
            &self.namespace,
            remapped.as_deref().unwrap_or(service_name),
            type_name,
            type_hash,
        );
        let handle = self
            .session
            .create_service(&info, QoSProfile::services_default())
            .map_err(NodeError::Transport)?;
        Ok(crate::executor::handles::RawServiceServer::new(handle))
    }

    /// Typeless service client. L1 counterpart of [`Self::create_client`].
    pub fn create_client_raw(
        &mut self,
        service_name: &str,
        type_name: &str,
        type_hash: &str,
    ) -> Result<crate::executor::handles::RawServiceClient, NodeError> {
        self.create_client_raw_sized::<
            { crate::config::DEFAULT_RX_BUF_SIZE },
            { crate::config::DEFAULT_RX_BUF_SIZE },
        >(service_name, type_name, type_hash)
    }

    /// Typeless service client with custom buffer sizes.
    pub fn create_client_raw_sized<const REQ_BUF: usize, const REPLY_BUF: usize>(
        &mut self,
        service_name: &str,
        type_name: &str,
        type_hash: &str,
    ) -> Result<crate::executor::handles::RawServiceClient<REQ_BUF, REPLY_BUF>, NodeError> {
        let remapped = self.remapped(service_name);
        let info = Self::service_info(
            self.domain_id,
            &self.name,
            &self.namespace,
            remapped.as_deref().unwrap_or(service_name),
            type_name,
            type_hash,
        );
        let handle = self
            .session
            .create_client(&info, QoSProfile::services_default())
            .map_err(|_| NodeError::Transport(TransportError::ServiceClientCreationFailed))?;
        Ok(crate::executor::handles::RawServiceClient::new(handle))
    }

    // -- Actions --

    /// Phase 122.3.c.6 — typeless action server. Builds the 5
    /// transport channels (`send_goal` / `cancel_goal` / `get_result`
    /// services + `feedback` / `status` publishers) and returns the
    /// raw `ActionServerCore` directly. Caller owns scheduling —
    /// drives `try_recv_goal_request` / `publish_feedback_raw` /
    /// `complete_goal_raw` / `try_handle_cancel` /
    /// `try_handle_get_result_raw` on the returned core.
    pub fn create_action_server_raw(
        &mut self,
        action_name: &str,
        type_name: &str,
        type_hash: &str,
    ) -> Result<
        super::action_core::ActionServerCore<
            { crate::config::DEFAULT_RX_BUF_SIZE },
            { crate::config::DEFAULT_RX_BUF_SIZE },
            { crate::config::DEFAULT_RX_BUF_SIZE },
            4,
        >,
        NodeError,
    > {
        self.create_action_server_raw_sized::<
            { crate::config::DEFAULT_RX_BUF_SIZE },
            { crate::config::DEFAULT_RX_BUF_SIZE },
            { crate::config::DEFAULT_RX_BUF_SIZE },
            4,
        >(action_name, type_name, type_hash)
    }

    /// Typeless action server with custom buffer + goal-slot sizes.
    pub fn create_action_server_raw_sized<
        const GOAL_BUF: usize,
        const RESULT_BUF: usize,
        const FEEDBACK_BUF: usize,
        const MAX_GOALS: usize,
    >(
        &mut self,
        action_name: &str,
        type_name: &str,
        type_hash: &str,
    ) -> Result<
        super::action_core::ActionServerCore<GOAL_BUF, RESULT_BUF, FEEDBACK_BUF, MAX_GOALS>,
        NodeError,
    > {
        let remapped = self.remapped(action_name);
        let action_name = remapped.as_deref().unwrap_or(action_name);
        let action_info = Self::action_info(self.domain_id, action_name, type_name, type_hash);

        // Issue 0454 / phase-354 W3 — per-CHANNEL DDS type, not the bare
        // action type (see `action_channel_type`). These two `_raw_sized`
        // constructors are the last holders of the phase-338 W3 defect: the
        // type name is baked into the keyexpr, so `…Fibonacci_` never matches a
        // peer advertising `…Fibonacci_SendGoal_`.
        let send_goal_type: heapless::String<256> =
            super::action_core::action_channel_type(type_name, "SendGoal");
        let send_goal_keyexpr: heapless::String<256> = action_info.send_goal_key();
        let send_goal_info = Self::service_info(
            self.domain_id,
            &self.name,
            &self.namespace,
            &send_goal_keyexpr,
            &send_goal_type,
            type_hash,
        );
        let send_goal_server = self
            .session
            .create_service(&send_goal_info, QoSProfile::services_default())
            .map_err(|_| NodeError::ActionCreationFailed)?;

        let cancel_goal_keyexpr: heapless::String<256> = action_info.cancel_goal_key();
        let cancel_goal_info = Self::service_info(
            self.domain_id,
            &self.name,
            &self.namespace,
            &cancel_goal_keyexpr,
            "action_msgs::srv::dds_::CancelGoal_",
            type_hash,
        );
        let cancel_goal_server = self
            .session
            .create_service(&cancel_goal_info, QoSProfile::services_default())
            .map_err(|_| NodeError::ActionCreationFailed)?;

        let get_result_type: heapless::String<256> =
            super::action_core::action_channel_type(type_name, "GetResult");
        let get_result_keyexpr: heapless::String<256> = action_info.get_result_key();
        let get_result_info = Self::service_info(
            self.domain_id,
            &self.name,
            &self.namespace,
            &get_result_keyexpr,
            &get_result_type,
            type_hash,
        );
        let get_result_server = self
            .session
            .create_service(&get_result_info, QoSProfile::services_default())
            .map_err(|_| NodeError::ActionCreationFailed)?;

        let feedback_type: heapless::String<256> =
            super::action_core::action_channel_type(type_name, "FeedbackMessage");
        let feedback_keyexpr: heapless::String<256> = action_info.feedback_key();
        let feedback_topic = Self::topic_info(
            self.domain_id,
            &self.name,
            &self.namespace,
            &feedback_keyexpr,
            &feedback_type,
            type_hash,
        );
        let feedback_publisher = self
            .session
            .create_publisher(&feedback_topic, QoSProfile::QOS_PROFILE_DEFAULT)
            .map_err(|_| NodeError::ActionCreationFailed)?;

        let status_keyexpr: heapless::String<256> = action_info.status_key();
        let status_topic = Self::topic_info(
            self.domain_id,
            &self.name,
            &self.namespace,
            &status_keyexpr,
            "action_msgs::msg::dds_::GoalStatusArray_",
            type_hash,
        );
        let status_publisher = self
            .session
            .create_publisher(&status_topic, QoSProfile::QOS_PROFILE_ACTION_STATUS_DEFAULT)
            .map_err(|_| NodeError::ActionCreationFailed)?;

        Ok(super::action_core::ActionServerCore {
            send_goal_server,
            cancel_goal_server,
            get_result_server,
            feedback_publisher,
            status_publisher,
            active_goals: heapless::Vec::new(),
            completed_results: heapless::Vec::new(),
            pending_get_results: heapless::Vec::new(),
            result_slab: [0u8; RESULT_BUF],
            result_slab_used: 0,
            goal_buffer: [0u8; GOAL_BUF],
            feedback_buffer: [0u8; FEEDBACK_BUF],
            cancel_buffer: [0u8; 256],
        })
    }

    /// Phase 122.3.c.6 — typeless action client. Same shape as
    /// `create_action_server_raw` but builds the 3 service clients
    /// + 1 feedback subscriber, returns the raw `ActionClientCore`.
    pub fn create_action_client_raw(
        &mut self,
        action_name: &str,
        type_name: &str,
        type_hash: &str,
    ) -> Result<
        super::action_core::ActionClientCore<
            { crate::config::DEFAULT_RX_BUF_SIZE },
            { crate::config::DEFAULT_RX_BUF_SIZE },
            { crate::config::DEFAULT_RX_BUF_SIZE },
        >,
        NodeError,
    > {
        self.create_action_client_raw_sized::<
            { crate::config::DEFAULT_RX_BUF_SIZE },
            { crate::config::DEFAULT_RX_BUF_SIZE },
            { crate::config::DEFAULT_RX_BUF_SIZE },
        >(action_name, type_name, type_hash)
    }

    /// Typeless action client with custom buffer sizes.
    pub fn create_action_client_raw_sized<
        const GOAL_BUF: usize,
        const RESULT_BUF: usize,
        const FEEDBACK_BUF: usize,
    >(
        &mut self,
        action_name: &str,
        type_name: &str,
        type_hash: &str,
    ) -> Result<super::action_core::ActionClientCore<GOAL_BUF, RESULT_BUF, FEEDBACK_BUF>, NodeError>
    {
        let remapped = self.remapped(action_name);
        let action_name = remapped.as_deref().unwrap_or(action_name);
        let action_info = Self::action_info(self.domain_id, action_name, type_name, type_hash);

        // Issue 0454 / phase-354 W3 — per-CHANNEL DDS type, not the bare
        // action type (see `action_channel_type`). These two `_raw_sized`
        // constructors are the last holders of the phase-338 W3 defect: the
        // type name is baked into the keyexpr, so `…Fibonacci_` never matches a
        // peer advertising `…Fibonacci_SendGoal_`.
        let send_goal_type: heapless::String<256> =
            super::action_core::action_channel_type(type_name, "SendGoal");
        let send_goal_keyexpr: heapless::String<256> = action_info.send_goal_key();
        let send_goal_info = Self::service_info(
            self.domain_id,
            &self.name,
            &self.namespace,
            &send_goal_keyexpr,
            &send_goal_type,
            type_hash,
        );
        let send_goal_client = self
            .session
            .create_client(&send_goal_info, QoSProfile::services_default())
            .map_err(|_| NodeError::ActionCreationFailed)?;

        let cancel_goal_keyexpr: heapless::String<256> = action_info.cancel_goal_key();
        let cancel_goal_info = Self::service_info(
            self.domain_id,
            &self.name,
            &self.namespace,
            &cancel_goal_keyexpr,
            "action_msgs::srv::dds_::CancelGoal_",
            type_hash,
        );
        let cancel_goal_client = self
            .session
            .create_client(&cancel_goal_info, QoSProfile::services_default())
            .map_err(|_| NodeError::ActionCreationFailed)?;

        let get_result_type: heapless::String<256> =
            super::action_core::action_channel_type(type_name, "GetResult");
        let get_result_keyexpr: heapless::String<256> = action_info.get_result_key();
        let get_result_info = Self::service_info(
            self.domain_id,
            &self.name,
            &self.namespace,
            &get_result_keyexpr,
            &get_result_type,
            type_hash,
        );
        let get_result_client = self
            .session
            .create_client(&get_result_info, QoSProfile::services_default())
            .map_err(|_| NodeError::ActionCreationFailed)?;

        let feedback_type: heapless::String<256> =
            super::action_core::action_channel_type(type_name, "FeedbackMessage");
        let feedback_keyexpr: heapless::String<256> = action_info.feedback_key();
        let feedback_topic = Self::topic_info(
            self.domain_id,
            &self.name,
            &self.namespace,
            &feedback_keyexpr,
            &feedback_type,
            type_hash,
        );
        let feedback_subscriber = self
            .session
            .create_subscription(&feedback_topic, QoSProfile::BEST_EFFORT)
            .map_err(|_| NodeError::ActionCreationFailed)?;

        Ok(super::action_core::ActionClientCore::new(
            send_goal_client,
            cancel_goal_client,
            get_result_client,
            feedback_subscriber,
        ))
    }

    /// Create an action server.
    pub fn create_action_server<A: RosAction>(
        &mut self,
        action_name: &str,
    ) -> Result<ActionServer<A>, NodeError>
    where
        A::Goal: MessageForRmw,
        A::Result: MessageForRmw,
        A::Feedback: MessageForRmw,
        A::SendGoalRequest: MessageForRmw,
        A::SendGoalResponse: MessageForRmw,
        A::GetResultRequest: MessageForRmw,
        A::GetResultResponse: MessageForRmw,
        A::FeedbackMessage: MessageForRmw,
    {
        self.create_action_server_sized::<A, { crate::config::DEFAULT_RX_BUF_SIZE }, { crate::config::DEFAULT_RX_BUF_SIZE }, { crate::config::DEFAULT_RX_BUF_SIZE }, 4>(action_name)
    }

    /// Create an action server with custom buffer sizes.
    pub fn create_action_server_sized<
        A: RosAction,
        const GOAL_BUF: usize,
        const RESULT_BUF: usize,
        const FEEDBACK_BUF: usize,
        const MAX_GOALS: usize,
    >(
        &mut self,
        action_name: &str,
    ) -> Result<ActionServer<A, GOAL_BUF, RESULT_BUF, FEEDBACK_BUF, MAX_GOALS>, NodeError>
    where
        A::Goal: MessageForRmw,
        A::Result: MessageForRmw,
        A::Feedback: MessageForRmw,
        A::SendGoalRequest: MessageForRmw,
        A::SendGoalResponse: MessageForRmw,
        A::GetResultRequest: MessageForRmw,
        A::GetResultResponse: MessageForRmw,
        A::FeedbackMessage: MessageForRmw,
    {
        // Phase 212.K.7.6.b + K.7.7.c — register the three user-facing
        // message types AND the five action-protocol envelope types under
        // cyclonedds. No-op for other RMWs. The envelopes are needed
        // because the action service shapes (`*_SendGoal_Request`,
        // `*_GetResult_Response`, …) are the actual on-wire CDR types,
        // and the C++ Cyclone bridge auto-prepends a cdds_request_header_t
        // for any TYPE_NAME ending `_Request`/`_Response`/`_Reply`.
        register_type::<A::Goal>()?;
        register_type::<A::Result>()?;
        register_type::<A::Feedback>()?;
        register_type::<A::SendGoalRequest>()?;
        register_type::<A::SendGoalResponse>()?;
        register_type::<A::GetResultRequest>()?;
        register_type::<A::GetResultResponse>()?;
        register_type::<A::FeedbackMessage>()?;
        // issue #234 — also register the fixed `action_msgs` protocol types
        // (`CancelGoal_{Request,Response}`, `GoalStatusArray`) the cancel service
        // + status publisher created below serialize. They are not `RosAction`
        // associated types (they live in `action_msgs`, which `nros-core` cannot
        // name), so the generated `impl RosAction::register_protocol_types` — which
        // routes them through the generic `nros_rmw::register_type_descriptor` seam —
        // registers them. Without this the cancel_goal service + status publisher
        // have no Cyclone descriptor → `ActionCreationFailed`. The callback executor
        // path (`executor/action.rs`) already did this; this node.rs path — the one
        // `create_action_server` materialises through — did not.
        A::register_protocol_types().map_err(|()| NodeError::ActionCreationFailed)?;
        let remapped = self.remapped(action_name);
        let action_name = remapped.as_deref().unwrap_or(action_name);
        let action_info =
            Self::action_info(self.domain_id, action_name, A::ACTION_NAME, A::ACTION_HASH);

        // Each underlying ServiceInfo / TopicInfo also carries the
        // node identity so the Zenoh shim declares a liveliness token
        // for it. Without `with_node_name` the shim's
        // `declare_entity_liveliness` short-circuits (`node_name.and_then`
        // → None) and `wait_for_action_server` has nothing to find.
        // Advertise the per-channel service / topic types ROS 2 matches on
        // (`<Action>_SendGoal` / `<Action>_GetResult` / `<Action>_FeedbackMessage`),
        // not the bare action type — see `action_core::action_service_base_type`.
        let send_goal_type = super::action_core::action_service_base_type(
            <A::SendGoalRequest as RosMessage>::TYPE_NAME,
            A::ACTION_NAME,
        );
        let get_result_type = super::action_core::action_service_base_type(
            <A::GetResultRequest as RosMessage>::TYPE_NAME,
            A::ACTION_NAME,
        );
        let feedback_type = <A::FeedbackMessage as RosMessage>::TYPE_NAME;

        let send_goal_keyexpr: heapless::String<256> = action_info.send_goal_key();
        let send_goal_info = Self::service_info(
            self.domain_id,
            &self.name,
            &self.namespace,
            &send_goal_keyexpr,
            send_goal_type,
            A::SEND_GOAL_SERVICE_HASH,
        );
        let send_goal_server = self
            .session
            .create_service(&send_goal_info, QoSProfile::services_default())
            .map_err(|_| NodeError::ActionCreationFailed)?;

        let cancel_goal_keyexpr: heapless::String<256> = action_info.cancel_goal_key();
        let cancel_goal_info = Self::service_info(
            self.domain_id,
            &self.name,
            &self.namespace,
            &cancel_goal_keyexpr,
            "action_msgs::srv::dds_::CancelGoal_",
            A::CANCEL_GOAL_SERVICE_HASH,
        );
        let cancel_goal_server = self
            .session
            .create_service(&cancel_goal_info, QoSProfile::services_default())
            .map_err(|_| NodeError::ActionCreationFailed)?;

        let get_result_keyexpr: heapless::String<256> = action_info.get_result_key();
        let get_result_info = Self::service_info(
            self.domain_id,
            &self.name,
            &self.namespace,
            &get_result_keyexpr,
            get_result_type,
            A::GET_RESULT_SERVICE_HASH,
        );
        let get_result_server = self
            .session
            .create_service(&get_result_info, QoSProfile::services_default())
            .map_err(|_| NodeError::ActionCreationFailed)?;

        let feedback_keyexpr: heapless::String<256> = action_info.feedback_key();
        let feedback_topic = Self::topic_info(
            self.domain_id,
            &self.name,
            &self.namespace,
            &feedback_keyexpr,
            feedback_type,
            <A::FeedbackMessage as RosMessage>::TYPE_HASH,
        );
        let feedback_publisher = self
            .session
            .create_publisher(&feedback_topic, QoSProfile::QOS_PROFILE_DEFAULT)
            .map_err(|_| NodeError::ActionCreationFailed)?;

        let status_keyexpr: heapless::String<256> = action_info.status_key();
        let status_topic = Self::topic_info(
            self.domain_id,
            &self.name,
            &self.namespace,
            &status_keyexpr,
            "action_msgs::msg::dds_::GoalStatusArray_",
            A::STATUS_MESSAGE_HASH,
        );
        let status_publisher = self
            .session
            .create_publisher(&status_topic, QoSProfile::QOS_PROFILE_ACTION_STATUS_DEFAULT)
            .map_err(|_| NodeError::ActionCreationFailed)?;

        Ok(ActionServer {
            core: super::action_core::ActionServerCore {
                send_goal_server,
                cancel_goal_server,
                get_result_server,
                feedback_publisher,
                status_publisher,
                active_goals: heapless::Vec::new(),
                completed_results: heapless::Vec::new(),
                pending_get_results: heapless::Vec::new(),
                result_slab: [0u8; RESULT_BUF],
                result_slab_used: 0,
                goal_buffer: [0u8; GOAL_BUF],
                feedback_buffer: [0u8; FEEDBACK_BUF],
                cancel_buffer: [0u8; 256],
            },
            typed_goals: heapless::Vec::new(),
            completed_goals: heapless::Vec::new(),
        })
    }

    /// Create an action client.
    pub fn create_action_client<A: RosAction>(
        &mut self,
        action_name: &str,
    ) -> Result<ActionClient<A>, NodeError>
    where
        A::Goal: MessageForRmw,
        A::Result: MessageForRmw,
        A::Feedback: MessageForRmw,
        A::SendGoalRequest: MessageForRmw,
        A::SendGoalResponse: MessageForRmw,
        A::GetResultRequest: MessageForRmw,
        A::GetResultResponse: MessageForRmw,
        A::FeedbackMessage: MessageForRmw,
    {
        self.create_action_client_sized::<A, { crate::config::DEFAULT_RX_BUF_SIZE }, { crate::config::DEFAULT_RX_BUF_SIZE }, { crate::config::DEFAULT_RX_BUF_SIZE }>(action_name)
    }

    /// Create an action client with custom buffer sizes.
    pub fn create_action_client_sized<
        A: RosAction,
        const GOAL_BUF: usize,
        const RESULT_BUF: usize,
        const FEEDBACK_BUF: usize,
    >(
        &mut self,
        action_name: &str,
    ) -> Result<ActionClient<A, GOAL_BUF, RESULT_BUF, FEEDBACK_BUF>, NodeError>
    where
        A::Goal: MessageForRmw,
        A::Result: MessageForRmw,
        A::Feedback: MessageForRmw,
        A::SendGoalRequest: MessageForRmw,
        A::SendGoalResponse: MessageForRmw,
        A::GetResultRequest: MessageForRmw,
        A::GetResultResponse: MessageForRmw,
        A::FeedbackMessage: MessageForRmw,
    {
        // Phase 212.K.7.6.b + K.7.7.c — see `create_action_server_sized`.
        register_type::<A::Goal>()?;
        register_type::<A::Result>()?;
        register_type::<A::Feedback>()?;
        register_type::<A::SendGoalRequest>()?;
        register_type::<A::SendGoalResponse>()?;
        register_type::<A::GetResultRequest>()?;
        register_type::<A::GetResultResponse>()?;
        register_type::<A::FeedbackMessage>()?;
        // issue #234 — register the `action_msgs` protocol types the cancel-goal
        // service client below serializes (`CancelGoal_{Request,Response}`; the impl
        // also registers `GoalStatusArray`, harmlessly unused client-side) via the
        // generic seam. Without it the cancel_goal client has no Cyclone descriptor
        // → `ActionCreationFailed`. Mirrors the server path.
        A::register_protocol_types().map_err(|()| NodeError::ActionCreationFailed)?;
        let remapped = self.remapped(action_name);
        let action_name = remapped.as_deref().unwrap_or(action_name);
        let action_info =
            Self::action_info(self.domain_id, action_name, A::ACTION_NAME, A::ACTION_HASH);

        // Mirror `create_action_server_sized`: thread node identity through
        // each underlying ServiceInfo / TopicInfo so the Zenoh shim
        // declares the matching client-side liveliness tokens (and so the
        // discovery wildcard built from `send_goal_info` ends up in the
        // same domain as the server's tokens).
        // Same per-channel typing as the server side so the client's requesters
        // and feedback reader match a real ROS 2 action server over DDS.
        let send_goal_type = super::action_core::action_service_base_type(
            <A::SendGoalRequest as RosMessage>::TYPE_NAME,
            A::ACTION_NAME,
        );
        let get_result_type = super::action_core::action_service_base_type(
            <A::GetResultRequest as RosMessage>::TYPE_NAME,
            A::ACTION_NAME,
        );
        let feedback_type = <A::FeedbackMessage as RosMessage>::TYPE_NAME;

        let send_goal_keyexpr: heapless::String<256> = action_info.send_goal_key();
        let send_goal_info = Self::service_info(
            self.domain_id,
            &self.name,
            &self.namespace,
            &send_goal_keyexpr,
            send_goal_type,
            A::SEND_GOAL_SERVICE_HASH,
        );
        let send_goal_client = self
            .session
            .create_client(&send_goal_info, QoSProfile::services_default())
            .map_err(|_| NodeError::ActionCreationFailed)?;

        let cancel_goal_keyexpr: heapless::String<256> = action_info.cancel_goal_key();
        let cancel_goal_info = Self::service_info(
            self.domain_id,
            &self.name,
            &self.namespace,
            &cancel_goal_keyexpr,
            "action_msgs::srv::dds_::CancelGoal_",
            A::CANCEL_GOAL_SERVICE_HASH,
        );
        let cancel_goal_client = self
            .session
            .create_client(&cancel_goal_info, QoSProfile::services_default())
            .map_err(|_| NodeError::ActionCreationFailed)?;

        let get_result_keyexpr: heapless::String<256> = action_info.get_result_key();
        let get_result_info = Self::service_info(
            self.domain_id,
            &self.name,
            &self.namespace,
            &get_result_keyexpr,
            get_result_type,
            A::GET_RESULT_SERVICE_HASH,
        );
        let get_result_client = self
            .session
            .create_client(&get_result_info, QoSProfile::services_default())
            .map_err(|_| NodeError::ActionCreationFailed)?;

        let feedback_keyexpr: heapless::String<256> = action_info.feedback_key();
        let feedback_topic = Self::topic_info(
            self.domain_id,
            &self.name,
            &self.namespace,
            &feedback_keyexpr,
            feedback_type,
            <A::FeedbackMessage as RosMessage>::TYPE_HASH,
        );
        let feedback_subscriber = self
            .session
            .create_subscription(&feedback_topic, QoSProfile::BEST_EFFORT)
            .map_err(|_| NodeError::ActionCreationFailed)?;

        Ok(ActionClient {
            core: super::action_core::ActionClientCore {
                send_goal_client,
                cancel_goal_client,
                get_result_client,
                feedback_subscriber,
                goal_buffer: [0u8; GOAL_BUF],
                result_buffer: [0u8; RESULT_BUF],
                feedback_buffer: [0u8; FEEDBACK_BUF],
                goal_counter: 0,
                in_flight_send_goal: false,
                in_flight_cancel: false,
                in_flight_get_result: false,
                pending_send_goal: super::action_core::ActionRequests::new(),
                pending_cancel: super::action_core::ActionRequests::new(),
                pending_get_result: super::action_core::ActionRequests::new(),
            },
            _phantom: PhantomData,
        })
    }
}

// ===================================================================
// Phase 189.M1 — entity builders (the `clone` tier)
// ===================================================================

/// Publisher builder — `node.publisher(topic)`. Choose `.typed::<M>()` or
/// `.generic(type, hash)`, optionally `.qos(..)`, then `.build()`.
pub struct PublisherBuilder<'n, 'a, 't> {
    node: &'n mut NodeHandle<'a>,
    topic: &'t str,
    qos: QoSProfile,
}

impl<'n, 'a, 't> PublisherBuilder<'n, 'a, 't> {
    /// Set the QoS (also settable on the typed/generic builder).
    pub fn qos(mut self, qos: QoSProfile) -> Self {
        self.qos = qos;
        self
    }

    /// Phase 282 (#145) — mark this publisher "express": its samples bypass
    /// transport tx batching (sent immediately even when the batching knob is
    /// on). For control-tier / latency-sensitive topics.
    pub fn tx_express(mut self, express: bool) -> Self {
        self.qos.tx_express = express;
        self
    }

    /// Typed publisher for a ROS message `M` (mirrors rclcpp/rclrs).
    pub fn typed<M: MessageForRmw>(self) -> TypedPublisherBuilder<'n, 'a, 't, M> {
        TypedPublisherBuilder {
            node: self.node,
            topic: self.topic,
            qos: self.qos,
            _phantom: PhantomData,
        }
    }

    /// Generic (type-erased) publisher — the rclcpp `create_generic_publisher`
    /// form; raw CDR bytes via `publish_raw`.
    pub fn generic(
        self,
        type_name: &'t str,
        type_hash: &'t str,
    ) -> GenericPublisherBuilder<'n, 'a, 't> {
        GenericPublisherBuilder {
            node: self.node,
            topic: self.topic,
            type_name,
            type_hash,
            qos: self.qos,
        }
    }
}

/// Typed publisher builder (`.typed::<M>()`).
pub struct TypedPublisherBuilder<'n, 'a, 't, M> {
    node: &'n mut NodeHandle<'a>,
    topic: &'t str,
    qos: QoSProfile,
    _phantom: PhantomData<M>,
}

impl<'n, 'a, 't, M: MessageForRmw> TypedPublisherBuilder<'n, 'a, 't, M> {
    pub fn qos(mut self, qos: QoSProfile) -> Self {
        self.qos = qos;
        self
    }

    /// Phase 282 (#145) — see [`PublisherBuilder::tx_express`].
    pub fn tx_express(mut self, express: bool) -> Self {
        self.qos.tx_express = express;
        self
    }

    pub fn build(self) -> Result<EmbeddedPublisher<M>, NodeError> {
        self.node
            .create_publisher_with_qos::<M>(self.topic, self.qos)
    }
}

/// Generic (type-erased) publisher builder (`.generic(type, hash)`).
pub struct GenericPublisherBuilder<'n, 'a, 't> {
    node: &'n mut NodeHandle<'a>,
    topic: &'t str,
    type_name: &'t str,
    type_hash: &'t str,
    qos: QoSProfile,
}

impl<'n, 'a, 't> GenericPublisherBuilder<'n, 'a, 't> {
    pub fn qos(mut self, qos: QoSProfile) -> Self {
        self.qos = qos;
        self
    }

    /// Phase 282 (#145) — see [`PublisherBuilder::tx_express`].
    pub fn tx_express(mut self, express: bool) -> Self {
        self.qos.tx_express = express;
        self
    }

    pub fn build(self) -> Result<crate::executor::handles::EmbeddedRawPublisher, NodeError> {
        self.node.create_publisher_raw_with_qos(
            self.topic,
            self.type_name,
            self.type_hash,
            self.qos,
        )
    }
}

// ============================================================================
// Phase 273 (RFC-0047) — CallbackGroup token (rclcpp/rclrs shape)
// ============================================================================

/// A first-class callback group — a **name-only token** (rclcpp/rclrs shape).
///
/// Created via [`Node::create_callback_group`].  Passed to the `_in`
/// entity-create variants (`create_timer_in_group`, `create_subscription_in_group`,
/// `create_publisher_in_group`) to label entities with a group name.
///
/// The group is just a name — the actual `SchedContext` binding is seeded at
/// boot by `Executor::bind_group_sched` (from `system.toml group_tiers`, phase
/// 273 W2) and resolved at entity-registration time by
/// `apply_node_default_sched` (phase 273 W1).  No concurrency type (Mutually-
/// Exclusive vs Reentrant) is stored here — that is RFC-0047 OQ1 follow-up.
pub struct CallbackGroup {
    name: heapless::String<32>,
}

impl CallbackGroup {
    /// The group name (e.g. `"ctrl"`, `"telem"`).
    pub fn name(&self) -> &str {
        &self.name
    }
}

/// A ROS 2 node — rclrs's `Node`, and the ONE node type (phase-483 W2).
///
/// `Executor::create_node(name)` returns it, and so does
/// `Executor::node_mut(id)`; a component and a standalone program hold the
/// same type. It carries rclrs's methods — `create_publisher`,
/// `create_subscription(topic, callback)`, `create_service`, timers, the
/// clock, `name` / `namespace` / `logger`, the graph queries — plus RTOS
/// extensions with names of their own (`create_polling_*`, the `_sized` and
/// `_raw` constructors).
///
/// ADOPT-BOUNDED: a node is a short-lived `&mut Executor` borrow, not rclrs's `Arc<Node>`.
/// Create entities, then drop it before acquiring the next node handle; entity
/// handles (`HandleId`, publishers) are owned and outlive it (no allocator —
/// see `docs/design/0022-entity-api-tiers.md` §Borrow model).
pub struct Node<'e, 's> {
    executor: &'e mut super::spin::Executor<'s>,
    node_id: super::node_record::NodeId,
}

// ============================================================================
// Parameters -- the node is the receiver (phase-515 R10)
// ============================================================================
//
// rclrs keeps parameter verbs on the node; ours were on the `Executor`, as a
// `_on(NodeId, ...)` form plus a bare form meaning the primary node, because
// the store is one table the executor owns (phase-426 W1). The store did not
// move -- these forward to the `_on` forms with this node's id -- but the
// RECEIVER a user writes is the node, as upstream's is, and the executor's
// bare forms are deprecated. Executor-wide verbs (`register_parameter_services`,
// `params`, `install_argv_params`, `flush_parameter_events`) stay there.
#[cfg(feature = "param-store")]
impl<'e, 's> Node<'e, 's> {
    /// rclrs's `Node::declare_parameter::<T>(name)`: a [`ParameterBuilder`]
    /// whose terminal (`mandatory` / `optional` / `read_only`) declares it.
    ///
    /// ADOPT-BOUNDED: returns `Result`, where rclrs's builder cannot fail. The
    /// store is created on first use; `Err(NodeError::NotInitialized)` means the
    /// platform heap refused it (issue 1706), which has to be reported.
    ///
    /// [`ParameterBuilder`]: nros_params::ParameterBuilder
    pub fn declare_parameter<'a, T: nros_params::ParameterVariant>(
        &'a mut self,
        name: &'a str,
    ) -> Result<nros_params::ParameterBuilder<'a, 's, T>, NodeError> {
        self.executor.parameter_on(self.node_id, name)
    }

    /// rclrs's `Node::use_undeclared_parameters()`: read and write any name,
    /// declared or not.
    pub fn use_undeclared_parameters(
        &mut self,
    ) -> Result<nros_params::Parameters<'_, 's>, NodeError> {
        self.executor.use_undeclared_parameters_on(self.node_id)
    }

    /// Declare `name` with `value` and a descriptor -- rclcpp's
    /// `declare_parameter(name, value, descriptor)`. `false` if it exists, the
    /// table is full, or the value is outside the descriptor's range.
    pub fn declare_parameter_with_descriptor(
        &mut self,
        name: &str,
        value: nros_params::ParameterValue,
        descriptor: nros_params::ParameterDescriptor,
    ) -> bool {
        self.executor
            .declare_parameter_with_descriptor_on(self.node_id, name, value, descriptor)
    }

    /// The current value of `name`, if declared.
    pub fn get_parameter(&self, name: &str) -> Option<&nros_params::ParameterValue> {
        self.executor.get_parameter_on(self.node_id, name)
    }

    /// Set `name`, through the same rules a remote set obeys.
    pub fn set_parameter(
        &mut self,
        name: &str,
        value: nros_params::ParameterValue,
    ) -> Result<(), nros_params::ParameterError> {
        self.executor.set_parameter_on(self.node_id, name, value)
    }

    /// Set several parameters all-or-nothing -- rclcpp's
    /// `set_parameters_atomically`.
    pub fn set_parameters_atomically<'x, I>(
        &mut self,
        items: I,
    ) -> Result<(), nros_params::ParameterError>
    where
        I: Iterator<Item = Option<(&'x str, nros_params::ParameterValue)>> + Clone,
    {
        self.executor
            .set_parameters_atomically_on(self.node_id, items)
    }

    /// Remove a declared parameter -- rclcpp's `undeclare_parameter`.
    pub fn undeclare_parameter(&mut self, name: &str) -> bool {
        self.executor.undeclare_parameter_on(self.node_id, name)
    }

    /// Whether a REMOTE set may create a name that was never declared.
    pub fn allow_undeclared_parameters(&mut self, allow: bool) {
        self.executor
            .allow_undeclared_parameters_on(self.node_id, allow)
    }

    /// The descriptor of `name` -- rclcpp's `describe_parameter`.
    pub fn describe_parameter(&self, name: &str) -> Option<&nros_params::ParameterDescriptor> {
        self.executor.describe_parameter_on(self.node_id, name)
    }

    /// The type of `name` -- rclcpp's `get_parameter_type`.
    pub fn get_parameter_type(&self, name: &str) -> Option<nros_params::ParameterType> {
        self.executor.get_parameter_type_on(self.node_id, name)
    }

    /// The names of this node's parameters under `prefix` -- rclcpp's
    /// `list_parameters`, as an iterator rather than an allocated list.
    pub fn list_parameters<'p>(&'p self, prefix: &'p str) -> impl Iterator<Item = &'p str> {
        self.executor.list_parameters_on(self.node_id, prefix)
    }

    /// Set the descriptor's description and constraint text.
    pub fn set_parameter_description(
        &mut self,
        name: &str,
        description: &str,
        additional_constraints: &str,
    ) -> bool {
        self.executor.set_parameter_description_on(
            self.node_id,
            name,
            description,
            additional_constraints,
        )
    }

    /// Make `name` read-only (or writable again).
    pub fn set_parameter_read_only(&mut self, name: &str, read_only: bool) -> bool {
        self.executor
            .set_parameter_read_only_on(self.node_id, name, read_only)
    }

    /// Constrain an integer parameter to `[min, max]` in steps of `step`.
    pub fn set_parameter_integer_range(
        &mut self,
        name: &str,
        min: i64,
        max: i64,
        step: i64,
    ) -> bool {
        self.executor
            .set_parameter_integer_range_on(self.node_id, name, min, max, step)
    }

    /// Constrain a floating-point parameter to `[min, max]` in steps of `step`.
    pub fn set_parameter_float_range(&mut self, name: &str, min: f64, max: f64, step: f64) -> bool {
        self.executor
            .set_parameter_float_range_on(self.node_id, name, min, max, step)
    }

    /// rclcpp's `add_on_set_parameters_callback`: `callback` may refuse a write
    /// to this node's parameters.
    pub fn add_on_set_parameters_callback(
        &mut self,
        callback: nros_params::OnSetParameterFn,
        ctx: nros_params::OnSetContext,
    ) -> Option<nros_params::OnSetParameterHandle> {
        self.executor
            .add_on_set_parameters_callback_on(self.node_id, callback, ctx)
    }

    /// rclcpp's `remove_on_set_parameters_callback`.
    pub fn remove_on_set_parameters_callback(
        &mut self,
        handle: nros_params::OnSetParameterHandle,
    ) -> bool {
        self.executor.remove_on_set_parameters_callback(handle)
    }

    // phase-515 R10 stage 1b -- the rest of the per-node parameter surface,
    // which only the executor's node-id-keyed forms carried before those moved
    // behind `ExecutorFfi`.

    /// The `-p` / `--params-file` overrides that apply to this node, one per
    /// name — rclcpp's `NodeParametersInterface::get_parameter_overrides()`
    /// (issue 2057).
    #[cfg(feature = "alloc")]
    pub fn argv_param_overrides(
        &self,
    ) -> impl Iterator<Item = (&str, nros_params::ParameterValue)> + '_ {
        self.executor.argv_param_overrides_on(self.node_id)
    }

    /// [`argv_param_overrides`](Self::argv_param_overrides), one call of `f`
    /// per override with the value borrowed; returns how many there were.
    #[cfg(feature = "alloc")]
    pub fn for_each_argv_param_override(
        &self,
        f: impl FnMut(&str, &nros_params::ParameterValue),
    ) -> usize {
        self.executor
            .for_each_argv_param_override_on(self.node_id, f)
    }

    /// rclc's LOCAL typed set: type and on-set hooks are checked, read-only
    /// and range are not (rclc's owner of a parameter is not bound by them).
    pub fn set_parameter_local(
        &mut self,
        name: &str,
        value: nros_params::ParameterValue,
    ) -> Result<(), nros_params::ParameterError> {
        self.executor
            .set_parameter_local_on(self.node_id, name, value)
    }

    /// Publish no `/parameter_events` for this node (rclcpp's
    /// `start_parameter_event_publisher(false)`).
    #[cfg(feature = "param-services")]
    pub fn disable_parameter_events(&mut self) {
        self.executor.disable_parameter_events_on(self.node_id)
    }

    /// Publish this node's parameter changes on `/parameter_events`.
    #[cfg(feature = "param-services")]
    pub fn enable_parameter_events(&mut self) -> Result<(), NodeError> {
        self.executor.enable_parameter_events_on(self.node_id)
    }

    /// A remote `NOT_SET` deletes the parameter (rclc) instead of being
    /// refused (rclcpp).
    pub fn delete_parameters_on_not_set(&mut self, on: bool) {
        self.executor
            .delete_parameters_on_not_set_on(self.node_id, on)
    }

    /// [`set_parameters_atomically`](Self::set_parameters_atomically) with
    /// element `i` filled into one scratch value by `fill(i, slot)`.
    pub fn set_parameters_atomically_with<'x>(
        &mut self,
        count: usize,
        fill: impl FnMut(usize, &mut nros_params::ParameterValue) -> Option<&'x str>,
    ) -> Result<(), nros_params::ParameterError> {
        self.executor
            .set_parameters_atomically_with_on(self.node_id, count, fill)
    }
}

impl<'e, 's> Node<'e, 's> {
    pub(crate) fn new(
        executor: &'e mut super::spin::Executor<'s>,
        node_id: super::node_record::NodeId,
    ) -> Self {
        Self { executor, node_id }
    }

    /// The node's name — rclrs's `Node::name()`.
    ///
    /// phase-483 W2. Read from the executor's node table, where the name the
    /// node was created with (or the launch file's override of it) lives.
    #[must_use]
    pub fn name(&self) -> &str {
        self.executor
            .node(self.node_id)
            .map(|r| r.name.as_str())
            .unwrap_or("")
    }

    /// The node's namespace — rclrs's `Node::namespace()`. `""` is the root.
    #[must_use]
    pub fn namespace(&self) -> &str {
        self.executor
            .node(self.node_id)
            .map(|r| r.namespace.as_str())
            .unwrap_or("")
    }

    /// `/<namespace>/<name>` — rclrs's `Node::fully_qualified_name()`.
    ///
    /// Bounded, where rclrs returns a `String`: no allocator here, and the
    /// two halves are each at most 64 bytes.
    pub fn fully_qualified_name(&self) -> Result<crate::names::ResolvedName, NodeError> {
        crate::names::fully_qualified_name(self.name(), self.namespace())
            .map_err(|()| NodeError::NameTooLong)
    }

    /// The node's ROS domain — rclrs's `Node::domain_id()`. Every node of an
    /// executor shares its session, so this is the executor's domain.
    #[must_use]
    pub fn domain_id(&self) -> u32 {
        self.executor.domain_id
    }

    /// The logger named for this node — rclrs's `Node::logger()`. See
    /// [`NodeHandle::logger`] for how the name is resolved.
    #[must_use]
    pub fn logger(&self) -> &'static nros_log::Logger {
        nros_log::resolve_node_logger(self.name())
    }

    /// The executor this node borrows — for the `nros` facade's declarative
    /// constructors, which register into it. Not a user API.
    #[doc(hidden)]
    pub fn __executor(&mut self) -> &mut super::spin::Executor<'s> {
        self.executor
    }

    /// This node's id in the executor's node table — an RTOS extension, for
    /// the APIs that address a node by id (`Executor::node_mut`, the
    /// per-node QoS override table).
    #[must_use]
    pub fn id(&self) -> super::node_record::NodeId {
        self.node_id
    }

    /// The node's clock — rclrs's `Node::get_clock()`, and rclcpp's.
    ///
    /// phase-417 G6. `cpp:Node::get_clock` has shipped since issue 0789 and
    /// only the Rust node lacked the accessor: a component reached the clock
    /// through the executor, or constructed one, and the ported line
    /// `node.get_clock().now()` had nothing to bind.
    ///
    /// **A delegation, not a second clock.** [`nros_core::clock::Clock`] is a
    /// one-field value naming a time SOURCE — it holds no state of its own, and
    /// `now()` on a ROS-time clock reads the process-global override that
    /// `install_ros_time_source` installs. So
    /// this returns the same answer as the executor's clock by construction
    /// rather than by agreement, which is the property the graph forwarders in
    /// this family are shaped for.
    ///
    /// ROS TIME, as rclcpp's and rclrs's node clocks are: with a `/clock`
    /// source installed it follows the simulator, and with none it reads the
    /// same wall clock `Clock::system()` does (issue 1334 put those two on one
    /// expression). A component that specifically wants the monotonic source
    /// asks for `Clock::steady()`.
    ///
    /// By VALUE, where rclcpp hands back a `Clock::SharedPtr`: there is no
    /// allocator here (RFC-0022) and the type is `Copy`, so a handle would be
    /// a pointer to one `u8`.
    pub fn get_clock(&self) -> nros_core::clock::Clock {
        nros_core::clock::Clock::ros_time()
    }

    /// Subscription builder (the `clone` tier). Pick a mode with `.typed::<M>()`
    /// or `.generic(type, hash)`, set knobs (`.qos`), then `.build(callback)`.
    pub fn subscription<'t>(&mut self, topic: &'t str) -> SubscriptionBuilder<'_, 'e, 't, 's> {
        SubscriptionBuilder {
            ctx: self,
            topic,
            qos: QoSProfile::default(),
        }
    }

    /// Publisher builder (the `clone` tier), symmetric with
    /// [`subscription`](Self::subscription). Pick `.typed::<M>()` or
    /// `.generic(type, hash)`, set `.qos()`, then `.build()`. The returned
    /// publisher handle is owned and outlives this `Node` — the bridge
    /// builds the dest publisher on one ctx, drops it, then registers the
    /// source subscription on another (see `0022-entity-api-tiers.md`).
    pub fn publisher<'t>(&mut self, topic: &'t str) -> CtxPublisherBuilder<'_, 'e, 't, 's> {
        CtxPublisherBuilder {
            ctx: self,
            topic,
            qos: QoSProfile::default(),
        }
    }

    /// Convenient typed publisher (the `fork` tier — rclcpp/rclrs shape).
    ///
    /// phase-515 R1 — takes rclrs's `impl IntoPrimitiveOptions`: a bare
    /// topic name (default QoS) or one with QoS steps chained on,
    /// `"chatter".keep_last(10).reliable()`.
    pub fn create_publisher<'o, M: MessageForRmw>(
        &mut self,
        options: impl crate::IntoPrimitiveOptions<'o>,
    ) -> Result<EmbeddedPublisher<M>, NodeError> {
        let options = options.into_primitive_options();
        self.create_publisher_with_qos::<M>(options.name, options.resolve(QoSProfile::default()))
    }

    /// Convenient generic (type-erased) publisher — rclcpp `create_generic_*`.
    pub fn create_generic_publisher(
        &mut self,
        topic: &str,
        type_name: &str,
        type_hash: &str,
    ) -> Result<crate::executor::handles::EmbeddedRawPublisher, NodeError> {
        self.create_generic_publisher_with_qos(topic, type_name, type_hash, QoSProfile::default())
    }

    /// Issue 0306 — the same, with an explicit profile. The declarative
    /// component path needs it: a node that declares
    /// `create_publisher_for_topic_with_qos(...)` carries its profile in
    /// `EntityMetadata::qos`, and the runtime used to drop it on the floor by
    /// calling the default-QoS constructor here.
    pub fn create_generic_publisher_with_qos(
        &mut self,
        topic: &str,
        type_name: &str,
        type_hash: &str,
        qos: QoSProfile,
    ) -> Result<crate::executor::handles::EmbeddedRawPublisher, NodeError> {
        self.executor
            .create_publisher_raw_on(self.node_id, topic, type_name, type_hash, qos)
    }

    /// Convenient typed subscription (the `fork` tier — rclcpp/rclrs shape).
    /// Sugar over the builder with default QoS + buffer.
    ///
    /// phase-515 R1 — `options` is rclrs's `impl IntoPrimitiveOptions`, as on
    /// [`create_publisher`](Self::create_publisher). ADOPT-BOUNDED (subscription):
    /// returns the entity's `HandleId`, not an owned rclrs `Subscription` — the
    /// entity lives in the executor's table, and the id is what cancels it.
    pub fn create_subscription<'o, M, F>(
        &mut self,
        options: impl crate::IntoPrimitiveOptions<'o>,
        callback: F,
    ) -> Result<super::types::HandleId, NodeError>
    where
        M: MessageForRmw + 'static,
        F: FnMut(&M) + 'static,
    {
        let options = options.into_primitive_options();
        self.executor
            .register_subscription_buffered_on::<M, F, { crate::config::DEFAULT_RX_BUF_SIZE }>(
                self.node_id,
                options.name,
                options.resolve(QoSProfile::default()),
                callback,
                None, // no group — node default
                None, // phase-403 W2: the configured default, unchanged
            )
    }

    /// Subscribe `/clock` and install every sample as this image's ROS time —
    /// phase-425 W3, the thing `rclcpp`'s `use_sim_time` parameter switches on.
    ///
    /// After this, `Clock::ros_time().now()` is the simulator's or the bag
    /// player's time, and a timer registered with `TimerClockSource::Ros`
    /// follows it: it stops while the simulation is paused and tracks the replay
    /// rate. Nothing else changes — a wall timer on the same executor keeps its
    /// own cadence, which is what a watchdog needs.
    ///
    /// QoS is `QoSProfile::clock_default()`, i.e. `rclcpp::ClockQoS`: best
    /// effort, keep-last 1, volatile. A late subscriber wants the NEXT sample,
    /// not a replay of the simulation's history.
    ///
    /// Returns the subscription handle. Cancelling it stops the source but does
    /// NOT clear the override: a node that unsubscribes mid-run keeps the last
    /// simulated time rather than jumping back to the wall clock, which every
    /// ROS-time timer would otherwise have to absorb as a jump.
    ///
    /// The override is process-global — one simulated clock per image — so
    /// calling this on a second node in the same image adds a second subscriber
    /// to the same global, which is redundant rather than wrong.
    #[cfg(all(feature = "sim-time", any(has_rmw, test)))]
    pub fn install_ros_time_source(&mut self) -> Result<super::types::HandleId, NodeError> {
        self.install_ros_time_source_on(crate::time_source::CLOCK_TOPIC)
    }

    /// [`install_ros_time_source`](Self::install_ros_time_source) against a
    /// topic other than `/clock` — a remapped or namespaced clock, which is what
    /// a launch file creates when two simulations share a graph.
    #[cfg(all(feature = "sim-time", any(has_rmw, test)))]
    pub fn install_ros_time_source_on(
        &mut self,
        topic: &str,
    ) -> Result<super::types::HandleId, NodeError> {
        // One spelling of the QoS and the sample conversion, on the executor:
        // the `use_sim_time` reconciliation (phase-425 W3b) installs the same
        // subscription, and two copies would be two places for the QoS to drift.
        self.executor.install_ros_time_source(self.node_id, topic)
    }

    // -----------------------------------------------------------------------
    // Phase 273 (RFC-0047) — callback-group API (rclcpp/rclrs shape)
    // -----------------------------------------------------------------------

    /// Create a named callback group — a thin token wrapping the group name.
    ///
    /// Pass the returned [`CallbackGroup`] to the `_in` create variants
    /// (`create_timer_in_group`, `create_subscription_in_group`, `create_publisher_in_group`)
    /// to label entities with this group. The executor binds the entity's
    /// callback to the `SchedContext` seeded via `bind_group_sched` for
    /// `(node_name, namespace, group_name)` (phase 273 W1/W2).
    ///
    /// ```ignore
    /// let ctrl = node.create_callback_group("ctrl");
    /// node.create_timer_in_group(&ctrl, TimerDuration::from_millis(10), || { /* … */ })?;
    /// ```
    ///
    /// Names longer than 32 bytes are silently truncated.
    pub fn create_callback_group(&self, name: &str) -> CallbackGroup {
        let mut s = heapless::String::<32>::new();
        // Silently truncate if the name is too long (defensive; caller
        // should use short ASCII group names like "ctrl"/"telem").
        for ch in name.chars() {
            if s.push(ch).is_err() {
                break;
            }
        }
        CallbackGroup { name: s }
    }

    /// Create a repeating timer **in** a callback group (phase 273, rclcpp shape).
    ///
    /// The timer's callback is bound to the `SchedContext` associated with
    /// `group` in the executor's `group_sched_table` for this node. If no
    /// entry was seeded for the group, the node's default `SchedContext` applies
    /// (same as `register_timer`). `period` fires the callback repeatedly.
    pub fn create_timer_in_group<F>(
        &mut self,
        group: &CallbackGroup,
        period: crate::timer::TimerDuration,
        callback: F,
    ) -> Result<super::types::HandleId, NodeError>
    where
        F: FnMut() + 'static,
    {
        self.executor
            .register_timer_on(Some(self.node_id), period, callback, Some(group.name()))
    }

    /// rclrs's `create_timer_repeating(period, callback)` — phase-515 R4.
    ///
    /// Fires `callback` every `period`, on this node's default `SchedContext`.
    /// rclrs 0.7.0 settled on the same three timer modes ours carries
    /// (repeating, one-shot, inert); these three verbs are its spelling of
    /// them. `period` is a [`TimerDuration`](crate::timer::TimerDuration) or a
    /// `core::time::Duration`.
    ///
    /// ADOPT-BOUNDED: returns the timer's [`HandleId`](super::types::HandleId),
    /// not an rclrs `Timer` object — the timer lives in the executor's static
    /// table, so `Executor::cancel_timer` / `reset_timer` / `timer_period_ms`
    /// take that id. No `IntoTimerOptions` (a clock choice is
    /// [`create_timer_on_clock`](Self::create_timer_on_clock)).
    pub fn create_timer_repeating<F>(
        &mut self,
        period: impl Into<crate::timer::TimerDuration>,
        callback: F,
    ) -> Result<super::types::HandleId, NodeError>
    where
        F: FnMut() + 'static,
    {
        self.executor.register_timer_entry(
            Some(self.node_id),
            period.into(),
            super::TimerClockSource::Steady,
            false,
            None,
            callback,
        )
    }

    /// rclrs's `create_timer_oneshot(period, callback)` — phase-515 R4.
    ///
    /// Fires `callback` once, `period` after creation, then goes inert;
    /// `Executor::reset_timer` re-arms it for one more shot. Same envelope as
    /// [`create_timer_repeating`](Self::create_timer_repeating).
    pub fn create_timer_oneshot<F>(
        &mut self,
        period: impl Into<crate::timer::TimerDuration>,
        callback: F,
    ) -> Result<super::types::HandleId, NodeError>
    where
        F: FnMut() + 'static,
    {
        self.executor.register_timer_entry(
            Some(self.node_id),
            period.into(),
            super::TimerClockSource::Steady,
            true,
            None,
            callback,
        )
    }

    /// rclrs's `create_timer_inert(period, callback)` — phase-515 R4.
    ///
    /// Creates the timer without starting it (rcl's `autostart = false`): it
    /// never fires until `Executor::reset_timer` starts it, after which it
    /// repeats every `period`. ADOPT-BOUNDED beyond
    /// [`create_timer_repeating`](Self::create_timer_repeating)'s envelope:
    /// rclrs can later switch an inert timer to one-shot; ours starts as a
    /// repeating timer, and a one-shot is created with
    /// [`create_timer_oneshot`](Self::create_timer_oneshot) instead.
    pub fn create_timer_inert<F>(
        &mut self,
        period: impl Into<crate::timer::TimerDuration>,
        callback: F,
    ) -> Result<super::types::HandleId, NodeError>
    where
        F: FnMut() + 'static,
    {
        let mut callback = callback;
        self.executor.register_timer_entry_with(
            Some(self.node_id),
            period.into(),
            super::TimerClockSource::Steady,
            false,
            false,
            None,
            None,
            core::ptr::null_mut(),
            move |_| move |_: &crate::timer::TimerCallInfo| callback(),
        )
    }

    /// Create a guard condition on this node — phase-417 W4.e, the Rust half of
    /// the one creation shape.
    ///
    /// A guard condition is the cross-thread / ISR wake source: any thread may
    /// call [`GuardCondition::trigger`](super::types::GuardCondition::trigger)
    /// and the `callback` runs on the executor's next `spin_once`. The flag
    /// itself lives in the executor's arena, which is what makes the returned
    /// handle `Send + Sync` with no allocation — so the NODE is the owner a
    /// caller names and the EXECUTOR is where the entity lands, exactly as for
    /// a timer or a subscription.
    ///
    /// Until 2026-09-18 the only Rust spelling was
    /// [`Executor::register_guard_condition`](crate::Executor::register_guard_condition),
    /// while C++ created one on the node and C needed three calls. That was the
    /// three-owner state RFC-0089 stage 4 exists to remove; the executor form
    /// survives as the ours-only primitive this calls.
    ///
    /// The returned [`HandleId`](super::types::HandleId) is the slot, for
    /// `trigger_one` / `bind_handle_to_sched_context`; the
    /// [`GuardCondition`](super::types::GuardCondition) is the trigger handle to
    /// hand to the other thread.
    ///
    /// ```ignore
    /// let (_id, guard) = node.create_guard_condition(|| { /* woken */ })?;
    /// // from an ISR or another task:
    /// guard.trigger();
    /// ```
    pub fn create_guard_condition<F>(
        &mut self,
        callback: F,
    ) -> Result<(super::types::HandleId, super::types::GuardCondition), NodeError>
    where
        F: FnMut() + 'static,
    {
        self.executor
            .register_guard_condition_on(Some(self.node_id), callback)
    }

    /// Create a repeating timer on a chosen CLOCK — phase-430 W4, the
    /// node-level spelling of
    /// [`Executor::register_timer_on_clock`](crate::Executor::register_timer_on_clock)
    /// and the Rust counterpart of C++'s `create_timer(clock, period, cb)`.
    ///
    /// [`create_timer_in_group`](Self::create_timer_in_group) and the executor's
    /// `register_timer` are the WALL case: they consume the spin delta the
    /// executor already measured, and no simulator can slow them down. A
    /// [`TimerClockSource::Ros`](crate::executor::TimerClockSource::Ros) timer instead follows `/clock`, so it stops
    /// while the simulator is paused and tracks a bag's replay rate; with no
    /// `/clock` source installed it reads system time, the same fallback
    /// `rclcpp::Clock` has.
    ///
    /// The timer is bound to this node, so the node's default `SchedContext`
    /// applies — [`create_timer_on_clock_in_group`](Self::create_timer_on_clock_in_group)
    /// is the callback-group form.
    ///
    /// ```ignore
    /// node.create_timer_on_clock(
    ///     TimerDuration::from_millis(100),
    ///     TimerClockSource::Ros,
    ///     || { /* fires once per 100 ms of SIMULATED time */ },
    /// )?;
    /// ```
    pub fn create_timer_on_clock<F>(
        &mut self,
        period: crate::timer::TimerDuration,
        source: crate::executor::TimerClockSource,
        callback: F,
    ) -> Result<super::types::HandleId, NodeError>
    where
        F: FnMut() + 'static,
    {
        self.executor.register_timer_entry(
            Some(self.node_id),
            period,
            source,
            false,
            None,
            callback,
        )
    }

    /// [`create_timer_on_clock`](Self::create_timer_on_clock) **in** a callback
    /// group — the clock axis on [`create_timer_in_group`](Self::create_timer_in_group)
    /// (phase-430 W4). The two axes are independent: the group decides which
    /// `SchedContext` the callback runs under, the clock decides what advances
    /// the timer towards its next activation.
    pub fn create_timer_on_clock_in_group<F>(
        &mut self,
        group: &CallbackGroup,
        period: crate::timer::TimerDuration,
        source: crate::executor::TimerClockSource,
        callback: F,
    ) -> Result<super::types::HandleId, NodeError>
    where
        F: FnMut() + 'static,
    {
        self.executor.register_timer_entry(
            Some(self.node_id),
            period,
            source,
            false,
            Some(group.name()),
            callback,
        )
    }

    /// Create a repeating timer whose callback receives the activation's
    /// [`TimerCallInfo`](crate::TimerCallInfo) — issue 1974, the Rust
    /// counterpart of Jazzy's `rclcpp::TimerInfoCallbackType`.
    ///
    /// `info.expected_call_time` is the deadline the executor was serving and
    /// `info.actual_call_time` when the callback ran, both on `source`'s clock;
    /// `info.lateness_ns()` is the jitter a real-time node otherwise measures
    /// by hand. `autostart == false` creates the timer cancelled (rclcpp's
    /// `create_timer(..., autostart = false)`); `Executor::reset_timer` on the
    /// returned handle starts it.
    ///
    /// ```ignore
    /// node.create_timer_with_info(
    ///     TimerDuration::from_millis(10),
    ///     TimerClockSource::Steady,
    ///     true,
    ///     |info| worst = worst.max(info.lateness_ns()),
    /// )?;
    /// ```
    pub fn create_timer_with_info<F>(
        &mut self,
        period: crate::timer::TimerDuration,
        source: crate::executor::TimerClockSource,
        autostart: bool,
        callback: F,
    ) -> Result<super::types::HandleId, NodeError>
    where
        F: FnMut(&crate::TimerCallInfo) + 'static,
    {
        self.executor.register_timer_with_info_on(
            Some(self.node_id),
            period,
            source,
            autostart,
            None,
            callback,
        )
    }

    /// Create a typed subscription **in** a callback group (phase 273, rclcpp shape).
    ///
    /// The subscription's callback is bound to the `SchedContext` associated
    /// with `group` in the `group_sched_table` for this node. If no entry was
    /// seeded, the node default applies.
    pub fn create_subscription_in_group<M, F>(
        &mut self,
        group: &CallbackGroup,
        topic: &str,
        callback: F,
    ) -> Result<super::types::HandleId, NodeError>
    where
        M: MessageForRmw + 'static,
        F: FnMut(&M) + 'static,
    {
        self.executor
            .register_subscription_buffered_on::<M, F, { crate::config::DEFAULT_RX_BUF_SIZE }>(
                self.node_id,
                topic,
                QoSProfile::default(),
                callback,
                Some(group.name()),
                None, // phase-403 W2: the configured default, unchanged
            )
    }

    /// Create a typed publisher **in** a callback group (phase 273, rclcpp shape).
    ///
    /// Publishers do not have an executor-dispatched callback, so the group
    /// name has no scheduling effect today (publishers are explicitly driven by
    /// the user via `publish()`). The API is provided for symmetry and
    /// forward-compatibility (intra-process / loaned-message knobs may use
    /// it in the future).
    pub fn create_publisher_in_group<M: MessageForRmw>(
        &mut self,
        _group: &CallbackGroup,
        topic: &str,
    ) -> Result<EmbeddedPublisher<M>, NodeError> {
        // Publishers carry no executor callback slot; group is forward-compat.
        self.create_publisher::<M>(topic)
    }

    /// RFC-0041 / Phase 239.1 — callback-based service client (rclcpp
    /// `async_send_request(req, cb)` analogue). The reply is delivered to
    /// `callback` at `spin_once` (no `Promise` poll). Returns a
    /// [`ServiceClientCallback`] send handle; dual-mode — the `Promise`-based
    /// `create_client` is unchanged.
    pub fn create_client_with_callback<Svc, F>(
        &mut self,
        service_name: &str,
        callback: F,
    ) -> Result<ServiceClientCallback<Svc>, NodeError>
    where
        Svc: RosService + 'static,
        Svc::Request: MessageForRmw,
        Svc::Reply: MessageForRmw,
        F: FnMut(&Svc::Reply) + 'static,
    {
        self.create_client_with_callback_sized::<
            Svc,
            F,
            { crate::config::DEFAULT_RX_BUF_SIZE },
            { crate::config::DEFAULT_RX_BUF_SIZE },
        >(service_name, callback)
    }

    /// Callback-based service client with custom buffer sizes (Phase 239.1).
    pub fn create_client_with_callback_sized<Svc, F, const REQ_BUF: usize, const REPLY_BUF: usize>(
        &mut self,
        service_name: &str,
        callback: F,
    ) -> Result<ServiceClientCallback<Svc, REQ_BUF, REPLY_BUF>, NodeError>
    where
        Svc: RosService + 'static,
        Svc::Request: MessageForRmw,
        Svc::Reply: MessageForRmw,
        F: FnMut(&Svc::Reply) + 'static,
    {
        self.register_client_callback::<Svc, _, REQ_BUF, REPLY_BUF>(
            service_name,
            super::arena::ReplyOnly(callback),
        )
    }

    /// [`create_client_with_callback`](Self::create_client_with_callback)
    /// whose closure also receives the reply's
    /// [`ServiceInfo`](nros_core::ServiceInfo) — issue 2148, the callback twin
    /// of [`Promise::take_with_info`](super::handles::Promise::take_with_info)
    /// and the client twin of
    /// [`create_service_with_info`](Self::create_service_with_info).
    ///
    /// rclrs's `(Response, ServiceInfo)` reply and upstream
    /// `rmw_take_response`'s `rmw_service_info_t`: `request_id` names the
    /// request this reply answers (this client's GID and the sequence number
    /// of its request), `source_timestamp` is when the SERVER sent the reply,
    /// on its clock, and `received_timestamp` is when this client received it.
    /// A field the backend cannot report is `None`, never a zero — see
    /// [`ServiceInfo`](nros_core::ServiceInfo).
    pub fn create_client_with_callback_with_info<Svc, F>(
        &mut self,
        service_name: &str,
        callback: F,
    ) -> Result<ServiceClientCallback<Svc>, NodeError>
    where
        Svc: RosService + 'static,
        Svc::Request: MessageForRmw,
        Svc::Reply: MessageForRmw,
        F: FnMut(&Svc::Reply, &nros_core::ServiceInfo) + 'static,
    {
        self.register_client_callback::<
            Svc,
            _,
            { crate::config::DEFAULT_RX_BUF_SIZE },
            { crate::config::DEFAULT_RX_BUF_SIZE },
        >(service_name, super::arena::ReplyWithInfo(callback))
    }

    /// The one registration behind both callback-client shapes (issue 2148).
    fn register_client_callback<Svc, C, const REQ_BUF: usize, const REPLY_BUF: usize>(
        &mut self,
        service_name: &str,
        callback: C,
    ) -> Result<ServiceClientCallback<Svc, REQ_BUF, REPLY_BUF>, NodeError>
    where
        Svc: RosService + 'static,
        Svc::Request: MessageForRmw,
        Svc::Reply: MessageForRmw,
        C: super::arena::ClientReplyCallback<Svc::Reply> + 'static,
    {
        register_type::<Svc::Request>()?;
        register_type::<Svc::Reply>()?;
        let (_id, hdr) = self
            .executor
            .register_service_client_callback::<Svc, C, REPLY_BUF>(
                Some(self.node_id),
                service_name,
                Svc::SERVICE_NAME,
                Svc::SERVICE_HASH,
                QoSProfile::services_default(),
                callback,
            )?;
        Ok(ServiceClientCallback::new(hdr))
    }

    /// RFC-0041 / Phase 239.2 — callback-based action client (rclcpp
    /// `SendGoalOptions{goal_response_callback, feedback_callback,
    /// result_callback}` analogue). Goal-response / feedback / result are
    /// delivered to the closures at `spin_once`. Returns an
    /// [`ActionClientCallback`] send handle (`send_goal` / `get_result`);
    /// dual-mode — the `Promise`-based `create_action_client`
    /// is unchanged.
    #[allow(clippy::type_complexity)]
    pub fn create_action_client_with_callbacks<A, GRespF, FbF, ResF>(
        &mut self,
        action_name: &str,
        on_goal_response: GRespF,
        on_feedback: FbF,
        on_result: ResF,
    ) -> Result<ActionClientCallback<A>, NodeError>
    where
        A: RosAction + 'static,
        A::Goal: MessageForRmw,
        A::Result: MessageForRmw,
        A::Feedback: MessageForRmw,
        GRespF: FnMut(&nros_core::GoalUuid, bool) + 'static,
        FbF: FnMut(&nros_core::GoalUuid, &A::Feedback) + 'static,
        ResF: FnMut(&nros_core::GoalUuid, nros_core::GoalStatusCode, &A::Result) + 'static,
    {
        self.create_action_client_with_callbacks_sized::<
            A,
            GRespF,
            FbF,
            ResF,
            { crate::config::DEFAULT_RX_BUF_SIZE },
            { crate::config::DEFAULT_RX_BUF_SIZE },
            { crate::config::DEFAULT_RX_BUF_SIZE },
        >(action_name, on_goal_response, on_feedback, on_result)
    }

    /// Callback-based action client with custom buffer sizes (Phase 239.2).
    #[allow(clippy::type_complexity)]
    pub fn create_action_client_with_callbacks_sized<
        A,
        GRespF,
        FbF,
        ResF,
        const GOAL_BUF: usize,
        const RESULT_BUF: usize,
        const FEEDBACK_BUF: usize,
    >(
        &mut self,
        action_name: &str,
        on_goal_response: GRespF,
        on_feedback: FbF,
        on_result: ResF,
    ) -> Result<ActionClientCallback<A, GOAL_BUF, RESULT_BUF, FEEDBACK_BUF>, NodeError>
    where
        A: RosAction + 'static,
        A::Goal: MessageForRmw,
        A::Result: MessageForRmw,
        A::Feedback: MessageForRmw,
        GRespF: FnMut(&nros_core::GoalUuid, bool) + 'static,
        FbF: FnMut(&nros_core::GoalUuid, &A::Feedback) + 'static,
        ResF: FnMut(&nros_core::GoalUuid, nros_core::GoalStatusCode, &A::Result) + 'static,
    {
        register_type::<A::Goal>()?;
        register_type::<A::Result>()?;
        register_type::<A::Feedback>()?;
        let (_id, core) = self
            .executor
            .register_action_client_callback::<A, GRespF, FbF, ResF, GOAL_BUF, RESULT_BUF, FEEDBACK_BUF>(
                Some(self.node_id),
                action_name,
                A::ACTION_NAME,
                A::ACTION_HASH,
                // Feedback is a stream → buffer a short QoS-depth history (Phase
                // 239.5). Goal-response / result are single-outstanding (gated).
                super::arena::DEFAULT_ACTION_FEEDBACK_DEPTH,
                on_goal_response,
                on_feedback,
                on_result,
            )?;
        Ok(ActionClientCallback::new(core))
    }

    /// Convenient generic (type-erased) subscription — rclcpp `create_generic_*`.
    pub fn create_generic_subscription<F>(
        &mut self,
        topic: &str,
        type_name: &str,
        type_hash: &str,
        callback: F,
    ) -> Result<super::types::HandleId, NodeError>
    where
        F: FnMut(&[u8]) + 'static,
    {
        self.create_generic_subscription_with_qos(
            topic,
            type_name,
            type_hash,
            QoSProfile::default(),
            callback,
        )
    }

    /// Issue 0306 — the same, with an explicit profile (see
    /// [`Self::create_generic_publisher_with_qos`] for why).
    pub fn create_generic_subscription_with_qos<F>(
        &mut self,
        topic: &str,
        type_name: &str,
        type_hash: &str,
        qos: QoSProfile,
        callback: F,
    ) -> Result<super::types::HandleId, NodeError>
    where
        F: FnMut(&[u8]) + 'static,
    {
        self.create_generic_subscription_with_qos_and_rx_hint(
            topic, type_name, type_hash, qos, None, callback,
        )
    }

    /// Issue 2208 -- [`Self::create_generic_subscription_with_qos`] for a
    /// caller that knows the message type's bound although it registers
    /// type-erased (the declarative runtime, which recorded it at
    /// declaration). `rx_hint` reaches the backend as `rx_buffer_hint`, which
    /// sizes zenoh-pico's receive class and, with a declared arrival curve, the
    /// XRCE agent's pace (RFC-0074 "On XRCE"). `None` states nothing.
    pub fn create_generic_subscription_with_qos_and_rx_hint<F>(
        &mut self,
        topic: &str,
        type_name: &str,
        type_hash: &str,
        qos: QoSProfile,
        rx_hint: Option<usize>,
        callback: F,
    ) -> Result<super::types::HandleId, NodeError>
    where
        F: FnMut(&[u8]) + 'static,
    {
        self.executor
            .register_subscription_buffered_raw_on::<F, { crate::config::DEFAULT_RX_BUF_SIZE }>(
                self.node_id,
                topic,
                type_name,
                type_hash,
                qos,
                rx_hint,
                callback,
            )
    }

    /// Phase 250 (Wave 2) — generic (type-erased) subscription that surfaces E2E
    /// [`IntegrityStatus`](nros_rmw::IntegrityStatus) (CRC + sequence gap/dup) to
    /// the callback (`FnMut(&[u8], &IntegrityStatus)`). The declarative-`Node`
    /// analog of the typed `.typed::<M>().safety()` builder: the validator lives
    /// in the `RmwSubscriber`, so the raw bytes + status arrive together without
    /// a typed `M`. Wired by the declarative runtime's `.safety()` opt-in.
    #[cfg(feature = "safety-e2e")]
    pub fn create_generic_subscription_with_integrity<F>(
        &mut self,
        topic: &str,
        type_name: &str,
        type_hash: &str,
        callback: F,
    ) -> Result<super::types::HandleId, NodeError>
    where
        F: FnMut(&[u8], &nros_rmw::IntegrityStatus) + 'static,
    {
        self.create_generic_subscription_with_integrity_and_rx_hint(
            topic, type_name, type_hash, None, callback,
        )
    }

    /// Issue 2208 -- [`Self::create_generic_subscription_with_integrity`]
    /// with the message type's bound, as
    /// [`Self::create_generic_subscription_with_qos_and_rx_hint`].
    #[cfg(feature = "safety-e2e")]
    pub fn create_generic_subscription_with_integrity_and_rx_hint<F>(
        &mut self,
        topic: &str,
        type_name: &str,
        type_hash: &str,
        rx_hint: Option<usize>,
        callback: F,
    ) -> Result<super::types::HandleId, NodeError>
    where
        F: FnMut(&[u8], &nros_rmw::IntegrityStatus) + 'static,
    {
        self.executor
            .register_subscription_buffered_raw_safety_on::<F, { crate::config::DEFAULT_RX_BUF_SIZE }>(
                self.node_id,
                topic,
                type_name,
                type_hash,
                QoSProfile::default(),
                rx_hint,
                callback,
            )
    }

    /// Convenient zero-copy subscription (Phase 229.6, issue 0007 / RFC-0033
    /// `view` mode).
    ///
    /// `B` is the code-generated VIEWABLE marker (e.g. `ImageViewable`, emitted
    /// alongside the `inline` `Image` for a `.msg` with a `view`-mode field).
    /// The marker is zero-sized and carries NO lifetime, which is the whole
    /// reason it exists: `B::View<'a>` does, and a generic parameter cannot be
    /// the lifetime-carrying type itself. That is also why phase-390 could not
    /// simply rename it to `{Msg}View` — the view struct already owns that
    /// name, and the marker points AT it.
    ///
    /// The callback receives `&B::View<'a>` — a message whose unbounded
    /// sequence/string fields point directly into the receive buffer (no
    /// `heapless::Vec` copy); valid only for the callback's duration.
    ///
    /// Uses `KEEP_LAST(1)` QoS → triple buffer, as view subscriptions require
    /// (a single well-defined slot for the callback). For an explicit deeper
    /// queue use the `inline` [`create_subscription`](Self::create_subscription);
    /// a view subscription registered with `KEEP_LAST(N>1)` is rejected.
    pub fn create_subscription_viewable<B, F>(
        &mut self,
        topic: &str,
        callback: F,
    ) -> Result<super::types::HandleId, NodeError>
    where
        B: nros_core::ViewableMessage + 'static,
        F: for<'a> FnMut(&B::View<'a>) + 'static,
    {
        self.executor
            .register_subscription_buffered_borrowed_on::<B, F, { crate::config::DEFAULT_RX_BUF_SIZE }>(
                self.node_id,
                topic,
                QoSProfile::default().keep_last(1),
                callback,
            )
    }

    /// Service-server builder (the `clone` tier) — `node.service(name)`.
    /// Set `.qos()` (defaults to the services profile = RELIABLE+VOLATILE+
    /// KEEP_LAST(10)), then `.build::<Svc, _>(callback)` (Phase 193.2).
    pub fn service<'t>(&mut self, name: &'t str) -> CtxServiceBuilder<'_, 'e, 't, 's> {
        CtxServiceBuilder {
            ctx: self,
            name,
            qos: QoSProfile::services_default(),
        }
    }

    /// Convenient service server (the `fork` tier — rclrs/rclcpp shape), default
    /// services QoS. Mirror of `create_subscription`.
    ///
    /// phase-515 R1 — `options` is rclrs's `impl IntoPrimitiveOptions`; a bare
    /// name keeps the services profile. ADOPT-BOUNDED (service): returns the
    /// entity's `HandleId`, not an owned rclrs `Service` — the entity lives in
    /// the executor's table.
    pub fn create_service<'o, Svc, F>(
        &mut self,
        options: impl crate::IntoPrimitiveOptions<'o>,
        callback: F,
    ) -> Result<super::types::HandleId, NodeError>
    where
        Svc: RosService + 'static,
        Svc::Request: crate::rmw_type_registry::MessageForRmw,
        Svc::Reply: crate::rmw_type_registry::MessageForRmw,
        F: FnMut(&Svc::Request) -> Svc::Reply + 'static,
    {
        let options = options.into_primitive_options();
        self.executor.register_service_sized_on::<
            Svc,
            F,
            { crate::config::DEFAULT_RX_BUF_SIZE },
            { crate::config::DEFAULT_RX_BUF_SIZE },
        >(
            self.node_id,
            options.name,
            options.resolve(QoSProfile::services_default()),
            callback,
        )
    }

    /// [`create_service`](Self::create_service) whose callback also receives
    /// the request's [`ServiceInfo`](nros_core::ServiceInfo) — rclrs's
    /// `(request, ServiceInfo)` service callback (issue 1942).
    ///
    /// The info says who sent the request (client GID + the client's sequence
    /// number) and when. Which fields arrive depends on the backend; one it
    /// cannot report is `None`, never a zero — see
    /// [`ServiceInfo`](nros_core::ServiceInfo).
    pub fn create_service_with_info<Svc, F>(
        &mut self,
        name: &str,
        callback: F,
    ) -> Result<super::types::HandleId, NodeError>
    where
        Svc: RosService + 'static,
        Svc::Request: crate::rmw_type_registry::MessageForRmw,
        Svc::Reply: crate::rmw_type_registry::MessageForRmw,
        F: FnMut(&Svc::Request, &nros_core::ServiceInfo) -> Svc::Reply + 'static,
    {
        self.service(name).service_info().build::<Svc, F>(callback)
    }

    // ====================================================================
    // Graph introspection — RFC-0036, phase-444.
    //
    // Upstream hangs the graph family off the NODE (`rclrs::Node::
    // get_node_names`, `rcl_get_node_names(node, ...)`). We open ONE transport
    // session per image and the Executor owns it, so the capability lives on
    // [`Executor`](super::spin::Executor) — see phase-381 W4. These are the
    // forwarders that let a ported node keep upstream's receiver and upstream's
    // verb, the same forwarders phase-417 stage 2b wrote for C++ `nros::Node`.
    //
    // They add no behaviour: each is a one-line delegation to the executor
    // this handle already borrows.
    //
    // THE CONTRACT EVERY ONE OF THEM KEEPS (RFC-0036 §"Static CONNECTION,
    // dynamic GRAPH"), stated once here and referenced from each:
    //
    //  * It reports what has been DISCOVERED, and NEVER BLOCKS. The backend
    //    keeps a standing view fed by `spin`, so a first call legitimately
    //    returns a partial graph and the view fills in over successive calls.
    //    Code waiting for a peer must poll, not call once and conclude.
    //  * An empty answer is "nobody seen yet", NEVER "nobody exists".
    //  * `Err(Transport(Unsupported))` — a backend that cannot tell you, such
    //    as XRCE — is a DIFFERENT answer from empty, and must not be collapsed
    //    into zero or into an empty enumeration.
    //
    // SHAPE DIVERGENCE from rclrs, and why: rclrs returns owned heap
    // containers (`Vec<NodeNameInfo>`, `TopicNamesAndTypes`). The graph has no
    // bound the caller can know, so a caller-sized buffer is an unanswerable
    // question and an owned container needs an allocator a 128 KiB target does
    // not have. We stream through a visitor instead, which costs one entry of
    // peak RAM. Every `&str` handed to a visitor is BORROWED for the duration
    // of that call — copy anything you keep. Returning `false` stops the walk.
    // ====================================================================

    /// Every node on the graph, with its namespace — rclrs's
    /// `Node::get_node_names`.
    ///
    /// `visit(name, namespace)` once per node. For the enclave as well, use
    /// [`get_node_names_with_enclaves`](Self::get_node_names_with_enclaves):
    /// one backend slot answers both, exactly as `rmw_get_node_names` and
    /// `rmw_get_node_names_with_enclaves` are one query upstream.
    ///
    /// Reports what has been DISCOVERED and never blocks, so an empty
    /// enumeration means "nobody seen yet", not "nobody exists";
    /// `Err(Transport(Unsupported))` from a backend with no graph is a
    /// different answer from empty. See the module note above.
    pub fn get_node_names(
        &mut self,
        visit: &mut dyn FnMut(&str, &str) -> bool,
    ) -> Result<(), NodeError> {
        self.executor
            .get_node_names(&mut |name, ns, _enclave| visit(name, ns))
    }

    /// Every node on the graph with its namespace AND its enclave — rclrs's
    /// `Node::get_node_names_with_enclaves`.
    ///
    /// `visit(name, namespace, enclave)`. `enclave` is `None` when the backend
    /// reports the node but not its enclave, which is a partial answer rather
    /// than an error — reporting the node without the enclave beats dropping
    /// it. Same discovery envelope as [`get_node_names`](Self::get_node_names).
    pub fn get_node_names_with_enclaves(
        &mut self,
        visit: &mut dyn FnMut(&str, &str, Option<&str>) -> bool,
    ) -> Result<(), NodeError> {
        self.executor.get_node_names(visit)
    }

    /// Every topic on the graph, with the types on it — rclrs's
    /// `Node::get_topic_names_and_types`.
    ///
    /// `visit(topic_name, types)` once per distinct TOPIC: a topic carrying
    /// two types is one call with two entries, not two calls. `types` may
    /// legitimately be empty on a partially discovered graph. Same discovery
    /// envelope as [`get_node_names`](Self::get_node_names).
    pub fn get_topic_names_and_types(
        &mut self,
        visit: &mut dyn FnMut(&str, &[&str]) -> bool,
    ) -> Result<(), NodeError> {
        self.executor.get_topic_names_and_types(visit)
    }

    /// Every service on the graph, with its types — rclrs's
    /// `Node::get_service_names_and_types`. As
    /// [`get_topic_names_and_types`](Self::get_topic_names_and_types), over
    /// servers and clients.
    pub fn get_service_names_and_types(
        &mut self,
        visit: &mut dyn FnMut(&str, &[&str]) -> bool,
    ) -> Result<(), NodeError> {
        self.executor.get_service_names_and_types(visit)
    }

    /// How many publishers are visible on `topic_name` — rclrs's
    /// `Node::count_publishers`.
    ///
    /// **This is the GRAPH's count, not this node's.** The number of
    /// publishers this node itself DECLARED is
    /// [`StandaloneNode::publisher_count`](crate::StandaloneNode), a different
    /// question with a different answer.
    ///
    /// A count reflects what has been DISCOVERED, so it can be low right after
    /// startup and is never a proof of absence. See
    /// [`get_node_names`](Self::get_node_names).
    pub fn count_publishers(&mut self, topic_name: &str) -> Result<usize, NodeError> {
        self.executor.count_publishers(topic_name)
    }

    /// How many subscriptions are visible on `topic_name` — rclrs's
    /// `Node::count_subscriptions`.
    ///
    /// **`subscriptions`, not `subscribers`** — this is rclrs's spelling, and
    /// the Rust surface takes its vocabulary from rclrs so a user porting Rust
    /// ROS 2 code types what they already know. The C surface says
    /// `subscribers` because rcl does, and the vtable slot says `subscribers`
    /// because upstream rmw does. Three layers, three upstreams, one word each
    /// — not drift.
    ///
    /// The graph's count, not this node's; see
    /// [`count_publishers`](Self::count_publishers) for that distinction and
    /// for the discovery caveats.
    pub fn count_subscriptions(&mut self, topic_name: &str) -> Result<usize, NodeError> {
        self.executor.count_subscribers(topic_name)
    }

    /// How many service clients are visible on `service_name` — rclcpp's
    /// (Jazzy) `Node::count_clients`; rclrs has no counterpart. Issue 1978.
    ///
    /// The graph's count, not this node's; see
    /// [`count_publishers`](Self::count_publishers) for that distinction and
    /// for the discovery caveats.
    pub fn count_clients(&mut self, service_name: &str) -> Result<usize, NodeError> {
        self.executor.count_clients(service_name)
    }

    /// How many service servers are visible on `service_name` — rclcpp's
    /// (Jazzy) `Node::count_services`. See [`count_clients`](Self::count_clients).
    pub fn count_services(&mut self, service_name: &str) -> Result<usize, NodeError> {
        self.executor.count_services(service_name)
    }

    /// What one named node PUBLISHES, with the types — rclrs's
    /// `Node::get_publisher_names_and_types_by_node`.
    ///
    /// A node the graph has not discovered yields no visits, which is not an
    /// error — see [`get_node_names`](Self::get_node_names) for why an empty
    /// answer means "not seen yet".
    pub fn get_publisher_names_and_types_by_node(
        &mut self,
        node_name: &str,
        node_namespace: &str,
        visit: &mut dyn FnMut(&str, &[&str]) -> bool,
    ) -> Result<(), NodeError> {
        self.executor
            .get_publisher_names_and_types_by_node(node_name, node_namespace, visit)
    }

    /// What one named node SUBSCRIBES to, with the types — rclrs's
    /// `Node::get_subscription_names_and_types_by_node`. See
    /// [`count_subscriptions`](Self::count_subscriptions) on the spelling.
    pub fn get_subscription_names_and_types_by_node(
        &mut self,
        node_name: &str,
        node_namespace: &str,
        visit: &mut dyn FnMut(&str, &[&str]) -> bool,
    ) -> Result<(), NodeError> {
        self.executor
            .get_subscription_names_and_types_by_node(node_name, node_namespace, visit)
    }

    /// What services one named node SERVES, with the types — rclrs's
    /// `Node::get_service_names_and_types_by_node`.
    pub fn get_service_names_and_types_by_node(
        &mut self,
        node_name: &str,
        node_namespace: &str,
        visit: &mut dyn FnMut(&str, &[&str]) -> bool,
    ) -> Result<(), NodeError> {
        self.executor
            .get_service_names_and_types_by_node(node_name, node_namespace, visit)
    }

    /// What services one named node CALLS, with the types — rclrs's
    /// `Node::get_client_names_and_types_by_node`.
    pub fn get_client_names_and_types_by_node(
        &mut self,
        node_name: &str,
        node_namespace: &str,
        visit: &mut dyn FnMut(&str, &[&str]) -> bool,
    ) -> Result<(), NodeError> {
        self.executor
            .get_client_names_and_types_by_node(node_name, node_namespace, visit)
    }

    /// The PUBLISHERS on `topic_name`, one visit each — rclrs's
    /// `Node::get_publishers_info_by_topic`.
    ///
    /// The last two members of the graph family to reach upstream's receiver.
    /// `visit(info)` per discovered endpoint; every `&str` on the
    /// [`GraphEndpointInfo`](nros_rmw::GraphEndpointInfo) is BORROWED for the
    /// duration of that call, so copy anything you keep, and returning `false`
    /// stops the walk.
    ///
    /// ADOPT-BOUNDED: rclrs returns an owned `Vec<TopicEndpointInfo>`; here the
    /// answer is STREAMED to `visit`, because the graph has no bound a caller could
    /// size a buffer to and there is no allocator for a `Vec`. It reports what has
    /// been DISCOVERED and never blocks, so an empty walk means "nobody seen yet";
    /// a backend with no graph answers `Err(Transport(Unsupported))`, never an
    /// empty `Ok`.
    ///
    /// It carries NO QoS, and that absence is the point: the GRANTED profile is
    /// what answers "why is nothing arriving", no backend can read a remote's
    /// back yet, and reporting the remote's DECLARED profile instead would be
    /// the confident wrong answer the field exists to avoid.
    pub fn get_publishers_info_by_topic(
        &mut self,
        topic_name: &str,
        visit: &mut dyn FnMut(&nros_rmw::GraphEndpointInfo<'_>) -> bool,
    ) -> Result<(), NodeError> {
        self.executor
            .get_publishers_info_by_topic(topic_name, visit)
    }

    /// The SUBSCRIPTIONS on `topic_name`, one visit each — rclrs's
    /// `Node::get_subscriptions_info_by_topic`. See
    /// [`get_publishers_info_by_topic`](Self::get_publishers_info_by_topic).
    ///
    /// Same envelope: STREAMED to `visit` rather than returned as a `Vec`, every
    /// `&str` borrowed for the visit, discovered-only and non-blocking, and
    /// `Err(Transport(Unsupported))` from a backend with no graph.
    pub fn get_subscriptions_info_by_topic(
        &mut self,
        topic_name: &str,
        visit: &mut dyn FnMut(&nros_rmw::GraphEndpointInfo<'_>) -> bool,
    ) -> Result<(), NodeError> {
        self.executor
            .get_subscriptions_info_by_topic(topic_name, visit)
    }

    // ------------------------------------------------------------------
    // phase-483 W2 — the constructors that only `NodeHandle` had. `Node`
    // is the ONE node type, so it carries them too; each forwards through
    // `Executor::with_node_try`, which opens a `NodeHandle` on this node's
    // session for the duration of the call. The polled subscription and
    // service constructors are `create_polling_*`: rclrs's
    // `create_subscription` / `create_service` take a callback, and this type
    // already has those under exactly those names.
    // ------------------------------------------------------------------

    /// [`create_publisher`](Self::create_publisher) with an explicit profile (phase-483 W2).
    ///
    /// Through [`NodeHandle::create_publisher_with_qos`], so a publisher made
    /// here gets what one made on the handle `Executor::create_node` used to
    /// return: the node's remap rules (launch, then `--ros-args`), the plan's
    /// QoS overrides and the declared-QoS check. `Executor::create_publisher_on`
    /// takes a name its caller already resolved, which is right for the C and
    /// component roads and wrong for a user's source name.
    pub fn create_publisher_with_qos<M: MessageForRmw>(
        &mut self,
        topic_name: &str,
        qos: QoSProfile,
    ) -> Result<EmbeddedPublisher<M>, NodeError> {
        self.executor.with_node_try(self.node_id, |h| {
            h.create_publisher_with_qos::<M>(topic_name, qos)
        })
    }

    /// [`NodeHandle::create_publisher_raw`] on this node (phase-483 W2).
    pub fn create_publisher_raw(
        &mut self,
        topic_name: &str,
        type_name: &str,
        type_hash: &str,
    ) -> Result<crate::executor::handles::EmbeddedRawPublisher, NodeError> {
        self.executor.with_node_try(self.node_id, |h| {
            h.create_publisher_raw(topic_name, type_name, type_hash)
        })
    }

    /// [`NodeHandle::create_publisher_raw_with_qos`] on this node (phase-483 W2).
    pub fn create_publisher_raw_with_qos(
        &mut self,
        topic_name: &str,
        type_name: &str,
        type_hash: &str,
        qos: QoSProfile,
    ) -> Result<crate::executor::handles::EmbeddedRawPublisher, NodeError> {
        self.executor.with_node_try(self.node_id, |h| {
            h.create_publisher_raw_with_qos(topic_name, type_name, type_hash, qos)
        })
    }

    /// The POLLED form of [`NodeHandle::create_subscription`] (phase-483 W2): the caller
    /// takes from it, where rclrs's `create_subscription` takes a callback.
    pub fn create_polling_subscription<M: MessageForRmw>(
        &mut self,
        topic_name: &str,
    ) -> Result<Subscription<M>, NodeError> {
        self.executor
            .with_node_try(self.node_id, |h| h.create_subscription::<M>(topic_name))
    }

    /// The POLLED form of [`NodeHandle::create_subscription_sized`] (phase-483 W2): the caller
    /// takes from it, where rclrs's `create_subscription` takes a callback.
    pub fn create_polling_subscription_sized<M: MessageForRmw, const RX_BUF: usize>(
        &mut self,
        topic_name: &str,
    ) -> Result<Subscription<M, RX_BUF>, NodeError> {
        self.executor.with_node_try(self.node_id, |h| {
            h.create_subscription_sized::<M, RX_BUF>(topic_name)
        })
    }

    /// The POLLED form of [`NodeHandle::create_subscription_with_qos`] (phase-483 W2): the caller
    /// takes from it, where rclrs's `create_subscription` takes a callback.
    pub fn create_polling_subscription_with_qos<M: MessageForRmw, const RX_BUF: usize>(
        &mut self,
        topic_name: &str,
        qos: QoSProfile,
    ) -> Result<Subscription<M, RX_BUF>, NodeError> {
        self.executor.with_node_try(self.node_id, |h| {
            h.create_subscription_with_qos::<M, RX_BUF>(topic_name, qos)
        })
    }

    /// The POLLED form of [`NodeHandle::create_subscription_raw`] (phase-483 W2): the caller
    /// takes from it, where rclrs's `create_subscription` takes a callback.
    pub fn create_polling_subscription_raw(
        &mut self,
        topic_name: &str,
        type_name: &str,
        type_hash: &str,
    ) -> Result<crate::executor::handles::RawSubscription, NodeError> {
        self.executor.with_node_try(self.node_id, |h| {
            h.create_subscription_raw(topic_name, type_name, type_hash)
        })
    }

    /// The POLLED form of [`NodeHandle::create_subscription_raw_sized`] (phase-483 W2): the caller
    /// takes from it, where rclrs's `create_subscription` takes a callback.
    pub fn create_polling_subscription_raw_sized<const RX_BUF: usize>(
        &mut self,
        topic_name: &str,
        type_name: &str,
        type_hash: &str,
    ) -> Result<crate::executor::handles::RawSubscription<RX_BUF>, NodeError> {
        self.executor.with_node_try(self.node_id, |h| {
            h.create_subscription_raw_sized::<RX_BUF>(topic_name, type_name, type_hash)
        })
    }

    /// The POLLED form of [`NodeHandle::create_service`] (phase-483 W2): the caller
    /// takes from it, where rclrs's `create_service` takes a callback.
    pub fn create_polling_service<Svc: RosService>(
        &mut self,
        service_name: &str,
    ) -> Result<EmbeddedServiceServer<Svc>, NodeError>
    where
        Svc::Request: MessageForRmw,
        Svc::Reply: MessageForRmw,
    {
        self.executor
            .with_node_try(self.node_id, |h| h.create_service::<Svc>(service_name))
    }

    /// The POLLED form of [`NodeHandle::create_service_with_qos`] (phase-483 W2): the caller
    /// takes from it, where rclrs's `create_service` takes a callback.
    pub fn create_polling_service_with_qos<Svc: RosService>(
        &mut self,
        service_name: &str,
        qos: QoSProfile,
    ) -> Result<EmbeddedServiceServer<Svc>, NodeError>
    where
        Svc::Request: MessageForRmw,
        Svc::Reply: MessageForRmw,
    {
        self.executor.with_node_try(self.node_id, |h| {
            h.create_service_with_qos::<Svc>(service_name, qos)
        })
    }

    /// The POLLED form of [`NodeHandle::create_service_sized`] (phase-483 W2): the caller
    /// takes from it, where rclrs's `create_service` takes a callback.
    pub fn create_polling_service_sized<
        Svc: RosService,
        const REQ_BUF: usize,
        const REPLY_BUF: usize,
    >(
        &mut self,
        service_name: &str,
        qos: QoSProfile,
    ) -> Result<EmbeddedServiceServer<Svc, REQ_BUF, REPLY_BUF>, NodeError>
    where
        Svc::Request: MessageForRmw,
        Svc::Reply: MessageForRmw,
    {
        self.executor.with_node_try(self.node_id, |h| {
            h.create_service_sized::<Svc, REQ_BUF, REPLY_BUF>(service_name, qos)
        })
    }

    /// [`NodeHandle::create_client`] on this node (phase-483 W2).
    ///
    /// phase-515 R1 — `options` is rclrs's `impl IntoPrimitiveOptions`; a bare
    /// name keeps the services profile. ADOPT-BOUNDED (client): returns an
    /// `EmbeddedServiceClient`, the executor-table handle, where rclrs returns
    /// a reference-counted `Client`.
    pub fn create_client<'o, Svc: RosService>(
        &mut self,
        options: impl crate::IntoPrimitiveOptions<'o>,
    ) -> Result<EmbeddedServiceClient<Svc>, NodeError>
    where
        Svc::Request: MessageForRmw,
        Svc::Reply: MessageForRmw,
    {
        let options = options.into_primitive_options();
        let qos = options.resolve(QoSProfile::services_default());
        self.executor.with_node_try(self.node_id, |h| {
            h.create_client_with_qos::<Svc>(options.name, qos)
        })
    }

    /// [`NodeHandle::create_client_with_qos`] on this node (phase-483 W2).
    pub fn create_client_with_qos<Svc: RosService>(
        &mut self,
        service_name: &str,
        qos: QoSProfile,
    ) -> Result<EmbeddedServiceClient<Svc>, NodeError>
    where
        Svc::Request: MessageForRmw,
        Svc::Reply: MessageForRmw,
    {
        self.executor.with_node_try(self.node_id, |h| {
            h.create_client_with_qos::<Svc>(service_name, qos)
        })
    }

    /// [`NodeHandle::create_client_sized`] on this node (phase-483 W2).
    pub fn create_client_sized<Svc: RosService, const REQ_BUF: usize, const REPLY_BUF: usize>(
        &mut self,
        service_name: &str,
        qos: QoSProfile,
    ) -> Result<EmbeddedServiceClient<Svc, REQ_BUF, REPLY_BUF>, NodeError>
    where
        Svc::Request: MessageForRmw,
        Svc::Reply: MessageForRmw,
    {
        self.executor.with_node_try(self.node_id, |h| {
            h.create_client_sized::<Svc, REQ_BUF, REPLY_BUF>(service_name, qos)
        })
    }

    /// The POLLED form of [`NodeHandle::create_service_raw`] (phase-483 W2): the caller
    /// takes from it, where rclrs's `create_service` takes a callback.
    pub fn create_polling_service_raw(
        &mut self,
        service_name: &str,
        type_name: &str,
        type_hash: &str,
    ) -> Result<crate::executor::handles::RawServiceServer, NodeError> {
        self.executor.with_node_try(self.node_id, |h| {
            h.create_service_raw(service_name, type_name, type_hash)
        })
    }

    /// The POLLED form of [`NodeHandle::create_service_raw_sized`] (phase-483 W2): the caller
    /// takes from it, where rclrs's `create_service` takes a callback.
    pub fn create_polling_service_raw_sized<const REQ_BUF: usize, const RESP_BUF: usize>(
        &mut self,
        service_name: &str,
        type_name: &str,
        type_hash: &str,
    ) -> Result<crate::executor::handles::RawServiceServer<REQ_BUF, RESP_BUF>, NodeError> {
        self.executor.with_node_try(self.node_id, |h| {
            h.create_service_raw_sized::<REQ_BUF, RESP_BUF>(service_name, type_name, type_hash)
        })
    }

    /// [`NodeHandle::create_client_raw`] on this node (phase-483 W2).
    pub fn create_client_raw(
        &mut self,
        service_name: &str,
        type_name: &str,
        type_hash: &str,
    ) -> Result<crate::executor::handles::RawServiceClient, NodeError> {
        self.executor.with_node_try(self.node_id, |h| {
            h.create_client_raw(service_name, type_name, type_hash)
        })
    }

    /// [`NodeHandle::create_client_raw_sized`] on this node (phase-483 W2).
    pub fn create_client_raw_sized<const REQ_BUF: usize, const REPLY_BUF: usize>(
        &mut self,
        service_name: &str,
        type_name: &str,
        type_hash: &str,
    ) -> Result<crate::executor::handles::RawServiceClient<REQ_BUF, REPLY_BUF>, NodeError> {
        self.executor.with_node_try(self.node_id, |h| {
            h.create_client_raw_sized::<REQ_BUF, REPLY_BUF>(service_name, type_name, type_hash)
        })
    }

    /// [`NodeHandle::create_action_server_raw`] on this node (phase-483 W2).
    pub fn create_action_server_raw(
        &mut self,
        action_name: &str,
        type_name: &str,
        type_hash: &str,
    ) -> Result<
        super::action_core::ActionServerCore<
            { crate::config::DEFAULT_RX_BUF_SIZE },
            { crate::config::DEFAULT_RX_BUF_SIZE },
            { crate::config::DEFAULT_RX_BUF_SIZE },
            4,
        >,
        NodeError,
    > {
        self.executor.with_node_try(self.node_id, |h| {
            h.create_action_server_raw(action_name, type_name, type_hash)
        })
    }

    /// [`NodeHandle::create_action_server_raw_sized`] on this node (phase-483 W2).
    pub fn create_action_server_raw_sized<
        const GOAL_BUF: usize,
        const RESULT_BUF: usize,
        const FEEDBACK_BUF: usize,
        const MAX_GOALS: usize,
    >(
        &mut self,
        action_name: &str,
        type_name: &str,
        type_hash: &str,
    ) -> Result<
        super::action_core::ActionServerCore<GOAL_BUF, RESULT_BUF, FEEDBACK_BUF, MAX_GOALS>,
        NodeError,
    > {
        self.executor.with_node_try(self.node_id, |h| {
            h.create_action_server_raw_sized::<GOAL_BUF, RESULT_BUF, FEEDBACK_BUF, MAX_GOALS>(
                action_name,
                type_name,
                type_hash,
            )
        })
    }

    /// [`NodeHandle::create_action_client_raw`] on this node (phase-483 W2).
    pub fn create_action_client_raw(
        &mut self,
        action_name: &str,
        type_name: &str,
        type_hash: &str,
    ) -> Result<
        super::action_core::ActionClientCore<
            { crate::config::DEFAULT_RX_BUF_SIZE },
            { crate::config::DEFAULT_RX_BUF_SIZE },
            { crate::config::DEFAULT_RX_BUF_SIZE },
        >,
        NodeError,
    > {
        self.executor.with_node_try(self.node_id, |h| {
            h.create_action_client_raw(action_name, type_name, type_hash)
        })
    }

    /// [`NodeHandle::create_action_client_raw_sized`] on this node (phase-483 W2).
    pub fn create_action_client_raw_sized<
        const GOAL_BUF: usize,
        const RESULT_BUF: usize,
        const FEEDBACK_BUF: usize,
    >(
        &mut self,
        action_name: &str,
        type_name: &str,
        type_hash: &str,
    ) -> Result<super::action_core::ActionClientCore<GOAL_BUF, RESULT_BUF, FEEDBACK_BUF>, NodeError>
    {
        self.executor.with_node_try(self.node_id, |h| {
            h.create_action_client_raw_sized::<GOAL_BUF, RESULT_BUF, FEEDBACK_BUF>(
                action_name,
                type_name,
                type_hash,
            )
        })
    }

    /// [`NodeHandle::create_action_server`] on this node (phase-483 W2).
    pub fn create_action_server<A: RosAction>(
        &mut self,
        action_name: &str,
    ) -> Result<ActionServer<A>, NodeError>
    where
        A::Goal: MessageForRmw,
        A::Result: MessageForRmw,
        A::Feedback: MessageForRmw,
        A::SendGoalRequest: MessageForRmw,
        A::SendGoalResponse: MessageForRmw,
        A::GetResultRequest: MessageForRmw,
        A::GetResultResponse: MessageForRmw,
        A::FeedbackMessage: MessageForRmw,
    {
        self.executor
            .with_node_try(self.node_id, |h| h.create_action_server::<A>(action_name))
    }

    /// [`NodeHandle::create_action_server_sized`] on this node (phase-483 W2).
    pub fn create_action_server_sized<
        A: RosAction,
        const GOAL_BUF: usize,
        const RESULT_BUF: usize,
        const FEEDBACK_BUF: usize,
        const MAX_GOALS: usize,
    >(
        &mut self,
        action_name: &str,
    ) -> Result<ActionServer<A, GOAL_BUF, RESULT_BUF, FEEDBACK_BUF, MAX_GOALS>, NodeError>
    where
        A::Goal: MessageForRmw,
        A::Result: MessageForRmw,
        A::Feedback: MessageForRmw,
        A::SendGoalRequest: MessageForRmw,
        A::SendGoalResponse: MessageForRmw,
        A::GetResultRequest: MessageForRmw,
        A::GetResultResponse: MessageForRmw,
        A::FeedbackMessage: MessageForRmw,
    {
        self.executor.with_node_try(self.node_id, |h| {
            h.create_action_server_sized::<A, GOAL_BUF, RESULT_BUF, FEEDBACK_BUF, MAX_GOALS>(
                action_name,
            )
        })
    }

    /// [`NodeHandle::create_action_client`] on this node (phase-483 W2).
    pub fn create_action_client<A: RosAction>(
        &mut self,
        action_name: &str,
    ) -> Result<ActionClient<A>, NodeError>
    where
        A::Goal: MessageForRmw,
        A::Result: MessageForRmw,
        A::Feedback: MessageForRmw,
        A::SendGoalRequest: MessageForRmw,
        A::SendGoalResponse: MessageForRmw,
        A::GetResultRequest: MessageForRmw,
        A::GetResultResponse: MessageForRmw,
        A::FeedbackMessage: MessageForRmw,
    {
        self.executor
            .with_node_try(self.node_id, |h| h.create_action_client::<A>(action_name))
    }

    /// [`NodeHandle::create_action_client_sized`] on this node (phase-483 W2).
    pub fn create_action_client_sized<
        A: RosAction,
        const GOAL_BUF: usize,
        const RESULT_BUF: usize,
        const FEEDBACK_BUF: usize,
    >(
        &mut self,
        action_name: &str,
    ) -> Result<ActionClient<A, GOAL_BUF, RESULT_BUF, FEEDBACK_BUF>, NodeError>
    where
        A::Goal: MessageForRmw,
        A::Result: MessageForRmw,
        A::Feedback: MessageForRmw,
        A::SendGoalRequest: MessageForRmw,
        A::SendGoalResponse: MessageForRmw,
        A::GetResultRequest: MessageForRmw,
        A::GetResultResponse: MessageForRmw,
        A::FeedbackMessage: MessageForRmw,
    {
        self.executor.with_node_try(self.node_id, |h| {
            h.create_action_client_sized::<A, GOAL_BUF, RESULT_BUF, FEEDBACK_BUF>(action_name)
        })
    }
}

/// Service-server builder on a [`Node`] — `node.service(name)`.
pub struct CtxServiceBuilder<'c, 'e, 't, 's> {
    ctx: &'c mut Node<'e, 's>,
    name: &'t str,
    qos: QoSProfile,
}

impl<'c, 'e, 't, 's> CtxServiceBuilder<'c, 'e, 't, 's> {
    /// Service QoS (applies to both the request + reply endpoints). Defaults to
    /// `QoSProfile::services_default()`.
    pub fn qos(mut self, qos: QoSProfile) -> Self {
        self.qos = qos;
        self
    }

    /// Surface the request's [`ServiceInfo`](nros_core::ServiceInfo) to the
    /// callback — `FnMut(&Request, &ServiceInfo) -> Reply` (issue 1942). The
    /// service-side twin of the subscription builder's `.message_info()`.
    pub fn service_info(self) -> CtxServiceInfoBuilder<'c, 'e, 't, 's> {
        CtxServiceInfoBuilder {
            ctx: self.ctx,
            name: self.name,
            qos: self.qos,
        }
    }

    pub fn build<Svc, F>(self, callback: F) -> Result<super::types::HandleId, NodeError>
    where
        Svc: RosService + 'static,
        Svc::Request: crate::rmw_type_registry::MessageForRmw,
        Svc::Reply: crate::rmw_type_registry::MessageForRmw,
        F: FnMut(&Svc::Request) -> Svc::Reply + 'static,
    {
        self.ctx.executor.register_service_sized_on::<
            Svc,
            F,
            { crate::config::DEFAULT_RX_BUF_SIZE },
            { crate::config::DEFAULT_RX_BUF_SIZE },
        >(self.ctx.node_id, self.name, self.qos, callback)
    }
}

/// Service-server builder with [`ServiceInfo`](nros_core::ServiceInfo)
/// surfaced (`node.service(name).service_info()`). Callback is
/// `FnMut(&Svc::Request, &ServiceInfo) -> Svc::Reply`.
pub struct CtxServiceInfoBuilder<'c, 'e, 't, 's> {
    ctx: &'c mut Node<'e, 's>,
    name: &'t str,
    qos: QoSProfile,
}

impl<'c, 'e, 't, 's> CtxServiceInfoBuilder<'c, 'e, 't, 's> {
    /// Service QoS (applies to both the request + reply endpoints). Defaults to
    /// `QoSProfile::services_default()`.
    pub fn qos(mut self, qos: QoSProfile) -> Self {
        self.qos = qos;
        self
    }

    pub fn build<Svc, F>(self, callback: F) -> Result<super::types::HandleId, NodeError>
    where
        Svc: RosService + 'static,
        Svc::Request: crate::rmw_type_registry::MessageForRmw,
        Svc::Reply: crate::rmw_type_registry::MessageForRmw,
        F: FnMut(&Svc::Request, &nros_core::ServiceInfo) -> Svc::Reply + 'static,
    {
        self.ctx.executor.register_service_with_info_sized_on::<
            Svc,
            F,
            { crate::config::DEFAULT_RX_BUF_SIZE },
            { crate::config::DEFAULT_RX_BUF_SIZE },
        >(self.ctx.node_id, self.name, self.qos, callback)
    }
}

/// Publisher builder on a [`Node`] — `node.publisher(topic)`.
pub struct CtxPublisherBuilder<'c, 'e, 't, 's> {
    ctx: &'c mut Node<'e, 's>,
    topic: &'t str,
    qos: QoSProfile,
}

impl<'c, 'e, 't, 's> CtxPublisherBuilder<'c, 'e, 't, 's> {
    pub fn qos(mut self, qos: QoSProfile) -> Self {
        self.qos = qos;
        self
    }

    /// Typed publisher for a ROS message `M`.
    pub fn typed<M: MessageForRmw>(self) -> CtxTypedPublisherBuilder<'c, 'e, 't, 's, M> {
        CtxTypedPublisherBuilder {
            ctx: self.ctx,
            topic: self.topic,
            qos: self.qos,
            _phantom: PhantomData,
        }
    }

    /// Generic (type-erased) publisher.
    pub fn generic(
        self,
        type_name: &'t str,
        type_hash: &'t str,
    ) -> CtxGenericPublisherBuilder<'c, 'e, 't, 's> {
        CtxGenericPublisherBuilder {
            ctx: self.ctx,
            topic: self.topic,
            type_name,
            type_hash,
            qos: self.qos,
        }
    }
}

/// Typed publisher builder on a `Node` (`.typed::<M>()`).
pub struct CtxTypedPublisherBuilder<'c, 'e, 't, 's, M> {
    ctx: &'c mut Node<'e, 's>,
    topic: &'t str,
    qos: QoSProfile,
    _phantom: PhantomData<M>,
}

impl<'c, 'e, 't, 's, M: MessageForRmw> CtxTypedPublisherBuilder<'c, 'e, 't, 's, M> {
    pub fn qos(mut self, qos: QoSProfile) -> Self {
        self.qos = qos;
        self
    }

    pub fn build(self) -> Result<EmbeddedPublisher<M>, NodeError> {
        self.ctx
            .executor
            .create_publisher_on::<M>(self.ctx.node_id, self.topic, self.qos)
    }
}

/// Generic publisher builder on a `Node` (`.generic(type, hash)`).
pub struct CtxGenericPublisherBuilder<'c, 'e, 't, 's> {
    ctx: &'c mut Node<'e, 's>,
    topic: &'t str,
    type_name: &'t str,
    type_hash: &'t str,
    qos: QoSProfile,
}

impl<'c, 'e, 't, 's> CtxGenericPublisherBuilder<'c, 'e, 't, 's> {
    pub fn qos(mut self, qos: QoSProfile) -> Self {
        self.qos = qos;
        self
    }

    pub fn build(self) -> Result<crate::executor::handles::EmbeddedRawPublisher, NodeError> {
        self.ctx.executor.create_publisher_raw_on(
            self.ctx.node_id,
            self.topic,
            self.type_name,
            self.type_hash,
            self.qos,
        )
    }
}

/// Subscription builder — `node.subscription(topic)`.
pub struct SubscriptionBuilder<'c, 'e, 't, 's> {
    ctx: &'c mut Node<'e, 's>,
    topic: &'t str,
    qos: QoSProfile,
}

impl<'c, 'e, 't, 's> SubscriptionBuilder<'c, 'e, 't, 's> {
    pub fn qos(mut self, qos: QoSProfile) -> Self {
        self.qos = qos;
        self
    }

    /// Typed subscription for a ROS message `M`.
    pub fn typed<M: MessageForRmw + 'static>(self) -> TypedSubscriptionBuilder<'c, 'e, 't, 's, M> {
        TypedSubscriptionBuilder {
            ctx: self.ctx,
            topic: self.topic,
            qos: self.qos,
            sched: None,
            rx_explicit: false,
            _phantom: PhantomData,
        }
    }

    /// Generic (type-erased) subscription — raw CDR bytes to the callback.
    pub fn generic(
        self,
        type_name: &'t str,
        type_hash: &'t str,
    ) -> GenericSubscriptionBuilder<'c, 'e, 't, 's> {
        GenericSubscriptionBuilder {
            ctx: self.ctx,
            topic: self.topic,
            type_name,
            type_hash,
            qos: self.qos,
            sched: None,
        }
    }
}

/// Typed subscription builder (`.typed::<M>()`). `RX` is the staging-buffer
/// size, set via `.rx_buffer::<N>()` (defaults to `DEFAULT_RX_BUF_SIZE`).
pub struct TypedSubscriptionBuilder<
    'c,
    'e,
    't,
    's,
    M,
    const RX: usize = { crate::config::DEFAULT_RX_BUF_SIZE },
> {
    ctx: &'c mut Node<'e, 's>,
    topic: &'t str,
    qos: QoSProfile,
    sched: Option<super::sched_context::SchedContextId>,
    /// phase-392 W3c -- did the CONSUMER name `RX`, or is it just the default?
    ///
    /// The two are the same const generic and mean opposite things now: an
    /// unnamed `RX` is a ceiling the derivation clamps to, a named one is the
    /// number the consumer asked for. `.rx_buffer::<N>()` is the only thing that
    /// sets this, so the flag cannot drift from "somebody wrote a number".
    rx_explicit: bool,
    _phantom: PhantomData<M>,
}

impl<'c, 'e, 't, 's, M: MessageForRmw + 'static, const RX: usize>
    TypedSubscriptionBuilder<'c, 'e, 't, 's, M, RX>
{
    pub fn qos(mut self, qos: QoSProfile) -> Self {
        self.qos = qos;
        self
    }

    /// Bind the subscription's callback to a scheduling context.
    pub fn sched_context(mut self, sc: super::sched_context::SchedContextId) -> Self {
        self.sched = Some(sc);
        self
    }

    /// Set the staging-buffer size (const-generic).
    ///
    /// phase-392 W3c -- **this is the opt-out.** A subscription that does not
    /// call it is sized from `M`'s own serialized bound; calling it says "give
    /// me exactly `N` bytes per slot" and the derivation is skipped. Naming a
    /// number is the only way to spend more arena than the type needs, which is
    /// the direction that should require saying so.
    ///
    /// It stays the way to raise a buffer above the type's bound -- for a peer
    /// that pads, or a type whose `.msg` bound is looser than what it really
    /// sends -- and `.rx_buffer_from_type()` after it re-enters the derivation
    /// with `N` as the ceiling.
    pub fn rx_buffer<const N: usize>(self) -> TypedSubscriptionBuilder<'c, 'e, 't, 's, M, N> {
        TypedSubscriptionBuilder {
            ctx: self.ctx,
            topic: self.topic,
            qos: self.qos,
            sched: self.sched,
            rx_explicit: true,
            _phantom: PhantomData,
        }
    }

    /// Phase 403 W2 -- size the receive buffer from `M`'s OWN serialized bound
    /// instead of the configured default.
    ///
    /// ```ignore
    /// node.subscription("/chatter")
    ///     .typed::<Int32>()
    ///     .rx_buffer_from_type()   // 12 bytes per slot, not 1024
    ///     .build(on_msg)?;
    /// ```
    ///
    /// **No longer the only way to get it (phase-392 W3c).** This was opt-in,
    /// and its own doc said making it the default "would move every image's
    /// arena occupancy at once, which is a decision, not a refactor". W3c is
    /// that decision: a plain `.typed::<M>().build()` now derives too, wherever
    /// the bound is reachable, and `.rx_buffer::<N>()` is the opt-out.
    ///
    /// Two jobs are left, and they are why this method stays:
    ///
    /// * **It REFUSES an unbounded `M` at build time.** The default path cannot
    ///   -- erroring there would make a bound mandatory for every message type
    ///   in every image at once -- so it keeps `RX` for a type with no bound.
    ///   Calling this says the bound must exist, and the `const` assert below
    ///   is the error that says so.
    /// * **It reaches the derivation where `MessageForRmw` carries no schema.**
    ///   Only a backend declaring `type-descriptors` makes the type-erased
    ///   default site able to read a bound at all (see
    ///   [`default_subscription_rx_bytes`](crate::rmw_type_registry::default_subscription_rx_bytes));
    ///   this method names `M: nros_serdes::schema::Message` itself, so it works
    ///   on every backend.
    ///
    /// **What it costs and what it buys.** The buffered path stores `depth <= 1
    /// ? 3 : depth+1` slots of this size in the executor arena, so a bounded
    /// type recovers `slots * (default - bound)` bytes per subscription. The
    /// number is the LARGER of the type's XCDR1 and XCDR2 bounds, because the
    /// stack writes XCDR1 but a peer may send XCDR2 (see
    /// [`subscription_rx_bytes`](crate::rmw_type_registry::subscription_rx_bytes)).
    ///
    /// **An unbounded `M` fails the BUILD.** Every message type is required to
    /// carry a bound -- stated in the `.msg` (`string<=64`) or capped in
    /// `nros-codegen.toml` -- so "this type has no bound" is a defect to report,
    /// not a case to absorb. The alternative, quietly keeping the configured
    /// default, is precisely the substitution phase 380 forbids: `None` means
    /// "no bound EXISTS", never "unknown", and a buffer sized from a fallback is
    /// the failure that rule was written to prevent. The C header already refuses
    /// the same thing the same way -- naming an unbounded type's
    /// `{PREFIX}_RX_MAX_SERIALIZED_SIZE` expands to a deliberate compile error
    /// carrying the member that costs the bound (`unbounded_token`) -- and this
    /// is that refusal on the Rust path. rustc names `M` in the instantiation
    /// note; the member comes from the codegen diagnostic for the same type.
    ///
    /// **The bound never GROWS the buffer.** A type larger than `RX` keeps `RX`;
    /// spending unbudgeted arena silently is the worse failure, and the
    /// too-small case already reports itself (`report_dropped_take`).
    ///
    /// Returns a builder without `.message_info()` / `.safety()` on purpose:
    /// those entries hold a real `[u8; RX]` array, so their size can only come
    /// from a const generic -- `nros::rx_buffer_for!(M)` at the call site -- and
    /// carrying the flag into them would silently drop it.
    pub fn rx_buffer_from_type(self) -> TypedSubBoundBuilder<'c, 'e, 't, 's, M, RX>
    where
        M: nros_serdes::schema::Message,
    {
        // The diagnostic lives HERE, at the opt-in, rather than in `build()`:
        // this is the call that asserts the type has a bound, so it is the call
        // the error should point at.
        const {
            assert!(
                crate::rmw_type_registry::subscription_rx_bytes::<M>(RX).is_some(),
                "this message type has NO maximum serialized size, so its receive \
                 buffer cannot be sized from it. Every message type must carry a \
                 bound: give the unbounded member one in the `.msg` \
                 (`string<=64`, `int32[<=8]`) or a `cap` in `nros-codegen.toml`. \
                 The generated C header for this type names the member that costs \
                 it the bound (`NROS_UNBOUNDED__<type>__field_<member>`)."
            )
        }
        TypedSubBoundBuilder {
            ctx: self.ctx,
            topic: self.topic,
            qos: self.qos,
            sched: self.sched,
            _phantom: PhantomData,
        }
    }

    /// Surface per-message [`MessageInfo`](nros_core::MessageInfo) (seq,
    /// publisher GID, timestamps) to the callback — `FnMut(&M, Option<&MessageInfo>)`,
    /// the rclrs shape. Distinct from the generic builder's `.message_info()`
    /// (which yields a `RawMessageInfo` with the wire attachment).
    pub fn message_info(self) -> TypedSubInfoBuilder<'c, 'e, 't, 's, M, RX> {
        TypedSubInfoBuilder {
            ctx: self.ctx,
            topic: self.topic,
            qos: self.qos,
            sched: self.sched,
            rx_explicit: self.rx_explicit,
            _phantom: PhantomData,
        }
    }

    /// Surface E2E-safety validation (CRC + sequence gap/duplicate) to the
    /// callback — `FnMut(&M, &IntegrityStatus)`.
    #[cfg(feature = "safety-e2e")]
    pub fn safety(self) -> TypedSubSafetyBuilder<'c, 'e, 't, 's, M, RX> {
        TypedSubSafetyBuilder {
            ctx: self.ctx,
            topic: self.topic,
            qos: self.qos,
            sched: self.sched,
            rx_explicit: self.rx_explicit,
            _phantom: PhantomData,
        }
    }

    pub fn build<F: FnMut(&M) + 'static>(
        self,
        callback: F,
    ) -> Result<super::types::HandleId, NodeError> {
        let handle = self
            .ctx
            .executor
            .register_subscription_buffered_on::<M, F, RX>(
                self.ctx.node_id,
                self.topic,
                self.qos,
                callback,
                None, // group threaded via create_subscription_in_group; builder uses sched override
                // phase-392 W3c -- `None` asks the choke point to DERIVE from
                // `M`; `Some(RX)` is the consumer's own number, and only
                // `.rx_buffer::<N>()` sets that flag.
                if self.rx_explicit { Some(RX) } else { None },
            )?;
        if let Some(sc) = self.sched {
            self.ctx.executor.bind_handle_to_sched_context(handle, sc)?;
        }
        Ok(handle)
    }
}

/// Phase 403 W2 -- a typed subscription whose buffer is sized by `M`'s own
/// serialized bound (`.typed::<M>().rx_buffer_from_type()`).
///
/// Terminal by construction: `.message_info()` and `.safety()` are absent
/// because their entries store `[u8; RX]` and cannot take a runtime size. `RX`
/// survives as the CEILING -- an unbounded `M`, or one whose bound exceeds `RX`,
/// gets `RX` unchanged.
pub struct TypedSubBoundBuilder<
    'c,
    'e,
    't,
    's,
    M,
    const RX: usize = { crate::config::DEFAULT_RX_BUF_SIZE },
> {
    ctx: &'c mut Node<'e, 's>,
    topic: &'t str,
    qos: QoSProfile,
    sched: Option<super::sched_context::SchedContextId>,
    _phantom: PhantomData<M>,
}

impl<'c, 'e, 't, 's, M: MessageForRmw + nros_serdes::schema::Message + 'static, const RX: usize>
    TypedSubBoundBuilder<'c, 'e, 't, 's, M, RX>
{
    pub fn qos(mut self, qos: QoSProfile) -> Self {
        self.qos = qos;
        self
    }

    /// Bind the subscription's callback to a scheduling context.
    pub fn sched_context(mut self, sc: super::sched_context::SchedContextId) -> Self {
        self.sched = Some(sc);
        self
    }

    pub fn build<F: FnMut(&M) + 'static>(
        self,
        callback: F,
    ) -> Result<super::types::HandleId, NodeError> {
        // Unreachable: `rx_buffer_from_type` is the only constructor of this
        // builder and its `const` assert already refused an unbounded `M`.
        let rx_bytes = crate::rmw_type_registry::subscription_rx_bytes::<M>(RX)
            .expect("rx_buffer_from_type asserts the bound exists at build time");
        let handle = self
            .ctx
            .executor
            .register_subscription_buffered_on::<M, F, RX>(
                self.ctx.node_id,
                self.topic,
                self.qos,
                callback,
                None,
                Some(rx_bytes),
            )?;
        if let Some(sc) = self.sched {
            self.ctx.executor.bind_handle_to_sched_context(handle, sc)?;
        }
        Ok(handle)
    }
}

/// Typed subscription builder with `MessageInfo` (`.typed::<M>().message_info()`).
/// Callback is `FnMut(&M, Option<&MessageInfo>)`.
pub struct TypedSubInfoBuilder<
    'c,
    'e,
    't,
    's,
    M,
    const RX: usize = { crate::config::DEFAULT_RX_BUF_SIZE },
> {
    ctx: &'c mut Node<'e, 's>,
    topic: &'t str,
    qos: QoSProfile,
    sched: Option<super::sched_context::SchedContextId>,
    /// Issue 1860 -- did the consumer NAME `RX` (`.rx_buffer::<N>()`), or is it
    /// the default the type's bound is clamped to? As `TypedSubscriptionBuilder`.
    rx_explicit: bool,
    _phantom: PhantomData<M>,
}

impl<'c, 'e, 't, 's, M: MessageForRmw + 'static, const RX: usize>
    TypedSubInfoBuilder<'c, 'e, 't, 's, M, RX>
{
    pub fn qos(mut self, qos: QoSProfile) -> Self {
        self.qos = qos;
        self
    }

    pub fn sched_context(mut self, sc: super::sched_context::SchedContextId) -> Self {
        self.sched = Some(sc);
        self
    }

    pub fn rx_buffer<const N: usize>(self) -> TypedSubInfoBuilder<'c, 'e, 't, 's, M, N> {
        TypedSubInfoBuilder {
            ctx: self.ctx,
            topic: self.topic,
            qos: self.qos,
            sched: self.sched,
            rx_explicit: true,
            _phantom: PhantomData,
        }
    }

    pub fn build<F: FnMut(&M, Option<&nros_core::MessageInfo>) + 'static>(
        self,
        callback: F,
    ) -> Result<super::types::HandleId, NodeError> {
        let handle = self
            .ctx
            .executor
            .register_subscription_with_info_sized_inner::<M, F, RX>(
                Some(self.ctx.node_id),
                self.topic,
                self.qos,
                callback,
                if self.rx_explicit { Some(RX) } else { None },
            )?;
        if let Some(sc) = self.sched {
            self.ctx.executor.bind_handle_to_sched_context(handle, sc)?;
        }
        Ok(handle)
    }
}

/// Typed subscription builder with E2E-safety validation
/// (`.typed::<M>().safety()`). Callback is `FnMut(&M, &IntegrityStatus)`.
#[cfg(feature = "safety-e2e")]
pub struct TypedSubSafetyBuilder<
    'c,
    'e,
    't,
    's,
    M,
    const RX: usize = { crate::config::DEFAULT_RX_BUF_SIZE },
> {
    ctx: &'c mut Node<'e, 's>,
    topic: &'t str,
    qos: QoSProfile,
    sched: Option<super::sched_context::SchedContextId>,
    /// Issue 1860 -- did the consumer NAME `RX` (`.rx_buffer::<N>()`), or is it
    /// the default the type's bound is clamped to? As `TypedSubscriptionBuilder`.
    rx_explicit: bool,
    _phantom: PhantomData<M>,
}

#[cfg(feature = "safety-e2e")]
impl<'c, 'e, 't, 's, M: MessageForRmw + 'static, const RX: usize>
    TypedSubSafetyBuilder<'c, 'e, 't, 's, M, RX>
{
    pub fn qos(mut self, qos: QoSProfile) -> Self {
        self.qos = qos;
        self
    }

    pub fn sched_context(mut self, sc: super::sched_context::SchedContextId) -> Self {
        self.sched = Some(sc);
        self
    }

    pub fn rx_buffer<const N: usize>(self) -> TypedSubSafetyBuilder<'c, 'e, 't, 's, M, N> {
        TypedSubSafetyBuilder {
            ctx: self.ctx,
            topic: self.topic,
            qos: self.qos,
            sched: self.sched,
            rx_explicit: true,
            _phantom: PhantomData,
        }
    }

    pub fn build<F: FnMut(&M, &nros_rmw::IntegrityStatus) + 'static>(
        self,
        callback: F,
    ) -> Result<super::types::HandleId, NodeError> {
        let handle = self
            .ctx
            .executor
            .register_subscription_with_safety_sized_inner::<M, F, RX>(
                Some(self.ctx.node_id),
                self.topic,
                self.qos,
                callback,
                if self.rx_explicit { Some(RX) } else { None },
            )?;
        if let Some(sc) = self.sched {
            self.ctx.executor.bind_handle_to_sched_context(handle, sc)?;
        }
        Ok(handle)
    }
}

/// Generic (type-erased) subscription builder (`.generic(type, hash)`).
pub struct GenericSubscriptionBuilder<
    'c,
    'e,
    't,
    's,
    const RX: usize = { crate::config::DEFAULT_RX_BUF_SIZE },
> {
    ctx: &'c mut Node<'e, 's>,
    topic: &'t str,
    type_name: &'t str,
    type_hash: &'t str,
    qos: QoSProfile,
    sched: Option<super::sched_context::SchedContextId>,
}

impl<'c, 'e, 't, 's, const RX: usize> GenericSubscriptionBuilder<'c, 'e, 't, 's, RX> {
    pub fn qos(mut self, qos: QoSProfile) -> Self {
        self.qos = qos;
        self
    }

    pub fn sched_context(mut self, sc: super::sched_context::SchedContextId) -> Self {
        self.sched = Some(sc);
        self
    }

    pub fn rx_buffer<const N: usize>(self) -> GenericSubscriptionBuilder<'c, 'e, 't, 's, N> {
        GenericSubscriptionBuilder {
            ctx: self.ctx,
            topic: self.topic,
            type_name: self.type_name,
            type_hash: self.type_hash,
            qos: self.qos,
            sched: self.sched,
        }
    }

    /// Surface the sample's wire attachment + metadata to the callback
    /// (`FnMut(&[u8], &RawMessageInfo)`). The cross-RMW bridge reads the
    /// bridge_origin tag from `info.attachment()` for echo suppression.
    pub fn message_info(self) -> GenericSubInfoBuilder<'c, 'e, 't, 's, RX> {
        GenericSubInfoBuilder {
            ctx: self.ctx,
            topic: self.topic,
            type_name: self.type_name,
            type_hash: self.type_hash,
            qos: self.qos,
            sched: self.sched,
        }
    }

    pub fn build<F: FnMut(&[u8]) + 'static>(
        self,
        callback: F,
    ) -> Result<super::types::HandleId, NodeError> {
        let handle = self
            .ctx
            .executor
            .register_subscription_buffered_raw_on::<F, RX>(
                self.ctx.node_id,
                self.topic,
                self.type_name,
                self.type_hash,
                self.qos,
                None,
                callback,
            )?;
        if let Some(sc) = self.sched {
            self.ctx.executor.bind_handle_to_sched_context(handle, sc)?;
        }
        Ok(handle)
    }
}

/// Generic subscription builder with `MessageInfo` surfaced
/// (`.message_info()`). Callback is `FnMut(&[u8], &RawMessageInfo)`.
pub struct GenericSubInfoBuilder<
    'c,
    'e,
    't,
    's,
    const RX: usize = { crate::config::DEFAULT_RX_BUF_SIZE },
> {
    ctx: &'c mut Node<'e, 's>,
    topic: &'t str,
    type_name: &'t str,
    type_hash: &'t str,
    qos: QoSProfile,
    sched: Option<super::sched_context::SchedContextId>,
}

impl<'c, 'e, 't, 's, const RX: usize> GenericSubInfoBuilder<'c, 'e, 't, 's, RX> {
    pub fn qos(mut self, qos: QoSProfile) -> Self {
        self.qos = qos;
        self
    }

    pub fn sched_context(mut self, sc: super::sched_context::SchedContextId) -> Self {
        self.sched = Some(sc);
        self
    }

    pub fn rx_buffer<const N: usize>(self) -> GenericSubInfoBuilder<'c, 'e, 't, 's, N> {
        GenericSubInfoBuilder {
            ctx: self.ctx,
            topic: self.topic,
            type_name: self.type_name,
            type_hash: self.type_hash,
            qos: self.qos,
            sched: self.sched,
        }
    }

    pub fn build<F: FnMut(&[u8], &nros_core::RawMessageInfo) + 'static>(
        self,
        callback: F,
    ) -> Result<super::types::HandleId, NodeError> {
        let handle = self
            .ctx
            .executor
            .register_subscription_buffered_raw_info_on::<F, RX>(
                self.ctx.node_id,
                self.topic,
                self.type_name,
                self.type_hash,
                self.qos,
                callback,
            )?;
        if let Some(sc) = self.sched {
            self.ctx.executor.bind_handle_to_sched_context(handle, sc)?;
        }
        Ok(handle)
    }
}

// `not(feature = "rmw-cffi")` — these tests use the `mock` backend
// (`crate::mock`, itself `cfg(all(test, not(rmw-cffi)))`); a workspace test
// build that unifies `rmw-cffi` on swaps `ConcreteSession` to the cffi session
// and drops `mock`, so the module must drop with it (matches the
// `mock_integration` gate in lifecycle_services.rs).
#[cfg(all(test, feature = "alloc", not(feature = "rmw-cffi")))]
mod builder_tests {
    use super::*;
    use crate::{executor::Executor, mock::MockSession};
    use nros_core::{CdrReader, CdrWriter, DeserError, Deserialize, SerError, Serialize};

    struct TestMsg;
    impl RosMessage for TestMsg {
        const TYPE_NAME: &'static str = "test/msg/TestMsg";
        const TYPE_HASH: &'static str = "test_hash";
    }
    impl Serialize for TestMsg {
        fn serialize(&self, _w: &mut CdrWriter) -> Result<(), SerError> {
            Ok(())
        }
    }
    impl Deserialize for TestMsg {
        fn deserialize(_r: &mut CdrReader) -> Result<Self, DeserError> {
            Ok(Self)
        }
    }
    // Phase 212.K.7.6.b — minimal single-field `Message` impl so
    // `TypedPublisherBuilder::build` resolves under the cyclonedds-tightened
    // bound AND the runtime register call succeeds. `DescriptorBuilder`
    // rejects empty `FIELDS` with `BuildError::EmptySchema`; pretend
    // there's one byte so the bridge stub returns a non-NULL pointer.
    // phase-380 W4 — was `#[cfg(rmw_needs_type_descriptors)]`: the schema is
    // now required by `MessageForRmw` on EVERY backend, because that is where
    // a subscription's build-time size bound comes from.
    impl nros_serdes::schema::Message for TestMsg {
        const TYPE_NAME: &'static str = "test/msg/TestMsg";
        const FIELDS: &'static [nros_serdes::schema::Field] = &[nros_serdes::schema::Field {
            name: "data",
            ty: nros_serdes::schema::FieldType::Uint8,
            offset: 0,
        }];
    }

    fn s(v: &str) -> heapless::String<64> {
        heapless::String::try_from(v).unwrap()
    }

    #[test]
    fn publisher_builder_typed_and_generic() {
        let mut session = MockSession::new();
        let mut node = NodeHandle::new(
            s("n"),
            s("/"),
            &mut session,
            crate::executor::spin::RemapView {
                launch: &[],
                runtime: &[],
            },
            0,
        );

        // typed: node.publisher(t).typed::<M>().qos(..).build()
        let _typed = node
            .publisher("/chatter")
            .typed::<TestMsg>()
            .qos(QoSProfile::default().keep_last(5))
            .build()
            .expect("typed publisher builds");

        // generic: node.publisher(t).qos(..).generic(type, hash).build()
        let _generic = node
            .publisher("/chatter")
            .qos(QoSProfile::default())
            .generic("std_msgs/msg/Int32", "hash")
            .build()
            .expect("generic publisher builds");
    }

    // The handle `Executor::create_node` returns resolves through the same two
    // tiers. Asserted on the name the BACKEND was handed (`MockPublisher`
    // records it), not on a helper's return value.

    fn argv_rule(from: &'static str, to: &'static str) -> crate::ros_args::RemapArg<'static> {
        crate::ros_args::RemapArg {
            node: None,
            from,
            to,
        }
    }

    #[test]
    fn create_node_handle_applies_an_argv_remap() {
        let mut exec: Executor = Executor::from_session(MockSession::new());
        crate::ros_args::install_argv_remaps(&mut exec, [argv_rule("chatter", "/from_argv")])
            .unwrap();
        let mut node = exec.create_node("talker").expect("node");
        let p = node
            .create_publisher::<TestMsg>("chatter")
            .expect("publisher");
        assert_eq!(p.handle.topic_name(), "/from_argv");
    }

    #[test]
    fn create_node_handle_prefers_a_launch_rule_over_argv() {
        let mut exec: Executor = Executor::from_session(MockSession::new());
        crate::ros_args::install_argv_remaps(&mut exec, [argv_rule("chatter", "/from_argv")])
            .unwrap();
        exec.declare_remap("talker", "/", "chatter", "/from_launch")
            .unwrap();
        let mut node = exec.create_node("talker").expect("node");
        let p = node
            .create_publisher::<TestMsg>("chatter")
            .expect("publisher");
        assert_eq!(p.handle.topic_name(), "/from_launch");
    }

    /// Issue 1971 -- the handle reads the flash launch table too, above argv.
    #[test]
    fn create_node_handle_prefers_a_flash_launch_rule_over_argv() {
        static LAUNCH: [crate::executor::LaunchRemap; 1] = [crate::executor::LaunchRemap::new(
            c"talker",
            c"/",
            c"chatter",
            c"/from_flash",
        )];
        let mut exec: Executor = Executor::from_session(MockSession::new());
        crate::ros_args::install_argv_remaps(&mut exec, [argv_rule("chatter", "/from_argv")])
            .unwrap();
        exec.set_launch_remaps(&LAUNCH);
        let mut node = exec.create_node("talker").expect("node");
        let p = node
            .create_publisher::<TestMsg>("chatter")
            .expect("publisher");
        assert_eq!(p.handle.topic_name(), "/from_flash");
    }

    #[test]
    fn create_node_handle_passes_an_unmatched_name_through_unchanged() {
        // No rule matches, so the handle hands the backend exactly what it was
        // given — NOT the expanded `/talker_ns/chatter`. This handle has never
        // expanded names, and a remap that does not fire must not start.
        let mut exec: Executor = Executor::from_session(MockSession::new());
        crate::ros_args::install_argv_remaps(&mut exec, [argv_rule("other", "/x")]).unwrap();
        let mut node = exec.create_node("talker").expect("node");
        let p = node
            .create_publisher::<TestMsg>("chatter")
            .expect("publisher");
        assert_eq!(p.handle.topic_name(), "chatter");
    }

    #[test]
    fn subscription_builder_and_convenient() {
        let mut exec: Executor = Executor::from_session(MockSession::new());
        let id = exec.node_builder("n").build().expect("node");

        // builder: typed
        let _h = exec
            .node_mut(id)
            .subscription("/chatter")
            .typed::<TestMsg>()
            .qos(QoSProfile::default().keep_last(5))
            .build(|_m: &TestMsg| {})
            .expect("typed subscription builds");

        // builder: generic (raw bytes)
        let _g = exec
            .node_mut(id)
            .subscription("/raw")
            .generic("std_msgs/msg/Int32", "hash")
            .build(|_b: &[u8]| {})
            .expect("generic subscription builds");

        // builder: sized + sched-context (slice 3 knobs)
        let sc = exec.default_sched_context_id();
        let _s = exec
            .node_mut(id)
            .subscription("/sized")
            .typed::<TestMsg>()
            .rx_buffer::<64>()
            .sched_context(sc)
            .build(|_m: &TestMsg| {})
            .expect("sized + sched subscription builds");

        // convenient (fork tier) — one node-ctx at a time, re-acquired
        let _c = exec
            .node_mut(id)
            .create_subscription::<TestMsg, _>("/conv", |_m: &TestMsg| {})
            .expect("convenient typed subscription builds");
    }

    #[test]
    fn generic_message_info_builder() {
        // slice 3b — the bridge echo path: generic sub whose callback
        // receives the wire attachment via RawMessageInfo.
        let mut exec: Executor = Executor::from_session(MockSession::new());
        let id = exec.node_builder("n").build().expect("node");

        let _i = exec
            .node_mut(id)
            .subscription("/info")
            .generic("std_msgs/msg/Int32", "hash")
            .message_info()
            .rx_buffer::<256>()
            .build(|_payload: &[u8], info: &nros_core::RawMessageInfo| {
                let _ = info.attachment();
            })
            .expect("generic + message_info subscription builds");
    }

    #[test]
    fn typed_message_info_builder() {
        // M2.a — typed .message_info() (rclrs shape FnMut(&M, Option<&MessageInfo>)),
        // replacing register_subscription_with_info.
        let mut exec: Executor = Executor::from_session(MockSession::new());
        let id = exec.node_builder("n").build().expect("node");
        let _h = exec
            .node_mut(id)
            .subscription("/chatter")
            .typed::<TestMsg>()
            .qos(QoSProfile::default().keep_last(5))
            .message_info()
            .build(|_m: &TestMsg, _info: Option<&nros_core::MessageInfo>| {})
            .expect("typed + message_info subscription builds");
    }

    #[cfg(feature = "safety-e2e")]
    #[test]
    fn typed_safety_builder() {
        // M2.a — typed .safety(), replacing register_subscription_with_safety.
        let mut exec: Executor = Executor::from_session(MockSession::new());
        let id = exec.node_builder("n").build().expect("node");
        let _h = exec
            .node_mut(id)
            .subscription("/chatter")
            .typed::<TestMsg>()
            .safety()
            .build(|_m: &TestMsg, _status: &nros_rmw::IntegrityStatus| {})
            .expect("typed + safety subscription builds");
    }

    #[test]
    fn generator_emitted_chain_compiles() {
        // Locks the exact builder chain the orchestration generator emits
        // for a subscriber (replaces register_subscription_raw_with_qos_sized_on).
        let mut exec: Executor = Executor::from_session(MockSession::new());
        let id = exec.node_builder("n").build().expect("node");
        let _h = exec
            .node_mut(id)
            .subscription("/topic")
            .generic("std_msgs/msg/Int32", "hash")
            .qos(QoSProfile::default().keep_last(1))
            .rx_buffer::<1024>()
            .build(|_data: &[u8]| {})
            .expect("generator-shape subscription builds");
    }

    #[test]
    fn nodectx_publisher_and_bridge_shape() {
        // Node publisher symmetry + the bridge two-ctx borrow pattern:
        // build the dest publisher on one Node (dropped), then register
        // the source subscription on another — the owned publisher outlives.
        let mut exec: Executor = Executor::from_session(MockSession::new());
        let id = exec.node_builder("n").build().expect("node");

        // convenient + builder publisher on Node
        let _p = exec
            .node_mut(id)
            .create_publisher::<TestMsg>("/p")
            .expect("ctx convenient publisher");
        let dest_pub = exec
            .node_mut(id)
            .publisher("/fwd")
            .generic("std_msgs/msg/Int32", "hash")
            .build()
            .expect("ctx generic publisher builds"); // Node dropped here

        // re-borrow exec for the source sub; closure owns dest_pub
        let _s = exec
            .node_mut(id)
            .subscription("/src")
            .generic("std_msgs/msg/Int32", "hash")
            .message_info()
            .build(move |payload: &[u8], _info: &nros_core::RawMessageInfo| {
                let _ = dest_pub.publish_serialized(payload);
            })
            .expect("bridge-shape source subscription builds");
    }
}