nros/

node_runtime.rs

//! Phase 212.M.5.a.2 — the executor-backed component runtime for nano-ros.
//!
//! [`MetadataRecorder`] (the planner sink) records what a [`Component`] /
//! [`ExecutableNode`] declares. This module is the
//! missing twin: it binds the same traits to a live [`Executor`] so
//! a Node pkg can actually run — nodes, publishers,
//! subscriptions, timers materialise as real executor handles, and
//! every fired callback dispatches into
//! [`ExecutableNode::on_callback`] with the right
//! [`CallbackId`].
//!
//! Shape:
//!
//! ```ignore
//! use nros::{Executor, ExecutorConfig};
//! use nros::node_runtime::ExecutorNodeRuntime;
//!
//! let cfg = ExecutorConfig::from_env().node_name("talker_main");
//! let executor = Executor::open(&cfg).unwrap();
//! let mut runtime = ExecutorNodeRuntime::from_executor(executor);
//! let _handle = runtime.register_node::<Talker>().unwrap();
//! runtime.spin().unwrap();
//! ```
//!
//! Owned-spin / board consumer: the macro-emitted `<pkg>::register(runtime)`
//! wrapper (Phase 258, Track 2) installs each Node onto the runtime's executor
//! through the uniform `install_node_typed` seam — the same seam the C/C++
//! typed entries use. The codegen `run_plan(runtime)` body + `nros::main!`
//! owned-spin loop drive one `register(runtime)?` per launch-XML `<node>`,
//! then `runtime.spin()`. (The retired Phase 212.M.5.a four-fn-ptr BSP-baker
//! ABI — ~~`register_dispatch_slot`~~ / ~~`nros_run_components`~~ — is gone.)
//!
//! ## Coverage today (Phase 212.M.5.a.2)
//!
//! Publishers, subscriptions, and repeating timers wire end-to-end:
//! the live executor delivers callbacks; the bound
//! [`ExecutableNode::on_callback`] body runs with a
//! [`CallbackCtx`] backed by the per-component publisher resolver.
//! Service servers / clients and action servers / clients wire
//! end-to-end too (Phase 212.M-F.23): `create_entity` registers them on
//! the executor with C-ABI trampolines that route inbound requests /
//! goals into the component's `on_callback`, and the tick-time client /
//! action surface ([`TickCtx`]) is backed by `RuntimeClientDispatch` /
//! `RuntimeActions` over the live executor. Parameters are still a
//! follow-up (registration succeeds; param callbacks don't fire yet).

#![cfg(feature = "rmw-cffi")]

// W5-endgame (issue 0843): the DECLARATION is gated too — a bare
// `extern crate alloc` links the alloc crate into every image and rustc then
// demands a `#[global_allocator]` even when nothing here allocates. This line
// is what kept the first heap-free-tier image from linking.
#[cfg(feature = "alloc")]
extern crate alloc;

#[cfg(feature = "alloc")]
use alloc::{boxed::Box, vec::Vec};
#[cfg(feature = "alloc")]
use core::time::Duration;
use core::{
    cell::{Ref, RefCell, UnsafeCell},
    marker::PhantomData,
    mem::MaybeUninit,
};
use nros_node::executor::ffi::{
    ActionClientCoreFfi as _, ActionClientRawHandleFfi as _, ActionServerRawHandleFfi as _,
};

// Via nros-core's re-export rather than a new direct dependency — every graph
// containing `nros` would otherwise move (16 leaf lockfiles, measured on W5.1).
use nros_core::heapless;
// phase-515 R9: the raw service/client entry points this runtime drives are
// the executor's FFI surface. Imported, never re-exported.
use nros_node::executor::ffi::ExecutorFfi as _;
use portable_atomic::{AtomicUsize, Ordering};
#[cfg(feature = "alloc")]
use portable_atomic_util::Arc;

/// phase-391 W5 — entity/callback identifier, fixed-capacity.
///
/// Same bound as `nros_node::names::ResolvedName` (`MAX_RESOLVED_NAME_LEN` =
/// 128): every string these registries hold is a resolved entity id, callback
/// id, node name or namespace, i.e. name-shaped. An id that does not fit is a
/// REGISTRATION error, never a truncation.
type IdStr = heapless::String<{ nros_node::names::MAX_RESOLVED_NAME_LEN }>;

/// W5-endgame alloc-off — nodes one component's `register()` may create.
/// Components create one node in every in-tree class; 4 leaves room without
/// costing anything that outlives registration (the sink is a stack local).
const MAX_SINK_NODES: usize = 4;

// phase-391 W5 (amended by W5-endgame step 2a): the per-cell registry bound,
// PER KIND, is the `MAX_CELL_ENTITIES` knob — now spelled as `ComponentCell`'s
// const-parameter DEFAULTS rather than a shared ~~`CELL_REG_CAP`~~ const, so the
// macro can name tighter per-class bounds while every existing spelling keeps
// the knob-capped layout. Was `DEFAULT_MAX_METADATA_ENTITIES` (32), borrowed
// from the metadata twin — but that figure is per-PLAN-shaped and made every
// cell ~20 KB up front. 8 is per-component-shaped; a component declaring more
// gets a loud registration error naming `NROS_RUNTIME_MAX_CELL_ENTITIES`.

/// issue 1130 -- a cell registry refused an entity. ONE spelling for all six
/// sites (four registries, three ctx slabs, minus the overlap), so none of them
/// can go back to a bare `Runtime`. Logged as well as returned: the install
/// seam collapses every non-`ExecutorFull` error to `NodeRegister(pkg)`, so the
/// console line is what a board with no debugger actually shows.
fn cell_full(kind: &str) -> NodeDeclError {
    nros_log::log_error!(
        nros_log::get_logger("nros"),
        "component cell registry full: this class creates more {} than its \
         registry holds. Raise the class's ENTITY_BOUNDS, or \
         NROS_RUNTIME_MAX_CELL_ENTITIES (build-time, default 8; derived from \
         the image's declared entities when nothing states it).",
        kind
    );
    NodeDeclError::CellRegistryFull
}

/// Owned copy of a name-shaped `&str`, or the registration error that says
/// which knob-less bound it burst.
fn id_str(s: &str) -> Result<IdStr, NodeDeclError> {
    IdStr::try_from(s).map_err(|_| NodeDeclError::Runtime)
}

use crate::{
    EmbeddedRawPublisher, Executor, GoalStatusCode, GoalUuid,
    node::{
        ActionExecutor, Callback, CallbackCtx, ClientDispatch, ExecutableNode, NodeContext,
        NodeDeclError, NodeOptions, NodeResult, PublisherResolver, TickCtx, TimerOp,
    },
    node_metadata::{
        CallbackEffectKind, CallbackId, EntityId, EntityKind, EntityMetadata, NodeId as MetaNodeId,
        NodeMetadataError,
    },
};

// Phase 212.N.7 closing sweep — ~~`component_register_symbol`~~ retired
// (no live callers after the BSP baker + macro extern emit were
// removed). The former re-export here is gone.

// =============================================================================
// Public types
// =============================================================================

/// Opaque handle returned by
/// [`ExecutorNodeRuntime::register_node`].
///
/// `C` distinguishes handles at the type level so a caller who keeps
/// the handle can later (post-M.5.a.3) recover a typed mut-state
/// borrow. For today the handle is purely a witness that registration
/// succeeded.
pub struct RegisteredNode<C: ExecutableNode> {
    component_idx: usize,
    _phantom: PhantomData<fn() -> C>,
}

impl<C: ExecutableNode> RegisteredNode<C> {
    /// Slot index of this component inside the runtime.
    pub fn slot(&self) -> usize {
        self.component_idx
    }
}

/// Errors returned by the runtime entry points.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ExecutorError {
    /// One of the components' register / lifecycle calls failed.
    Node(NodeDeclError),
    /// The executor's spin loop returned an unexpected error.
    SpinFailed,
}

impl From<NodeDeclError> for ExecutorError {
    fn from(e: NodeDeclError) -> Self {
        Self::Node(e)
    }
}

// =============================================================================
// Internal slot — type-erases the component's `State` so the runtime
// can hold a heterogeneous vec.
// =============================================================================

