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, QosSettings, 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,
};

// ============================================================================
// 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,
    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
    /// `QosSettings` 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,
        domain_id: u32,
    ) -> Self {
        Self {
            name,
            namespace,
            session,
            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
    }

    /// Phase 88.12 — return the [`nros_log::Logger`] keyed on the
    /// node name.
    ///
    /// 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. Otherwise the call returns
    /// [`nros_log::DEFAULT_LOGGER`], keeping the API total.
    ///
    /// ```ignore
    /// // Pre-register if you want a dedicated threshold:
    /// 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::nros_info!(logger, "started; domain = {}", node.domain_id());
    /// ```
    #[must_use]
    pub fn logger(&self) -> &'static nros_log::Logger {
        nros_log::get_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
    // `with_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)
    }

    // -- 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, QosSettings::default())
    }

    /// Create a publisher with custom QoS settings.
    pub fn create_publisher_with_qos<M: MessageForRmw>(
        &mut self,
        topic_name: &str,
        qos: QosSettings,
    ) -> Result<EmbeddedPublisher<M>, NodeError> {
        // 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.
        qos.validate_against(nros_rmw::Session::supported_qos_policies(self.session))
            .map_err(NodeError::Transport)?;
        let topic = Self::topic_info(
            self.domain_id,
            &self.name,
            &self.namespace,
            topic_name,
            <M as RosMessage>::TYPE_NAME,
            <M as RosMessage>::TYPE_HASH,
        );
        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,
            _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, QosSettings::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: QosSettings,
    ) -> 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,
        );
        qos.validate_against(nros_rmw::Session::supported_qos_policies(self.session))
            .map_err(NodeError::Transport)?;
        let topic = Self::topic_info(
            self.domain_id,
            &self.name,
            &self.namespace,
            topic_name,
            type_name,
            type_hash,
        );
        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(),
        })
    }

    /// 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: QosSettings::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, QosSettings::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: QosSettings,
    ) -> Result<Subscription<M, RX_BUF>, NodeError> {
        // 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,
        );
        qos.validate_against(nros_rmw::Session::supported_qos_policies(self.session))
            .map_err(NodeError::Transport)?;
        let topic = Self::topic_info(
            self.domain_id,
            &self.name,
            &self.namespace,
            topic_name,
            <M as RosMessage>::TYPE_NAME,
            <M as RosMessage>::TYPE_HASH,
        );
        let handle = self
            .session
            .create_subscriber(&topic, qos)
            .map_err(|_| NodeError::Transport(TransportError::SubscriberCreationFailed))?;
        // W3b.5 — attach the contracted endpoint's age cell (stamped
        // types only; needs an epoch source).
        let age_mon = match (<M as RosMessage>::STAMP_OFFSET, self.epoch_us_fn) {
            (Some(_), Some(epoch)) => self
                .age_monitors
                .iter()
                .find(|a| a.topic == topic_name)
                .map(|a| (a.cell, epoch)),
            _ => 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 = QosSettings::default().apply_overrides(
            topic_name,
            nros_rmw::QosOverrideRole::Subscription,
            self.qos_overrides,
        );
        qos.validate_against(nros_rmw::Session::supported_qos_policies(self.session))
            .map_err(NodeError::Transport)?;
        let topic = Self::topic_info(
            self.domain_id,
            &self.name,
            &self.namespace,
            topic_name,
            type_name,
            type_hash,
        );
        let handle = self
            .session
            .create_subscriber(&topic, qos)
            .map_err(|_| NodeError::Transport(TransportError::SubscriberCreationFailed))?;
        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, QosSettings::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: QosSettings,
    ) -> 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: QosSettings,
    ) -> 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.
        qos.validate_against(nros_rmw::Session::supported_qos_policies(self.session))
            .map_err(NodeError::Transport)?;
        let info = Self::service_info(
            self.domain_id,
            &self.name,
            &self.namespace,
            service_name,
            Svc::SERVICE_NAME,
            Svc::SERVICE_HASH,
        );
        let handle = self
            .session
            .create_service_server(&info, qos)
            .map_err(|_| NodeError::Transport(TransportError::ServiceServerCreationFailed))?;
        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, QosSettings::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: QosSettings,
    ) -> 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: QosSettings,
    ) -> 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.
        qos.validate_against(nros_rmw::Session::supported_qos_policies(self.session))
            .map_err(NodeError::Transport)?;
        let info = Self::service_info(
            self.domain_id,
            &self.name,
            &self.namespace,
            service_name,
            Svc::SERVICE_NAME,
            Svc::SERVICE_HASH,
        );
        let handle = self
            .session
            .create_service_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,
            _phantom: PhantomData,
        })
    }

    /// Typeless service server. L1 counterpart of [`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 info = Self::service_info(
            self.domain_id,
            &self.name,
            &self.namespace,
            service_name,
            type_name,
            type_hash,
        );
        let handle = self
            .session
            .create_service_server(&info, QosSettings::services_default())
            .map_err(|_| NodeError::Transport(TransportError::ServiceServerCreationFailed))?;
        Ok(crate::executor::handles::RawServiceServer::new(handle))
    }

    /// Typeless service client. L1 counterpart of [`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 info = Self::service_info(
            self.domain_id,
            &self.name,
            &self.namespace,
            service_name,
            type_name,
            type_hash,
        );
        let handle = self
            .session
            .create_service_client(&info, QosSettings::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 action_info = Self::action_info(self.domain_id, action_name, type_name, type_hash);

        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,
            type_name,
            type_hash,
        );
        let send_goal_server = self
            .session
            .create_service_server(&send_goal_info, QosSettings::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_server(&cancel_goal_info, QosSettings::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,
            type_name,
            type_hash,
        );
        let get_result_server = self
            .session
            .create_service_server(&get_result_info, QosSettings::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,
            type_name,
            type_hash,
        );
        let feedback_publisher = self
            .session
            .create_publisher(&feedback_topic, QosSettings::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,
                QosSettings::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 action_info = Self::action_info(self.domain_id, action_name, type_name, type_hash);

        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,
            type_name,
            type_hash,
        );
        let send_goal_client = self
            .session
            .create_service_client(&send_goal_info, QosSettings::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_service_client(&cancel_goal_info, QosSettings::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,
            type_name,
            type_hash,
        );
        let get_result_client = self
            .session
            .create_service_client(&get_result_info, QosSettings::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,
            type_name,
            type_hash,
        );
        let feedback_subscriber = self
            .session
            .create_subscriber(&feedback_topic, QosSettings::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>()?;
        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::ACTION_HASH,
        );
        let send_goal_server = self
            .session
            .create_service_server(&send_goal_info, QosSettings::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::ACTION_HASH,
        );
        let cancel_goal_server = self
            .session
            .create_service_server(&cancel_goal_info, QosSettings::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::ACTION_HASH,
        );
        let get_result_server = self
            .session
            .create_service_server(&get_result_info, QosSettings::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::ACTION_HASH,
        );
        let feedback_publisher = self
            .session
            .create_publisher(&feedback_topic, QosSettings::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::ACTION_HASH,
        );
        let status_publisher = self
            .session
            .create_publisher(
                &status_topic,
                QosSettings::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>()?;
        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::ACTION_HASH,
        );
        let send_goal_client = self
            .session
            .create_service_client(&send_goal_info, QosSettings::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::ACTION_HASH,
        );
        let cancel_goal_client = self
            .session
            .create_service_client(&cancel_goal_info, QosSettings::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::ACTION_HASH,
        );
        let get_result_client = self
            .session
            .create_service_client(&get_result_info, QosSettings::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::ACTION_HASH,
        );
        let feedback_subscriber = self
            .session
            .create_subscriber(&feedback_topic, QosSettings::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,
            },
            _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: QosSettings,
}

impl<'n, 'a, 't> PublisherBuilder<'n, 'a, 't> {
    /// Set the QoS (also settable on the typed/generic builder).
    pub fn qos(mut self, qos: QosSettings) -> 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: QosSettings,
    _phantom: PhantomData<M>,
}

impl<'n, 'a, 't, M: MessageForRmw> TypedPublisherBuilder<'n, 'a, 't, M> {
    pub fn qos(mut self, qos: QosSettings) -> 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: QosSettings,
}

impl<'n, 'a, 't> GenericPublisherBuilder<'n, 'a, 't> {
    pub fn qos(mut self, qos: QosSettings) -> 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 [`NodeCtx::create_callback_group`].  Passed to the `_in`
/// entity-create variants (`create_timer_in`, `create_subscription_in`,
/// `create_publisher_in`) 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
    }
}