trait ComponentSlot {
    fn dispatch(&mut self, cb_id: &str, ctx: &mut CallbackCtx<'_>);
    fn tick(&mut self, ctx: &mut TickCtx<'_>);
}

struct TypedSlot<C: ExecutableNode> {
    state: C::State,
    _phantom: PhantomData<fn() -> C>,
}

impl<C: ExecutableNode> ComponentSlot for TypedSlot<C> {
    fn dispatch(&mut self, cb_id: &str, ctx: &mut CallbackCtx<'_>) {
        C::on_callback(
            &mut self.state,
            Callback::__from_id(CallbackId::new(cb_id)),
            ctx,
        );
    }
    fn tick(&mut self, ctx: &mut TickCtx<'_>) {
        C::tick(&mut self.state, ctx);
    }
}

// Phase 258 (Track 2, w5) — ~~`BspDispatchSlot`~~ (the type-erased
// four-fn-ptr BSP dispatch slot) is gone with the retired
// ~~`register_dispatch_slot`~~ / ~~`nros_run_components`~~ BSP-baker path. The
// only remaining `ComponentSlot` impl is `TypedSlot<C>` above (used by
// `register_node` / `register_node_borrowed` / `install_node_typed`).

/// phase-391 W5.3b — per-class slot storage the `nros::node!` macro emits.
///
/// One `static` of this type per macro expansion, so `C` is CONCRETE at the
/// emission site and the storage is sized `size_of::<TypedSlot<C>>()` exactly —
/// no byte-budget guessing on the FFI install path. The type is public API for
/// the MACRO EMIT only; it never crosses the C ABI (the trampoline keeps its
/// `(ptr, ptr, ptr) -> i32` shape), which is what keeps the const-generic
/// parameter legal under the "no const generics on FFI-visible types" rule.
///
/// `N` is the per-class INSTANCE cap: the launch path bakes one identity per
/// plan node and can name the same class twice, so this is an array, not one
/// slot. Taking past `N` is a registration error (`take` returns `None`), the
/// same Full shape as the executor's node table. Slots are one-shot — `next`
/// never rewinds — so storage is never re-initialised under a stale reference.
pub struct ComponentSlotStorage<
    C: ExecutableNode,
    const N: usize = { crate::config::MAX_CLASS_INSTANCES },
    const PUBS: usize = { crate::config::MAX_CELL_ENTITIES },
    const SVCS: usize = { crate::config::MAX_CELL_ENTITIES },
    const ACTC: usize = { crate::config::MAX_CELL_ENTITIES },
    const ACTS: usize = { crate::config::MAX_CELL_ENTITIES },
    const SSRV: usize = { crate::config::MAX_CELL_ENTITIES },
    const TIMS: usize = { crate::config::MAX_CELL_ENTITIES },
> {
    slots: [UnsafeCell<MaybeUninit<TypedSlot<C>>>; N],
    /// W5-endgame step 2b (issue 0857) — the per-instance CELL lives here too,
    /// so the whole component (state + registries) is `.bss`, not heap. Handed
    /// out pairwise with its slot by `take`; the cell's `slot` field then
    /// borrows the sibling region, which is sound because the two arrays are
    /// distinct `UnsafeCell`s claimed by the same monotonic index.
    cells: [UnsafeCell<MaybeUninit<ComponentCell<PUBS, SVCS, ACTC, ACTS, SSRV, TIMS>>>; N],
    next: AtomicUsize,
}

// SAFETY: `take` hands each slot out at most once (monotonic `fetch_add`), so
// no two references to the same `UnsafeCell` contents ever coexist.
unsafe impl<
    C: ExecutableNode,
    const N: usize,
    const PUBS: usize,
    const SVCS: usize,
    const ACTC: usize,
    const ACTS: usize,
    const SSRV: usize,
    const TIMS: usize,
> Sync for ComponentSlotStorage<C, N, PUBS, SVCS, ACTC, ACTS, SSRV, TIMS>
{
}

impl<
    C: ExecutableNode,
    const N: usize,
    const PUBS: usize,
    const SVCS: usize,
    const ACTC: usize,
    const ACTS: usize,
    const SSRV: usize,
    const TIMS: usize,
> ComponentSlotStorage<C, N, PUBS, SVCS, ACTC, ACTS, SSRV, TIMS>
{
    /// Const constructor — the macro emits `static STORE: ... = ...::new();`.
    #[allow(clippy::new_without_default)]
    pub const fn new() -> Self {
        Self {
            slots: [const { UnsafeCell::new(MaybeUninit::uninit()) }; N],
            cells: [const { UnsafeCell::new(MaybeUninit::uninit()) }; N],
            next: AtomicUsize::new(0),
        }
    }

    /// Hand out the next uninitialised slot, or `None` past the instance cap.
    ///
    /// `clippy::mut_from_ref` fires because a `&mut` derived from a `&self` is
    /// usually unsound. It is not here, and the reason is already written down
    /// on the `unsafe impl Sync` above: the monotonic `fetch_add` hands each
    /// index out at most once, so no two `&mut` to one `UnsafeCell` can coexist,
    /// and `i >= N` bounds it. Allowed rather than restructured — the lint
    /// cannot see the atomic, and the invariant is the point of the type.
    #[allow(clippy::mut_from_ref)]
    fn take(
        &'static self,
    ) -> Option<(
        &'static mut MaybeUninit<TypedSlot<C>>,
        &'static mut MaybeUninit<ComponentCell<PUBS, SVCS, ACTC, ACTS, SSRV, TIMS>>,
    )> {
        let i = self.next.fetch_add(1, Ordering::Relaxed);
        if i >= N {
            return None;
        }
        // SAFETY: `i` was claimed exactly once by the fetch_add above, so these
        // are the only references that will ever exist to `slots[i]`'s /
        // `cells[i]`'s contents; the storage is a `static`, so `'static` is
        // honest.
        Some(unsafe { (&mut *self.slots[i].get(), &mut *self.cells[i].get()) })
    }
}

/// Placement-init a [`TypedSlot<C>`] into `slot` and erase it.
///
/// Both registration paths end here: the FFI path with a slot from the
/// macro-emitted [`ComponentSlotStorage`], the dynamic path with one carved
/// from the runtime's caller-supplied backing.
fn place_slot<C: ExecutableNode>(
    slot: &'static mut MaybeUninit<TypedSlot<C>>,
) -> &'static mut dyn ComponentSlot
where
    C::State: 'static,
{
    slot.write(TypedSlot::<C> {
        state: C::init(),
        _phantom: PhantomData,
    })
}

/// Issue 2075 — one controllable declared timer in a component cell.
#[derive(Clone, Copy)]
struct TimerSlot {
    /// FNV-1a 64 of the timer's entity id. A hash, not an [`IdStr`]: an id
    /// string is `MAX_RESOLVED_NAME_LEN` bytes, a row here is 16 + 1.
    key: u64,
    handle: crate::HandleId,
    /// The op a context requested since the last tick, applied by
    /// [`apply_timer_ops`]. The last request wins.
    pending: Option<TimerOp>,
}

/// Issue 2075 — the key a timer's entity id is filed under (FNV-1a 64).
fn timer_key(entity_id: &str) -> u64 {
    let mut h: u64 = 0xcbf2_9ce4_8422_2325;
    for b in entity_id.as_bytes() {
        h ^= u64::from(*b);
        h = h.wrapping_mul(0x0000_0100_0000_01b3);
    }
    h
}

/// phase-391 W5-endgame step 2b (issue 0857) — the NON-GENERIC head of every
/// `ComponentCell<..>`, guaranteed FIRST FIELD by the cell's `#[repr(C)]`.
///
/// The executor's enrolled-component state is a thin pointer to a cell whose
/// const parameters only the enrolling monomorphization knows. The tick/drop
/// trampolines are monomorphized alongside it and cast back to the concrete
/// type; the one consumer that iterates ALL enrolled states regardless of
/// class — the dispatch-stats fold — casts to this header instead. Both casts
/// are sound because the header is the first field of a `repr(C)` struct, so
/// a cell pointer IS a header pointer.
struct CellHeader {
    callback_dispatches: AtomicUsize,
    message_dispatches: AtomicUsize,
}

/// Issue 1865 — the NON-GENERIC prefix of every `ComponentCell<..>`: its
/// header and its slot, the two fields that precede every const-generic one.
///
/// [`ExecutorNodeRuntime::dispatch_callback`] must reach the component state
/// of an ENROLLED cell (the install seam's) to hand it a deferred callback,
/// and like the stats fold it cannot name the cell's const parameters. Under
/// `#[repr(C)]` a field's offset depends only on the fields before it, so a
/// `repr(C)` struct that repeats the first two fields has their offsets —
/// asserted below rather than trusted, because reordering the cell's fields
/// would otherwise turn this cast into a read of the wrong bytes.
#[repr(C)]
struct CellPrefix {
    header: CellHeader,
    slot: RefCell<&'static mut dyn ComponentSlot>,
}

const _: () = {
    assert!(
        core::mem::offset_of!(ComponentCell<0, 0, 0, 0, 0, 0>, header)
            == core::mem::offset_of!(CellPrefix, header)
    );
    assert!(
        core::mem::offset_of!(ComponentCell<0, 0, 0, 0, 0, 0>, slot)
            == core::mem::offset_of!(CellPrefix, slot)
    );
    assert!(
        core::mem::offset_of!(ComponentCell<1, 1, 1, 1, 1, 1>, slot)
            == core::mem::offset_of!(CellPrefix, slot)
    );
};

/// Shared per-component cell. Subscription / timer closures registered
/// against the executor hold a [`CellHandle`] so they can dispatch +
/// publish back through the resolver.
///
/// phase-391 W5-endgame step 2a (issue 0857) — generic over the four registry
/// capacities, defaulted to the `MAX_CELL_ENTITIES` knob so every existing
/// spelling (`ComponentCell`) keeps meaning the pool-capped layout. The
/// per-class macro emission names tighter bounds; everything downstream of
/// construction consumes the cell through the non-generic [`CellView`], so
/// the const params never cross an FFI or public signature.
#[repr(C)]
struct ComponentCell<
    const PUBS: usize = { crate::config::MAX_CELL_ENTITIES },
    const SVCS: usize = { crate::config::MAX_CELL_ENTITIES },
    const ACTC: usize = { crate::config::MAX_CELL_ENTITIES },
    const ACTS: usize = { crate::config::MAX_CELL_ENTITIES },
    const SSRV: usize = { crate::config::MAX_CELL_ENTITIES },
    const TIMS: usize = { crate::config::MAX_CELL_ENTITIES },
> {
    /// W5-endgame step 2b — MUST stay first; see [`CellHeader`].
    header: CellHeader,
    /// phase-391 W5.3b — BORROWED from per-class or pool storage, not boxed.
    /// The storage is `'static` (macro-emitted static / caller backing), so the
    /// reference outlives every closure that dispatches through it.
    slot: RefCell<&'static mut dyn ComponentSlot>,
    publishers: RefCell<heapless::Vec<(IdStr, EmbeddedRawPublisher), PUBS>>,
    // Phase 212.M-F.23 — declarative service/action CLIENT + action-SERVER
    // handles, keyed by stable entity id, resolved during tick dispatch.
    // Built at registration time on the single-node runtime (they mirrored the
    // orchestration's ~~`GenClientDispatch`~~/~~`GenActionExec`~~ arrays, both
    // retired). Service- and
    // action-SERVER request/goal dispatch is owned by the executor (the
    // trampolines registered in `create_entity`); only the action-server
    // handle is kept here so the tick can complete goals / publish feedback.
    service_clients: RefCell<heapless::Vec<(IdStr, crate::HandleId), SVCS>>,
    action_clients: RefCell<heapless::Vec<(IdStr, usize), ACTC>>,
    action_servers: RefCell<heapless::Vec<(IdStr, crate::internals::ActionServerRawHandle), ACTS>>,
    /// Issue 2075 — the component's controllable declared timers, keyed by
    /// the FNV-1a hash of the entity id, with the op a context requested.
    /// Bounded by the class's `ENTITY_BOUNDS.timers`; a timer past the bound
    /// registers and fires but is not in here.
    timers: RefCell<heapless::Vec<TimerSlot, TIMS>>,
    // W5-endgame ctx slabs (issue 0857) — the leaked trampoline contexts the
    // sink used to `Box::into_raw` live INSIDE the cell, sized by the class's
    // declared bounds, so the macro path allocates nothing. Monotonic
    // counters; a full slab is a loud registration error. The entries die
    // with the cell (executor drop), after which no trampoline can fire.
    svc_ctxs: [UnsafeCell<MaybeUninit<ServiceServerCtx>>; SSRV],
    svc_ctxs_used: core::cell::Cell<usize>,
    act_srv_ctxs: [UnsafeCell<MaybeUninit<ActionServerCtx>>; ACTS],
    act_srv_ctxs_used: core::cell::Cell<usize>,
    act_cli_ctxs: [UnsafeCell<MaybeUninit<ActionClientCtx>>; ACTC],
    act_cli_ctxs_used: core::cell::Cell<usize>,
    // Phase 264 W4c — raw pointer to the executor's volatile parameter store, so a
    // subscription/timer/service/action callback can read `ctx.parameter::<T>(name)`.
    // The callback closures + leaked trampolines hold only a `CellHandle` (the
    // executor is unreachable when they fire), so the store address is threaded HERE by
    // `apply_param_services`' post-pass (mirrors the `run_ticks` disjoint borrow). Null
    // until param services are registered. Stable for the executor's life: the server
    // lives in a `Box<ParamState>`, and since phase-382 W2' its SLOTS live in
    // caller-owned storage it merely borrows — a fixed-length table either way, so a
    // declare never moves anything this pointer or a stored `&ParameterValue` names.
    #[cfg(feature = "param-store")]
    param_server: core::cell::Cell<*const nros_params::ParameterServer<'static>>,
    // phase-426 W3 — WHICH node in that store this component's callbacks read.
    //
    // The store is keyed by node and the six parameter services are registered
    // per node, so a cell holding the store address and nothing else can only
    // answer for the executor's first node: every component on a multi-node
    // executor read the primary's values. Set from the sink's own
    // `create_node` result once `register()` has run — the raw index rather
    // than a `NodeKey`, because `Cell<T>` wants a `Copy` payload and the key is
    // exactly this byte. `NodeId::PRIMARY` (0) until then, which is what a
    // component that declares no node means.
    #[cfg(feature = "param-store")]
    param_node: core::cell::Cell<u8>,
}

impl<
    const PUBS: usize,
    const SVCS: usize,
    const ACTC: usize,
    const ACTS: usize,
    const SSRV: usize,
    const TIMS: usize,
> Drop for ComponentCell<PUBS, SVCS, ACTC, ACTS, SSRV, TIMS>
{
    fn drop(&mut self) {
        // phase-391 W5.3b — the slot is BORROWED from one-shot storage, so the
        // component state's destructor no longer rides a `Box` drop. Run it
        // here instead: the cell drops exactly once (the executor's drop
        // trampoline runs it in place on the enrolled path; the dynamic
        // path's `Arc` drops it on runtime drop), and the
        // storage slot is never handed out again (`take`/`next_slot` counters
        // are monotonic), so nothing can observe the dropped bytes.
        // W5-endgame ctx slabs — run the placed ctxs' destructors (their
        // `CellHandle` may hold an `Arc` on the pooled arm). Monotonic
        // counters bound the initialized prefix exactly.
        for i in 0..self.svc_ctxs_used.get() {
            // SAFETY: entries [0, used) were initialized by `place_service_ctx`
            // and are dropped exactly once, here.
            unsafe { (*self.svc_ctxs[i].get()).assume_init_drop() };
        }
        for i in 0..self.act_srv_ctxs_used.get() {
            // SAFETY: as above, for `place_action_server_ctx`.
            unsafe { (*self.act_srv_ctxs[i].get()).assume_init_drop() };
        }
        for i in 0..self.act_cli_ctxs_used.get() {
            // SAFETY: as above, for `place_action_client_ctx`.
            unsafe { (*self.act_cli_ctxs[i].get()).assume_init_drop() };
        }
        let slot: &mut &'static mut dyn ComponentSlot = self.slot.get_mut();
        let p: *mut dyn ComponentSlot = &raw mut **slot;
        // SAFETY: `p` targets storage uniquely owned by this cell (handed out
        // at most once), reachable only through the reference being dropped
        // with us; it is dropped at most once because the cell is.
        unsafe { core::ptr::drop_in_place(p) };
    }
}

impl<
    const PUBS: usize,
    const SVCS: usize,
    const ACTC: usize,
    const ACTS: usize,
    const SSRV: usize,
    const TIMS: usize,
> ComponentCell<PUBS, SVCS, ACTC, ACTS, SSRV, TIMS>
{
    /// Phase 264 W4c — the executor's parameter store, or `None` until
    /// `apply_param_services` threads it in. The deref is sound: single-threaded
    /// executor, param services mutate the server only outside callback dispatch.
    #[cfg(feature = "param-store")]
    fn param_server(&self) -> Option<&nros_params::ParameterServer<'static>> {
        let ptr = self.param_server.get();
        if ptr.is_null() {
            None
        } else {
            // SAFETY: `ptr` is the address of the executor's boxed `ParameterServer`
            // (stable for the executor's life); param services mutate it before/after
            // dispatch (`spin.rs` pre/post), never during, so no aliasing `&mut` is live.
            Some(unsafe { &*ptr })
        }
    }
}

/// phase-391 W5-endgame step 1 (issue 0857) — the NON-GENERIC view every
/// dispatch/tick path consumes instead of the concrete [`ComponentCell`].
///
/// Why a trait and not the cell: the endgame emits a per-class cell whose
/// registries are sized to the class's DECLARED entity counts (const generics,
/// which the "never public to other languages" rule confines to the macro
/// emission), while the dynamic `register_node` path keeps pool-backed cells at
/// the knob caps. Two cell layouts, ONE dispatch implementation — this trait is
/// the seam that makes the split affordable. `Ref<'_, [T]>` erases the
/// `heapless::Vec` capacity; `try_with_slot_mut` erases the slot's storage
/// shape (borrowed `&'static mut dyn` today, fused inline in the per-class
/// static tomorrow).
///
/// Object-safe on purpose: the FFI trampolines carry a THIN `*mut c_void`, so
/// each concrete cell type casts back to itself and only then widens to
/// `&dyn CellView`.
trait CellView {
    /// Bump the dispatch counters (`message` = payload was non-empty).
    fn note_dispatch(&self, message: bool);
    /// The publisher registered under `entity_id`, if this component declared
    /// one. Holds the registry's shared borrow for the returned `Ref`'s life.
    fn publisher(&self, entity_id: &str) -> Option<Ref<'_, EmbeddedRawPublisher>>;
    fn service_clients(&self) -> Ref<'_, [(IdStr, crate::HandleId)]>;
    fn action_clients(&self) -> Ref<'_, [(IdStr, usize)]>;
    fn action_servers(&self) -> Ref<'_, [(IdStr, crate::internals::ActionServerRawHandle)]>;
    /// Run `f` over the component slot unless it is already mutably borrowed
    /// (a re-entrant dispatch on the same cell) — in which case the dispatch
    /// is dropped, exactly as the pre-trait `try_borrow_mut` spelled it.
    fn try_with_slot_mut(&self, f: &mut dyn FnMut(&mut dyn ComponentSlot));
    /// The executor's parameter store, or `None` until `apply_param_services`
    /// threads it in.
    #[cfg(feature = "param-store")]
    fn view_param_server(&self) -> Option<&nros_params::ParameterServer<'static>>;
    /// phase-426 W3 — the node key this component's parameter reads are about.
    /// Travels with the store, because a store keyed by node and a reader that
    /// cannot name one is how every component read the primary's values.
    #[cfg(feature = "param-store")]
    fn view_param_node(&self) -> nros_params::NodeKey;
    /// The `(callbacks, messages)` dispatch counters. Only the dynamic
    /// runtime's stats fold reads them through the trait; the macro path's
    /// fold reads the `CellHeader` directly.
    #[cfg(feature = "alloc")]
    fn dispatch_counts(&self) -> (usize, usize);
    /// Registration-time registry appends. `Err(())` = registry full — the
    /// component declared more entities of that kind than its cell's bound,
    /// which the caller reports loudly (never a silent drop).
    fn push_publisher(&self, id: IdStr, handle: EmbeddedRawPublisher) -> Result<(), ()>;
    fn push_service_client(&self, id: IdStr, handle: crate::HandleId) -> Result<(), ()>;
    fn push_action_client(&self, id: IdStr, entry_index: usize) -> Result<(), ()>;
    fn push_action_server(
        &self,
        id: IdStr,
        handle: crate::internals::ActionServerRawHandle,
    ) -> Result<(), ()>;
    /// W5-endgame ctx slabs — place a trampoline context into the cell's slab
    /// and return its stable address for the executor's C-ABI registration.
    /// `Err(())` = slab full (the class declared fewer of that kind than
    /// `register()` creates). The entry lives — and is dropped — with the cell.
    fn place_service_ctx(&self, ctx: ServiceServerCtx) -> Result<*mut core::ffi::c_void, ()>;
    fn place_action_server_ctx(&self, ctx: ActionServerCtx) -> Result<*mut core::ffi::c_void, ()>;
    fn place_action_client_ctx(&self, ctx: ActionClientCtx) -> Result<*mut core::ffi::c_void, ()>;
    /// Issue 2075 — file a declared timer's handle under its id's key.
    /// `Ok(true)` = filed; `Ok(false)` = the table is full, so the timer runs
    /// but cannot be controlled by id; `Err(())` = another timer of this
    /// component already has that key (an FNV-1a collision between two
    /// distinct ids — the metadata already refuses a duplicate id).
    fn push_timer(&self, key: u64, handle: crate::HandleId) -> Result<bool, ()>;
    /// Issue 2075 — record `op` for the timer filed under `key`; `false` when
    /// none is.
    fn request_timer_op(&self, key: u64, op: TimerOp) -> bool;
    /// Issue 2075 — hand every pending op to `f` and clear it.
    fn drain_timer_ops(&self, f: &mut dyn FnMut(crate::HandleId, TimerOp));
}

impl<
    const PUBS: usize,
    const SVCS: usize,
    const ACTC: usize,
    const ACTS: usize,
    const SSRV: usize,
    const TIMS: usize,
> CellView for ComponentCell<PUBS, SVCS, ACTC, ACTS, SSRV, TIMS>
{
    fn note_dispatch(&self, message: bool) {
        self.header
            .callback_dispatches
            .fetch_add(1, Ordering::Relaxed);
        if message {
            self.header
                .message_dispatches
                .fetch_add(1, Ordering::Relaxed);
        }
    }

    #[cfg(feature = "alloc")]
    fn dispatch_counts(&self) -> (usize, usize) {
        (
            self.header.callback_dispatches.load(Ordering::Relaxed),
            self.header.message_dispatches.load(Ordering::Relaxed),
        )
    }

    fn push_publisher(&self, id: IdStr, handle: EmbeddedRawPublisher) -> Result<(), ()> {
        self.publishers
            .borrow_mut()
            .push((id, handle))
            .map_err(|_| ())
    }

    fn push_service_client(&self, id: IdStr, handle: crate::HandleId) -> Result<(), ()> {
        self.service_clients
            .borrow_mut()
            .push((id, handle))
            .map_err(|_| ())
    }

    fn push_action_client(&self, id: IdStr, entry_index: usize) -> Result<(), ()> {
        self.action_clients
            .borrow_mut()
            .push((id, entry_index))
            .map_err(|_| ())
    }

    fn push_action_server(
        &self,
        id: IdStr,
        handle: crate::internals::ActionServerRawHandle,
    ) -> Result<(), ()> {
        self.action_servers
            .borrow_mut()
            .push((id, handle))
            .map_err(|_| ())
    }

    fn place_service_ctx(&self, ctx: ServiceServerCtx) -> Result<*mut core::ffi::c_void, ()> {
        let i = self.svc_ctxs_used.get();
        if i >= SSRV {
            return Err(());
        }
        self.svc_ctxs_used.set(i + 1);
        // SAFETY: `i` is claimed exactly once (monotonic counter, single-
        // threaded executor contract), so no other reference to this entry
        // exists; the address is stable because the cell never moves after
        // placement.
        Ok(
            unsafe { (*self.svc_ctxs[i].get()).write(ctx) as *mut ServiceServerCtx }
                as *mut core::ffi::c_void,
        )
    }

    fn place_action_server_ctx(&self, ctx: ActionServerCtx) -> Result<*mut core::ffi::c_void, ()> {
        let i = self.act_srv_ctxs_used.get();
        if i >= ACTS {
            return Err(());
        }
        self.act_srv_ctxs_used.set(i + 1);
        // SAFETY: as `place_service_ctx`.
        Ok(
            unsafe { (*self.act_srv_ctxs[i].get()).write(ctx) as *mut ActionServerCtx }
                as *mut core::ffi::c_void,
        )
    }

    fn place_action_client_ctx(&self, ctx: ActionClientCtx) -> Result<*mut core::ffi::c_void, ()> {
        let i = self.act_cli_ctxs_used.get();
        if i >= ACTC {
            return Err(());
        }
        self.act_cli_ctxs_used.set(i + 1);
        // SAFETY: as `place_service_ctx`.
        Ok(
            unsafe { (*self.act_cli_ctxs[i].get()).write(ctx) as *mut ActionClientCtx }
                as *mut core::ffi::c_void,
        )
    }

    fn push_timer(&self, key: u64, handle: crate::HandleId) -> Result<bool, ()> {
        let mut timers = self.timers.borrow_mut();
        if timers.iter().any(|t| t.key == key) {
            return Err(());
        }
        Ok(timers
            .push(TimerSlot {
                key,
                handle,
                pending: None,
            })
            .is_ok())
    }

    fn request_timer_op(&self, key: u64, op: TimerOp) -> bool {
        // `try_borrow_mut`: a request made while the table is being drained
        // (a timer op reaching back into the component) is refused rather
        // than panicking; nothing on the drain path calls back in today.
        let Ok(mut timers) = self.timers.try_borrow_mut() else {
            return false;
        };
        match timers.iter_mut().find(|t| t.key == key) {
            Some(t) => {
                t.pending = Some(op);
                true
            }
            None => false,
        }
    }

    fn drain_timer_ops(&self, f: &mut dyn FnMut(crate::HandleId, TimerOp)) {
        let mut timers = self.timers.borrow_mut();
        for t in timers.iter_mut() {
            if let Some(op) = t.pending.take() {
                f(t.handle, op);
            }
        }
    }

    fn publisher(&self, entity_id: &str) -> Option<Ref<'_, EmbeddedRawPublisher>> {
        Ref::filter_map(self.publishers.borrow(), |pubs| {
            pubs.iter().find(|(id, _)| id == entity_id).map(|(_, p)| p)
        })
        .ok()
    }

    fn service_clients(&self) -> Ref<'_, [(IdStr, crate::HandleId)]> {
        Ref::map(self.service_clients.borrow(), |v| v.as_slice())
    }

    fn action_clients(&self) -> Ref<'_, [(IdStr, usize)]> {
        Ref::map(self.action_clients.borrow(), |v| v.as_slice())
    }

    fn action_servers(&self) -> Ref<'_, [(IdStr, crate::internals::ActionServerRawHandle)]> {
        Ref::map(self.action_servers.borrow(), |v| v.as_slice())
    }

    fn try_with_slot_mut(&self, f: &mut dyn FnMut(&mut dyn ComponentSlot)) {
        if let Ok(mut slot) = self.slot.try_borrow_mut() {
            f(*slot);
        }
    }

    #[cfg(feature = "param-store")]
    fn view_param_server(&self) -> Option<&nros_params::ParameterServer<'static>> {
        self.param_server()
    }

    #[cfg(feature = "param-store")]
    fn view_param_node(&self) -> nros_params::NodeKey {
        nros_params::NodeKey::new(self.param_node.get())
    }
}

/// phase-391 W5-endgame step 2b — the two ways a cell is held, one spelling.
///
/// The macro/FFI install path places its cell in per-class static storage and
/// holds `&'static`; the dynamic `register_node` path (alloc-gated, hosted)
/// keeps the `Arc` it always had — its cells are not 0857's embedded heap
/// cost, and `Arc` gives them the drop-on-runtime-drop lifetime the dynamic
/// path needs. Closures and leaked ctxs clone the HANDLE (a ref copy or an
/// `Arc` bump) and dispatch through [`CellHandle::view`].
#[derive(Clone)]
enum CellHandle {
    Static(&'static dyn CellView),
    #[cfg(feature = "alloc")]
    Pooled(Arc<ComponentCell>),
}

impl CellHandle {
    fn view(&self) -> &dyn CellView {
        match self {
            CellHandle::Static(v) => *v,
            #[cfg(feature = "alloc")]
            CellHandle::Pooled(cell) => cell.as_ref(),
        }
    }
}

/// `PublisherResolver` implementation backed by a [`CellView`].
struct CellResolver<'a> {
    cell: &'a dyn CellView,
}

impl PublisherResolver for CellResolver<'_> {
    fn publish_raw(&self, entity_id: &str, data: &[u8]) -> NodeResult<()> {
        // issue 0736 — the two arms report DIFFERENT failures. A miss on
        // the cell's publisher lookup (~~`lookup_publisher`~~ once) means this
        // component declared no publisher for the
        // entity, so nothing ever reached the transport; a `publish_raw` error
        // means the transport had the sample and refused it. Both used to
        // return `Runtime`, which is why a console full of "publish FAILED"
        // could not distinguish a wiring bug from a congested link.
        match self.cell.publisher(entity_id) {
            Some(p) => p
                .publish_serialized(data)
                .map_err(|_| NodeDeclError::Runtime),
            None => Err(NodeDeclError::UnknownPublisher),
        }
    }

    fn request_timer(&self, entity_id: &str, op: TimerOp) -> NodeResult<()> {
        if self.cell.request_timer_op(timer_key(entity_id), op) {
            Ok(())
        } else {
            Err(NodeDeclError::UnknownTimer)
        }
    }
}