/// An executor-borrowing node handle — `exec.node(id)`. Hosts the
/// callback-registering entity builders (subscriptions register into the
/// executor's dispatch arena). It is a **short-lived `&mut Executor` borrow**:
/// create entities, then drop it before acquiring the next node handle; entity
/// handles (`HandleId`, publishers) are owned and outlive it (no `Arc` — see
/// `docs/design/0022-entity-api-tiers.md` §Borrow model).
pub struct NodeCtx<'e, 's> {
    executor: &'e mut super::spin::Executor<'s>,
    node_id: super::node_record::NodeId,
}

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

    /// 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: QosSettings::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 `NodeCtx` — 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: QosSettings::default(),
        }
    }

    /// Convenient typed publisher (the `fork` tier — rclcpp/rclrs shape).
    pub fn create_publisher<M: MessageForRmw>(
        &mut self,
        topic: &str,
    ) -> Result<EmbeddedPublisher<M>, NodeError> {
        self.executor
            .create_publisher_on::<M>(self.node_id, topic, QosSettings::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.executor.create_publisher_raw_on(
            self.node_id,
            topic,
            type_name,
            type_hash,
            QosSettings::default(),
        )
    }

    /// Convenient typed subscription (the `fork` tier — rclcpp/rclrs shape).
    /// Sugar over the builder with default QoS + buffer.
    pub fn create_subscription<M, F>(
        &mut self,
        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,
                QosSettings::default(),
                callback,
                None, // no group — node default
            )
    }

    // -----------------------------------------------------------------------
    // 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`, `create_subscription_in`, `create_publisher_in`)
    /// 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(&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<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()))
    }

    /// 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<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,
                QosSettings::default(),
                callback,
                Some(group.name()),
            )
    }

    /// 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<M: MessageForRmw>(
        &mut self,
        _group: &CallbackGroup,
        topic: &str,
    ) -> Result<EmbeddedPublisher<M>, NodeError> {
        // Publishers carry no executor callback slot; group is forward-compat.
        self.executor
            .create_publisher_on::<M>(self.node_id, topic, QosSettings::default())
    }

    /// 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`](Self::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,
    {
        register_type::<Svc::Request>()?;
        register_type::<Svc::Reply>()?;
        let (_id, hdr) = self
            .executor
            .register_service_client_callback::<Svc, F, REPLY_BUF>(
                Some(self.node_id),
                service_name,
                Svc::SERVICE_NAME,
                Svc::SERVICE_HASH,
                QosSettings::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`](Self::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::GoalId, bool) + 'static,
        FbF: FnMut(&nros_core::GoalId, &A::Feedback) + 'static,
        ResF: FnMut(&nros_core::GoalId, nros_core::GoalStatus, &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::GoalId, bool) + 'static,
        FbF: FnMut(&nros_core::GoalId, &A::Feedback) + 'static,
        ResF: FnMut(&nros_core::GoalId, nros_core::GoalStatus, &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).
                8u16,
                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.executor
            .register_subscription_buffered_raw_on::<F, { crate::config::DEFAULT_RX_BUF_SIZE }>(
                self.node_id,
                topic,
                type_name,
                type_hash,
                QosSettings::default(),
                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.executor
            .register_subscription_buffered_raw_safety_on::<F, { crate::config::DEFAULT_RX_BUF_SIZE }>(
                self.node_id,
                topic,
                type_name,
                type_hash,
                QosSettings::default(),
                callback,
            )
    }

    /// Convenient borrowed (zero-copy) subscription (Phase 229.6, issue 0007 /
    /// RFC-0033 `borrowed` mode).
    ///
    /// `B` is the code-generated borrowed-message marker (e.g. `ImageBorrow`,
    /// emitted alongside the owned `Image` for a `.msg` with a `borrowed`-mode
    /// field). The callback receives `&B::View<'a>` — a lifetime-carrying
    /// message whose unbounded sequence/string fields borrow directly from the
    /// receive buffer (no `heapless::Vec` copy); the view is valid only for the
    /// callback's duration.
    ///
    /// Uses `KEEP_LAST(1)` QoS → triple buffer, as borrowed subscriptions
    /// require (a single well-defined slot for the callback's borrow). For an
    /// explicit deeper queue use the owned
    /// [`create_subscription`](Self::create_subscription); a borrowed
    /// subscription registered with `KEEP_LAST(N>1)` is rejected.
    pub fn create_subscription_borrowed<B, F>(
        &mut self,
        topic: &str,
        callback: F,
    ) -> Result<super::types::HandleId, NodeError>
    where
        B: nros_core::BorrowedMessage + '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,
                QosSettings::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: QosSettings::services_default(),
        }
    }

    /// Convenient service server (the `fork` tier — rclrs/rclcpp shape), default
    /// services QoS. Mirror of `create_subscription`.
    pub fn create_service<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) -> Svc::Reply + 'static,
    {
        self.executor.register_service_sized_on::<
            Svc,
            F,
            { crate::config::DEFAULT_RX_BUF_SIZE },
            { crate::config::DEFAULT_RX_BUF_SIZE },
        >(self.node_id, name, QosSettings::services_default(), callback)
    }
}

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