/// Issue 2075 — apply the timer ops this cell's contexts requested, against
/// the executor. Runs on both sides of the component's tick: before it for
/// the ops callbacks requested during the pass's dispatch, after it for those
/// the tick itself requested. The executor's own `reset_timer` /
/// `cancel_timer` are the ops (a reset re-seeds a clock timer's reading, so
/// a timer declared stopped starts a FRESH period). A timer already released
/// answers an error, which is dropped: there is nothing left to control.
fn apply_timer_ops(cell: &dyn CellView, exec_ptr: *mut Executor<'static>) {
    cell.drain_timer_ops(&mut |handle, op| {
        // SAFETY: as `tick_one_cell`'s other executor uses: no callback is
        // running and nothing else holds the executor between dispatches.
        let executor = unsafe { &mut *exec_ptr };
        let _ = match op {
            TimerOp::Reset => executor.reset_timer(handle),
            TimerOp::Cancel => executor.cancel_timer(handle),
        };
    });
}

// Phase 212.M-F.23 — the ~~`UnsupportedActions`~~ / ~~`UnsupportedClients`~~ tick-side
// stubs are retired. Real service/action client + action-server dispatch on the
// single-node runtime lives in `RuntimeClientDispatch` / `RuntimeActions`
// (below), wired into `run_ticks`.

// =============================================================================
// ExecutorNodeRuntime
// =============================================================================

/// Executor-backed component runtime.
///
/// Owns the [`Executor`] and one slot per registered component. The
/// register / spin lifecycle:
///
/// 1. [`from_executor`](Self::from_executor) wraps an open
///    [`Executor`].
/// 2. [`register_node`](Self::register_node) builds the
///    component's `State`, runs [`Component::register`](crate::node::Component::register)
///    against the live executor, whose registration sink materialises nodes /
///    pubs / subs / timers on the real executor, and wires each
///    subscription + timer callback to dispatch into
///    [`ExecutableNode::on_callback`] with the right
///    [`CallbackId`].
/// 3. [`spin`](Self::spin) / [`spin_once`](Self::spin_once) drive the
///    executor; between iterations every registered component's
///    [`ExecutableNode::tick`] runs.
#[cfg(feature = "alloc")]
pub struct ExecutorNodeRuntime {
    executor: Executor<'static>,
    components: Vec<Arc<ComponentCell>>,
    /// Issue 2020 -- the level that registered each of `components`, by
    /// index (`Executor::registering_level` at registration; `None` before a
    /// level runner set one). Only that level's pass ticks the component.
    component_levels: Vec<Option<u8>>,
    /// phase-391 W5.3b — slot storage for [`register_node`](Self::register_node).
    /// Always present: the leaking constructor allocates it once, `new_in`
    /// carves the caller's backing.
    pool: ComponentPool,
    /// phase-521 W6 — whether a component registered through
    /// [`register_node`](Self::register_node) has a working `tick`
    /// ([`ExecutableNode::HAS_TICK`]); see [`Self::entry_spin_budget`].
    ticking: bool,
}

/// Bump view over the runtime's slot backing. Raw pointer, not a slice, so the
/// runtime stays movable — the SLOTS never move; they live in the backing.
#[cfg(feature = "alloc")]
struct ComponentPool {
    base: *mut MaybeUninit<u8>,
    len_bytes: usize,
    per_slot: usize,
    capacity: usize,
    next: usize,
}

// SAFETY: the pool is confined to the runtime (`&mut self` on every use); the
// raw pointer targets `'static` backing handed over at construction.
#[cfg(feature = "alloc")]
unsafe impl Send for ComponentPool {}

#[cfg(feature = "alloc")]
impl ComponentPool {
    /// Next disjoint slot window, or `None` when the pool is full.
    fn next_slot(&mut self) -> Option<&'static mut [MaybeUninit<u8>]> {
        if self.next >= self.capacity {
            return None;
        }
        let off = self.next * self.per_slot;
        debug_assert!(off + self.per_slot <= self.len_bytes);
        self.next += 1;
        // SAFETY: `base` points at `'static` backing of `len_bytes`; the
        // window is in-bounds by the checks above and DISJOINT from every
        // earlier one because `next` only increases.
        Some(unsafe { core::slice::from_raw_parts_mut(self.base.add(off), self.per_slot) })
    }
}

#[cfg(feature = "alloc")]
impl ExecutorNodeRuntime {
    /// Wrap an already-built [`Executor`], LEAKING the slot backing.
    ///
    /// The convenience constructor — allocates `RuntimeSizing::DEFAULT`'s
    /// backing once and never frees it, exactly the relationship
    /// `Executor::from_session` has to `Executor::open_in`. Use
    /// [`new_in`](Self::new_in) on an image that must not allocate.
    pub fn from_executor(executor: Executor<'static>) -> Self {
        let sizing = crate::runtime_storage::RuntimeSizing::DEFAULT;
        let backing: &'static mut [MaybeUninit<u64>] =
            Vec::leak(alloc::vec![MaybeUninit::uninit(); sizing.u64_len()]);
        // SAFETY: freshly leaked — `'static`, uniquely owned, exactly
        // `u64_len()` words.
        unsafe { Self::new_in(executor, backing, sizing) }
    }

    /// Wrap an already-built [`Executor`] over CALLER-SUPPLIED slot storage.
    ///
    /// Size `backing` with [`crate::runtime_storage::RuntimeSizing::u64_len`]; a short one panics,
    /// naming both sizes (fail-loud on every profile — a short backing is
    /// silent corruption, the `executor::storage::carve` / issue #131 lesson).
    ///
    /// # Safety
    /// `backing` must be uniquely owned by this runtime for its whole life:
    /// slots carved from it are handed out as `&'static mut`, so aliasing it
    /// anywhere else is undefined behaviour.
    pub unsafe fn new_in(
        executor: Executor<'static>,
        backing: &'static mut [MaybeUninit<u64>],
        sizing: crate::runtime_storage::RuntimeSizing,
    ) -> Self {
        let need = sizing.u64_len();
        assert!(
            backing.len() >= need,
            "component pool backing too small: {} u64 words < {} required for {} slot(s) \
             of {} bytes — size it with RuntimeSizing::u64_len()",
            backing.len(),
            need,
            sizing.components,
            sizing.slot_bytes,
        );
        Self {
            executor,
            components: Vec::new(),
            component_levels: Vec::new(),
            pool: ComponentPool {
                base: backing.as_mut_ptr() as *mut MaybeUninit<u8>,
                len_bytes: backing.len() * 8,
                per_slot: (sizing.slot_bytes.div_ceil(8) * 8).max(1),
                capacity: sizing.components,
                next: 0,
            },
            ticking: false,
        }
    }

    /// phase-521 W6 (RFC-0113 D2) — the budget this runtime's entry loop
    /// passes to [`spin_once`](Self::spin_once):
    /// [`Executor::entry_spin_budget`], or the 10 ms cadence while a component
    /// registered here has a working `tick` (it runs once per spin).
    pub fn entry_spin_budget(&self) -> Duration {
        if self.ticking {
            nros_node::ENTRY_POLL_BUDGET
        } else {
            self.executor.entry_spin_budget()
        }
    }

    /// Borrow the underlying executor.
    pub fn executor(&self) -> &Executor<'static> {
        &self.executor
    }

    /// phase-509 P10 -- the executor's address, through a raw runtime
    /// pointer and without a reference: what a level runner takes the
    /// executor's lock through before it touches the runtime
    /// ([`crate::levels::LevelCore`]).
    ///
    /// # Safety
    /// `this` points at a live runtime.
    pub unsafe fn executor_ptr(this: *mut Self) -> *mut Executor<'static> {
        // SAFETY: only the field address is computed.
        unsafe { core::ptr::addr_of_mut!((*this).executor) }
    }

    /// phase-509 P10 -- [`spin_once`](Self::spin_once) on a runtime other
    /// level workers share: the executor's lock is taken first, the pass
    /// releases it around each user callback
    /// ([`Executor::spin_once_shared`]), and the components' ticks run under
    /// it. Issue 2020 -- on a level image, only the boot level's components
    /// (and any registered before a level was set): each other level's pass
    /// ticks its own ([`crate::levels::LevelCore::spin_level_once`]).
    ///
    /// # Safety
    /// `this` points at a live runtime whose executor has the lock installed
    /// if any other thread can reach it, and no reference to the runtime is
    /// live in the caller.
    #[cfg(feature = "rmw-cffi")]
    pub unsafe fn spin_once_shared(
        this: *mut Self,
        timeout: Duration,
    ) -> nros_node::SpinOnceResult {
        // SAFETY: the caller's contract; the lock is taken before any
        // reference to the runtime exists.
        let exec = unsafe { Self::executor_ptr(this) };
        let _g = unsafe { nros_node::executor::exec_lock::ExecGuard::acquire(exec) };
        let result = unsafe { Executor::spin_once_shared(exec, timeout) };
        // SAFETY: under the lock no other thread touches the runtime.
        unsafe { (*this).run_ticks() };
        result
    }

    /// Mutably borrow the underlying executor — for advanced wiring
    /// (parameter services, custom guard conditions). Don't use during
    /// [`spin`](Self::spin) from another thread; the runtime is
    /// single-threaded.
    pub fn executor_mut(&mut self) -> &mut Executor<'static> {
        &mut self.executor
    }

    /// RFC-0052 / phase-296 W5.4 — lower a tier's RTOS-agnostic scheduling
    /// policy onto this executor's DEFAULT scheduling context. One `Executor`
    /// per tier means "the tier's policy" == "this executor's default SC";
    /// per-group/per-handle bindings still take precedence.
    ///
    /// **Portable** across every board — call from each board's `run_tiers`
    /// after building the runtime. Takes the tier fields as primitives (not a
    /// `TierSpec`) so `nros` needs no board/platform dependency; a board passes
    /// `tier.class`, `tier.period_us`, … straight through.
    ///
    /// `real_time` + `budget_us` + `period_us` → [`SchedClass::Sporadic`](crate::SchedClass::Sporadic);
    /// `best_effort` → [`SchedClass::BestEffort`](crate::SchedClass::BestEffort); `time_triggered` +
    /// `period_us` → the cyclic dispatcher (major frame = period, window =
    /// `budget_us` or the whole frame); `deadline_us` sets the SC deadline and
    /// `deadline_policy` its action. A tier with no class/budget/deadline
    /// leaves the default `Fifo` SC untouched (byte-identical pre-W3 behavior).
    pub fn apply_tier_sched_policy(
        &mut self,
        class: Option<&str>,
        period_us: Option<u64>,
        budget_us: Option<u64>,
        deadline_us: Option<u64>,
        deadline_policy: Option<&str>,
    ) {
        use crate::SchedContext;
        // Common backend (RFC-0052): the tier→SchedContext lowering lives ONCE
        // in `SchedContext::from_tier_policy`, shared with the C / C++ entries
        // (`nros_{c,cpp}_create_sched_context_from_policy`) so the mapping never
        // drifts between languages. `None` → keep the default `Fifo` SC.
        let Some((sc, tt_frame)) = SchedContext::from_tier_policy(
            class,
            period_us,
            budget_us,
            deadline_us,
            deadline_policy,
        ) else {
            return;
        };
        if let Some(frame_us) = tt_frame {
            self.executor_mut()
                .register_time_triggered_dispatcher(frame_us);
        }
        self.executor_mut().set_default_sched_context(sc);
    }

    /// Issue 2018 (RFC-0108 As-built) -- lower a LEVEL's policy onto its own
    /// scheduling context, the level-runner twin of
    /// [`apply_tier_sched_policy`](Self::apply_tier_sched_policy). Call it on
    /// the registering level (after the runner opened or adopted it), before
    /// its setup: what the setup registers then binds to the level's context
    /// instead of the executor's default, which every level on the core
    /// shares.
    ///
    /// `kernel_enforces_budget` says the kernel throttles the level's thread
    /// with the same budget and period (NuttX `SCHED_SPORADIC`, Linux
    /// `SCHED_DEADLINE`), so the executor does not enforce it a second time
    /// ([`Executor::defer_budget_to_kernel`](crate::Executor::defer_budget_to_kernel)).
    /// Returns the context, or `None` when the level declares no policy (it
    /// keeps the default) or no context slot was free (said on the console:
    /// the level then runs on the default context, unbudgeted by the
    /// executor).
    pub fn apply_level_sched_policy(
        &mut self,
        class: Option<&str>,
        period_us: Option<u64>,
        budget_us: Option<u64>,
        deadline_us: Option<u64>,
        deadline_policy: Option<&str>,
        kernel_enforces_budget: bool,
    ) -> Option<crate::SchedContextId> {
        use crate::SchedContext;
        let (sc, tt_frame) = SchedContext::from_tier_policy(
            class,
            period_us,
            budget_us,
            deadline_us,
            deadline_policy,
        )?;
        if let Some(frame_us) = tt_frame {
            self.executor_mut()
                .register_time_triggered_dispatcher(frame_us);
        }
        match self.executor_mut().set_level_sched_context(sc) {
            Ok(id) => {
                if kernel_enforces_budget {
                    self.executor_mut().defer_budget_to_kernel(id);
                }
                Some(id)
            }
            Err(e) => {
                nros_log::log_warn!(
                    nros_log::get_logger("nros"),
                    "a level's sched context was not created ({}): the level runs on \
                     the default context, unbudgeted by the executor",
                    e
                );
                None
            }
        }
    }

    /// Number of registered components.
    pub fn component_count(&self) -> usize {
        self.components.len()
    }

    /// Register a [`Component`](crate::node::Component) (which must also be
    /// [`ExecutableNode`]) into this runtime. Builds the
    /// component's `State` (via [`ExecutableNode::init`]) and
    /// walks [`Component::register`](crate::node::Component::register) over the live executor — every
    /// declared node / pub / sub / timer materialises as a real
    /// executor handle, and subscription + timer callbacks are wired
    /// to dispatch into [`ExecutableNode::on_callback`].
    pub fn register_node<C: ExecutableNode + 'static>(&mut self) -> NodeResult<RegisteredNode<C>>
    where
        C::State: 'static,
    {
        // phase-391 W5.3b — draw slot storage from the runtime's pool instead
        // of boxing. Full → the executor-table-Full class (raise
        // NROS_RUNTIME_MAX_COMPONENTS); too big for `slot_bytes` → likewise a
        // registration error (raise NROS_RUNTIME_COMPONENT_SLOT_BYTES), never
        // a truncation.
        let raw = self.pool.next_slot().ok_or(NodeDeclError::ExecutorFull)?;
        if raw.len() < core::mem::size_of::<TypedSlot<C>>()
            || !(raw.as_ptr() as usize).is_multiple_of(core::mem::align_of::<TypedSlot<C>>())
        {
            return Err(NodeDeclError::Runtime);
        }
        // SAFETY: sized + aligned for `TypedSlot<C>` by the check above;
        // uniquely owned (`next_slot` windows are disjoint); `'static` backing.
        let slot_mu: &'static mut MaybeUninit<TypedSlot<C>> =
            unsafe { &mut *(raw.as_mut_ptr() as *mut MaybeUninit<TypedSlot<C>>) };
        let cell = Arc::new(ComponentCell {
            slot: RefCell::new(place_slot::<C>(slot_mu)),
            publishers: RefCell::new(heapless::Vec::new()),
            service_clients: RefCell::new(heapless::Vec::new()),
            action_clients: RefCell::new(heapless::Vec::new()),
            action_servers: RefCell::new(heapless::Vec::new()),
            timers: RefCell::new(heapless::Vec::new()),
            svc_ctxs: [const { UnsafeCell::new(MaybeUninit::uninit()) }; _],
            svc_ctxs_used: core::cell::Cell::new(0),
            act_srv_ctxs: [const { UnsafeCell::new(MaybeUninit::uninit()) }; _],
            act_srv_ctxs_used: core::cell::Cell::new(0),
            act_cli_ctxs: [const { UnsafeCell::new(MaybeUninit::uninit()) }; _],
            act_cli_ctxs_used: core::cell::Cell::new(0),
            header: CellHeader {
                callback_dispatches: AtomicUsize::new(0),
                message_dispatches: AtomicUsize::new(0),
            },
            // W4c — set by `apply_param_services` once the store exists.
            #[cfg(feature = "param-store")]
            param_server: core::cell::Cell::new(core::ptr::null()),
            // phase-426 W3 — overwritten below with the node `register()`
            // actually created.
            #[cfg(feature = "param-store")]
            param_node: core::cell::Cell::new(0),
        });
        let component_idx = self.components.len();
        self.ticking |= C::HAS_TICK;
        self.components.push(cell.clone());
        self.component_levels
            .push(self.executor.registering_level());

        let mut sink = ExecutorSink {
            cell: CellHandle::Pooled(cell.clone()),
            nodes: heapless::Vec::new(),
            node_identity: None, // direct API — no launch injection
            remaps: &[],         // direct API — no launch remaps
            qos_overrides: &[],  // direct API — no plan overrides
            launch_params: &[],  // direct API -- no launch params
        };
        let result = {
            let mut frame = crate::node::ComponentFrame::new(&mut sink);
            let mut context = NodeContext::new(C::NAME, &mut self.executor, &mut frame);
            C::register(&mut context)
        };
        // phase-426 W3 — the node this component actually created, so its
        // callbacks read ITS parameters and not the primary's. First node: a
        // component declares one in every shipped shape, and the reads are
        // per COMPONENT, which is the granularity a cell has.
        #[cfg(feature = "param-store")]
        if let Some(node) = sink.nodes.first() {
            cell.param_node.set(node.node_id.raw());
        }
        if result.is_err() {
            // Roll back the slot push so `component_count` stays
            // consistent with what users observe.
            self.components.pop();
            self.component_levels.pop();
        }
        result?;

        // W4c — capture the executor's volatile param store on the cell (if param
        // services were registered before this call) so the node's callbacks can read
        // `ctx.parameter::<T>(name)`. Mirrors the install-seam path.
        #[cfg(feature = "param-store")]
        if let Some(server) = self.executor.params() {
            cell.param_server.set(server as *const _);
        }

        Ok(RegisteredNode {
            component_idx,
            _phantom: PhantomData,
        })
    }

    // Phase 258 (Track 2, w5) — ~~`register_dispatch_slot`~~ (the four-fn-ptr
    // BSP registration) is gone with the retired ~~`register_dispatch_slot_dyn`~~
    // bridge + ~~`nros_run_components`~~. Owned-spin / BSP entries now register
    // through `install_node_typed` (the uniform install seam).

    /// Drive one executor iteration + a `tick` per registered
    /// component.
    pub fn spin_once(&mut self, timeout: Duration) -> Result<(), ExecutorError> {
        let _result = self.executor.spin_once(timeout);
        self.run_ticks();
        Ok(())
    }

    /// [`Self::spin_once`], returning the executor's own per-iteration counts
    /// instead of discarding them — issue 0572.
    ///
    /// `spin_once` throws away a `SpinOnceResult` that already carries exactly
    /// what a stalled tier needs to be diagnosed: how many timers fired, how
    /// many subscription callbacks ran, and how many errored. Without it, "the
    /// tier's timer never fires" and "the tier's callback runs and its publish
    /// fails" are the same observation from outside the guest — a silent topic.
    pub fn spin_once_counted(
        &mut self,
        timeout: Duration,
    ) -> Result<nros_node::SpinOnceResult, ExecutorError> {
        let result = self.executor.spin_once(timeout);
        self.run_ticks();
        Ok(result)
    }

    /// Phase 216.B.3 / C.3 follow-up — route a signaled callback to
    /// every registered component slot.
    ///
    /// The RTIC (`nros-board-rtic-stm32f4`) and Embassy
    /// (`nros-board-embassy-stm32f4`) dispatch tasks dequeue a
    /// [`nros_platform::SignaledCallback`] envelope from their SPSC
    /// queue / Embassy channel and need a routing entry point that
    /// hands the callback off to the right Node's `on_callback`
    /// trampoline. This method is that entry point.
    ///
    /// # Strategy — linear scan
    ///
    /// Each registered slot dispatches through the codegen-emitted
    /// `d()` trampoline from `nros::node!()` (a ~~`dispatch_fn`~~ pointer once;
    /// see
    /// `packages/core/nros-macros/src/lib.rs`). That trampoline calls
    /// `<NodeTy as ExecutableNode>::on_callback`, whose body
    /// `match`es on the callback's own tag set
    /// (`Subscription` / `Timer` / `Service` / `Action` ids) and is a
    /// no-op for non-matching `cb_id`s. So a linear scan across every
    /// slot is correct — each slot self-filters and at most one
    /// component actually acts on a given `cb_id`. A focused
    /// `cb_id → slot` index is a separate follow-up; the trampoline's
    /// tag dispatch already gates the real work cheaply (string
    /// compare on statically known literals), so the linear scan is
    /// the minimum-viable wiring that closes the conceptual gap left
    /// by the B.3 / C.3 skeleton emits.
    ///
    /// # Borrow semantics
    ///
    /// Each `ComponentCell`'s slot lives behind a [`RefCell`]; the
    /// per-slot dispatch takes `try_borrow_mut` and is a no-op on
    /// re-entrancy. The runtime is single-threaded by construction
    /// (the dispatch task owns it via `&mut self`), so the borrow
    /// always succeeds in normal flow.
    ///
    /// # Both populations
    ///
    /// Issue 1865 — a component reaches this runtime one of two ways, and the
    /// two are disjoint (see the `observed_callback_counts` fold, issue #140):
    /// [`register_node`](Self::register_node) pushes it to `components`, while
    /// the install seam every `nros::node!` `register(runtime)` goes through
    /// (`install_node_typed*` → `register_node_borrowed`) ENROLLS it in the
    /// executor and never pushes. This scanned `components` alone, so since
    /// the RTIC entry moved onto `register(runtime)` (phase-289) every signaled
    /// callback the RTIC/Embassy dispatch task forwarded reached no component.
    pub fn dispatch_callback(&mut self, cb_id: &str, ctx: &mut CallbackCtx<'_>) {
        for cell in &self.components {
            if let Ok(mut slot) = cell.slot.try_borrow_mut() {
                slot.dispatch(cb_id, ctx);
            }
        }
        for state in self.executor.enrolled_component_states() {
            // SAFETY: every enrolled state is a placed `ComponentCell` from
            // `register_node_borrowed` (the only enroll site), kept alive until
            // `Executor::drop`; `CellPrefix` is its layout-checked prefix.
            let cell = unsafe { &*(state as *const CellPrefix) };
            if let Ok(mut slot) = cell.slot.try_borrow_mut() {
                slot.dispatch(cb_id, ctx);
            }
        }
    }

    /// Spin until the executor's halt flag is raised.
    ///
    /// phase-359 W10 — this used to be `#[cfg(feature = "std")]` and described
    /// itself as hosted-only, but the body is `Duration` + `spin_once` +
    /// `is_halted`, none of which need an OS. The gate was describing a
    /// CONVENTION (a BSP usually wants its own loop so it can interleave
    /// board work) as if it were a requirement, and a bare-metal image that
    /// wants exactly this loop had to hand-roll it to get it.
    ///
    /// It does need `alloc`: the halt flag is an `Arc`, so a core-only image
    /// has no flag to poll.
    #[cfg(feature = "alloc")]
    pub fn spin(&mut self) -> Result<(), ExecutorError> {
        // 10 ms tick cadence — matches the existing executor spin
        // budgeting (see `SpinOptions::poll_interval`); short enough that
        // component `tick` hooks observe latency under one cycle.
        let tick = Duration::from_millis(10);
        while !self.executor.is_halted() {
            let _ = self.executor.spin_once(tick);
            self.run_ticks();
        }
        Ok(())
    }

    /// Halt a running [`spin`](Self::spin). Idempotent.
    ///
    /// phase-417 W4.c — forwards to `Executor::cancel()`, the rclcpp spelling.
    /// `Executor::halt` is now a deprecated forwarder onto it, so naming the old
    /// spelling here would be this crate warning at itself.
    #[cfg(feature = "alloc")]
    pub fn halt(&self) {
        self.executor.cancel();
    }

    fn run_ticks(&mut self) {
        self.run_ticks_for(None);
    }

    /// The registered components' ticks for one pass. Issue 2020 -- only
    /// those the pass's level owns (`Executor::pass_ticks_owner`):
    /// `band_level` is a level worker's band pass, `None` the executor's own
    /// worker (the boot level, or the only one).
    pub(crate) fn run_ticks_for(&mut self, band_level: Option<u8>) {
        // Per-component tick — each component's resolver is its own cell.
        // Phase 212.M-F.23: the tick reaches the executor (service-client
        // call_raw poll, action-server complete/feedback) through a raw
        // pointer so `&self.components` and `&mut self.executor` (disjoint
        // fields) can be live at once.
        let exec_ptr: *mut Executor<'static> = &mut self.executor;
        for (cell, owner) in self.components.iter().zip(&self.component_levels) {
            // SAFETY: a shared read of the executor between ticks; no tick
            // is running and the `&mut` above is not used meanwhile.
            if !unsafe { &*exec_ptr }.pass_ticks_owner(band_level, *owner) {
                continue;
            }
            tick_one_cell(cell.as_ref(), exec_ptr);
        }
    }
}

/// Phase 258 (Track 2, 2a) — drive one component cell's `tick` against the
/// executor. The single source of truth for the per-component tick body,
/// shared by [`ExecutorNodeRuntime::run_ticks`] (the owned-runtime path) and
/// the executor-enrolled [`component_tick_trampoline`] (the `install` path).
///
/// `exec_ptr` is reached through a raw `*mut Executor<'static>` so the caller can hold
/// the component (`&ComponentCell`) and the executor live at once — they are
/// disjoint, and `RuntimeActions` / `RuntimeClientDispatch` reborrow `&mut`
/// per call (see their docs).
fn tick_one_cell(cell: &dyn CellView, exec_ptr: *mut Executor<'static>) {
    let resolver = CellResolver { cell };
    let service_clients = cell.service_clients();
    let action_clients = cell.action_clients();
    let action_servers = cell.action_servers();
    let mut actions = RuntimeActions {
        executor: exec_ptr,
        handles: &action_servers,
    };
    let mut clients = RuntimeClientDispatch {
        executor: exec_ptr,
        services: &service_clients,
        actions: &action_clients,
    };
    let mut ctx = TickCtx::new(&resolver, &mut actions, &mut clients);
    // W4c — `tick` reads `ctx.parameter::<T>(name)` from the store via the cell pointer
    // (the store is a separate `Box<ParamState>` allocation, so this does NOT alias the
    // `&mut Executor<'static>` the action/client tick calls reborrow through `exec_ptr`).
    #[cfg(feature = "param-store")]
    ctx.set_param_server(cell.view_param_server(), cell.view_param_node());
    // Issue 2075 — ops the pass's callbacks requested, then the tick, then
    // the ops the tick requested.
    apply_timer_ops(cell, exec_ptr);
    cell.try_with_slot_mut(&mut |slot| slot.tick(&mut ctx));
    apply_timer_ops(cell, exec_ptr);
}

/// Phase 258 (Track 2, 2a) — executor `ComponentSlot.tick` trampoline. Casts
/// the enrolled state back to the component cell + `exec_ctx` back to the
/// executor and drives one tick. The layering-clean `extern "C"` shim the
/// `nros-node` [`Executor`] calls each `spin_once` (it can't name `nros`'s
/// [`ComponentCell`] — see [`register_node_borrowed`]'s enroll).
///
/// # Safety
/// `state` must be the placed `ComponentCell` enrolled via
/// [`Executor::enroll_component`] from [`register_node_borrowed`] (per-class
/// static storage or the alloc convenience's leaked box — live until
/// `component_drop_trampoline`); `exec_ctx` must be the live
/// `*mut Executor<'static>` the executor passes itself.
unsafe extern "C" fn component_tick_trampoline<
    const PUBS: usize,
    const SVCS: usize,
    const ACTC: usize,
    const ACTS: usize,
    const SSRV: usize,
    const TIMS: usize,
>(
    state: *mut core::ffi::c_void,
    exec_ctx: *mut core::ffi::c_void,
) {
    // SAFETY: `state` is a live placed `ComponentCell` (kept in place until
    // the drop trampoline); this monomorphization was enrolled alongside it,
    // so the cast target is the cell's true type.
    let cell = unsafe { &*(state as *const ComponentCell<PUBS, SVCS, ACTC, ACTS, SSRV, TIMS>) };
    tick_one_cell(cell, exec_ctx as *mut Executor<'static>);
}

/// Phase 258 (Track 2, 2a; W5-endgame step 2b) — executor `ComponentSlot.drop`
/// trampoline. Drops the placed `ComponentCell` (which drops the component
/// state through its borrowed slot) IN PLACE — the storage itself is a
/// `static` (or the alloc convenience's leaked box, whose bytes are never
/// freed, matching its slot's documented leak). Run exactly once on
/// `Executor::drop`; the storage's monotonic `take` never re-hands the
/// region out, so nothing can observe the dropped bytes.
///
/// # Safety
/// `state` must be the placed `ComponentCell` enrolled via
/// [`Executor::enroll_component`], not yet dropped.
unsafe extern "C" fn component_drop_trampoline<
    const PUBS: usize,
    const SVCS: usize,
    const ACTC: usize,
    const ACTS: usize,
    const SSRV: usize,
    const TIMS: usize,
>(
    state: *mut core::ffi::c_void,
) {
    // SAFETY: enroll's contract above; dropped exactly once by the executor,
    // through the monomorphization enrolled with the cell.
    unsafe {
        core::ptr::drop_in_place(state as *mut ComponentCell<PUBS, SVCS, ACTC, ACTS, SSRV, TIMS>)
    };
}

// =============================================================================
// Phase 212.N.7 step-3.3 — bridge to platform-side `NodeDispatchRuntime`.
// =============================================================================
//
// `nros_platform::NodeDispatchRuntime` is the board-side sink: object-safe +
// `no_std`. `BoardEntry::run` installs this `ExecutorNodeRuntime` impl on the
// per-boot `RuntimeCtx::runtime` slot. The owned-spin entry reaches the live
// executor through `executor_handle()` (a raw pointer crosses the layering wall
// cleanly) and installs via `nros::__macro::install_node_typed`. Phase 258 (w5) retired
// the old ~~`register_dispatch_slot_dyn`~~ four-fn-ptr bridge.

#[cfg(feature = "alloc")]
impl ::nros_platform::NodeDispatchRuntime for ExecutorNodeRuntime {
    fn spin_once(&mut self, timeout_ms: u32) -> Result<(), ()> {
        Self::spin_once(self, Duration::from_millis(timeout_ms.into())).map_err(|_| ())
    }

    fn entry_spin_budget_ms(&self) -> u32 {
        Self::entry_spin_budget(self)
            .as_millis()
            .min(u32::MAX as u128) as u32
    }

    fn executor_handle(&mut self) -> *mut core::ffi::c_void {
        // Phase 258 (Track 2, 2a) — hand the owned-spin entry a raw pointer to
        // the executor this runtime owns, so a Node pkg's `register(runtime)`
        // can install through `nros::__macro::install_node_typed` (same seam as the
        // C/C++ typed entries). The pointer is valid for the runtime's life
        // (the executor is an inline field); the install call uses it only
        // during registration, before any concurrent spin.
        &mut self.executor as *mut Executor<'static> as *mut core::ffi::c_void
    }

    // phase-446 W6 -- record the contract's declared parameters on the owned
    // executor; `ExecutorSink`'s parameter arm checks each declaration against
    // them. Without `param-services` there is no store, so the default no-op.
    #[cfg(feature = "param-store")]
    fn apply_declared_params(
        &mut self,
        nodes: &'static [(&'static str, &'static str)],
        params: &'static [(&'static str, &'static str, u8)],
    ) -> Result<(), &'static str> {
        self.executor
            .set_declared_params(nros_params::DeclaredParams::new(nodes, params));
        Ok(())
    }

    // Phase 264 W2 — register the REP-2002 lifecycle services + drive boot
    // autostart on the owned executor (mirrors `generate.rs::render_lifecycle_fn`).
    // Only compiled with `lifecycle-services`; without it the trait default no-op
    // applies, so a `[lifecycle]` block is silently inert (the Entry opts in by
    // enabling `nros/lifecycle-services`). `nros::main!` calls this when
    // `system.toml` declares `[lifecycle]`.
    #[cfg(feature = "lifecycle-services")]
    fn apply_lifecycle(&mut self, autostart: u8) -> Result<(), &'static str> {
        // issue 0460 — carry WHY. The caller can only say which capability
        // failed; without this the whole diagnostic was
        // `NodeRegister("lifecycle")`.
        self.executor
            .register_lifecycle_services()
            .map_err(|e| capability_reason(&e))?;
        if autostart >= 1
            && let Some(sm) = self.executor.lifecycle_state_machine_mut()
        {
            // No transition callbacks registered → each transition takes the
            // default-success path (REP-2002 skeleton), as the bake does.
            unsafe {
                let _ = sm.trigger_transition(crate::LifecycleTransition::Configure);
                if autostart >= 2 {
                    let _ = sm.trigger_transition(crate::LifecycleTransition::Activate);
                }
            }
        }
        Ok(())
    }

    // Phase 264 W4b -- register the 6 ROS 2 parameter services on the owned executor
    // and create the volatile param store. issue 1272 -- it no longer seeds the
    // launch `<param>` initials: those used to arrive here as one flat list and all
    // land on the primary node; now `ExecutorSink::seed_launch_params` declares each
    // node's own on the node `create_node` builds. Only
    // compiled with `param-services`; without it the trait default no-op applies, so a
    // `[param_services]` block is silently inert (the Entry opts in by enabling
    // `nros/param-services`). `nros::main!` calls this when `system.toml` declares
    // `[param_services]`. Reconfigured values (via `ros2 param set`) live in RAM until
    // the next boot — persistence is out of scope (issue 0080).
    // W4c note: `nros::main!` emits this BEFORE the per-node `register` calls, so the
    // store exists when each cell is created — `register_node_borrowed` / `register_node`
    // then capture the (stable, boxed) `ParameterServer` address on the cell, letting a
    // callback read `ctx.parameter::<T>(name)`. (The macro-path cells live in the
    // executor's tick registry, not `self.components`, so a post-pass here wouldn't reach
    // them — capture-at-registration does.)
    #[cfg(feature = "param-services")]
    fn apply_param_services(&mut self) -> Result<(), &'static str> {
        self.executor
            .register_parameter_services()
            .map_err(|e| capability_reason(&e))
    }

    // phase-461 W6 — the `params` axis: build the store before the per-node
    // `register` calls (so each cell captures it, as above) and register no
    // service. `ros2 param` reaches nothing in such an image, by declaration.
    #[cfg(feature = "param-store")]
    fn apply_param_store(&mut self) -> Result<(), &'static str> {
        self.executor
            .register_parameter_store()
            .map_err(|e| capability_reason(&e))
    }

    fn observed_callback_counts(&self) -> (usize, usize) {
        let direct = self
            .components
            .iter()
            .fold((0, 0), |(callbacks, messages), cell| {
                let (c, m) = cell.dispatch_counts();
                (callbacks + c, messages + m)
            });
        // issue #140 — install-seam components (`nros::node!` →
        // `install_node_typed*` → `register_node_borrowed`) never enter
        // `self.components`; their cells live only as the executor's enrolled
        // component slots (leaked `Arc<ComponentCell>`s). Without folding them
        // the hosted spin reported callbacks=0 for every macro-baked entry
        // (multihost robot2 et al.) while dispatch demonstrably ran. The two
        // populations are disjoint by construction: `register_node` pushes to
        // `components` and does not enroll; `register_node_borrowed` enrolls
        // and does not push.
        self.executor
            .enrolled_component_states()
            .fold(direct, |(callbacks, messages), state| {
                // SAFETY: every enrolled state is a leaked `Arc<ComponentCell>`
                // from `register_node_borrowed` (the only enroll site); the
                // executor keeps it alive until `Executor::drop`, and we only
                // read its atomic counters here.
                let header = unsafe { &*(state as *const CellHeader) };
                (
                    callbacks + header.callback_dispatches.load(Ordering::Relaxed),
                    messages + header.message_dispatches.load(Ordering::Relaxed),
                )
            })
    }
}

// =============================================================================
// Internal sink — bridges `NodeRuntime` declarations onto the
// live executor.
// =============================================================================