impl<'c, 'e, 't, 's> CtxServiceBuilder<'c, 'e, 't, 's> {
    /// Service QoS (applies to both the request + reply endpoints). Defaults to
    /// `QosSettings::services_default()`.
    pub fn qos(mut self, qos: QosSettings) -> 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) -> 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)
    }
}

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

impl<'c, 'e, 't, 's> CtxPublisherBuilder<'c, 'e, 't, 's> {
    pub fn qos(mut self, qos: QosSettings) -> 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 `NodeCtx` (`.typed::<M>()`).
pub struct CtxTypedPublisherBuilder<'c, 'e, 't, 's, M> {
    ctx: &'c mut NodeCtx<'e, 's>,
    topic: &'t str,
    qos: QosSettings,
    _phantom: PhantomData<M>,
}

impl<'c, 'e, 't, 's, M: MessageForRmw> CtxTypedPublisherBuilder<'c, 'e, 't, 's, M> {
    pub fn qos(mut self, qos: QosSettings) -> 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 `NodeCtx` (`.generic(type, hash)`).
pub struct CtxGenericPublisherBuilder<'c, 'e, 't, 's> {
    ctx: &'c mut NodeCtx<'e, 's>,
    topic: &'t str,
    type_name: &'t str,
    type_hash: &'t str,
    qos: QosSettings,
}

impl<'c, 'e, 't, 's> CtxGenericPublisherBuilder<'c, 'e, 't, 's> {
    pub fn qos(mut self, qos: QosSettings) -> 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 NodeCtx<'e, 's>,
    topic: &'t str,
    qos: QosSettings,
}

impl<'c, 'e, 't, 's> SubscriptionBuilder<'c, 'e, 't, 's> {
    pub fn qos(mut self, qos: QosSettings) -> 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,
            _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 NodeCtx<'e, 's>,
    topic: &'t str,
    qos: QosSettings,
    sched: Option<super::sched_context::SchedContextId>,
    _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: QosSettings) -> 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).
    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,
            _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,
            _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,
            _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; builder uses sched override
            )?;
        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 NodeCtx<'e, 's>,
    topic: &'t str,
    qos: QosSettings,
    sched: Option<super::sched_context::SchedContextId>,
    _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: QosSettings) -> 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,
            _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 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 NodeCtx<'e, 's>,
    topic: &'t str,
    qos: QosSettings,
    sched: Option<super::sched_context::SchedContextId>,
    _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: QosSettings) -> 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,
            _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 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 NodeCtx<'e, 's>,
    topic: &'t str,
    type_name: &'t str,
    type_hash: &'t str,
    qos: QosSettings,
    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: QosSettings) -> 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,
                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 NodeCtx<'e, 's>,
    topic: &'t str,
    type_name: &'t str,
    type_hash: &'t str,
    qos: QosSettings,
    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: QosSettings) -> 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, 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.
    #[cfg(rmw_needs_type_descriptors)]
    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, 0);

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

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

    #[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(QosSettings::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(QosSettings::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(QosSettings::default().keep_last(1))
            .rx_buffer::<1024>()
            .build(|_data: &[u8]| {})
            .expect("generator-shape subscription builds");
    }

    #[test]
    fn nodectx_publisher_and_bridge_shape() {
        // NodeCtx publisher symmetry + the bridge two-ctx borrow pattern:
        // build the dest publisher on one NodeCtx (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 NodeCtx
        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"); // NodeCtx 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_raw(payload);
            })
            .expect("bridge-shape source subscription builds");
    }
}