/// phase-483 W3 — the live component runtime's registration sink. It no
/// longer holds the executor: the [`crate::Node`] a component's `register`
/// holds borrows it, so the sink receives it per call (`FrameSink`).
struct ExecutorSink<'a> {
    cell: CellHandle,
    /// Per-registration node mapping: stable id → executor `NodeId` plus the
    /// EFFECTIVE `(name, namespace)` the node was created with (launch identity
    /// or the `NodeOptions` default) — phase-306 W3 needs it to expand
    /// `~`/relative entity names per node.
    /// W5-endgame alloc-off — heapless: a component class declares a small,
    /// statically bounded number of NODES (nearly always one). Full = loud
    /// registration error, the registries' rule.
    nodes: heapless::Vec<SinkNode, MAX_SINK_NODES>,
    /// Phase 268 W1 — launch-injected node identity `(name, namespace)` baked by
    /// `nros::main!` per component. When `Some`, `create_node` uses this identity
    /// instead of the `NodeOptions` default; `None` → default stands (backward-compat).
    node_identity: Option<(&'static str, &'static str)>,
    /// Phase 305 W3 (issue 0255) — launch `<remap from= to=/>` rules baked by
    /// `nros::main!` for this component. Applied in `create_entity` via the
    /// shared `node_metadata::resolve_name` seam (exact-FQN match, first rule
    /// wins). Empty → names still get ROS 2 expansion, no substitution.
    remaps: &'a [(&'a str, &'a str)],
    /// Issue #52 — the component's baked QoS-override codes, installed on each
    /// node this sink creates (`Executor::set_node_qos_overrides`) BEFORE the
    /// component declares entities, so `create_publisher`/`create_subscription`
    /// fold the matching ones in. Empty → nothing installed, zero cost.
    qos_overrides: &'static [nros_node::executor::node_record::QoSOverrideCode],
    /// issue 1272 -- the launch `<param>` initials `nros::main!` baked for this
    /// component (`RuntimeCtx::params`). `create_node` seeds them on the node it
    /// builds (see [`ExecutorSink::seed_launch_params`]). Empty for the direct
    /// API, which has no launch.
    #[cfg_attr(not(feature = "param-store"), allow(dead_code))]
    launch_params: &'a [(&'a str, &'a str)],
}

struct SinkNode {
    stable_id: IdStr,
    node_id: nros_node::executor::NodeId,
    name: IdStr,
    namespace: IdStr,
}

impl ExecutorSink<'_> {
    fn lookup_node(&self, stable_id: &str) -> Option<&SinkNode> {
        self.nodes.iter().find(|n| n.stable_id == stable_id)
    }

    /// issue 1272 -- declare this component's launch `<param>` initials on
    /// `node`, the node `create_node` has just built.
    ///
    /// Before 1272 `nros::main!` flattened every node's initials into one list
    /// and `apply_param_services` declared them all on `NodeId::PRIMARY`, ahead
    /// of the `register` calls: node 0 held every node's values, and a second
    /// node's identical name was refused as a duplicate. Seeding here keys each
    /// value on the `NodeId` construction actually returned, so no index has
    /// to be predicted and none can drift.
    ///
    /// Only when the executor already has a parameter store, i.e. the system
    /// declared `[param_services]` and `apply_param_services` ran before this
    /// `register` call. That is the condition the flat seed had.
    ///
    /// A `false` from the store is a refusal only if the name is still absent
    /// afterwards: the store also answers `false` for "already declared".
    #[cfg(feature = "param-store")]
    fn seed_launch_params(
        &mut self,
        executor: &mut Executor<'static>,
        node: nros_node::executor::NodeId,
    ) -> NodeResult<()> {
        if executor.params().is_none() {
            return Ok(());
        }
        let params = self.launch_params;
        for (name, raw) in params {
            // phase-446 seam: the value's type is inferred from the launch
            // string; a type declared in the contract replaces this call.
            let value = infer_param_value(raw);
            if !executor.declare_parameter_on(node, name, value)
                && executor.get_parameter_on(node, name).is_none()
            {
                nros_log::log_error!(
                    nros_log::get_logger("nros"),
                    "launch parameter '{name}' for node index {} was refused by the \
                     parameter store",
                    node.raw()
                );
                return Err(NodeDeclError::Runtime);
            }
        }
        Ok(())
    }
}

impl crate::node::FrameSink for ExecutorSink<'_> {
    fn create_node(
        &mut self,
        executor: &mut Executor<'static>,
        id: MetaNodeId<'_>,
        options: NodeOptions<'_>,
    ) -> NodeResult<nros_node::executor::NodeId> {
        if self.nodes.iter().any(|n| n.stable_id == id.as_str()) {
            return Err(NodeDeclError::Runtime);
        }
        // Phase 268 W1 — launch wins over the NodeOptions default (RFC-0046).
        // When `nros::main!` injected an identity for this component, use it;
        // otherwise fall back to what the Node declared in its `create_node` call.
        let (name, ns) = match self.node_identity {
            Some((n, s)) => (n, s),
            None => (options.name, options.namespace),
        };
        let node_id = executor
            .node_builder(name)
            .namespace(ns)
            .domain_id(options.domain_id)
            .build()
            .map_err(decl_err_from_node)?;
        // Issue 1419 — open this node in the census recorder so the entities
        // `register()` declares next attribute to it (the RMW seam carries no
        // node). The namespace the node LANDED in, read back from the
        // executor, as the C and C++ node entry points report it. No-op unless
        // `metadata-mode` is on -- and under `ACTIVE`, because the namespace
        // lookup is work an empty hook body does not remove (phase-463 W5 I4).
        if crate::census_hooks::ACTIVE {
            let landed = match executor.node(node_id).map(|r| r.namespace.as_str()) {
                Some(s) if !s.is_empty() => s,
                _ => "/",
            };
            crate::census_hooks::on_node_create(name, landed, options.domain_id);
        }
        // Issue #52 — install the bake BEFORE the component declares any
        // entity on this node; entities created earlier could not be folded.
        if !self.qos_overrides.is_empty() {
            executor.set_node_qos_overrides(node_id, self.qos_overrides);
        }
        // issue 1272 -- the launch values belong to the component's launch
        // node, which is the FIRST node it builds (the same node its cell reads
        // parameters from, see `register_node_borrowed`). Seeded before the
        // component declares any entity, so its own declarations adopt them.
        #[cfg(feature = "param-store")]
        if self.nodes.is_empty() {
            self.seed_launch_params(executor, node_id)?;
        }
        self.nodes
            .push(SinkNode {
                stable_id: id_str(id.as_str())?,
                node_id,
                name: id_str(name)?,
                namespace: id_str(ns)?,
            })
            .map_err(|_| NodeDeclError::Runtime)?;
        Ok(node_id)
    }

    fn create_entity(
        &mut self,
        executor: &mut Executor<'static>,
        metadata: EntityMetadata,
    ) -> NodeResult<()> {
        // Phase 228.C tier gate: when this executor runs a specific tier
        // (`active_groups` set by codegen), an entity whose callback group
        // is not active on this tier is a no-op — no RMW handle, no slot.
        // An unlabeled entity (`callback_group == None`) is wildcard-eligible
        // and always registers; the degenerate single-tier executor leaves
        // `active_groups == None`, so every entity registers (byte-identical
        // to pre-228 output).
        //
        // issue 1172 — the lookup moved ABOVE this gate because the gate is now
        // node-qualified. It used to ask "is this GROUP active", which cannot
        // tell node A's `ctrl` from node B's `ctrl`: two nodes may legitimately
        // declare the same group id on different tiers, and the tier then
        // admitted entities it did not own. Reordering is safe — a missing node
        // was already `NodeDeclError::Runtime` three lines below, so the only
        // change is which error wins for an entity that is BOTH off-tier and
        // owned by an unknown node, and that combination is a bug either way.
        let (node, node_name, node_ns) = {
            let entry = self
                .lookup_node(metadata.node_id.as_str())
                .ok_or(NodeDeclError::Runtime)?;
            (entry.node_id, entry.name.clone(), entry.namespace.clone())
        };
        if let Some(group) = metadata.callback_group.as_ref()
            && !executor.group_active(node_name.as_str(), node_ns.as_str(), group.as_str())
        {
            return Ok(());
        }
        // Phase 305 W3 (issue 0255) — expand `~`/relative source names against
        // the owning node's identity and apply the launch remap rules (shared
        // `nros_node::names` seam) before any name reaches the wire.
        // Timers have no wire name; parameter names are NOT remapped (matches
        // ROS 2 basic name remapping — param remaps are a separate rule class).
        //
        // Two tiers: `self.remaps` is the launch projection and therefore
        // AUTHORITATIVE; the executor's `--ros-args` rules (installed by
        // `nros::Context::create_executor` from `init_with_args`) are the
        // FALLBACK, consulted only when no launch rule matched — RFC-0046's
        // precedence, and rcl's local-before-global. Executors no `Context`
        // seeded carry no argv rules, so for every generated image this is the
        // launch slice alone, as it always was.
        let resolved_name = match metadata.kind {
            EntityKind::Timer | EntityKind::Parameter => None,
            _ => Some(
                nros_node::names::resolve_name_layered(
                    metadata.source_name.as_str(),
                    &node_name,
                    &node_ns,
                    self.remaps.iter().copied(),
                    nros_node::ros_args::argv_fallback(&*executor, &node_name, &node_ns),
                )
                .map_err(|_| NodeDeclError::Runtime)?,
            ),
        };
        let entity_name = resolved_name.as_ref().map(|r| r.as_str()).unwrap_or("");
        match metadata.kind {
            EntityKind::Publisher => {
                // Issue 0306 — the node's DECLARED profile
                // (`create_publisher_for_topic_with_qos`) rides
                // `metadata.qos`; this used to call the default-QoS
                // constructor, so every declarative entity was created
                // `QoSProfile::default()` and a node's own QoS was silently
                // discarded. Plan overrides still win: they fold in below this,
                // inside the executor's create path.
                let handle = executor
                    .node_mut(node)
                    .create_generic_publisher_with_qos(
                        entity_name,
                        metadata.type_name,
                        metadata.type_hash,
                        metadata.qos,
                    )
                    .map_err(decl_err_from_node)?;
                let id_owned = id_str(metadata.id.as_str())?;
                // Registry full = this component declared more entities than
                // CELL_REG_CAP — a registration error, never a silent drop.
                self.cell
                    .view()
                    .push_publisher(id_owned, handle)
                    .map_err(|_| cell_full("publishers"))?;
                Ok(())
            }
            EntityKind::Subscription => {
                let cb_id = metadata
                    .callback_id
                    .as_ref()
                    .ok_or(NodeDeclError::Runtime)?;
                let cb_id_owned = id_str(cb_id.as_str())?;
                let cell = self.cell.clone();
                // phase-457 W5 (issue 1522) — THE branch, and it is on the
                // CLASSIFIER, not on `metadata.safety`.
                //
                // The metadata recorder states this subscription's
                // `in_place_capable` from the same call, because a Rust probe
                // opens no executor and so observes no registration
                // (`registration_observer` never fires on this road). Testing
                // the flag here as well would make the probe's answer a second
                // opinion about this site rather than a reading of it — which
                // is exactly what W3 deleted from the CLI. Gated by
                // `check-declared-subscription-shape`: every variant of
                // `DeclaredSubscriptionShape` must be named in this arm, so a
                // third lowering cannot fall through to the basic path while
                // the probe keeps stating the basic path's answer.
                let shape = metadata.declared_subscription_shape();
                #[cfg(not(feature = "safety-e2e"))]
                let _ = shape;
                // Phase 250 (Wave 2b) — a `.safety()` subscription registers via
                // the integrity-aware generic path so `CallbackCtx::integrity()`
                // surfaces CRC + sequence gap/dup. Gated: when `safety-e2e` is off
                // `declared_subscription_shape` masks the flag, never answers
                // `BufferedRawSafety`, and the basic path below runs.
                #[cfg(feature = "safety-e2e")]
                use nros_node::executor::declared_shape::DeclaredSubscriptionShape;
                #[cfg(feature = "safety-e2e")]
                if shape == DeclaredSubscriptionShape::BufferedRawSafety {
                    let cell_s = self.cell.clone();
                    let cb_s = cb_id_owned.clone();
                    executor
                        .node_mut(node)
                        .create_generic_subscription_with_integrity_and_rx_hint(
                            entity_name,
                            metadata.type_name,
                            metadata.type_hash,
                            metadata.rx_bound,
                            move |payload: &[u8], status: &nros_node::IntegrityStatus| {
                                dispatch_into_cell_with_integrity(
                                    cell_s.view(),
                                    &cb_s,
                                    payload,
                                    status,
                                );
                            },
                        )
                        .map_err(decl_err_from_node)?;
                    return Ok(());
                }
                // `DeclaredSubscriptionShape::BufferedRaw` — the only shape
                // left, named so the gate can see that this arm handles it.
                //
                // Issue 0306 — same as the publisher branch: honour the
                // node's declared profile instead of defaulting it.
                executor
                    .node_mut(node)
                    // Issue 2208 -- with the type's bound the declaration
                    // recorded, so the backend gets the same `rx_buffer_hint`
                    // the typed executor path gives it.
                    .create_generic_subscription_with_qos_and_rx_hint(
                        entity_name,
                        metadata.type_name,
                        metadata.type_hash,
                        metadata.qos,
                        metadata.rx_bound,
                        move |payload: &[u8]| {
                            dispatch_into_cell(cell.view(), &cb_id_owned, payload);
                        },
                    )
                    .map_err(decl_err_from_node)?;
                Ok(())
            }
            EntityKind::Timer => {
                let cb_id = metadata
                    .callback_id
                    .as_ref()
                    .ok_or(NodeDeclError::Runtime)?;
                let cb_id_owned = id_str(cb_id.as_str())?;
                // Issue #505 — prefer the microsecond field; fall back to
                // the millisecond one only for metadata that predates it.
                let period = match metadata.period_us {
                    Some(us) => nros_node::TimerDuration::from_micros(us),
                    None => nros_node::TimerDuration::from_millis(
                        metadata.period_ms.ok_or(NodeDeclError::Runtime)?,
                    ),
                };
                let cell = self.cell.clone();
                // phase-430 W4 — ONE registrar for both spellings. A timer
                // declared with no clock carries `TimerClockSource::Steady`,
                // which is the arena entry `register_timer` built anyway (same
                // fields, same seed), so the wall case is unchanged and a `Ros`
                // one now reaches the executor at all. Before W4 this site
                // called `register_timer` unconditionally and the declarative
                // surface could not express a clock, which is why a
                // `nros::main!` component had no way to own a ROS-time timer.
                //
                // Issue 2038 — and ONE registrar for the call info and the
                // autostart flag too: `register_timer_with_info` is the entry
                // `register_timer_on_clock` builds (same arena entry, same
                // seed) with the dispatcher's `TimerCallInfo` handed to the
                // closure and rcl's `autostart` honoured, the call
                // `Node::create_timer_with_info` makes. The info reaches the
                // component as `CallbackCtx::timer_info`.
                let autostart = metadata.timer_autostart;
                let handle = executor
                    .register_timer_with_info(
                        period,
                        metadata.timer_clock,
                        autostart,
                        move |info: &nros_node::TimerCallInfo| {
                            dispatch_timer_into_cell(cell.view(), &cb_id_owned, info);
                        },
                    )
                    .map_err(decl_err_from_node)?;
                // Issue 2075 — file the handle so a context can reset or
                // cancel it by id. A full table leaves the timer running and
                // uncontrollable (`UnknownTimer` names the bound when it is
                // asked for); a key collision is a loud registration error.
                self.cell
                    .view()
                    .push_timer(timer_key(metadata.id.as_str()), handle)
                    .map_err(|()| NodeDeclError::Metadata(NodeMetadataError::DuplicateId))?;
                // Issue 1419 — a Rust timer never reaches the RMW, so this hook
                // is how it enters the census. Recorded AFTER the tier gate
                // above, so a tiered entry's census holds each timer once, on
                // the tier that owns it. `Wall` on the steady clock, `Clock`
                // on any other — the split the C and C++ entry points report.
                crate::census_hooks::on_timer_create(
                    match metadata.timer_clock {
                        nros_node::executor::TimerClockSource::Steady => {
                            crate::node_metadata::TimerKind::Wall
                        }
                        _ => crate::node_metadata::TimerKind::Clock,
                    },
                    period.as_millis(),
                );
                Ok(())
            }
            // Phase 212.M-F.23 — service / action client + server dispatch on
            // the single-node runtime. The executor-level `register_*_on`
            // calls add an arena dispatch entry, so inbound requests / goals
            // are serviced inside `spin_once`; the leaked `*Ctx` trampoline
            // contexts bridge back into the component's `on_callback`. Client
            // handles are stashed in the cell for the tick-side dispatch
            // (`RuntimeClientDispatch` / `RuntimeActions` in `run_ticks`).
            EntityKind::ServiceServer => {
                let cb_id = metadata
                    .callback_id
                    .as_ref()
                    .ok_or(NodeDeclError::Runtime)?;
                let ctx = self
                    .cell
                    .view()
                    .place_service_ctx(ServiceServerCtx {
                        cell: self.cell.clone(),
                        callback_id: id_str(cb_id.as_str())?,
                    })
                    .map_err(|_| cell_full("service servers"))?;
                executor
                    .register_service_raw_sized_on::<1024, 1024>(
                        node,
                        entity_name,
                        metadata.type_name,
                        metadata.type_hash,
                        crate::QoSProfile::services_default(),
                        service_server_trampoline,
                        ctx,
                    )
                    .map_err(decl_err_from_node)?;
                Ok(())
            }
            EntityKind::ServiceClient => {
                let hid = executor
                    .register_service_client_raw_sized_on::<1024>(
                        node,
                        entity_name,
                        metadata.type_name,
                        metadata.type_hash,
                        crate::QoSProfile::services_default(),
                        None,
                        core::ptr::null_mut(),
                    )
                    .map_err(decl_err_from_node)?;
                self.cell
                    .view()
                    .push_service_client(id_str(metadata.id.as_str())?, hid)
                    .map_err(|_| cell_full("service clients"))?;
                Ok(())
            }
            EntityKind::ActionServer => {
                let goal_cb = metadata
                    .callback_id
                    .as_ref()
                    .ok_or(NodeDeclError::Runtime)?;
                let cancel_cb = metadata
                    .action_cancel_callback_id
                    .as_ref()
                    .ok_or(NodeDeclError::Runtime)?;
                let accepted_cb = metadata
                    .action_accepted_callback_id
                    .as_ref()
                    .map(|c| id_str(c.as_str()))
                    .transpose()?;
                let ctx = self
                    .cell
                    .view()
                    .place_action_server_ctx(ActionServerCtx {
                        cell: self.cell.clone(),
                        goal_callback_id: id_str(goal_cb.as_str())?,
                        cancel_callback_id: id_str(cancel_cb.as_str())?,
                        accepted_callback_id: accepted_cb,
                    })
                    .map_err(|_| cell_full("action servers"))?;
                let handle = executor
                    .register_action_server_raw_sized::<1024, 1024, 1024, 4>(
                        crate::internals::RawActionServerSpec {
                            node_id: Some(node),
                            action_name: entity_name,
                            type_name: metadata.type_name,
                            type_hash: metadata.type_hash,
                            hashes: metadata.action_hashes.ok_or(NodeDeclError::Runtime)?,
                            qos: crate::QoSProfile::services_default(),
                            goal_callback: action_goal_trampoline,
                            cancel_callback: action_cancel_trampoline,
                            accepted_callback: Some(action_accepted_trampoline),
                            context: ctx,
                        },
                    )
                    .map_err(decl_err_from_node)?;
                self.cell
                    .view()
                    .push_action_server(id_str(metadata.id.as_str())?, handle)
                    .map_err(|_| cell_full("action servers"))?;
                Ok(())
            }
            EntityKind::ActionClient => {
                // A bound `callback_id` (set by
                // `create_action_client_with_callbacks_for_name`) delivers the
                // terminal goal result to the component via `on_callback`; the
                // optional `action_accepted_callback_id` slot carries the
                // feedback callback (reused — unused on a client). The executor
                // auto-drives accept → feedback → result during spin and invokes
                // these trampolines. No callbacks → send-goal only.
                let (result_callback, feedback_callback, ctx) = match metadata.callback_id.as_ref()
                {
                    Some(result_cb) => {
                        let feedback_cb = metadata
                            .action_accepted_callback_id
                            .as_ref()
                            .map(|c| id_str(c.as_str()))
                            .transpose()?;
                        let ctx = self
                            .cell
                            .view()
                            .place_action_client_ctx(ActionClientCtx {
                                cell: self.cell.clone(),
                                result_callback_id: id_str(result_cb.as_str())?,
                                feedback_callback_id: feedback_cb.clone(),
                            })
                            .map_err(|_| cell_full("action clients"))?;
                        let fb = feedback_cb.map(|_| action_feedback_trampoline as _);
                        (Some(action_result_trampoline as _), fb, ctx)
                    }
                    None => (None, None, core::ptr::null_mut()),
                };
                let handle = executor
                    .register_action_client_raw_sized::<1024, 1024, 1024>(
                        crate::internals::RawActionClientSpec {
                            node_id: Some(node),
                            action_name: entity_name,
                            type_name: metadata.type_name,
                            type_hash: metadata.type_hash,
                            hashes: metadata.action_hashes.ok_or(NodeDeclError::Runtime)?,
                            goal_response_callback: None,
                            feedback_callback,
                            result_callback,
                            cancel_response_callback: None,
                            context: ctx,
                        },
                    )
                    .map_err(decl_err_from_node)?;
                self.cell
                    .view()
                    .push_action_client(id_str(metadata.id.as_str())?, handle.entry_index())
                    .map_err(|_| cell_full("action clients"))?;
                Ok(())
            }
            EntityKind::Parameter => {
                // Issue 1419 — the declaration as the code made it, BEFORE the
                // store answers and whether or not `param-services` gives this
                // image a store at all: the census describes the code.
                crate::census_hooks::on_param_declare(
                    metadata.source_name.as_str(),
                    &param_default_to_value(metadata.parameter_default.as_ref()),
                );
                // Phase 212.M-F.23 Wave 2 — declarative parameter dispatch on
                // the single-node runtime. The first declared parameter lazily
                // stands up the 6 ROS 2 parameter services for this executor's
                // node; `spin_once` drives those service servers thereafter
                // (`#[cfg(param-services)]` block at spin.rs). The declared
                // source default seeds the value. With `param-services` off the
                // arm is a no-op (entity declared, no RMW handle) — byte-
                // identical to the pre-Wave-2 behavior.
                #[cfg(feature = "param-store")]
                {
                    // Issue 1897 -- a store-only build has no services to
                    // stand up; it builds the store and serves nothing.
                    if executor.params().is_none() {
                        #[cfg(feature = "param-services")]
                        executor
                            .register_parameter_services()
                            .map_err(decl_err_from_node)?;
                        #[cfg(not(feature = "param-services"))]
                        executor
                            .register_parameter_store()
                            .map_err(decl_err_from_node)?;
                    }
                    let value = param_default_to_value(metadata.parameter_default.as_ref());
                    let name = metadata.source_name.as_str();
                    // phase-446 W6 -- the code must declare what the node's
                    // contract declares: the store was sized from it (W4). A
                    // node whose contract has no `params:` is not checked.
                    if let Err(m) = executor.declared_params().check(
                        node_ns.as_str(),
                        node_name.as_str(),
                        name,
                        value.param_type(),
                    ) {
                        let passed = nros_params::declared::type_name(value.param_type());
                        match m.declared {
                            None => nros_log::log_error!(
                                nros_log::get_logger("nros"),
                                "node '{}': parameter '{}' is not declared in its contract \
                                 ({}, `params:`). The parameter store is sized from that \
                                 declaration; declare it there with its type, or stop \
                                 declaring it in the code. (phase-446 W6)",
                                m.node,
                                name,
                                m.contract
                            ),
                            Some(declared) => nros_log::log_error!(
                                nros_log::get_logger("nros"),
                                "node '{}': parameter '{}' is declared `{}` in its contract \
                                 ({}) but the code declares it as `{}`. Fix the contract or \
                                 the code so they state one type. (phase-446 W6)",
                                m.node,
                                name,
                                nros_params::declared::type_name(declared),
                                m.contract,
                                passed
                            ),
                        }
                        return Err(NodeDeclError::Runtime);
                    }
                    // Same rule as `seed_launch_params`: already-declared is
                    // not a refusal (the launch seed may have declared it
                    // first); refused-and-absent is — and it gets its own
                    // variant, not the opaque `Runtime` (issue 1260). On THIS
                    // entity's node, not the executor's primary (issue 1272):
                    // a second node declaring a name the first also declares
                    // is its own parameter, not a duplicate.
                    if !executor.declare_parameter_on(node, name, value)
                        && executor.get_parameter_on(node, name).is_none()
                    {
                        nros_log::log_error!(
                            nros_log::get_logger("nros"),
                            "declared parameter '{name}' was refused by the parameter store"
                        );
                        return Err(NodeDeclError::ParameterRejected);
                    }
                }
                Ok(())
            }
        }
    }

    fn record_callback_effect(
        &mut self,
        _executor: &mut Executor<'static>,
        _callback_id: CallbackId<'_>,
        _kind: CallbackEffectKind,
        _entity_id: EntityId<'_>,
    ) -> NodeResult<()> {
        // Planner concern only — the live runtime doesn't need the
        // effect graph at spin time.
        Ok(())
    }
}

/// Lower a source-recorded [`ParameterDefault`] into the executor-facing
/// [`nros_params::ParameterValue`] used to seed a declared parameter. Scalar
/// defaults carry their value directly; the array variants record only the
/// declared type (no element data at the source layer) so they seed as
/// `NotSet` — the parameter is still declared, just without a concrete array
/// default. A `Double` default is stored as a string at the metadata layer and
/// parsed here (unparseable → `0.0`).
fn param_default_to_value(
    default: Option<&crate::node_metadata::ParameterDefault>,
) -> nros_params::ParameterValue {
    use crate::node_metadata::ParameterDefault;
    use nros_params::ParameterValue;
    match default {
        None => ParameterValue::NotSet,
        Some(ParameterDefault::Bool(b)) => ParameterValue::Bool(*b),
        Some(ParameterDefault::Integer(i)) => ParameterValue::Integer(*i),
        Some(ParameterDefault::Double(s)) => {
            ParameterValue::Double(s.as_str().parse::<f64>().unwrap_or(0.0))
        }
        Some(ParameterDefault::String(s)) => {
            ParameterValue::from_string(s.as_str()).unwrap_or(ParameterValue::NotSet)
        }
        Some(
            ParameterDefault::BoolArray
            | ParameterDefault::IntegerArray
            | ParameterDefault::DoubleArray
            | ParameterDefault::StringArray,
        ) => ParameterValue::NotSet,
    }
}

/// Phase 264 W4b — infer a [`nros_params::ParameterValue`] from a raw launch
/// `<param value=…/>` string. ROS 2 launch `<param>` values are untyped strings; the
/// macro path has no per-param type attribute, so infer: `true`/`false` → `Bool`, an
/// `i64` literal → `Integer`, an `f64` literal → `Double`, otherwise `String`. This
/// mirrors the type a `ros2 param set` of the same literal would land on, so the baked
/// initial and a CLI override agree on type.
#[cfg(feature = "param-store")]
fn infer_param_value(raw: &str) -> nros_params::ParameterValue {
    use nros_params::ParameterValue;
    match raw {
        "true" => return ParameterValue::from_bool(true),
        "false" => return ParameterValue::from_bool(false),
        _ => {}
    }
    if let Ok(i) = raw.parse::<i64>() {
        return ParameterValue::from_integer(i);
    }
    if let Ok(f) = raw.parse::<f64>() {
        return ParameterValue::from_double(f);
    }
    ParameterValue::from_string(raw).unwrap_or(ParameterValue::NotSet)
}

fn dispatch_into_cell(cell: &dyn CellView, cb_id: &str, payload: &[u8]) {
    cell.note_dispatch(!payload.is_empty());
    let resolver = CellResolver { cell };
    let mut ctx = CallbackCtx::new(payload, &resolver);
    // W4c — let the callback read `ctx.parameter::<T>(name)` from the executor's store
    // (threaded onto the cell by `apply_param_services`; `None` until then).
    #[cfg(feature = "param-store")]
    ctx.set_param_server(cell.view_param_server(), cell.view_param_node());
    // If the slot is already borrowed (a re-entrant publish from a
    // tick hook on the same cell, etc.) the view drops this dispatch. In
    // practice the borrow succeeds because subscription / timer
    // callbacks run sequentially under the single-threaded executor.
    cell.try_with_slot_mut(&mut |slot| slot.dispatch(cb_id, &mut ctx));
}

/// Issue 2038 — dispatch a declared timer's activation into the component's
/// `on_callback`, with its call info readable as `CallbackCtx::timer_info`.
/// The timer twin of [`dispatch_into_cell`] (a timer has no payload).
fn dispatch_timer_into_cell(cell: &dyn CellView, cb_id: &str, info: &nros_node::TimerCallInfo) {
    cell.note_dispatch(false);
    let resolver = CellResolver { cell };
    let mut ctx = CallbackCtx::new(&[], &resolver);
    ctx.set_timer_info(*info);
    #[cfg(feature = "param-store")]
    ctx.set_param_server(cell.view_param_server(), cell.view_param_node());
    cell.try_with_slot_mut(&mut |slot| slot.dispatch(cb_id, &mut ctx));
}

/// Phase 250 (Wave 2b) — dispatch a `.safety()` subscription message into the
/// component's `on_callback` with its E2E [`IntegrityStatus`] attached, read via
/// `CallbackCtx::integrity()`. The integrity-aware twin of [`dispatch_into_cell`].
#[cfg(feature = "safety-e2e")]
fn dispatch_into_cell_with_integrity(
    cell: &dyn CellView,
    cb_id: &str,
    payload: &[u8],
    status: &nros_node::IntegrityStatus,
) {
    cell.note_dispatch(!payload.is_empty());
    let resolver = CellResolver { cell };
    let mut ctx = CallbackCtx::new_with_integrity(payload, &resolver, status);
    // W4c — param store for a `.safety()` subscription callback too.
    #[cfg(feature = "param-store")]
    ctx.set_param_server(cell.view_param_server(), cell.view_param_node());
    cell.try_with_slot_mut(&mut |slot| slot.dispatch(cb_id, &mut ctx));
}

// =============================================================================
// Phase 212.M-F.23 — service / action SERVER trampolines + tick-side client /
// action dispatch.
//
// The executor's raw service/action-server registration takes C-ABI fn
// pointers, so the runtime leaks a `*Ctx` (lives for the runtime's lifetime,
// like the executor) holding the owning `ComponentCell` + the declared
// callback ids. Each trampoline rebuilds a `CallbackCtx` and routes into the
// component's `on_callback`, exactly as the orchestration codegen's
// svc_tramp_* / goal_tramp_* do for the Entry path.
// =============================================================================

/// Leaked context for a service-server arena callback.
struct ServiceServerCtx {
    cell: CellHandle,
    callback_id: IdStr,
}

/// Leaked context for an action-server arena callback (goal + cancel + the
/// optional accepted hook all share one).
struct ActionServerCtx {
    cell: CellHandle,
    goal_callback_id: IdStr,
    cancel_callback_id: IdStr,
    accepted_callback_id: Option<IdStr>,
}

/// Leaked context for an action-CLIENT result + feedback callbacks.
struct ActionClientCtx {
    cell: CellHandle,
    result_callback_id: IdStr,
    feedback_callback_id: Option<IdStr>,
}

/// Action-client result callback: the executor's spin auto-drives the goal to
/// completion and hands the terminal result CDR here; route it into the
/// component's `on_callback` (read with `CallbackCtx::message`).
unsafe extern "C" fn action_result_trampoline(
    _goal_id: *const GoalUuid,
    _status: GoalStatusCode,
    result_data: *const u8,
    result_len: usize,
    ctx: *mut core::ffi::c_void,
) {
    let actx = unsafe { &*(ctx as *const ActionClientCtx) };
    let result_slice = unsafe { core::slice::from_raw_parts(result_data, result_len) };
    dispatch_into_cell(actx.cell.view(), &actx.result_callback_id, result_slice);
}

/// Action-client feedback callback: route each feedback CDR into the
/// component's `on_callback` under the bound feedback callback id.
unsafe extern "C" fn action_feedback_trampoline(
    _goal_id: *const GoalUuid,
    feedback_data: *const u8,
    feedback_len: usize,
    ctx: *mut core::ffi::c_void,
) {
    let actx = unsafe { &*(ctx as *const ActionClientCtx) };
    let Some(cb_id) = actx.feedback_callback_id.as_ref() else {
        return;
    };
    let feedback_slice = unsafe { core::slice::from_raw_parts(feedback_data, feedback_len) };
    dispatch_into_cell(actx.cell.view(), cb_id, feedback_slice);
}

/// Service-server request callback: deserialize-side runs in the component's
/// `on_callback` via `CallbackCtx::with_reply`; the executor sends the reply
/// from the bytes written into `resp`.
unsafe extern "C" fn service_server_trampoline(
    req: *const u8,
    req_len: usize,
    resp: *mut u8,
    resp_cap: usize,
    resp_len: *mut usize,
    ctx: *mut core::ffi::c_void,
) -> bool {
    let sctx = unsafe { &*(ctx as *const ServiceServerCtx) };
    let req_slice = unsafe { core::slice::from_raw_parts(req, req_len) };
    let resp_slice = unsafe { core::slice::from_raw_parts_mut(resp, resp_cap) };
    let mut written = 0usize;
    let view = sctx.cell.view();
    let resolver = CellResolver { cell: view };
    let mut cb = CallbackCtx::with_reply(req_slice, &resolver, resp_slice, &mut written);
    // W4c — service-server callback can read `ctx.parameter::<T>(name)` too.
    #[cfg(feature = "param-store")]
    cb.set_param_server(view.view_param_server(), view.view_param_node());
    view.try_with_slot_mut(&mut |slot| slot.dispatch(&sctx.callback_id, &mut cb));
    unsafe { *resp_len = written };
    true
}

/// Action-server goal callback → component `on_callback` with a goal decision.
unsafe extern "C" fn action_goal_trampoline(
    _goal_id: *const GoalUuid,
    goal_data: *const u8,
    goal_len: usize,
    ctx: *mut core::ffi::c_void,
) -> crate::GoalResponse {
    let actx = unsafe { &*(ctx as *const ActionServerCtx) };
    let goal_slice = unsafe { core::slice::from_raw_parts(goal_data, goal_len) };
    let mut resp = crate::GoalResponse::Reject;
    let view = actx.cell.view();
    let resolver = CellResolver { cell: view };
    let mut cb = CallbackCtx::with_goal_decision(goal_slice, &resolver, &mut resp);
    // W4c — action goal callback can read `ctx.parameter::<T>(name)` too.
    #[cfg(feature = "param-store")]
    cb.set_param_server(view.view_param_server(), view.view_param_node());
    view.try_with_slot_mut(&mut |slot| slot.dispatch(&actx.goal_callback_id, &mut cb));
    resp
}

/// Action-server cancel callback → component `on_callback` with a cancel
/// decision. The cancel callback has no goal payload.
unsafe extern "C" fn action_cancel_trampoline(
    _goal_id: *const GoalUuid,
    _status: GoalStatusCode,
    ctx: *mut core::ffi::c_void,
) -> crate::CancelResponse {
    let actx = unsafe { &*(ctx as *const ActionServerCtx) };
    let mut resp = crate::CancelResponse::Reject;
    let view = actx.cell.view();
    let resolver = CellResolver { cell: view };
    let mut cb = CallbackCtx::with_cancel_decision(&[], &resolver, &mut resp);
    // W4c — action cancel callback can read `ctx.parameter::<T>(name)` too.
    #[cfg(feature = "param-store")]
    cb.set_param_server(view.view_param_server(), view.view_param_node());
    view.try_with_slot_mut(&mut |slot| slot.dispatch(&actx.cancel_callback_id, &mut cb));
    resp
}

/// Action-server accepted hook → component `on_callback` (no decision, no
/// payload). No-op when the component didn't declare an accepted callback.
unsafe extern "C" fn action_accepted_trampoline(
    _goal_id: *const GoalUuid,
    ctx: *mut core::ffi::c_void,
) {
    let actx = unsafe { &*(ctx as *const ActionServerCtx) };
    let Some(cb_id) = actx.accepted_callback_id.as_ref() else {
        return;
    };
    dispatch_into_cell(actx.cell.view(), cb_id, &[]);
}

/// Tick-side service/action CLIENT dispatch — the single-node runtime's own
/// (it mirrored the orchestration's ~~`GenClientDispatch`~~, since retired). Resolves the per-component client
/// handle arrays + a `*mut Executor<'static>` (the tick borrows `&components` while
/// needing `&mut executor`, so the executor is reached through a raw pointer,
/// reborrowed `&mut` per call; no aliasing — `executor` and `components` are
/// disjoint fields).
struct RuntimeClientDispatch<'a> {
    executor: *mut Executor<'static>,
    services: &'a [(IdStr, crate::HandleId)],
    actions: &'a [(IdStr, usize)],
}

impl RuntimeClientDispatch<'_> {
    fn service(&self, entity: &str) -> NodeResult<crate::HandleId> {
        self.services
            .iter()
            .find(|(e, _)| e == entity)
            .map(|(_, h)| *h)
            .ok_or(NodeDeclError::Runtime)
    }

    fn action_entry(&self, entity: &str) -> NodeResult<usize> {
        self.actions
            .iter()
            .find(|(e, _)| e == entity)
            .map(|(_, i)| *i)
            .ok_or(NodeDeclError::Runtime)
    }
}

impl ClientDispatch for RuntimeClientDispatch<'_> {
    fn call_raw(
        &mut self,
        service_entity: &str,
        request_cdr: &[u8],
        response_buf: &mut [u8],
    ) -> NodeResult<usize> {
        use nros_rmw::ClientTrait;
        let hid = self.service(service_entity)?;
        {
            let executor = unsafe { &mut *self.executor };
            let slot = executor.resolve_handle(hid).ok_or(NodeDeclError::Runtime)?;
            let entry =
                unsafe { executor.service_client_entry_mut(slot) }.ok_or(NodeDeclError::Runtime)?;
            let seq = entry
                .handle
                .send_request_raw(request_cdr)
                .map_err(|_| NodeDeclError::Runtime)?;
            // Issue 2009 — only the reply naming this request completes the
            // call; an earlier call's late reply is dropped.
            entry.record_send(seq, false);
        }
        // Bounded wait — caps total time so the tick loop stays responsive.
        for _ in 0..200 {
            let executor = unsafe { &mut *self.executor };
            executor.spin_once(core::time::Duration::from_millis(10));
            let slot = executor.resolve_handle(hid).ok_or(NodeDeclError::Runtime)?;
            let entry =
                unsafe { executor.service_client_entry_mut(slot) }.ok_or(NodeDeclError::Runtime)?;
            match entry.take_reply(response_buf) {
                Ok(Some(len)) => return Ok(len),
                Ok(None) => continue,
                Err(_) => return Err(NodeDeclError::Runtime),
            }
        }
        // Timed out: forget the call, so its reply — if it still comes — is
        // dropped rather than returned by the next `call_raw` (issue 2009).
        let executor = unsafe { &mut *self.executor };
        if let Some(slot) = executor.resolve_handle(hid)
            && let Some(entry) = unsafe { executor.service_client_entry_mut(slot) }
        {
            entry.sequences.clear();
        }
        Err(NodeDeclError::Runtime)
    }

    fn service_is_ready(&self, service_entity: &str) -> NodeResult<bool> {
        use nros_rmw::ClientTrait;
        let hid = self.service(service_entity)?;
        let executor = unsafe { &mut *self.executor };
        let slot = executor.resolve_handle(hid).ok_or(NodeDeclError::Runtime)?;
        let entry =
            unsafe { executor.service_client_entry_mut(slot) }.ok_or(NodeDeclError::Runtime)?;
        // `Err` here is the backend's "cannot say" (XRCE has no discovery
        // channel); it is folded into the same `Runtime` the trait default
        // returns, so a caller has ONE branch for "did not get an answer".
        entry
            .handle
            .service_is_ready()
            .map_err(|_| NodeDeclError::Runtime)
    }

    fn action_server_is_ready(&self, action_entity: &str) -> NodeResult<bool> {
        let entry_index = self.action_entry(action_entity)?;
        let executor = unsafe { &mut *self.executor };
        let core = unsafe { executor.action_client_core_mut(entry_index) }
            .ok_or(NodeDeclError::Runtime)?;
        // As for `service_is_ready`: the backend's "cannot say" folds into the
        // trait default's `Runtime`, one branch for "no answer".
        core.server_readiness().map_err(|_| NodeDeclError::Runtime)
    }

    fn send_goal_raw(&mut self, action_entity: &str, goal_cdr: &[u8]) -> NodeResult<GoalUuid> {
        let entry_index = self.action_entry(action_entity)?;
        let executor = unsafe { &mut *self.executor };
        let core = unsafe { executor.action_client_core_mut(entry_index) }
            .ok_or(NodeDeclError::Runtime)?;
        let goal_id = core.send_goal_raw(goal_cdr).map_err(decl_err_from_node)?;
        // rclcpp-style: request the result immediately. The server queues the
        // get_result request until the goal terminates, then replies — the
        // executor's spin auto-delivers it to the bound result callback (the
        // executor never auto-sends this request, so the declarative client
        // must). Best-effort: a transport hiccup just means no result callback.
        let _ = core.send_get_result_request(&goal_id);
        Ok(goal_id)
    }
}

/// Tick-side action-SERVER execution (it mirrored the retired
/// ~~`GenActionExec`~~). Lets a
/// component complete goals / publish feedback / enumerate active goals from
/// its `tick` via `TickCtx`.
struct RuntimeActions<'a> {
    executor: *mut Executor<'static>,
    handles: &'a [(IdStr, crate::internals::ActionServerRawHandle)],
}

impl RuntimeActions<'_> {
    fn handle(&self, entity: &str) -> NodeResult<crate::internals::ActionServerRawHandle> {
        self.handles
            .iter()
            .find(|(e, _)| e == entity)
            .map(|(_, h)| *h)
            .ok_or(NodeDeclError::Runtime)
    }
}

impl ActionExecutor for RuntimeActions<'_> {
    fn complete_goal_raw(
        &mut self,
        action_entity: &str,
        goal_id: &GoalUuid,
        status: GoalStatusCode,
        result: &[u8],
    ) -> NodeResult<()> {
        let handle = self.handle(action_entity)?;
        let executor = unsafe { &mut *self.executor };
        // issue 0796 — a result too large for the server's RESULT_BUF is
        // reported rather than silently dropped.
        handle
            .complete_goal_raw(executor, goal_id, status, result)
            .map_err(|_| NodeDeclError::Runtime)
    }

    fn publish_feedback_raw(
        &mut self,
        action_entity: &str,
        goal_id: &GoalUuid,
        feedback: &[u8],
    ) -> NodeResult<()> {
        let handle = self.handle(action_entity)?;
        let executor = unsafe { &mut *self.executor };
        handle
            .publish_feedback_raw(executor, goal_id, feedback)
            .map_err(|_| NodeDeclError::Runtime)
    }

    fn for_each_active_goal(
        &self,
        action_entity: &str,
        visit: &mut dyn FnMut(&GoalUuid, GoalStatusCode),
    ) {
        if let Ok(handle) = self.handle(action_entity) {
            let executor = unsafe { &*self.executor };
            handle.for_each_active_goal(executor, |g| visit(&g.goal_id, g.status));
        }
    }
}

// Phase 258 (Track 2, w5) — the typed BSP fn-ptr aliases (~~`NodeRegisterFn`~~ /
// ~~`NodeInitFn`~~ / ~~`NodeDispatchFn`~~ / ~~`NodeTickFn`~~) are gone with the retired
// ~~`register_dispatch_slot`~~ / ~~`nros_run_components`~~ BSP-baker path. The
// macro-emitted `register(runtime)` wrapper now installs via the
// `install_node_typed` seam (Track 2 w4).

/// Phase 257 (W0-B) — register an [`ExecutableNode`] `C` against a **borrowed**
/// executor (the shared cffi `Executor` a foreign-language typed entry hands in via
/// its `nros::global_handle()` / `Node::executor_handle()`), returning the live
/// [`ComponentCell`].
///
/// Unlike [`ExecutorNodeRuntime::register_node`] this owns neither the executor nor a
/// components list: the executor's per-entity callbacks hold `CellHandle`s
/// clones (see [`ExecutorSink`]), so the cell stays alive for the executor's lifetime
/// via dispatch alone — the caller may drop the returned cell (pub/sub/timer nodes, the
/// W0-B target) or stash it to drive `tick` (service-client/action nodes; phase-257 D2).
/// The node self-creates its node (its `Node::NAME`) on the shared executor (phase-257
/// D7 Option C — Rust nodes in a foreign entry self-name, no entry-side qos-override).
/// Issue 0095 — preserve executor callback-table exhaustion through the
/// `NodeError → NodeDeclError` collapse so the register seam (and ultimately
/// the `nros::main!` entry) can name `NROS_EXECUTOR_MAX_CBS` instead of an
/// opaque `NodeRegister`. Every other `NodeError` stays `Runtime`.
fn decl_err_from_node(e: nros_node::NodeError) -> NodeDeclError {
    // Issue 0428 — `NodeDeclError::Runtime` collapses every non-`ExecutorFull`
    // cause into one opaque `NodeRegister("<pkg>")`, "four collapses away from the
    // cause" (node.rs). That is how a Cyclone descriptor/transport failure was
    // mis-diagnosed twice. Surface the real variant on the error path (rare — this
    // only runs when a declaration is already failing), so the register seam names
    // WHAT failed, not just WHERE.
    // issue 0589 — `nros_log`, not `std::eprintln!`: std stdio is fatal on Zephyr
    // native_sim, and this crate cannot gate on the platform. Routing it here
    // also means a `no_std` image gets the diagnostic, which the old
    // `cfg(feature = "std")` arm never delivered.
    if !matches!(e, nros_node::NodeError::ExecutorFull) {
        nros_log::log_error!(
            nros_log::get_logger("nros"),
            "node declaration failed — NodeError::{e:?}"
        );
    }
    // The same diagnostic through the `log` facade (every RTOS board bridges
    // it); without this the collapse left only an opaque `NodeRegister("<pkg>")`
    // on embedded — e.g. a static subscriber-pool exhaustion
    // (NROS_RMW_SUBSCRIBER_SLOTS) surfaced with no cause at all. phase-359 W10
    // stage B: this was `not(std)`, and `nros` has no `std` any more; on a host
    // with a `log` logger installed the line appears twice, on an error path
    // that only runs when a declaration is already failing.
    if !matches!(e, nros_node::NodeError::ExecutorFull) {
        log::error!("nros: node declaration failed — NodeError::{e:?}");
    }
    match e {
        nros_node::NodeError::ExecutorFull => NodeDeclError::ExecutorFull,
        _ => NodeDeclError::Runtime,
    }
}

fn register_node_borrowed<
    'p,
    C: ExecutableNode + 'static,
    const PUBS: usize,
    const SVCS: usize,
    const ACTC: usize,
    const ACTS: usize,
    const SSRV: usize,
    const TIMS: usize,
>(
    executor: &mut Executor<'static>,
    params: &'p [(&'p str, &'p str)],
    node_identity: Option<(&'static str, &'static str)>,
    remaps: &'p [(&'p str, &'p str)],
    qos_overrides: &'static [nros_node::executor::node_record::QoSOverrideCode],
    slot_mu: &'static mut MaybeUninit<TypedSlot<C>>,
    cell_mu: &'static mut MaybeUninit<ComponentCell<PUBS, SVCS, ACTC, ACTS, SSRV, TIMS>>,
) -> NodeResult<&'static ComponentCell<PUBS, SVCS, ACTC, ACTS, SSRV, TIMS>>
where
    C::State: 'static,
{
    // W5-endgame step 2b (issue 0857) — the cell is PLACED into caller
    // storage (the per-class static's paired region, or the alloc
    // convenience's leaked box), never `Arc`'d: 0857 measured the Arc'd cell
    // at ~17.5 KiB of heap per component. Closures and leaked ctxs hold a
    // `CellHandle::Static` copy; the executor's drop trampoline runs the
    // cell's destructor in place.
    let cell: &'static ComponentCell<PUBS, SVCS, ACTC, ACTS, SSRV, TIMS> =
        cell_mu.write(ComponentCell {
            slot: RefCell::new(place_slot::<C>(slot_mu)),
            publishers: RefCell::new(heapless::Vec::new()),
            service_clients: RefCell::new(heapless::Vec::new()),
            action_clients: RefCell::new(heapless::Vec::new()),
            action_servers: RefCell::new(heapless::Vec::new()),
            timers: RefCell::new(heapless::Vec::new()),
            svc_ctxs: [const { UnsafeCell::new(MaybeUninit::uninit()) }; _],
            svc_ctxs_used: core::cell::Cell::new(0),
            act_srv_ctxs: [const { UnsafeCell::new(MaybeUninit::uninit()) }; _],
            act_srv_ctxs_used: core::cell::Cell::new(0),
            act_cli_ctxs: [const { UnsafeCell::new(MaybeUninit::uninit()) }; _],
            act_cli_ctxs_used: core::cell::Cell::new(0),
            header: CellHeader {
                callback_dispatches: AtomicUsize::new(0),
                message_dispatches: AtomicUsize::new(0),
            },
            // W4c — set by `apply_param_services` once the store exists (it runs after this).
            #[cfg(feature = "param-store")]
            param_server: core::cell::Cell::new(core::ptr::null()),
            // phase-426 W3 — overwritten below with the node `register()` created.
            #[cfg(feature = "param-store")]
            param_node: core::cell::Cell::new(0),
        });
    let mut sink = ExecutorSink {
        cell: CellHandle::Static(cell),
        nodes: heapless::Vec::new(),
        // Phase 268 W1 — thread the per-component identity bake (RFC-0046).
        node_identity,
        // Phase 305 W3 (issue 0255) — thread the per-component remap bake.
        remaps,
        // Issue #52 — thread the per-component QoS-override bake.
        qos_overrides,
        // issue 1272 -- the launch `<param>` initials, seeded on the node
        // `create_node` builds rather than on the executor's primary node.
        launch_params: params,
    };
    {
        let mut frame = crate::node::ComponentFrame::new(&mut sink);
        let mut context = NodeContext::new(C::NAME, &mut *executor, &mut frame);
        // Phase 264 W4a — seed the baked launch-param initials so `register()` can read
        // them via `NodeContext::param`.
        context.set_params(params);
        C::register(&mut context)?;
    }
    // phase-426 W3 — see the sibling in `register_node`: the cell records the
    // node it created, so `ctx.parameter::<T>(name)` reads that node's slot.
    #[cfg(feature = "param-store")]
    if let Some(node) = sink.nodes.first() {
        cell.param_node.set(node.node_id.raw());
    }

    // Phase 258 (Track 2, 2a) — enroll the cell in the executor's component
    // tick registry so `install`'d nodes tick (closes phase-257 D2: poll-only
    // service-client/action nodes have no callbacks keeping the cell alive AND
    // never ran `tick`). The executor runs `tick` each `spin_once` and runs
    // the cell's destructor in place on `Executor::drop`. Harmless for
    // pub/sub/timer-only nodes — their tick body is a no-op. Registry-full
    // (`MAX_NODES`) proceeds without a tick slot; the placed cell then simply
    // lives (and leaks its component state's destructor) with its storage,
    // which a monotonic `take` never re-hands out.
    let raw =
        cell as *const ComponentCell<PUBS, SVCS, ACTC, ACTS, SSRV, TIMS> as *mut core::ffi::c_void;
    // SAFETY: `raw` is the freshly placed cell; the enrolled trampoline
    // monomorphizations match its exact type (borrow on tick, drop_in_place
    // on executor drop).
    let _ = unsafe {
        executor.enroll_component_ticking(
            raw,
            component_tick_trampoline::<PUBS, SVCS, ACTC, ACTS, SSRV, TIMS>,
            component_drop_trampoline::<PUBS, SVCS, ACTC, ACTS, SSRV, TIMS>,
            C::HAS_TICK,
        )
    };

    // W4c — capture the executor's volatile param store on the cell so this node's
    // callbacks can read `ctx.parameter::<T>(name)`. Non-null only when `nros::main!`
    // emitted `apply_param_services` BEFORE this register call (`[param_services]`
    // declared); otherwise the store is absent and the field stays null.
    #[cfg(feature = "param-store")]
    if let Some(server) = executor.params() {
        cell.param_server.set(server as *const _);
    }

    Ok(cell)
}

/// Issue 1535 — the install seam's return code for a refused executor handle:
/// `-1` for null (unchanged), `-4` for an nros-cpp context that was not
/// unwrapped with `nros_cpp_executor_inner`.
fn install_handle_refusal_code(refusal: crate::executor_handle::HandleRefusal) -> i32 {
    match refusal {
        crate::executor_handle::HandleRefusal::Null => -1,
        crate::executor_handle::HandleRefusal::CppContext => -4,
    }
}

/// Phase 257 (W0-B) — C-ABI typed component install. Recovers the shared `Executor`
/// from the handle a typed entry passes — a bare `*mut Executor<'static>`. A C/C++
/// entry's `rclcpp::global_handle()` is NOT one: it is an nros-cpp context, and
/// the entry unwraps it with `nros_cpp_executor_inner` first (issue 1535). Registers
/// `C` on it via `register_node_borrowed` (private helper). The component's `ComponentCell`
/// is PLACED in caller/leaked storage (W5-endgame step 2b) and dropped in place by the
/// executor on `Executor::drop`. Returns `0` on success, `-1` on a null handle or a
/// registration error, `-4` on an nros-cpp context that was not unwrapped.
///
/// This backs the `__nros_component_<pkg>_install(node, executor, self)` symbol
/// `nros::node!()` emits — the uniform cross-language install seam (phase-257 D6).
///
/// # Safety
/// `executor` must be the live `*mut Executor<'static>` handle a typed entry passes (its
/// `nros::global_handle()` / a node's `executor_handle()`), valid for the call.
#[cfg(feature = "alloc")]
pub unsafe fn install_node_typed<C: ExecutableNode + 'static>(
    executor: *mut core::ffi::c_void,
) -> i32
where
    C::State: 'static,
{
    // SAFETY: forwarded per this fn's contract; no baked params.
    unsafe { install_node_typed_with_params::<C>(executor, &[]) }
}

/// W4a — same as [`install_node_typed`] but seeds the node's [`NodeContext`] with the
/// launch-baked `<param>` initial values, so a `register`/`init`-time `ctx.param(name)`
/// observes the compile-time launch value (RFC-0004 §10). `params` must outlive the call.
///
/// # Safety
/// `executor` must be the live `*mut Executor<'static>` handle a typed entry passes, valid for the call.
#[cfg(feature = "alloc")]
pub unsafe fn install_node_typed_with_params<C: ExecutableNode + 'static>(
    executor: *mut core::ffi::c_void,
    params: &[(&str, &str)],
) -> i32
where
    C::State: 'static,
{
    // SAFETY: forwarded per this fn's contract; no identity injection.
    unsafe { install_node_typed_with_node_identity::<C>(executor, params, None) }
}

/// Phase 268 W1 — same as [`install_node_typed_with_params`] but also injects the
/// launch `<node name= namespace=>` identity so `ExecutorSink::create_node` uses it
/// instead of the `NodeOptions` default (RFC-0046). `None` → backward-compatible
/// (NodeOptions stands). `nros::node!()` calls this variant to carry the identity from
/// `RuntimeCtx::node_identity` set by `nros::main!` per component. `params` and
/// `node_identity` strings must outlive the call (both are `'static` in the macro emit).
///
/// # Safety
/// `executor` must be the live `*mut Executor<'static>` handle a typed entry passes, valid for the call.
#[cfg(feature = "alloc")]
pub unsafe fn install_node_typed_with_node_identity<C: ExecutableNode + 'static>(
    executor: *mut core::ffi::c_void,
    params: &[(&str, &str)],
    node_identity: Option<(&'static str, &'static str)>,
) -> i32
where
    C::State: 'static,
{
    // SAFETY: forwarded per this fn's contract; no remap rules.
    unsafe { install_node_typed_with_launch::<C>(executor, params, node_identity, &[], &[]) }
}

/// Phase 305 W3 (issue 0255) — same as [`install_node_typed_with_node_identity`]
/// but also carries the launch `<remap from= to=/>` rules `nros::main!` baked for
/// this component (`RuntimeCtx::remaps`). `ExecutorSink::create_entity` applies
/// them (plus `~`/relative expansion) before any entity name reaches the wire.
/// All slices must outlive the call (`'static` promoted in the macro emit).
///
/// # Safety
/// `executor` must be the live `*mut Executor<'static>` handle a typed entry passes, valid for the call.
#[cfg(feature = "alloc")]
pub unsafe fn install_node_typed_with_launch<C: ExecutableNode + 'static>(
    executor: *mut core::ffi::c_void,
    params: &[(&str, &str)],
    node_identity: Option<(&'static str, &'static str)>,
    remaps: &[(&str, &str)],
    qos_overrides: &'static [nros_node::executor::node_record::QoSOverrideCode],
) -> i32
where
    C::State: 'static,
{
    // Issue 1535 — the ONE handle check: null, or an nros-cpp context a C/C++
    // entry forgot to unwrap, is refused before anything is written through it.
    let exec: &mut Executor<'static> = match unsafe {
        crate::executor_handle::executor_from_handle(executor, "install_node_typed")
    } {
        Ok(exec) => exec,
        Err(refusal) => return install_handle_refusal_code(refusal),
    };
    // phase-391 W5.3b — the alloc convenience: leak exactly-sized slot storage
    // per call, the same relationship `from_executor` has to `new_in`. The
    // macro-emitted entries call the `_in` twin with their per-class static
    // instead, so GENERATED images do not take this branch.
    let slot_mu: &'static mut MaybeUninit<TypedSlot<C>> =
        Box::leak(Box::new(MaybeUninit::uninit()));
    // W5-endgame step 2b — the cell leaks the same way; its destructor still
    // runs (executor drop trampoline, in place), only the bytes stay.
    let cell_mu: &'static mut MaybeUninit<ComponentCell> =
        Box::leak(Box::new(MaybeUninit::uninit()));
    match register_node_borrowed::<C, _, _, _, _, _, _>(
        exec,
        params,
        node_identity,
        remaps,
        qos_overrides,
        slot_mu,
        cell_mu,
    ) {
        Ok(_cell) => 0,
        // Issue 0095 — distinct code for executor-table exhaustion so the macro
        // register seam can name `NROS_EXECUTOR_MAX_CBS` instead of an opaque
        // `NodeRegister`. Every other failure stays the generic `-1`.
        Err(NodeDeclError::ExecutorFull) => -2,
        Err(_) => -1,
    }
}

/// phase-391 W5.3b — [`install_node_typed_with_launch`] over the CALLER's
/// per-class slot storage: the heap-free FFI install path. The `nros::node!`
/// macro emits one `static ComponentSlotStorage<C>` per expansion and calls
/// this twin; the C ABI trampoline above it keeps its `(ptr, ptr, ptr) -> i32`
/// shape, so nothing foreign changes.
///
/// Returns `-3` when the class's instance cap is exhausted — raise
/// `NROS_RUNTIME_MAX_CLASS_INSTANCES`. (Distinct from `-2`, the executor
/// callback-table Full.) Returns `-4` for an nros-cpp context handle that the
/// C/C++ caller did not unwrap with `nros_cpp_executor_inner` (issue 1535).
///
/// # Safety
/// `executor` must be null, an nros-cpp context (refused), or the live
/// `*mut Executor<'static>` handle a typed entry passes, valid for the call.
pub unsafe fn install_node_typed_with_launch_in<
    C: ExecutableNode + 'static,
    const N: usize,
    const PUBS: usize,
    const SVCS: usize,
    const ACTC: usize,
    const ACTS: usize,
    const SSRV: usize,
    const TIMS: usize,
>(
    executor: *mut core::ffi::c_void,
    store: &'static ComponentSlotStorage<C, N, PUBS, SVCS, ACTC, ACTS, SSRV, TIMS>,
    params: &[(&str, &str)],
    node_identity: Option<(&'static str, &'static str)>,
    remaps: &[(&str, &str)],
    qos_overrides: &'static [nros_node::executor::node_record::QoSOverrideCode],
) -> i32
where
    C::State: 'static,
{
    // Issue 1535 — checked BEFORE the class's storage is taken, so a refused
    // handle leaves the slot for a corrected retry.
    let exec: &mut Executor<'static> = match unsafe {
        crate::executor_handle::executor_from_handle(executor, "install_node_typed_in")
    } {
        Ok(exec) => exec,
        Err(refusal) => return install_handle_refusal_code(refusal),
    };
    let Some((slot_mu, cell_mu)) = store.take() else {
        return -3;
    };
    match register_node_borrowed::<C, _, _, _, _, _, _>(
        exec,
        params,
        node_identity,
        remaps,
        qos_overrides,
        slot_mu,
        cell_mu,
    ) {
        Ok(_cell) => 0,
        Err(NodeDeclError::ExecutorFull) => -2,
        Err(_) => -1,
    }
}

/// phase-391 W5.3b — [`install_node_typed`] over caller storage; see
/// [`install_node_typed_with_launch_in`].
///
/// # Safety
/// As [`install_node_typed_with_launch_in`].
pub unsafe fn install_node_typed_in<
    C: ExecutableNode + 'static,
    const N: usize,
    const PUBS: usize,
    const SVCS: usize,
    const ACTC: usize,
    const ACTS: usize,
    const SSRV: usize,
    const TIMS: usize,
>(
    executor: *mut core::ffi::c_void,
    store: &'static ComponentSlotStorage<C, N, PUBS, SVCS, ACTC, ACTS, SSRV, TIMS>,
) -> i32
where
    C::State: 'static,
{
    // SAFETY: forwarded per this fn's contract.
    unsafe { install_node_typed_with_launch_in(executor, store, &[], None, &[], &[]) }
}

// Phase 258 (Track 2, w5) — ~~`nros_run_components`~~ (the BSP shim that registered
// every component via the four-fn-ptr ~~`register_dispatch_slot`~~ then spun) is
// gone. It had no callers; owned-spin / BSP entries register through the
// `install_node_typed` seam + drive `ExecutorNodeRuntime::spin`.

/// issue 0460 — map a `NodeError` to a static reason a capability failure can
/// carry across the `nros-platform` trait boundary (which cannot name
/// `NodeError`). Static strings, so this works on `no_std` targets with no
/// allocator and no `log` dependency in this crate.
///
/// Both matches are EXHAUSTIVE on purpose. A wildcard arm is how this issue got
/// its diagnostic — `NodeRegister("lifecycle")` named the capability and
/// nothing about the cause — and a `_ =>` here would reintroduce exactly that
/// for whichever variant is added next. The compile error is the point.
///
/// Gated on `alloc` as well, because its only callers are
/// `ExecutorNodeRuntime`'s capability hooks, and that runtime is `alloc`-only.
/// A heap-free C++ image with a carved parameter store (`param-store`, no
/// `alloc`) compiled it with nothing calling it, which `-D warnings` refuses
/// (issue 2115).
#[cfg(all(
    feature = "alloc",
    any(feature = "lifecycle-services", feature = "param-store")
))]
fn capability_reason(e: &nros_node::NodeError) -> &'static str {
    use nros_node::NodeError;
    match e {
        NodeError::Transport(t) => capability_transport_reason(t),
        NodeError::NameTooLong => {
            "NameTooLong (the node FQN or a service name overflowed its fixed buffer)"
        }
        NodeError::ExecutorFull => {
            "ExecutorFull (no callback slot left — raise CONFIG_NROS_EXECUTOR_MAX_CBS; \
             a capability's services are not counted by the model)"
        }
        NodeError::NodeTableFull => "NodeTableFull (NROS_EXECUTOR_MAX_NODES reached)",
        NodeError::NotInitialized => "NotInitialized (a required subsystem was never registered)",
        NodeError::BackendMismatch => "BackendMismatch (the RMW is not the executor's session)",
        NodeError::BufferTooSmall => "BufferTooSmall",
        NodeError::Serialization => "Serialization",
        NodeError::Deserialization => "Deserialization",
        NodeError::ActionCreationFailed => "ActionCreationFailed",
        NodeError::ServiceRequestFailed => "ServiceRequestFailed",
        NodeError::ServiceReplyFailed => "ServiceReplyFailed",
        NodeError::Timeout => "Timeout",
        NodeError::RequestInFlight => "RequestInFlight",
        NodeError::NoSchedContextSlot => "NoSchedContextSlot",
        NodeError::InvalidSchedContextBinding => "InvalidSchedContextBinding",
        // issue 0790 added this variant and did not add the arm; the exhaustive
        // match above did exactly what its doc comment says it is for, and the
        // lanes that were run could not see it (`--all-targets` enables `test`,
        // which pulls in a different cfg path, and `cargo doc` does no
        // exhaustiveness checking, so the parity lane stayed green over it).
        NodeError::ShutdownCallbacksFull => {
            "ShutdownCallbacksFull (the executor's pre/on-shutdown callback table is \
             full — raise NROS_EXECUTOR_MAX_SHUTDOWN_CBS)"
        }
        // phase-454 W10. The two numbers and the topic are already on the log
        // by the time this renders — this string is the CLASS, and a reason
        // that repeated the numbers without the topic would be worse than one
        // that points at the line which has all three.
        NodeError::DeclaredDepthMismatch => {
            "DeclaredDepthMismatch (this subscription's QoS depth disagrees with the depth \
             the contract sidecar declares for that topic — the log line above names the \
             topic and both numbers; depth multiplies the executor arena, so the two must \
             state one number)"
        }
        // issue 1256. Same shape: the policy, the topic and both values are on
        // the log line above.
        NodeError::DeclaredQosMismatch => {
            "DeclaredQosMismatch (this subscription's QoS reliability or durability \
             disagrees with what the contract sidecar declares for that topic — the log \
             line above names the policy, the topic and both values; an incompatible-QoS \
             match never delivers)"
        }
    }
}

/// The `NodeError::Transport` payload, which is the half that actually names
/// what the RMW refused. `Backend(&'static str)` passes its own diagnostic
/// straight through — it is already the string the backend wanted to say.
///
/// Unlike `capability_reason` above this one keeps a wildcard, for
/// `TransportError::BackendDynamic`, which is gated on `nros-rmw/alloc`. This
/// function is compiled only with `nros/alloc`, which turns that on, so the
/// variant always exists here; the wildcard stands for it rather than naming
/// it, so its text is what any unmapped variant looks like. (Before issue 2115
/// this function was compiled without `alloc` too, where the variant does not
/// exist and the wildcard was an `unreachable_patterns` error.)
#[cfg(all(
    feature = "alloc",
    any(feature = "lifecycle-services", feature = "param-store")
))]
fn capability_transport_reason(t: &nros_rmw::TransportError) -> &'static str {
    use nros_rmw::TransportError as T;
    match t {
        T::Backend(s) => s,
        T::ServiceServerCreationFailed => {
            "Transport::ServiceServerCreationFailed (the RMW refused to declare the queryable)"
        }
        T::ServiceClientCreationFailed => "Transport::ServiceClientCreationFailed",
        T::PublisherCreationFailed => "Transport::PublisherCreationFailed",
        T::SubscriberCreationFailed => "Transport::SubscriberCreationFailed",
        T::ConnectionFailed => "Transport::ConnectionFailed (no session to the router)",
        T::Disconnected => "Transport::Disconnected",
        T::TopicNameInvalid => "Transport::TopicNameInvalid",
        T::NodeNameNonExistent => "Transport::NodeNameNonExistent",
        // phase-417 G7 — the variant carries the refused `QoSPolicyMask` bit
        // now, and this function's return type has no room for it: a
        // `&'static str` cannot be composed, and a second table of twelve
        // pre-written sentences is exactly the parallel vocabulary the bit's
        // own name exists to avoid. The policy is already on the line ABOVE
        // this one — `nros_node`'s `validate_qos_or_report` logs it at ERROR
        // at the refusing create — so this string is the CLASS and points at
        // the line that has the name, the same shape `DeclaredDepthMismatch`
        // uses two arms up. A Rust caller holding the error reads
        // `err.qos_policy_name()` and needs neither.
        T::IncompatibleQos(_) => {
            "Transport::IncompatibleQos (the backend does not honour a policy this \
             profile states — the log line above names which)"
        }
        T::IncompatibleAbi => "Transport::IncompatibleAbi",
        T::InvalidConfig => "Transport::InvalidConfig",
        T::InvalidArgument => "Transport::InvalidArgument",
        T::Unsupported => "Transport::Unsupported (the backend does not implement this)",
        T::BadAlloc => "Transport::BadAlloc",
        T::PublishFailed => "Transport::PublishFailed",
        T::ServiceRequestFailed => "Transport::ServiceRequestFailed",
        T::ServiceReplyFailed => "Transport::ServiceReplyFailed",
        T::SerializationError => "Transport::SerializationError",
        T::DeserializationError => "Transport::DeserializationError",
        T::BufferTooSmall => "Transport::BufferTooSmall",
        T::MessageTooLarge => "Transport::MessageTooLarge",
        T::Timeout => "Transport::Timeout",
        T::WouldBlock => "Transport::WouldBlock",
        T::TooLarge => "Transport::TooLarge",
        T::TaskStartFailed => "Transport::TaskStartFailed",
        T::PollFailed => "Transport::PollFailed",
        T::KeepaliveFailed => "Transport::KeepaliveFailed",
        T::JoinFailed => "Transport::JoinFailed",
        T::LoanNotSupported => "Transport::LoanNotSupported",
        T::NoData => "Transport::NoData",
        _ => "Transport::<variant with no reason mapping — add it here>",
    }
}

// =============================================================================
// Tests
// =============================================================================
//
// Concrete `Executor` construction needs a real RMW backend session
// (with `rmw-cffi` on, `Executor::from_session` takes the cffi
// session). MockSession only exists when `rmw-cffi` is off — so the
// unit tests that exercise live timer firing live in
// `packages/testing/nros-tests/tests/phase212_m5a2_component_runtime.rs`
// gated behind the `component-runtime-test` feature (pulls
// `nros-rmw-zenoh`). The compile-only smoke here verifies the public
// types are reachable through the umbrella surface.

/// phase-509 P5 (RFC-0108 D2) -- seed the contract rank of the node the
/// NEXT `register` call creates, `(name, namespace)`, on the executor the
/// runtime holds: 0 is the most urgent, as the derivation normalises it
/// (`nros_orchestration_ir::levels::node_ranks`). `nros::main!` emits it before
/// each ranked node's `register`, the Rust twin of the C/C++ packs'
/// `nros_cpp_bind_node_rank`.
///
/// A runtime with no executor handle (no live executor) has nothing to
/// order and seeds nothing. A binding that does not fit the node-name table
/// is refused with the node's name, never dropped.
pub fn seed_node_rank(
    runtime: &mut nros_platform::RuntimeCtx<'_>,
    name: &'static str,
    namespace: &'static str,
    rank: u8,
) -> Result<(), nros_platform::RuntimeError> {
    let handle = runtime.runtime.executor_handle();
    if handle.is_null() {
        return Ok(());
    }
    // SAFETY: a non-null handle from the runtime sink is its live executor
    // (or an nros-cpp context, which the one handle check refuses), and
    // nothing else borrows it during this call.
    let exec = unsafe { crate::executor_handle::executor_from_handle(handle, "seed_node_rank") }
        .map_err(|_| nros_platform::RuntimeError::NodeRegister(name))?;
    exec.bind_node_rank(name, namespace, rank)
        .map_err(|_| nros_platform::RuntimeError::NodeRegister(name))
}

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

    #[test]
    fn handle_slot_is_observable() {
        // Trivial smoke — the handle type carries the slot index.
        let h = RegisteredNode::<DummyComp> {
            component_idx: 7,
            _phantom: PhantomData,
        };
        assert_eq!(h.slot(), 7);
    }

    /// Issue 0857 — the per-class store pays its DECLARED capacity, and the
    /// publisher registry is where the bytes are.
    ///
    /// The same quantity `nm -S` reads off a linked image, without the link
    /// step. Measured both ways on `examples/workspaces/rust`'s
    /// `linux_service_server_entry` (2026-09-06, one build dir, one file
    /// changed): `__NROS_COMPONENT_service_server_pkg_SLOT_STORE` is 50,824 B
    /// of `.bss` when the class declares nothing and 568 B when it declares
    /// `EntityBounds::exact(0, 1, 0, 0, 0)`.
    ///
    /// The assertions are on the SHAPE rather than on those two literals,
    /// because both move with the knob, with `DEFAULT_LOAN_BUF` and with the
    /// target's pointer width. What must not move is the direction: one
    /// publisher slot is kilobytes, and declaring less costs less.
    #[test]
    fn a_declared_cell_pays_for_what_it_declares() {
        use core::mem::size_of;
        const N: usize = crate::config::MAX_CLASS_INSTANCES;
        type Zero = ComponentSlotStorage<DummyComp, N, 0, 0, 0, 0, 0, 0>;
        type OnePub = ComponentSlotStorage<DummyComp, N, 1, 0, 0, 0, 0, 0>;
        type Capped = ComponentSlotStorage<DummyComp>;
        // Issue 2075 -- a controllable-timer row is a hash and a handle, not
        // a name: at most 32 B an instance (16 B of payload plus the op and
        // padding), so the table the knob default carries is small.
        type OneTimer = ComponentSlotStorage<DummyComp, N, 0, 0, 0, 0, 0, 1>;
        assert!(
            size_of::<OneTimer>() - size_of::<Zero>() <= 32 * N,
            "a timer row costs {} B for {N} instances; it should be a hash + a handle",
            size_of::<OneTimer>() - size_of::<Zero>()
        );

        let zero = size_of::<Zero>();
        let one_pub = size_of::<OnePub>();
        let capped = size_of::<Capped>();

        // `EmbeddedRawPublisher` embeds a `TxArena<DEFAULT_LOAN_BUF>`, so ONE
        // publisher slot is the whole reason the knob-capped default is
        // measured in tens of kilobytes. If this ever drops below a kilobyte
        // the issue's cost model has changed and the sweep is worth revisiting.
        assert!(
            one_pub - zero >= 1024 * N,
            "a publisher registry slot should cost >= 1 KiB per instance; \
             zero={zero} one_pub={one_pub} N={N}"
        );

        // Declaring nothing must cost strictly less than the knob caps —
        // unless the knob itself is 0, in which case they are the same store.
        if crate::config::MAX_CELL_ENTITIES > 0 {
            assert!(
                zero < capped,
                "declaring nothing ({zero} B) is not cheaper than the knob caps \
                 ({capped} B at MAX_CELL_ENTITIES={})",
                crate::config::MAX_CELL_ENTITIES
            );
            assert!(
                capped - zero >= (one_pub - zero) * crate::config::MAX_CELL_ENTITIES,
                "the knob-capped store should carry MAX_CELL_ENTITIES publisher \
                 slots at least; zero={zero} one_pub={one_pub} capped={capped}"
            );
        }
    }

    struct DummyComp;
    impl Component for DummyComp {
        const NAME: &'static str = "dummy";
        fn register(_ctx: &mut NodeContext<'_>) -> NodeResult<()> {
            Ok(())
        }
    }
    impl ExecutableNode for DummyComp {
        type State = ();
        fn init() -> Self::State {}
        fn on_callback(_s: &mut (), _cb: Callback<'_>, _ctx: &mut CallbackCtx<'_>) {}
    }
}