nros_params/

server.rs

//! Parameter server implementation
//!
//! Provides a static storage parameter server for embedded systems.
//! Parameters are stored in a fixed-size array with compile-time capacity.

use crate::{
    ParameterError,
    types::{
        FloatingPointRange, IntegerRange, MAX_PARAM_NAME_LEN, NodeFlags, NodeKey, Parameter,
        ParameterDescriptor, ParameterRange, ParameterType, ParameterValue,
    },
};
use heapless::String;

// MAX_PARAMETERS is generated by build.rs and included via types.rs.
use crate::types::MAX_PARAMETERS;

/// phase-491 W3 (RFC-0105 D8) — what a pending `/parameter_events` entry
/// reports. The value is read from the store when the event is published, so
/// an event says what the parameter IS, not what it was when it moved.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ParameterEventKind {
    /// Declared (locally, or by a remote set on an undeclared-allowing node).
    New,
    /// Its value changed.
    Changed,
    /// Undeclared.
    Deleted,
    /// Issue 2023 — the node's start-up event when it holds no parameter at
    /// the moment events go on: the node's name and three EMPTY lists. A
    /// node that does hold parameters announces itself with one `New` per
    /// parameter instead (see [`ParameterServer::pop_initial_parameter_event`](crate::ParameterServerOps::pop_initial_parameter_event)).
    Announce,
}

/// One change waiting to be published on `/parameter_events`.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PendingParameterEvent {
    /// The node whose parameter moved.
    pub node: NodeKey,
    /// What happened to it.
    pub kind: ParameterEventKind,
    /// Its name.
    pub name: String<MAX_PARAM_NAME_LEN>,
}

/// How many changes the store holds between two flushes of
/// `/parameter_events`.
///
/// A flush runs after every C or C++ parameter call and on every spin, so this only
/// has to cover the changes ONE spin's parameter services make; a burst past
/// it is counted ([`ParameterServer::take_parameter_events_dropped`](crate::ParameterServerOps::take_parameter_events_dropped)) and
/// reported, never blocked on. A second change to a name already pending
/// merges into its entry, since the event carries the value at publish time.
pub const PARAMETER_EVENT_LOG_LEN: usize = 8;

/// Entry in the parameter storage.
///
/// phase-382 W2' — crate-private ON PURPOSE. The storage is caller-owned now,
/// but the SLOT layout is not part of the deal: consumers name
/// [`ParameterTable`] / [`ParameterStorage`], never this.
#[derive(Debug, Clone)]
pub(crate) struct ParameterEntry {
    /// phase-426 W1 — which node declared it. The slot arena is shared (one
    /// fixed table per executor, no per-node allocation); this is what keeps
    /// two nodes' identically-named parameters apart inside it.
    node: NodeKey,
    /// The parameter (name + value)
    param: Parameter,
    /// Optional descriptor with constraints
    descriptor: Option<ParameterDescriptor>,
}

/// Why a declaration was refused — phase-417 W4.a.
///
/// rclrs separates DECLARE-time failures from set-time ones, and this is that
/// separation: a write's `Result<(), ParameterError>` answers "may this WRITE happen", this
/// answers "may this parameter EXIST". Ours is a subset of rclrs's, and the
/// difference is structural rather than an omission: rclrs's
/// `NoValueAvailable`, `OverrideValueTypeMismatch` and `PriorValueTypeMismatch`
/// are about a launch OVERRIDE arriving before the declaration and disagreeing
/// with it, and nano-ros seeds overrides through the same
/// [`ParameterServer::apply`](crate::ParameterServerOps::apply) path a remote set uses — so a disagreeing
/// override is a `ParameterError`, on the write that carried it, and
/// cannot reach a declaration.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum DeclarationError {
    /// This node already declared this name (rclrs `AlreadyDeclared`).
    AlreadyDeclared,
    /// The descriptor's range is ill-formed — `min > max`, or a negative step
    /// (rclrs `InvalidRange`).
    InvalidRange,
    /// The default value is outside the descriptor's range (rclrs
    /// `InitialValueOutOfRange`).
    InitialValueOutOfRange,
    /// Every slot of the caller-owned table is taken. No rclrs correspondent:
    /// rclrs's store grows, ours is the storage the caller placed.
    StorageFull,
    /// The name is longer than `NROS_MAX_PARAM_NAME_LEN`. No rclrs
    /// correspondent, same reason.
    NameTooLong,
    /// A read-only parameter was declared with no default (rclrs
    /// `NoValueAvailable`; phase-515 R8). A read-only value can never be set
    /// later, so it must arrive with the declaration.
    NoValueAvailable,
}

/// phase-515 R8 -- the typed builder's terminals return [`DeclarationError`]
/// now, as rclrs's do. This keeps `?` into a [`ParameterError`]-returning
/// caller compiling, and maps each case to the `ParameterError` such a caller
/// saw before the change.
impl From<DeclarationError> for crate::ParameterError {
    fn from(e: DeclarationError) -> Self {
        match e {
            DeclarationError::AlreadyDeclared => crate::ParameterError::TypeMismatch,
            DeclarationError::InvalidRange => crate::ParameterError::InvalidRange,
            DeclarationError::InitialValueOutOfRange => crate::ParameterError::OutOfRange,
            DeclarationError::StorageFull | DeclarationError::NameTooLong => {
                crate::ParameterError::StorageFull
            }
            DeclarationError::NoValueAvailable => crate::ParameterError::NotFound,
        }
    }
}

/// An on-set-parameters callback — phase-417 W4.a.
///
/// `false` REFUSES the write, which is rclcpp's
/// `add_on_set_parameters_callback` contract (a `SetParametersResult` whose
/// `successful` is false) and the contract C's legacy
/// C's rclc parameter server carries too (phase-491, with the old value).
///
/// A `fn` pointer plus an [`OnSetContext`], not a closure: this crate is
/// `no_std` with no allocator, so there is nothing to box a `FnMut` into.
///
/// phase-491 W2 (RFC-0105 D6) — the hook sees BOTH sides of the change,
/// rclc's `rclc_parameter_callback_t` contract: `old` is `None` when the write
/// would CREATE the parameter (an undeclared remote set), `new` is `None` when
/// it would DELETE it (a remote `NOT_SET` on a node that asked for
/// [`set_delete_on_not_set`](crate::ParameterServerOps::set_delete_on_not_set)).
/// rclcpp's `add_on_set_parameters_callback` reads only `new`, so the C++ and
/// Rust facades ignore `old`.
pub type OnSetParameterFn = fn(
    node: NodeKey,
    name: &str,
    old: Option<&ParameterValue>,
    new: Option<&ParameterValue>,
    ctx: OnSetContext,
) -> bool;

/// Two machine words of caller context, carried verbatim to an
/// [`OnSetParameterFn`].
///
/// TWO, not one, because of the consumer that drives this API: a C or C++
/// registration is a FUNCTION POINTER plus a user `void*`, and the shim needs
/// both to reach the caller's code. One word would force each wrapper to keep
/// a static side table mapping an index back to the pair — a second, smaller
/// registry with its own bound and its own exhaustion story, which is the
/// class phase-426 spent six work items removing.
///
/// `usize`, not `*mut c_void`: a raw pointer in this struct would strip
/// [`ParameterServer`]'s auto traits, and the store lives inside the executor.
/// The C and C++ shims cast through it and say so.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct OnSetContext {
    /// First word. The wrappers put the user's `void* context` here, so a
    /// Rust caller with one pointer to carry uses this one and ignores `b`.
    pub a: usize,
    /// Second word. The wrappers put the C function pointer here.
    pub b: usize,
}

impl OnSetContext {
    /// One word of context; `b` is zero.
    pub const fn new(a: usize) -> Self {
        Self { a, b: 0 }
    }

    /// Both words.
    pub const fn pair(a: usize, b: usize) -> Self {
        Self { a, b }
    }
}

/// A registered [`OnSetParameterFn`], returned by
/// [`ParameterServer::add_on_set_parameters_callback`](crate::ParameterServerOps::add_on_set_parameters_callback) and consumed by
/// [`ParameterServer::remove_on_set_parameters_callback`](crate::ParameterServerOps::remove_on_set_parameters_callback).
///
/// rclcpp hands back an owning `OnSetParametersCallbackHandle` shared_ptr whose
/// destruction unregisters. With no allocator there is nothing to own, so this
/// is a plain index token and unregistering is explicit.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct OnSetParameterHandle(u16);

impl OnSetParameterHandle {
    /// The token as a plain integer, for an FFI that must hand it to C.
    ///
    /// An opaque `u16` rather than a pointer: it indexes the bounded registry
    /// and nothing else, so a wrong value is a `false` from
    /// [`ParameterServer::remove_on_set_parameters_callback`](crate::ParameterServerOps::remove_on_set_parameters_callback) rather than a
    /// wild write.
    pub const fn raw(self) -> u16 {
        self.0
    }

    /// Rebuild a handle a C caller kept. See [`Self::raw`].
    pub const fn from_raw(raw: u16) -> Self {
        Self(raw)
    }
}

#[derive(Clone, Copy)]
struct OnSetCallbackSlot {
    node: NodeKey,
    callback: OnSetParameterFn,
    ctx: OnSetContext,
}

/// How many on-set-parameters callbacks ONE store holds, across every node.
///
/// phase-417 W4.a — a BOUNDED registry with a stated capacity, because the
/// unbounded upstream shape needs an allocator this crate does not have. Four,
/// and the number is chosen rather than inherited: a slot costs
/// `size_of::<Option<OnSetCallbackSlot>>()` — 32 bytes on a 64-bit host, 16 on
/// a 32-bit target — so the whole registry is 128 bytes next to a 285,184-byte
/// default table, and four is one hook for each node of a four-node composed
/// image (RFC-0047's shape) with none left to spare by accident. Registering a
/// fifth REFUSES with `None` rather than dropping one silently.
///
/// It is deliberately NOT a build knob. Every other parameter capacity sizes
/// the table a user's data lands in; this one sizes a table of code the image
/// itself registers, which is known at build time by reading the source. A
/// knob here would be a number nobody can derive and everybody has to state.
pub const MAX_ON_SET_CALLBACKS: usize = 4;

/// The slot table a [`ParameterServer`] borrows.
///
/// phase-382 W2' — the server's storage is CALLER-OWNED, and this newtype is
/// what makes that expressible without publishing the slot type. A slot is a
/// `ParameterEntry`, which stays PRIVATE (it is the store's internal slot
/// layout, not an API); a consumer that only needs to *name* the storage —
/// `nros-node`, which carves it out of the executor backing in W3' — names a
/// `ParameterTable` instead.
///
/// **The table's LENGTH is the server's capacity.** [`MAX_PARAMETERS`] is the
/// DEFAULT capacity (see [`ParameterStorage`]), not a bound the borrow
/// re-imposes: a caller who hands over a longer or shorter table gets exactly
/// that many slots, and every capacity question the server answers
/// ([`is_full`](crate::ParameterServerOps::is_full), the declare path) reads it from
/// here.
pub struct ParameterTable<'s> {
    entries: &'s mut [Option<ParameterEntry>],
}

impl<'s> ParameterTable<'s> {
    /// Wrap a caller-owned slice of slots.
    pub(crate) fn new(entries: &'s mut [Option<ParameterEntry>]) -> Self {
        Self { entries }
    }

    /// Number of slots — the capacity of a server built on this table.
    pub fn capacity(&self) -> usize {
        self.entries.len()
    }

    /// Number of slots currently holding a parameter.
    pub fn occupied(&self) -> usize {
        self.entries.iter().filter(|slot| slot.is_some()).count()
    }
}

/// One slot of a [`ParameterTable`], opaque.
///
/// phase-382 W3' — what a caller CARVES when it places a table in memory it
/// already owns (the executor's backing). The slot layout is
/// `Option<ParameterEntry>` and stays private; this wrapper publishes only its
/// size and alignment (`size_of::<ParameterSlot>()`), which is exactly what a
/// region placement needs, and [`ParameterTable::init_in`] is the one way to
/// turn carved, uninitialised slots into a table.
///
/// `repr(transparent)`, so a `[ParameterSlot]` IS a `[Option<ParameterEntry>]`
/// and the table borrows it without a copy.
#[repr(transparent)]
pub struct ParameterSlot(Option<ParameterEntry>);

impl<'s> ParameterTable<'s> {
    /// Initialise `slots` to empty IN PLACE and borrow them as a table.
    ///
    /// phase-382 W3' — the home the executor carves for its store. Each slot
    /// is written through its own pointer, so the largest stack temporary is
    /// one slot, never the table (issue 0756's hazard, which a carved region
    /// avoids by construction as long as nothing builds the table by value).
    /// Any length works; the length is the capacity, as for every table.
    pub fn init_in(slots: &'s mut [core::mem::MaybeUninit<ParameterSlot>]) -> Self {
        for slot in slots.iter_mut() {
            slot.write(ParameterSlot(None));
        }
        let len = slots.len();
        let base = slots.as_mut_ptr().cast::<Option<ParameterEntry>>();
        // SAFETY: every element was written above, so the slice is initialised;
        // `ParameterSlot` is `repr(transparent)` over `Option<ParameterEntry>`,
        // so the element layouts are identical; the borrow is `slots`' own,
        // for `'s`, and `slots` is not reachable again while it lives.
        let entries = unsafe { core::slice::from_raw_parts_mut(base, len) };
        Self::new(entries)
    }
}

impl core::fmt::Debug for ParameterTable<'_> {
    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
        f.debug_struct("ParameterTable")
            .field("capacity", &self.capacity())
            .finish_non_exhaustive()
    }
}

/// Backing storage a caller places, then lends to a [`ParameterServer`].
///
/// `N` defaults to [`MAX_PARAMETERS`], the build-time knob
/// (`NROS_MAX_PARAMETERS` / `CONFIG_NROS_MAX_PARAMETERS`). It is a DEFAULT,
/// not a cap — `ParameterStorage::<8>::new()` and `ParameterStorage::<256>`
/// are equally valid, and the server takes its capacity from whatever table it
/// is handed.
///
/// The intended shapes are a `static`, a caller-owned struct field, or (with
/// `alloc`) one allocation made once at start-up:
///
/// ```
/// use nros_params::{
///     NodeKey, ParameterServer, ParameterServerOps as _, ParameterStorage, ParameterValue,
/// };
///
/// let mut storage: ParameterStorage<8> = ParameterStorage::new();
/// let mut server = ParameterServer::new_in(storage.as_table());
/// assert!(server.declare(NodeKey::PRIMARY, "max_speed", ParameterValue::Double(1.0)));
/// ```
///
/// # Why this type exists at all — issue 0756
///
/// This is where the size lives, and it is enormous: `ParameterValue` is sized
/// by its largest variant
/// (`StringArray(Vec<String<MAX_STRING_VALUE_LEN>, MAX_ARRAY_LEN>)`), so every
/// slot costs ~8.5 KiB whatever it actually holds. Measured: **285,184 bytes**
/// at the default 32 slots, **2,281,472** at 256. (The eight extra bytes in
/// issue 0756's 285,192 were the `count` that stayed behind on
/// [`ParameterServer`], which is 24 bytes total now.)
///
/// Before phase-382 W2' that bulk was INSIDE `ParameterServer`, so
/// `Box::new(ParameterServer::new())` built a multi-hundred-KiB value on the
/// STACK before copying it into the allocation — Rust has no placement-new —
/// which silently overran a thread stack sized for anything smaller. On the
/// Zephyr lane an image built with 256 slots booted to
/// `dds_create_participant` and hung with no fault and no output, while 32 —
/// which fits the 512 KiB main stack the cyclonedds snippet asks for — ran
/// clean. The mitigation then was `ParameterServer::init_in_place`, a
/// placement initialiser that bounded the largest temporary at one
/// `Option<ParameterEntry>`.
///
/// W2' dissolves the *server* half of that: `ParameterServer` is now a table
/// borrow plus a count (tens of bytes), so nothing about constructing a server
/// depends on the knob. The bulk did not vanish, it MOVED here — and so did
/// the hazard, verbatim: `Box::new(ParameterStorage::new())` is the same
/// stack-first copy the issue was about. [`init_in_place`] is therefore kept,
/// on the type that is actually big, and any heap-placed storage must go
/// through it. A caller who places the storage as a `static` or lets W3' carve
/// it out of the executor backing never materialises it by value and does not
/// need it.
///
/// [`init_in_place`]: ParameterStorage::init_in_place
pub struct ParameterStorage<const N: usize = MAX_PARAMETERS> {
    entries: [Option<ParameterEntry>; N],
}

impl<const N: usize> ParameterStorage<N> {
    /// Create empty storage.
    ///
    /// `const`, so a `static NROS_PARAMS: ParameterStorage = ParameterStorage::new();`
    /// lands in `.bss` with no initialiser and no stack temporary.
    // No `Default`: it would spell this same construction as a by-value
    // expression, which is exactly the `Box::new(Default::default())` shape
    // issue 0756 is about (see the type docs). Callers name `new()` — or
    // `init_in_place` when the home is a heap allocation.
    #[allow(clippy::new_without_default)]
    #[allow(clippy::large_stack_arrays)] // Intentional: this IS the storage.
    pub const fn new() -> Self {
        Self {
            entries: [const { None }; N],
        }
    }

    /// Slots in this storage.
    pub const fn capacity(&self) -> usize {
        N
    }

    /// Lend the storage to a [`ParameterServer`].
    pub fn as_table(&mut self) -> ParameterTable<'_> {
        ParameterTable::new(&mut self.entries)
    }

    /// Initialise storage directly into `dst`, without ever materialising one
    /// by value.
    ///
    /// Issue 0756 — see this type's documentation for why that matters. The
    /// short version: this type is 285,184 bytes at the default 32 slots and
    /// 2,281,472 at 256, and Rust has no placement-new, so
    /// `Box::new(ParameterStorage::new())` builds all of it on the caller's
    /// stack first. Initialising through the allocation bounds the largest
    /// stack temporary at one `Option<ParameterEntry>`, so the
    /// `NROS_MAX_PARAMETERS` knob no longer decides whether boot survives.
    ///
    /// # Safety
    ///
    /// `dst` must be non-null, correctly aligned, and valid for writes of
    /// `size_of::<Self>()` bytes. The pointee may be uninitialised on entry;
    /// it is fully initialised on return.
    pub unsafe fn init_in_place(dst: *mut Self) {
        // addr_of_mut! throughout: `(*dst).entries` as a place expression
        // would require the pointee to already be initialised.
        // Safety: delegated to this function's own contract on `dst`.
        unsafe {
            let entries = core::ptr::addr_of_mut!((*dst).entries).cast::<Option<ParameterEntry>>();
            for i in 0..N {
                entries.add(i).write(None);
            }
        }
    }
}

/// Parameter server over caller-owned storage
///
/// Stores parameters in a [`ParameterTable`] the caller lends it; the table's
/// length is the capacity. [`ParameterStorage`] is the usual way to produce
/// one, defaulting to the build-time `MAX_PARAMETERS` slots.
///
/// # Keyed by node — phase-426 W1
///
/// Every per-parameter method takes a [`NodeKey`] first. The slot arena is
/// shared (one table per executor), but a name is only ever looked up within
/// one node, so two nodes composed onto one executor (RFC-0047) hold
/// independent values for the same name — which is what upstream does, where
/// each node owns a table and a set of six services.
///
/// # Example
///
/// ```
/// use nros_params::{
///     NodeKey, ParameterServer, ParameterServerOps as _, ParameterStorage, ParameterValue,
/// };
///
/// let mut storage = ParameterStorage::<8>::new();
/// let mut server = ParameterServer::new_in(storage.as_table());
/// let talker = NodeKey::new(0);
/// let listener = NodeKey::new(1);
///
/// // Declare a parameter with initial value
/// server.declare(talker, "rate", ParameterValue::Double(1.0));
/// server.declare(listener, "rate", ParameterValue::Double(2.0));
///
/// // Get parameter value
/// assert_eq!(
///     server.get(talker, "rate").and_then(|v| v.as_double()),
///     Some(1.0)
/// );
/// assert_eq!(
///     server.get(listener, "rate").and_then(|v| v.as_double()),
///     Some(2.0)
/// );
///
/// // Set parameter value
/// server.set(talker, "rate", ParameterValue::Double(3.0));
/// ```
pub struct ParameterServer<'s> {
    /// Borrowed parameter storage
    table: ParameterTable<'s>,
    /// Number of parameters currently stored, across every node.
    count: usize,
    /// May a wire-facing set ([`apply`](Self::apply)) DECLARE a name the node
    /// does not hold? Off by default, as upstream
    /// (`allow_undeclared_parameters` in rclcpp's `NodeOptions`,
    /// `allow_undeclared` in rclrs) — issue 1151.
    ///
    /// phase-426 W1 — PER NODE, because upstream's is a node option. One node
    /// opting in must not decide what a remote set may do to its sibling.
    allow_undeclared: NodeFlags,
    /// phase-491 W2 — nodes whose remote `NOT_SET` DELETES the parameter
    /// (rclc's set-service behaviour) instead of being refused as a type
    /// change (rclcpp's, for a statically typed parameter). Set by the C
    /// `rclc_parameter_server_t`; per node because the two client libraries
    /// disagree and both live in one store.
    delete_on_not_set: NodeFlags,
    /// phase-446 F2 -- a declared descriptor holds a truncated description
    /// nobody has reported yet. Lets [`Self::take_truncated_descriptions`]
    /// answer "nothing" without walking the table, which is what lets the
    /// executor ask on every spin.
    truncation_pending: bool,
    /// phase-491 W3 — nodes whose changes are logged for `/parameter_events`.
    events_enabled: NodeFlags,
    /// phase-491 W3 — changes not yet published, oldest first.
    event_log: heapless::Deque<PendingParameterEvent, PARAMETER_EVENT_LOG_LEN>,
    /// phase-491 W3 — changes that found the log full since the last read.
    events_dropped: u32,
    /// Issue 2023 — nodes whose events just went ON and whose start-up
    /// event (every parameter they already hold, as `New`) is not yet
    /// drained. rclcpp's publisher exists before the first declaration, so
    /// a subscriber sees every parameter arrive; ours goes on later (the
    /// node table is built after the store), and this is what keeps the two
    /// equal without a log entry per parameter.
    events_initial: NodeFlags,
    /// Issue 2023 — how far the start-up drain of [`Self::events_initial`]'s
    /// lowest node has got, as an index into that node's parameters.
    initial_cursor: usize,
    /// phase-417 W4.a — the bounded on-set-parameters registry. See
    /// [`MAX_ON_SET_CALLBACKS`] for why it is bounded and why 4.
    on_set: [Option<OnSetCallbackSlot>; MAX_ON_SET_CALLBACKS],
}

impl<'s> ParameterServer<'s> {
    /// phase-446 F2 -- report each declared parameter whose description was
    /// truncated to `NROS_MAX_PARAM_DESCRIPTION_LEN`, ONCE, and forget it.
    ///
    /// This crate has no logger, so it records the truncation on the
    /// descriptor and the executor, which has one, drains it here and logs.
    /// The truncation itself happens where the text is given
    /// ([`ParameterDescriptor::set_description`]), and a declaration can come
    /// through several builders; the store is where every one of them lands.
    /// Returns how many were reported.
    pub(crate) fn take_truncated_descriptions(
        &mut self,
        mut report: impl FnMut(NodeKey, &str),
    ) -> usize {
        if !self.truncation_pending {
            return 0;
        }
        self.truncation_pending = false;
        let mut n = 0;
        for entry in self.table.entries.iter_mut().flatten() {
            if let Some(d) = entry.descriptor.as_mut()
                && d.description_truncated()
            {
                d.clear_description_truncated();
                report(entry.node, entry.param.name.as_str());
                n += 1;
            }
        }
        n
    }

    /// Note a descriptor entering the store (phase-446 F2).
    #[inline]
    fn note_descriptor(&mut self, descriptor: Option<&ParameterDescriptor>) {
        if descriptor.is_some_and(ParameterDescriptor::description_truncated) {
            self.truncation_pending = true;
        }
    }

    /// Create a parameter server over `table`.
    ///
    /// The server ADOPTS whatever the table already holds: `count` is
    /// recovered from the occupied slots rather than assumed zero. That is
    /// what makes a table safe to hand to a second server after the first is
    /// dropped (and what lets storage arrive pre-seeded), where assuming empty
    /// would desynchronise the count from the slots — `is_full` wrong in one
    /// direction, `remove` underflowing in the other. Fresh
    /// [`ParameterStorage`] is all-`None`, so the usual path costs one scan
    /// and starts at zero.
    pub(crate) fn new_in(table: ParameterTable<'s>) -> Self {
        let count = table.occupied();
        Self {
            table,
            count,
            allow_undeclared: NodeFlags::new(),
            delete_on_not_set: NodeFlags::new(),
            on_set: [None; MAX_ON_SET_CALLBACKS],
            // An adopted table may already hold a truncated description, so
            // the first drain looks rather than assuming none.
            truncation_pending: true,
            events_enabled: NodeFlags::new(),
            event_log: heapless::Deque::new(),
            events_dropped: 0,
            events_initial: NodeFlags::new(),
            initial_cursor: 0,
        }
    }

    /// Let a wire-facing set ([`apply`](Self::apply)) create a parameter
    /// `node` never declared. Default `false`: a remote set on a misspelled
    /// name used to report success and create the misspelling (issue 1151);
    /// now it is refused with a reason unless this is switched on, as
    /// upstream.
    ///
    /// phase-426 W1 — scoped to `node`. Switching it on for `/talker` leaves
    /// `/listener` refusing undeclared names.
    pub(crate) fn set_allow_undeclared(&mut self, node: NodeKey, allow: bool) {
        self.allow_undeclared.set(node, allow);
    }

    /// Whether [`apply`](Self::apply) may declare an unknown name on `node`.
    pub(crate) fn allows_undeclared(&self, node: NodeKey) -> bool {
        self.allow_undeclared.get(node)
    }

    /// phase-491 W3 — log `node`'s changes for `/parameter_events`
    /// (rclc's `notify_changed_over_dds`). Off by default, so a node nobody
    /// asked events for pays nothing.
    ///
    /// Issue 2023 — switching it ON (from off) also queues the node's
    /// START-UP event: every parameter the node already holds is reported as
    /// `New` by [`pop_initial_parameter_event`](Self::pop_initial_parameter_event),
    /// or one [`ParameterEventKind::Announce`] when it holds none, so a
    /// subscriber sees what rclcpp's would although our publisher comes up
    /// after the first declarations.
    pub(crate) fn set_parameter_events(&mut self, node: NodeKey, on: bool) {
        if on && !self.events_enabled.get(node) {
            self.events_initial.set(node, true);
        }
        if !on {
            self.events_initial.set(node, false);
            if self.events_initial.first().is_none() {
                self.initial_cursor = 0;
            }
        }
        self.events_enabled.set(node, on);
    }

    /// Whether `node`'s changes are logged for `/parameter_events`.
    pub(crate) fn parameter_events_enabled(&self, node: NodeKey) -> bool {
        self.events_enabled.get(node)
    }

    /// Issue 2023 — the next entry of a node's START-UP event, queued when its
    /// events went on: one `New` per parameter it held then (read at drain
    /// time, so it reports the current value), or one `Announce` for a node
    /// that holds none. Drain this BEFORE [`pop_parameter_event`](Self::pop_parameter_event);
    /// while a node's start-up event is pending its changes are not logged
    /// separately, because the drain reports them.
    ///
    /// Bounded and allocation-free: a cursor into the fixed table, no copy of
    /// the node's parameter list.
    pub(crate) fn pop_initial_parameter_event(&mut self) -> Option<PendingParameterEvent> {
        loop {
            let node = self.events_initial.first()?;
            let k = self.initial_cursor;
            let next = self.iter(node).nth(k).map(|p| p.name.clone());
            match next {
                Some(name) => {
                    self.initial_cursor = k + 1;
                    return Some(PendingParameterEvent {
                        node,
                        kind: ParameterEventKind::New,
                        name,
                    });
                }
                None => {
                    self.events_initial.set(node, false);
                    self.initial_cursor = 0;
                    if k == 0 {
                        return Some(PendingParameterEvent {
                            node,
                            kind: ParameterEventKind::Announce,
                            name: String::new(),
                        });
                    }
                }
            }
        }
    }

    /// phase-491 W3 — the oldest change waiting for `/parameter_events`.
    pub(crate) fn pop_parameter_event(&mut self) -> Option<PendingParameterEvent> {
        self.event_log.pop_front()
    }

    /// phase-491 W3 — how many changes found the log full since the last
    /// call, and reset the count.
    pub(crate) fn take_parameter_events_dropped(&mut self) -> u32 {
        core::mem::take(&mut self.events_dropped)
    }

    /// Record a change for `/parameter_events`, merging into an entry already
    /// pending for the same name.
    fn note_event(&mut self, node: NodeKey, kind: ParameterEventKind, name: &str) {
        // Issue 2023 — a node whose start-up event is still pending is
        // reported by that drain, which reads the table as it stands.
        if !self.events_enabled.get(node) || self.events_initial.get(node) {
            return;
        }
        if let Some(e) = self
            .event_log
            .iter_mut()
            .find(|e| e.node == node && e.name.as_str() == name)
        {
            // New-then-changed is still new; anything then deleted is deleted;
            // deleted-then-declared is new again.
            e.kind = match (e.kind, kind) {
                (ParameterEventKind::New, ParameterEventKind::Changed) => ParameterEventKind::New,
                (_, k) => k,
            };
            return;
        }
        let mut stored = String::new();
        if stored.push_str(name).is_err()
            || self
                .event_log
                .push_back(PendingParameterEvent {
                    node,
                    kind,
                    name: stored,
                })
                .is_err()
        {
            self.events_dropped = self.events_dropped.saturating_add(1);
        }
    }

    /// phase-491 W2 — make a wire-facing `NOT_SET` on `node` DELETE the
    /// parameter, after the on-set hooks accept `(old, None)`. This is rclc's
    /// parameter server (`rclc_parameter_server_set_service_callback`); the
    /// default, off, keeps rclcpp's refusal.
    pub(crate) fn set_delete_on_not_set(&mut self, node: NodeKey, on: bool) {
        self.delete_on_not_set.set(node, on);
    }

    /// Get the number of parameters `node` has stored.
    pub(crate) fn len(&self, node: NodeKey) -> usize {
        self.entries_of(node).count()
    }

    /// Check if `node` has no parameters
    pub(crate) fn is_empty(&self, node: NodeKey) -> bool {
        self.len(node) == 0
    }

    /// Parameters held across EVERY node — how much of the shared arena is in
    /// use. [`len`](Self::len) is the per-node question; this one is the
    /// capacity question's other half.
    pub(crate) fn total_len(&self) -> usize {
        self.count
    }

    /// Capacity of the borrowed table.
    ///
    /// phase-382 W2' — this is the TABLE's length, not `MAX_PARAMETERS`. The
    /// build-time knob only picks the default size of a [`ParameterStorage`].
    ///
    /// phase-426 W1 — the arena is SHARED across nodes, so this is not a
    /// per-node budget: `MAX_PARAMETERS` bounds the image, not each node.
    pub(crate) fn capacity(&self) -> usize {
        self.table.capacity()
    }

    /// Check if the server is at capacity (across every node)
    pub(crate) fn is_full(&self) -> bool {
        self.count >= self.table.capacity()
    }

    /// Occupied slots belonging to `node`.
    fn entries_of(&self, node: NodeKey) -> impl Iterator<Item = &ParameterEntry> {
        self.table
            .entries
            .iter()
            .filter_map(|slot| slot.as_ref())
            .filter(move |e| e.node == node)
    }

    /// Find the index of `node`'s parameter by name.
    ///
    /// phase-426 W1 — the node is part of the key, so a sibling's identically
    /// named parameter is simply not found here.
    fn find_index(&self, node: NodeKey, name: &str) -> Option<usize> {
        self.table.entries.iter().position(|entry| {
            entry
                .as_ref()
                .map(|e| e.node == node && e.param.name.as_str() == name)
                .unwrap_or(false)
        })
    }

    /// Find an empty slot
    fn find_empty_slot(&self) -> Option<usize> {
        self.table.entries.iter().position(|entry| entry.is_none())
    }

    /// Declare a new parameter with a value on `node`
    ///
    /// If the parameter already exists, this does nothing and returns false.
    /// Returns true if the parameter was declared successfully.
    ///
    /// phase-428 W6 remainder: this is the method `Executor::declare_parameter`
    /// delegates to, so marking only the wrapper would leave the class half
    /// fixed one crate down.
    #[must_use]
    pub(crate) fn declare(&mut self, node: NodeKey, name: &str, value: ParameterValue) -> bool {
        self.declare_with_descriptor(node, name, value, None)
    }

    /// [`declare`](Self::declare), borrowing the value (issue 2070): the
    /// spelling a large value should take, since it is copied into its slot
    /// once instead of through every frame.
    #[must_use]
    pub(crate) fn declare_ref(
        &mut self,
        node: NodeKey,
        name: &str,
        value: &ParameterValue,
    ) -> bool {
        self.try_declare_ref(node, name, value, None).is_ok()
    }

    /// Declare a new parameter with value and descriptor on `node`
    ///
    /// The descriptor provides metadata like description, constraints, and read-only flag.
    ///
    /// phase-428 W6 remainder: see [`Self::declare`]. Every `false` arm below
    /// is a REFUSAL (name taken, table full, value rejected), never a question.
    #[must_use]
    pub(crate) fn declare_with_descriptor(
        &mut self,
        node: NodeKey,
        name: &str,
        value: ParameterValue,
        descriptor: Option<ParameterDescriptor>,
    ) -> bool {
        self.try_declare_with_descriptor(node, name, value, descriptor)
            .is_ok()
    }

    /// [`declare`](Self::declare), saying WHY it refused — phase-417 W4.a.
    ///
    /// rclrs separates declare-time failures from set-time ones
    /// ([`DeclarationError`] against [`ParameterError`]); ours returned a
    /// bare `bool` for "the name was taken", "the table was full" and "the
    /// default is outside the range" alike, so a caller could not tell a
    /// programming error from a capacity one. The `bool` spellings above
    /// forward to this, so there is exactly one declare path.
    pub(crate) fn try_declare(
        &mut self,
        node: NodeKey,
        name: &str,
        value: ParameterValue,
    ) -> Result<(), DeclarationError> {
        self.try_declare_with_descriptor(node, name, value, None)
    }

    /// [`declare_with_descriptor`](Self::declare_with_descriptor), saying WHY
    /// it refused. See [`try_declare`](Self::try_declare).
    pub(crate) fn try_declare_with_descriptor(
        &mut self,
        node: NodeKey,
        name: &str,
        value: ParameterValue,
        descriptor: Option<ParameterDescriptor>,
    ) -> Result<(), DeclarationError> {
        self.try_declare_ref(node, name, &value, descriptor)
    }

    /// [`try_declare_with_descriptor`](Self::try_declare_with_descriptor),
    /// borrowing the value — issue 2070. THE declare path; the by-value
    /// spellings forward here.
    ///
    /// A [`ParameterValue`] is as large as its largest variant (`StringArray`:
    /// `MAX_ARRAY_LEN` x `MAX_STRING_VALUE_LEN`, ~8 KiB at the defaults), and
    /// the by-value path moved it through every frame and built it again in
    /// [`Parameter::new`] and in the slot's `Some(..)`: ~57 KiB of stack for
    /// one `declare_parameter<double>` on aarch64, ~67 KiB on Cortex-M3,
    /// which overflowed a 64 KiB RTOS app stack into the kernel's globals.
    /// Here the slot is filled with an empty value and the caller's is copied
    /// into it once, in place ([`ParameterValue::assign_from`]).
    pub(crate) fn try_declare_ref(
        &mut self,
        node: NodeKey,
        name: &str,
        value: &ParameterValue,
        descriptor: Option<ParameterDescriptor>,
    ) -> Result<(), DeclarationError> {
        // Check if already exists
        if self.find_index(node, name).is_some() {
            return Err(DeclarationError::AlreadyDeclared);
        }

        // Issue 1150 — an ill-formed range, or a default the range rejects,
        // is refused at declaration, as upstream. Same predicate as every set.
        Self::declaration_consistency(descriptor.as_ref(), value)?;

        // Find an empty slot
        let slot = match self.find_empty_slot() {
            Some(idx) => idx,
            None => return Err(DeclarationError::StorageFull),
        };

        // The name is checked before the slot is touched, so a refusal
        // leaves the table as it was.
        let mut stored_name = String::new();
        if stored_name.push_str(name).is_err() {
            return Err(DeclarationError::NameTooLong);
        }

        self.note_descriptor(descriptor.as_ref());
        // Issue 2070 -- the slot is written in place and the value copied in
        // once: an entry built on the stack and moved in is a second ~8 KiB
        // `ParameterValue` in this frame.
        let cell = &mut self.table.entries[slot];
        *cell = Some(ParameterEntry {
            node,
            param: Parameter {
                name: stored_name,
                value: ParameterValue::NotSet,
            },
            descriptor,
        });
        if let Some(entry) = cell {
            entry.param.value.assign_from(value);
        }
        self.count += 1;
        self.note_event(node, ParameterEventKind::New, name);
        Ok(())
    }

    /// The declare-time half of issue 1150: the range must be well-formed
    /// and the initial value must satisfy it. ONE spelling for both declare
    /// paths.
    ///
    /// phase-417 W4.a — it returns rclrs's two distinct names rather than a
    /// `bool`, because the remedies differ: `InvalidRange` means fix the
    /// range, `InitialValueOutOfRange` means fix the default. The `bool`
    /// spelling below is kept for the caller that only has a
    /// `ParameterError` channel, so there is still one predicate.
    pub(crate) fn declaration_consistency(
        descriptor: Option<&ParameterDescriptor>,
        value: &ParameterValue,
    ) -> Result<(), DeclarationError> {
        match descriptor {
            Some(desc) if !desc.range.is_valid() => Err(DeclarationError::InvalidRange),
            Some(desc) if !desc.validate_range(value) => {
                Err(DeclarationError::InitialValueOutOfRange)
            }
            _ => Ok(()),
        }
    }

    fn declaration_is_consistent(
        descriptor: Option<&ParameterDescriptor>,
        value: &ParameterValue,
    ) -> bool {
        Self::declaration_consistency(descriptor, value).is_ok()
    }

    /// Would `set(name, value)` succeed on an EXISTING entry? Read-only,
    /// then type, then range — the one ordering every set path uses.
    fn check_set_existing(
        entry: &ParameterEntry,
        value: &ParameterValue,
    ) -> Result<(), ParameterError> {
        if let Some(ref desc) = entry.descriptor {
            if desc.read_only {
                return Err(ParameterError::ReadOnly);
            }
            if !desc.dynamic_typing && desc.param_type != value.param_type() {
                return Err(ParameterError::TypeMismatch);
            }
            if !desc.validate_range(value) {
                return Err(ParameterError::OutOfRange);
            }
        }
        Ok(())
    }

    /// Dry-run of [`apply`](Self::apply): the verdict a wire-facing set would
    /// get, without changing anything. `SetParametersAtomically` runs this
    /// over the whole batch before applying any of it.
    pub(crate) fn check_apply(
        &self,
        node: NodeKey,
        name: &str,
        value: &ParameterValue,
    ) -> Result<(), ParameterError> {
        match self
            .find_index(node, name)
            .and_then(|idx| self.table.entries[idx].as_ref())
        {
            Some(entry) => Self::check_set_existing(entry, value),
            None if !self.allows_undeclared(node) => Err(ParameterError::Undeclared),
            None if self.is_full() => Err(ParameterError::StorageFull),
            None if name.len() > MAX_PARAM_NAME_LEN => Err(ParameterError::StorageFull),
            None => Ok(()),
        }
    }

    /// THE wire-facing set — what one element of `SetParameters` /
    /// `SetParametersAtomically` does (issue 1151).
    ///
    /// An existing parameter is set through [`set`](Self::set). An unknown
    /// name is [`Undeclared`](ParameterError::Undeclared) unless
    /// [`set_allow_undeclared`](Self::set_allow_undeclared) was switched on
    /// FOR THAT NODE, in which case it is declared (or `StorageFull`). Every
    /// service handler and every facade writer goes through here so the
    /// by-value oracle, the streaming path, the atomic pre-check and the C
    /// setters cannot disagree about what "undeclared" means.
    pub(crate) fn apply(
        &mut self,
        node: NodeKey,
        name: &str,
        value: ParameterValue,
    ) -> Result<(), ParameterError> {
        self.apply_ref(node, name, &value)
    }

    /// [`apply`](Self::apply), borrowing the value -- issue 2104, and THE
    /// apply the by-value spelling forwards to. The value is copied once, into
    /// its slot (`ParameterValue::assign_from`); moved in, a ~8 KiB value is a
    /// copy in every frame it passes, which is what put a C++ array set at
    /// 16.6 KiB of Cortex-M3 stack.
    pub(crate) fn apply_ref(
        &mut self,
        node: NodeKey,
        name: &str,
        value: &ParameterValue,
    ) -> Result<(), ParameterError> {
        if matches!(value, ParameterValue::NotSet) && self.delete_on_not_set.get(node) {
            return self.apply_delete(node, name);
        }
        self.check_apply(node, name, value)?;
        // phase-417 W4.a — the accept/reject hook, AFTER the store's own rules
        // and BEFORE the write, which is rclcpp's ordering: a callback never
        // sees a write the type, range or read-only rules already refused, and
        // a callback that refuses leaves the value untouched. `apply` is the
        // one seam every writer funnels through (the six services, the C and
        // C++ facades, `Executor::set_parameter`), so hooking it here is what
        // makes one registration reach all of them.
        if !self.run_on_set_callbacks(node, name, self.get(node, name), Some(value)) {
            return Err(ParameterError::Rejected);
        }
        if self.find_index(node, name).is_some() {
            self.set_ref(node, name, value)
        } else if self.declare_ref(node, name, value) {
            Ok(())
        } else {
            Err(ParameterError::StorageFull)
        }
    }

    /// [`apply`](Self::apply), taking the value from a caller-owned STAGING
    /// slot instead of by value — RFC-0105 D10.2 / issue 1812.
    ///
    /// A `ParameterValue` is sized by its largest variant (~8.5 KiB at the
    /// default bounds), so `apply(value)` puts one on the caller's stack, and
    /// the wire handler that decoded it put another there before that. A
    /// remote set decodes into a slot carved from the executor backing and
    /// hands it here: the same rules, hooks and order as `apply`, and the
    /// write is a SWAP between the slot and the stored value, so no
    /// value-sized temporary exists on any path that sets an existing
    /// parameter. On return `staged` holds the previous value (or whatever it
    /// held, when the set was refused); its content is scratch.
    ///
    /// The one path that still builds an entry is the undeclared add
    /// (`allow_undeclared`), which places an EMPTY entry first and then swaps
    /// the value into it; see `declare_placeholder`.
    pub(crate) fn apply_staged(
        &mut self,
        node: NodeKey,
        name: &str,
        staged: &mut ParameterValue,
    ) -> Result<(), ParameterError> {
        if matches!(staged, ParameterValue::NotSet) && self.delete_on_not_set.get(node) {
            return self.apply_delete(node, name);
        }
        self.check_apply(node, name, staged)?;
        if !self.run_on_set_callbacks(node, name, self.get(node, name), Some(staged)) {
            return Err(ParameterError::Rejected);
        }
        let (idx, kind) = match self.find_index(node, name) {
            Some(idx) => (idx, ParameterEventKind::Changed),
            None => match self.declare_placeholder(node, name) {
                Some(idx) => (idx, ParameterEventKind::New),
                None => return Err(ParameterError::StorageFull),
            },
        };
        match self.table.entries[idx].as_mut() {
            Some(entry) => {
                core::mem::swap(&mut entry.param.value, staged);
                self.note_event(node, kind, name);
                Ok(())
            }
            None => Err(ParameterError::NotFound),
        }
    }

    /// Place a `NotSet` entry named `name` on `node` and return its index —
    /// the undeclared-add half of [`apply_staged`](Self::apply_staged).
    ///
    /// Out of line on purpose: an unoptimised build materialises the entry it
    /// writes, and that frame belongs on the rare path only, never on the
    /// frame every remote set runs in.
    #[inline(never)]
    fn declare_placeholder(&mut self, node: NodeKey, name: &str) -> Option<usize> {
        let slot = self.find_empty_slot()?;
        let mut stored = String::new();
        stored.push_str(name).ok()?;
        self.table.entries[slot] = Some(ParameterEntry {
            node,
            param: Parameter {
                name: stored,
                value: ParameterValue::NotSet,
            },
            descriptor: None,
        });
        self.count += 1;
        Some(slot)
    }

    /// The delete half of [`apply`](Self::apply) for a node that asked for
    /// [`set_delete_on_not_set`](Self::set_delete_on_not_set): read-only
    /// refuses, the hooks see `(old, None)`, then the parameter is removed.
    fn apply_delete(&mut self, node: NodeKey, name: &str) -> Result<(), ParameterError> {
        let Some(entry) = self
            .find_index(node, name)
            .and_then(|idx| self.table.entries[idx].as_ref())
        else {
            return Err(ParameterError::NotFound);
        };
        if entry.descriptor.as_ref().is_some_and(|d| d.read_only) {
            return Err(ParameterError::ReadOnly);
        }
        if !self.run_on_set_callbacks(node, name, self.get(node, name), None) {
            return Err(ParameterError::Rejected);
        }
        self.remove(node, name);
        Ok(())
    }

    /// phase-491 W2 (RFC-0105 D6/D7) — rclc's LOCAL typed set,
    /// `rclc_parameter_set_{bool,int,double}`.
    ///
    /// rclc's owner of a parameter is not bound by its own read-only flag or
    /// range: read-only is "read-only to the OUTSIDE", and rclc never checks a
    /// range at all. So this is type, then the hooks with `(old, new)`, then
    /// the write — and nothing else. Remote writes keep going through
    /// [`apply`](Self::apply), which enforces both.
    ///
    /// The type check compares against the STORED value's type unless the
    /// descriptor allows dynamic typing, because rclc refuses a typed set of
    /// the wrong type (`RCLC_PARAMETER_TYPE_MISMATCH`) whether or not a
    /// descriptor was ever attached.
    pub(crate) fn set_local(
        &mut self,
        node: NodeKey,
        name: &str,
        value: ParameterValue,
    ) -> Result<(), ParameterError> {
        let Some(idx) = self.find_index(node, name) else {
            return Err(ParameterError::NotFound);
        };
        let Some(entry) = self.table.entries[idx].as_ref() else {
            return Err(ParameterError::NotFound);
        };
        let dynamic = entry.descriptor.as_ref().is_some_and(|d| d.dynamic_typing);
        if !dynamic && entry.param.value.param_type() != value.param_type() {
            return Err(ParameterError::TypeMismatch);
        }
        if !self.run_on_set_callbacks(node, name, Some(&entry.param.value), Some(&value)) {
            return Err(ParameterError::Rejected);
        }
        if let Some(entry) = self.table.entries[idx].as_mut() {
            entry.param.value = value;
        }
        self.note_event(node, ParameterEventKind::Changed, name);
        Ok(())
    }

    /// Run every registered hook for `node`. `false` from any one of them
    /// refuses the write; the rest are not consulted, as rclcpp stops at the
    /// first unsuccessful result.
    fn run_on_set_callbacks(
        &self,
        node: NodeKey,
        name: &str,
        old: Option<&ParameterValue>,
        new: Option<&ParameterValue>,
    ) -> bool {
        for slot in self.on_set.iter().flatten() {
            if slot.node == node && !(slot.callback)(node, name, old, new, slot.ctx) {
                return false;
            }
        }
        true
    }

    /// Register an accept/reject hook for writes to `node` — rclcpp's
    /// `add_on_set_parameters_callback` (phase-417 W4.a).
    ///
    /// The callback runs on every [`apply`](Self::apply): a remote
    /// `ros2 param set`, a C++
    /// `Node::set_parameter`, a Rust `Executor::set_parameter`. Returning
    /// `false` refuses the write with [`ParameterError::Rejected`] and
    /// leaves the stored value alone.
    ///
    /// `None` when all [`MAX_ON_SET_CALLBACKS`] slots are taken — a refusal,
    /// never a silent eviction.
    pub(crate) fn add_on_set_parameters_callback(
        &mut self,
        node: NodeKey,
        callback: OnSetParameterFn,
        ctx: OnSetContext,
    ) -> Option<OnSetParameterHandle> {
        let idx = self.on_set.iter().position(Option::is_none)?;
        self.on_set[idx] = Some(OnSetCallbackSlot {
            node,
            callback,
            ctx,
        });
        Some(OnSetParameterHandle(idx as u16))
    }

    /// Unregister a hook — rclcpp's `remove_on_set_parameters_callback`.
    /// `false` if the slot was already empty.
    pub(crate) fn remove_on_set_parameters_callback(
        &mut self,
        handle: OnSetParameterHandle,
    ) -> bool {
        let idx = handle.0 as usize;
        if idx >= MAX_ON_SET_CALLBACKS || self.on_set[idx].is_none() {
            return false;
        }
        self.on_set[idx] = None;
        true
    }

    /// How many hook slots are free. Lets a caller ask before registering
    /// rather than discovering the bound as a `None`.
    pub(crate) fn on_set_callbacks_free(&self) -> usize {
        self.on_set.iter().filter(|s| s.is_none()).count()
    }

    // ---- descriptor mutation on a DECLARED parameter (phase-417 W4.a) ------
    //
    // rclc's shape: `rclc_add_parameter_description`,
    // `rclc_add_parameter_constraint_double` / `_integer` and
    // `rclc_set_parameter_read_only` all attach metadata AFTER the declaration,
    // so a C caller that cannot build a descriptor value still gets one. Ours
    // are the same verbs against the one store, and they are what the C and C++
    // descriptor surfaces forward to — a wrapper that built its own descriptor
    // table would be the second store phase-426 removed.
    //
    // Each creates the descriptor on demand: a parameter declared without one
    // has `None` in its slot, and refusing to describe it would make "declare
    // then describe" work only in the order rclc does not use.

    /// The descriptor of `node`'s `name`, creating an empty one typed from the
    /// stored value if the declaration carried none.
    fn descriptor_mut(&mut self, node: NodeKey, name: &str) -> Option<&mut ParameterDescriptor> {
        let idx = self.find_index(node, name)?;
        let entry = self.table.entries[idx].as_mut()?;
        if entry.descriptor.is_none() {
            entry.descriptor = Some(ParameterDescriptor::new(
                name,
                entry.param.value.param_type(),
            )?);
        }
        entry.descriptor.as_mut()
    }

    /// Attach a description and free-text constraints to a declared parameter
    /// — rclc's `rclc_add_parameter_description`. Either text may be empty.
    /// `false` if the parameter is not declared on `node`.
    pub(crate) fn set_parameter_description(
        &mut self,
        node: NodeKey,
        name: &str,
        description: &str,
        additional_constraints: &str,
    ) -> bool {
        let mut truncated = false;
        let found = match self.descriptor_mut(node, name) {
            Some(desc) => {
                truncated |= desc.set_description(description);
                truncated |= desc.set_additional_constraints(additional_constraints);
                true
            }
            None => false,
        };
        // Same report path a declared-with-descriptor truncation takes, so the
        // executor logs it once whichever verb supplied the text.
        self.truncation_pending |= truncated;
        found
    }

    /// Mark a declared parameter read-only, so every later write is
    /// [`ParameterError::ReadOnly`] — rclc's
    /// `rclc_set_parameter_read_only`. `false` if not declared on `node`.
    pub(crate) fn set_parameter_read_only(
        &mut self,
        node: NodeKey,
        name: &str,
        read_only: bool,
    ) -> bool {
        match self.descriptor_mut(node, name) {
            Some(desc) => {
                desc.read_only = read_only;
                true
            }
            None => false,
        }
    }

    /// Attach an integer range to a declared parameter — rclc's
    /// `rclc_add_parameter_constraint_integer`.
    ///
    /// Refuses an ill-formed range, and refuses one the CURRENT value does not
    /// satisfy: attaching a constraint that the stored value already violates
    /// would leave the store in a state no `set` could have produced, and
    /// `ros2 param describe` would then advertise a bound the value breaks.
    pub(crate) fn set_parameter_integer_range(
        &mut self,
        node: NodeKey,
        name: &str,
        min: i64,
        max: i64,
        step: i64,
    ) -> bool {
        self.set_parameter_range(
            node,
            name,
            ParameterRange::Integer(IntegerRange::new(min, max, step)),
        )
    }

    /// Attach a floating-point range to a declared parameter — rclc's
    /// `rclc_add_parameter_constraint_double`. See
    /// [`set_parameter_integer_range`](Self::set_parameter_integer_range).
    pub(crate) fn set_parameter_float_range(
        &mut self,
        node: NodeKey,
        name: &str,
        min: f64,
        max: f64,
        step: f64,
    ) -> bool {
        self.set_parameter_range(
            node,
            name,
            ParameterRange::FloatingPoint(FloatingPointRange::new(min, max, step)),
        )
    }

    /// The one range-attaching path both spellings above use.
    fn set_parameter_range(&mut self, node: NodeKey, name: &str, range: ParameterRange) -> bool {
        if !range.is_valid() {
            return false;
        }
        let idx = match self.find_index(node, name) {
            Some(idx) => idx,
            None => return false,
        };
        // The stored value has to satisfy the range BEFORE it is attached, and
        // reading it needs the entry, so the check happens here rather than
        // inside `descriptor_mut`. Borrowed, not cloned (issue 2104): the clone
        // was a 16.9 KiB frame on Cortex-M3 for a range check.
        match self.table.entries[idx].as_ref() {
            Some(entry) if range.contains(&entry.param.value) => {}
            _ => return false,
        }
        match self.descriptor_mut(node, name) {
            Some(desc) => {
                desc.range = range;
                true
            }
            None => false,
        }
    }

    /// All-or-nothing multi-set — `~/set_parameters_atomically`'s local twin
    /// (phase-417 W4.a).
    ///
    /// We SERVED this and offered it in no language, so the only way to reach
    /// our own atomic set was over the wire. This is the same two passes the
    /// service handler makes: every item is checked against the CURRENT state,
    /// and only if all pass is any applied.
    ///
    /// `items` yields `None` for an element that could not even be turned into
    /// a value — which is how the service handler reports a wire value that
    /// does not fit the store's capacity (issue 0323) — and a `None` refuses
    /// the batch like any other failing check.
    ///
    /// It is `Clone` because the two passes walk it twice; a slice iterator is.
    ///
    /// Returns `Ok(())`, or the FIRST verdict that
    /// refused. Nothing is written on a refusal.
    ///
    /// The per-item check is against the state BEFORE the batch, not against
    /// the state each earlier item would leave. That is the service handler's
    /// contract verbatim, and it matters in one case: a batch that declares
    /// more undeclared names than there are free slots passes the check and
    /// fills what it can. Making the pre-check cumulative would need a
    /// speculative copy of the table, which is the allocation this store does
    /// not have.
    pub(crate) fn apply_atomically<'x, I>(
        &mut self,
        node: NodeKey,
        items: I,
    ) -> Result<(), ParameterError>
    where
        I: Iterator<Item = Option<(&'x str, ParameterValue)>> + Clone,
    {
        // Issue 2104 -- ONE implementation of the two passes, in
        // `apply_atomically_with`; this adapts an iterator to its fill
        // callback, restarting the iterator at each pass's first element.
        let count = items.clone().count();
        let mut it = items.clone();
        self.apply_atomically_with(node, count, |i, slot| {
            if i == 0 {
                it = items.clone();
            }
            let (name, value) = it.next()??;
            slot.assign_from(&value);
            Some(name)
        })
    }

    /// [`apply_atomically`](Self::apply_atomically), with element `i` FILLED
    /// into one scratch value by `fill(i, slot)` -- issue 2104.
    ///
    /// `fill` returns the element's name, or `None` for an element that could
    /// not be turned into a value (which refuses the batch, as a `None` item
    /// does). It is called for `0..count` in order, once per pass, so it must
    /// fill the same value for the same `i` both times.
    ///
    /// The iterator form hands each ~8 KiB value out by value, so a C++
    /// `set_parameters_atomically` held one in its closure's frame and more in
    /// its own (25 KiB on Cortex-M3). Here the only value is `slot`, which a
    /// caller fills in place (`ParameterValue::assign_string`, …).
    pub(crate) fn apply_atomically_with<'x>(
        &mut self,
        node: NodeKey,
        count: usize,
        mut fill: impl FnMut(usize, &mut ParameterValue) -> Option<&'x str>,
    ) -> Result<(), ParameterError> {
        let mut slot = ParameterValue::NotSet;
        for i in 0..count {
            let Some(name) = fill(i, &mut slot) else {
                return Err(ParameterError::TypeMismatch);
            };
            self.check_apply(node, name, &slot)?;
            // The hook is consulted in the CHECK pass, so a callback that
            // refuses one element refuses the batch before anything is
            // written. `apply` will consult it again on the write pass; a hook
            // that answers differently to the same value twice is the caller's
            // own inconsistency, and the second answer is the one that stands.
            if !self.run_on_set_callbacks(node, name, self.get(node, name), Some(&slot)) {
                return Err(ParameterError::Rejected);
            }
        }
        for i in 0..count {
            let Some(name) = fill(i, &mut slot) else {
                continue;
            };
            let _ = self.apply_ref(node, name, &slot);
        }
        Ok(())
    }

    /// Get a parameter value by name
    pub(crate) fn get(&self, node: NodeKey, name: &str) -> Option<&ParameterValue> {
        self.find_index(node, name)
            .and_then(|idx| self.table.entries[idx].as_ref())
            .map(|entry| &entry.param.value)
    }

    /// Get a parameter by name
    pub(crate) fn get_parameter(&self, node: NodeKey, name: &str) -> Option<&Parameter> {
        self.find_index(node, name)
            .and_then(|idx| self.table.entries[idx].as_ref())
            .map(|entry| &entry.param)
    }

    /// Get a parameter descriptor by name
    pub(crate) fn get_descriptor(&self, node: NodeKey, name: &str) -> Option<&ParameterDescriptor> {
        self.find_index(node, name)
            .and_then(|idx| self.table.entries[idx].as_ref())
            .and_then(|entry| entry.descriptor.as_ref())
    }

    /// Set a parameter value
    ///
    /// Returns the result of the set operation.
    pub(crate) fn set(
        &mut self,
        node: NodeKey,
        name: &str,
        value: ParameterValue,
    ) -> Result<(), ParameterError> {
        self.set_ref(node, name, &value)
    }

    /// [`set`](Self::set), borrowing the value (issue 2104): it is copied
    /// once, into the stored value.
    pub(crate) fn set_ref(
        &mut self,
        node: NodeKey,
        name: &str,
        value: &ParameterValue,
    ) -> Result<(), ParameterError> {
        let idx = match self.find_index(node, name) {
            Some(idx) => idx,
            None => return Err(ParameterError::NotFound),
        };

        let entry = match self.table.entries[idx].as_mut() {
            Some(e) => e,
            None => return Err(ParameterError::NotFound),
        };

        // Read-only, type, range (min/max/step) — one predicate, shared with
        // the atomic pre-check.
        Self::check_set_existing(entry, value)?;

        entry.param.value.assign_from(value);
        self.note_event(node, ParameterEventKind::Changed, name);
        Ok(())
    }

    /// Unset an optional parameter (set to NotSet)
    ///
    /// This bypasses the type check to allow optional parameters to be unset.
    /// Returns an error if the parameter is read-only.
    pub(crate) fn unset(&mut self, node: NodeKey, name: &str) -> Result<(), ParameterError> {
        let idx = match self.find_index(node, name) {
            Some(idx) => idx,
            None => return Err(ParameterError::NotFound),
        };

        let entry = match self.table.entries[idx].as_mut() {
            Some(e) => e,
            None => return Err(ParameterError::NotFound),
        };

        // Check if read-only
        if let Some(ref desc) = entry.descriptor
            && desc.read_only
        {
            return Err(ParameterError::ReadOnly);
        }

        entry.param.value = ParameterValue::NotSet;
        self.note_event(node, ParameterEventKind::Changed, name);
        Ok(())
    }

    /// Set or declare a parameter — an EXPLICIT auto-declare for in-image
    /// callers.
    ///
    /// If the parameter exists, sets its value. Otherwise, declares it. This
    /// ignores `allow_undeclared` on purpose: the flag governs what a REMOTE
    /// set may do; a Rust caller naming this method has opted in by naming
    /// it. Nothing on the wire path — and, since phase-426 W2, nothing on the
    /// C or C++ facade path either — reaches here (issue 1151).
    pub(crate) fn set_or_declare(
        &mut self,
        node: NodeKey,
        name: &str,
        value: ParameterValue,
    ) -> Result<(), ParameterError> {
        if self.find_index(node, name).is_some() {
            self.set(node, name, value)
        } else if self.declare(node, name, value) {
            Ok(())
        } else {
            Err(ParameterError::StorageFull)
        }
    }

    /// Check if a parameter exists
    pub(crate) fn has(&self, node: NodeKey, name: &str) -> bool {
        self.find_index(node, name).is_some()
    }

    /// Remove a parameter
    ///
    /// Returns true if the parameter was removed.
    pub(crate) fn remove(&mut self, node: NodeKey, name: &str) -> bool {
        if let Some(idx) = self.find_index(node, name) {
            self.table.entries[idx] = None;
            self.count -= 1;
            self.note_event(node, ParameterEventKind::Deleted, name);
            true
        } else {
            false
        }
    }

    /// Get the type of a parameter
    pub(crate) fn get_type(&self, node: NodeKey, name: &str) -> Option<ParameterType> {
        self.get(node, name).map(|v| v.param_type())
    }

    /// Iterate over `node`'s parameters
    pub(crate) fn iter(&self, node: NodeKey) -> impl Iterator<Item = &Parameter> {
        self.entries_of(node).map(|e| &e.param)
    }

    /// List `node`'s parameter names
    pub(crate) fn list_names(&self, node: NodeKey) -> impl Iterator<Item = &str> {
        self.iter(node).map(|p| p.name.as_str())
    }

    /// List `node`'s parameter names with a given prefix
    pub(crate) fn list_with_prefix<'a>(
        &'a self,
        node: NodeKey,
        prefix: &'a str,
    ) -> impl Iterator<Item = &'a str> {
        self.list_names(node)
            .filter(move |name| name.starts_with(prefix))
    }

    /// Get a bool parameter value
    pub(crate) fn get_bool(&self, node: NodeKey, name: &str) -> Option<bool> {
        self.get(node, name).and_then(|v| v.as_bool())
    }

    /// Get an integer parameter value
    pub(crate) fn get_integer(&self, node: NodeKey, name: &str) -> Option<i64> {
        self.get(node, name).and_then(|v| v.as_integer())
    }

    /// Get a double parameter value
    pub(crate) fn get_double(&self, node: NodeKey, name: &str) -> Option<f64> {
        self.get(node, name).and_then(|v| v.as_double())
    }

    /// Get a string parameter value
    pub(crate) fn get_string(&self, node: NodeKey, name: &str) -> Option<&str> {
        self.get(node, name).and_then(|v| v.as_string())
    }

    /// Set a bool parameter value
    pub(crate) fn set_bool(
        &mut self,
        node: NodeKey,
        name: &str,
        value: bool,
    ) -> Result<(), ParameterError> {
        self.set(node, name, ParameterValue::Bool(value))
    }

    /// Set an integer parameter value
    pub(crate) fn set_integer(
        &mut self,
        node: NodeKey,
        name: &str,
        value: i64,
    ) -> Result<(), ParameterError> {
        self.set(node, name, ParameterValue::Integer(value))
    }

    /// Set a double parameter value
    pub(crate) fn set_double(
        &mut self,
        node: NodeKey,
        name: &str,
        value: f64,
    ) -> Result<(), ParameterError> {
        self.set(node, name, ParameterValue::Double(value))
    }

    /// Set a string parameter value
    pub(crate) fn set_string(
        &mut self,
        node: NodeKey,
        name: &str,
        value: &str,
    ) -> Result<(), ParameterError> {
        match ParameterValue::from_string(value) {
            Some(v) => self.set(node, name, v),
            None => Err(ParameterError::StorageFull), // String too long
        }
    }

    /// Declare a parameter with a descriptor and optional initial value.
    ///
    /// This is a more comprehensive declaration method used by the typed parameter API.
    ///
    /// # Arguments
    /// - `node`: The node the parameter belongs to.
    /// - `descriptor`: The metadata for the parameter.
    /// - `initial_value`: An optional initial value for the parameter. If `None`,
    ///   the parameter will be initialized to `ParameterValue::NotSet`.
    pub(crate) fn declare_parameter(
        &mut self,
        node: NodeKey,
        descriptor: ParameterDescriptor,
        initial_value: Option<&ParameterValue>,
    ) -> Result<(), ParameterError> {
        let name = descriptor.name.as_str();

        // Check if already exists
        if self.find_index(node, name).is_some() {
            return Err(ParameterError::TypeMismatch); // Indicates already declared
        }

        let param_value = initial_value.cloned().unwrap_or_default();

        // Issue 1150 — same declare-time check as `declare_with_descriptor`.
        if !Self::declaration_is_consistent(Some(&descriptor), &param_value) {
            return Err(ParameterError::InvalidRange);
        }

        // Find an empty slot
        let slot = match self.find_empty_slot() {
            Some(idx) => idx,
            None => return Err(ParameterError::StorageFull),
        };

        let param = Parameter::new(name, param_value).ok_or(ParameterError::StorageFull)?; // name too long

        self.note_descriptor(Some(&descriptor));
        self.table.entries[slot] = Some(ParameterEntry {
            node,
            param,
            descriptor: Some(descriptor),
        });
        self.count += 1;
        Ok(())
    }

    /// Get the current value of a parameter.
    ///
    /// Returns `Some(ParameterValue)` if the parameter exists, `None` otherwise.
    /// The `ParameterValue` is cloned to avoid lifetime issues with `&mut self`.
    pub(crate) fn get_parameter_value(&self, node: NodeKey, name: &str) -> Option<ParameterValue> {
        self.get(node, name).cloned()
    }

    /// Set the value of a parameter.
    ///
    /// Returns `Ok(())` on success, or an error if the parameter
    /// is read-only, type mismatches, or is out of range.
    pub(crate) fn set_parameter_value(
        &mut self,
        node: NodeKey,
        name: &str,
        value: ParameterValue,
    ) -> Result<(), ParameterError> {
        self.set(node, name, value)
    }
}

/// Builder for declaring parameters with a fluent API
pub struct LegacyParameterBuilder<'a, 's> {
    server: &'a mut ParameterServer<'s>,
    /// phase-426 W1 — the node this declaration belongs to, fixed when the
    /// builder is made so no chained call can lose it.
    node: NodeKey,
    name: String<MAX_PARAM_NAME_LEN>,
    value: ParameterValue,
    descriptor: Option<ParameterDescriptor>,
}

impl<'a, 's> LegacyParameterBuilder<'a, 's> {
    /// Create a new parameter builder
    pub fn new(
        server: &'a mut ParameterServer<'s>,
        node: NodeKey,
        name: &str,
        value: ParameterValue,
    ) -> Option<Self> {
        let mut n = String::new();
        n.push_str(name).ok()?;
        Some(Self {
            server,
            node,
            name: n,
            value,
            descriptor: None,
        })
    }

    /// Set the parameter description
    pub fn description(mut self, desc: &str) -> Self {
        let param_type = self.value.param_type();
        if self.descriptor.is_none() {
            self.descriptor = ParameterDescriptor::new(self.name.as_str(), param_type);
        }
        // phase-446 F2 -- truncated at a character boundary and recorded,
        // where `let _ = push_str` stored nothing for a too-long description.
        if let Some(ref mut d) = self.descriptor {
            d.set_description(desc);
        }
        self
    }

    /// Set the parameter as read-only
    pub fn read_only(mut self) -> Self {
        let param_type = self.value.param_type();
        if self.descriptor.is_none() {
            self.descriptor = ParameterDescriptor::new(self.name.as_str(), param_type);
        }
        if let Some(ref mut d) = self.descriptor {
            d.read_only = true;
        }
        self
    }

    /// Set integer range constraints
    pub fn integer_range(mut self, min: i64, max: i64, step: i64) -> Self {
        let param_type = self.value.param_type();
        if self.descriptor.is_none() {
            self.descriptor = ParameterDescriptor::new(self.name.as_str(), param_type);
        }
        if let Some(ref mut d) = self.descriptor {
            d.range = crate::types::ParameterRange::Integer(crate::types::IntegerRange::new(
                min, max, step,
            ));
        }
        self
    }

    /// Set float range constraints
    pub fn float_range(mut self, min: f64, max: f64, step: f64) -> Self {
        let param_type = self.value.param_type();
        if self.descriptor.is_none() {
            self.descriptor = ParameterDescriptor::new(self.name.as_str(), param_type);
        }
        if let Some(ref mut d) = self.descriptor {
            d.range = crate::types::ParameterRange::FloatingPoint(
                crate::types::FloatingPointRange::new(min, max, step),
            );
        }
        self
    }

    /// Declare the parameter
    ///
    /// phase-428 W6 remainder: the terminal of a builder chain, so a dropped
    /// answer is a parameter that was configured in full and then not declared.
    #[must_use]
    pub fn declare(self) -> bool {
        self.server.declare_with_descriptor(
            self.node,
            self.name.as_str(),
            self.value,
            self.descriptor,
        )
    }
}

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

    /// phase-426 W1 — the node these single-node tests are about.
    const NODE: NodeKey = NodeKey::PRIMARY;
    use crate::types::{ParameterDescriptor, ParameterType};

    // ── issue 2070: a borrowed declaration stores exactly what it was lent ──

    fn every_variant() -> std::vec::Vec<ParameterValue> {
        let mut strings = ParameterValue::StringArray(Default::default());
        if let ParameterValue::StringArray(v) = &mut strings {
            for e in ["a", "", "frame_id"] {
                let mut s = heapless::String::new();
                s.push_str(e).unwrap();
                v.push(s).unwrap();
            }
        }
        std::vec![
            ParameterValue::Bool(true),
            ParameterValue::Integer(-7),
            ParameterValue::Double(2.5),
            ParameterValue::from_string("base_link").unwrap(),
            ParameterValue::from_byte_array(&[1, 2, 3]).unwrap(),
            ParameterValue::from_bool_array(&[true, false]).unwrap(),
            ParameterValue::from_integer_array(&[4, 5, 6]).unwrap(),
            ParameterValue::from_double_array(&[0.5, -1.0]).unwrap(),
            strings,
        ]
    }

    /// `declare_ref` copies the value into its slot once, and what the slot
    /// then holds is the value it was lent, for every variant.
    #[test]
    fn a_borrowed_declaration_stores_the_value_it_was_lent() {
        let values = every_variant();
        let mut storage = ParameterStorage::<16>::new();
        let mut server = ParameterServer::new_in(storage.as_table());
        for (i, v) in values.iter().enumerate() {
            let name = std::format!("p{i}");
            assert!(server.declare_ref(NODE, &name, v));
            assert_eq!(
                std::format!("{:?}", server.get(NODE, &name).unwrap()),
                std::format!("{v:?}")
            );
        }
        assert_eq!(server.len(NODE), values.len());
        // A second declaration of a name is refused and leaves the first.
        assert!(!server.declare_ref(NODE, "p0", &ParameterValue::Integer(1)));
        assert!(matches!(
            server.get(NODE, "p0"),
            Some(ParameterValue::Bool(true))
        ));
    }

    /// `assign_from` replaces whatever the destination held, including a
    /// different variant and a longer value of the same one.
    #[test]
    fn assign_from_replaces_any_previous_value() {
        let values = every_variant();
        for dst_init in &values {
            for src in &values {
                let mut dst = dst_init.clone();
                dst.assign_from(src);
                assert_eq!(std::format!("{dst:?}"), std::format!("{src:?}"));
            }
        }
        let mut long = ParameterValue::from_integer_array(&[1, 2, 3, 4]).unwrap();
        long.assign_from(&ParameterValue::from_integer_array(&[9]).unwrap());
        assert_eq!(
            std::format!("{long:?}"),
            std::format!("{:?}", ParameterValue::from_integer_array(&[9]).unwrap())
        );
    }

    // ── phase-446 F2: a truncated description is reported once ──

    /// Every declare path records a truncated description on the stored
    /// descriptor -- the plain descriptor, the typed builder and the legacy
    /// builder -- and `take_truncated_descriptions` hands each one over
    /// exactly once, with its node and name. A description that fits is
    /// never reported.
    #[test]
    fn each_truncated_description_is_handed_over_once() {
        let long = "y".repeat(crate::MAX_PARAM_DESCRIPTION_LEN + 1);
        let other = NodeKey::new(1);
        let mut storage = ParameterStorage::<8>::new();
        let mut server = ParameterServer::new_in(storage.as_table());
        assert_eq!(server.take_truncated_descriptions(|_, _| ()), 0);

        let d = ParameterDescriptor::new("plain", ParameterType::Integer)
            .unwrap()
            .with_description(&long);
        assert!(server.declare_with_descriptor(NODE, "plain", ParameterValue::Integer(1), Some(d)));
        let fits = ParameterDescriptor::new("short", ParameterType::Integer)
            .unwrap()
            .with_description("");
        assert!(server.declare_with_descriptor(
            NODE,
            "short",
            ParameterValue::Integer(1),
            Some(fits)
        ));
        crate::ParameterBuilder::<i64>::new(&mut server, other, "typed")
            .default(3)
            .description(&long)
            .optional()
            .expect("a long description no longer refuses the declaration");
        assert!(
            LegacyParameterBuilder::new(&mut server, NODE, "legacy", ParameterValue::Bool(true))
                .unwrap()
                .description(&long)
                .declare()
        );

        let mut seen: heapless::Vec<(usize, heapless::String<16>), 8> = heapless::Vec::new();
        let n = server.take_truncated_descriptions(|node, name| {
            let mut s = heapless::String::new();
            s.push_str(name).unwrap();
            seen.push((node.index(), s)).unwrap();
        });
        assert_eq!(n, 3, "{seen:?}");
        for (node, name) in [(0, "plain"), (1, "typed"), (0, "legacy")] {
            assert!(
                seen.iter().any(|(k, s)| *k == node && s.as_str() == name),
                "{name} on node {node} not reported: {seen:?}"
            );
        }
        assert!(!seen.iter().any(|(_, s)| s.as_str() == "short"));
        assert_eq!(
            server.take_truncated_descriptions(|_, _| ()),
            0,
            "once per parameter"
        );
        let kept = server.get_descriptor(other, "typed").unwrap();
        assert_eq!(kept.description.len(), crate::MAX_PARAM_DESCRIPTION_LEN);
    }

    // ── phase-426 W1: the store is keyed by NODE ──

    /// Two nodes composed onto one executor (RFC-0047) declaring the same
    /// name. Before W1 the table was flat, so the second `declare` returned
    /// `false` (already exists) and both nodes read ONE value: `/talker`'s
    /// `rate` and `/listener`'s `rate` were the same parameter.
    #[test]
    fn two_nodes_hold_independent_values_for_one_name() {
        let talker = NodeKey::new(0);
        let listener = NodeKey::new(1);
        let mut storage: ParameterStorage = ParameterStorage::new();
        let mut server = ParameterServer::new_in(storage.as_table());

        assert!(server.declare(talker, "rate", ParameterValue::Integer(10)));
        // The pre-W1 collision: a flat table refuses this as a duplicate.
        assert!(
            server.declare(listener, "rate", ParameterValue::Integer(20)),
            "a sibling node's identical name is a DIFFERENT parameter"
        );

        assert_eq!(server.get_integer(talker, "rate"), Some(10));
        assert_eq!(server.get_integer(listener, "rate"), Some(20));

        // A set on one is invisible to the other.
        assert_eq!(
            server.set(talker, "rate", ParameterValue::Integer(11)),
            Ok(())
        );
        assert_eq!(server.get_integer(talker, "rate"), Some(11));
        assert_eq!(server.get_integer(listener, "rate"), Some(20));

        // And each node enumerates only its own.
        assert_eq!(server.len(talker), 1);
        assert_eq!(server.len(listener), 1);
        assert_eq!(server.total_len(), 2, "one shared arena, two entries");
    }

    /// A name one node declared is not a name its sibling has.
    #[test]
    fn a_sibling_cannot_read_or_write_a_name_it_never_declared() {
        let a = NodeKey::new(0);
        let b = NodeKey::new(3);
        let mut storage: ParameterStorage = ParameterStorage::new();
        let mut server = ParameterServer::new_in(storage.as_table());
        assert!(server.declare(a, "gain", ParameterValue::Double(1.0)));

        assert!(!server.has(b, "gain"));
        assert!(server.get(b, "gain").is_none());
        assert_eq!(
            server.set(b, "gain", ParameterValue::Double(2.0)),
            Err(ParameterError::NotFound)
        );
        assert_eq!(
            server.check_apply(b, "gain", &ParameterValue::Double(2.0)),
            Err(ParameterError::Undeclared)
        );
        // Untouched.
        assert_eq!(server.get_double(a, "gain"), Some(1.0));
    }

    /// `allow_undeclared` is a NODE option upstream, so one node opting in
    /// must not decide what a remote set may do to its sibling (issue 1151 +
    /// phase-426 W1).
    #[test]
    fn allow_undeclared_does_not_leak_to_a_sibling_node() {
        let permissive = NodeKey::new(1);
        let strict = NodeKey::new(2);
        let mut storage: ParameterStorage = ParameterStorage::new();
        let mut server = ParameterServer::new_in(storage.as_table());
        server.set_allow_undeclared(permissive, true);

        assert!(server.allows_undeclared(permissive));
        assert!(!server.allows_undeclared(strict));

        assert_eq!(
            server.apply(permissive, "fresh", ParameterValue::Integer(7)),
            Ok(())
        );
        assert_eq!(
            server.apply(strict, "fresh", ParameterValue::Integer(7)),
            Err(ParameterError::Undeclared),
            "the sibling still refuses a name IT never declared"
        );
        assert_eq!(server.get_integer(permissive, "fresh"), Some(7));
        assert!(!server.has(strict, "fresh"));
    }

    /// The flag reaches every key in a `u8`'s range, not just the low ones —
    /// the bitmask is `[u64; 4]` precisely so there is no ceiling to find at
    /// runtime.
    #[test]
    fn the_allow_undeclared_flag_covers_the_whole_key_space() {
        let mut storage: ParameterStorage = ParameterStorage::new();
        let mut server = ParameterServer::new_in(storage.as_table());
        for raw in [0u8, 63, 64, 127, 128, 191, 192, 255] {
            let node = NodeKey::new(raw);
            assert!(!server.allows_undeclared(node), "raw {raw}");
            server.set_allow_undeclared(node, true);
            assert!(server.allows_undeclared(node), "raw {raw}");
            server.set_allow_undeclared(node, false);
            assert!(!server.allows_undeclared(node), "raw {raw}");
        }
    }

    /// The arena is shared, so capacity is an IMAGE bound, and the node that
    /// runs into it is told `StorageFull` rather than being silently given a
    /// sibling's slot.
    #[test]
    fn the_slot_arena_is_shared_across_nodes() {
        let a = NodeKey::new(0);
        let b = NodeKey::new(1);
        let mut storage = ParameterStorage::<2>::new();
        let mut server = ParameterServer::new_in(storage.as_table());
        assert!(server.declare(a, "one", ParameterValue::Integer(1)));
        assert!(server.declare(b, "one", ParameterValue::Integer(2)));
        assert!(server.is_full());
        assert!(!server.declare(a, "two", ParameterValue::Integer(3)));
        assert_eq!(server.total_len(), 2);
        assert_eq!(server.len(a), 1);
        assert_eq!(server.len(b), 1);
    }

    /// `remove` takes the right node's slot.
    #[test]
    fn remove_is_keyed_by_node() {
        let a = NodeKey::new(0);
        let b = NodeKey::new(1);
        let mut storage: ParameterStorage = ParameterStorage::new();
        let mut server = ParameterServer::new_in(storage.as_table());
        assert!(server.declare(a, "x", ParameterValue::Integer(1)));
        assert!(server.declare(b, "x", ParameterValue::Integer(2)));

        assert!(server.remove(a, "x"));
        assert!(!server.has(a, "x"));
        assert_eq!(server.get_integer(b, "x"), Some(2));
        assert_eq!(server.total_len(), 1);
    }

    // ── issue 1150: step reaches `set`, and declare validates its range ──

    #[test]
    fn set_rejects_a_value_off_the_integer_step() {
        let mut storage: ParameterStorage = ParameterStorage::new();
        let mut server = ParameterServer::new_in(storage.as_table());
        let desc = ParameterDescriptor::new("rate", ParameterType::Integer)
            .unwrap()
            .with_integer_range(0, 100, 10);
        assert!(server.declare_with_descriptor(
            NODE,
            "rate",
            ParameterValue::Integer(20),
            Some(desc)
        ));

        assert_eq!(
            server.set(NODE, "rate", ParameterValue::Integer(30)),
            Ok(())
        );
        // In range by min/max, off the lattice: the old `contains` said yes.
        assert_eq!(
            server.set(NODE, "rate", ParameterValue::Integer(35)),
            Err(ParameterError::OutOfRange)
        );
        assert_eq!(server.get_integer(NODE, "rate"), Some(30));
    }

    #[test]
    fn set_rejects_a_value_off_the_float_step() {
        let mut storage: ParameterStorage = ParameterStorage::new();
        let mut server = ParameterServer::new_in(storage.as_table());
        let desc = ParameterDescriptor::new("gain", ParameterType::Double)
            .unwrap()
            .with_float_range(0.0, 1.0, 0.25);
        assert!(server.declare_with_descriptor(
            NODE,
            "gain",
            ParameterValue::Double(0.5),
            Some(desc)
        ));

        assert_eq!(server.set_double(NODE, "gain", 0.75), Ok(()));
        assert_eq!(
            server.set_double(NODE, "gain", 0.6),
            Err(ParameterError::OutOfRange)
        );
        assert_eq!(server.get_double(NODE, "gain"), Some(0.75));
    }

    #[test]
    fn declare_refuses_an_ill_formed_range() {
        let mut storage: ParameterStorage = ParameterStorage::new();
        let mut server = ParameterServer::new_in(storage.as_table());
        let desc = ParameterDescriptor::new("bad", ParameterType::Integer)
            .unwrap()
            .with_integer_range(0, 100, -1);
        assert!(!server.declare_with_descriptor(
            NODE,
            "bad",
            ParameterValue::Integer(0),
            Some(desc.clone())
        ));
        assert!(!server.has(NODE, "bad"));
        assert_eq!(
            server.declare_parameter(NODE, desc, Some(&ParameterValue::Integer(0))),
            Err(ParameterError::InvalidRange)
        );
        assert!(!server.has(NODE, "bad"));
    }

    #[test]
    fn declare_refuses_a_default_its_own_range_rejects() {
        let mut storage: ParameterStorage = ParameterStorage::new();
        let mut server = ParameterServer::new_in(storage.as_table());
        let desc = ParameterDescriptor::new("odd", ParameterType::Integer)
            .unwrap()
            .with_integer_range(0, 100, 2);
        assert!(!server.declare_with_descriptor(
            NODE,
            "odd",
            ParameterValue::Integer(3),
            Some(desc.clone())
        ));
        assert!(server.declare_with_descriptor(
            NODE,
            "odd",
            ParameterValue::Integer(4),
            Some(desc)
        ));
    }

    // ── issue 1151: a wire-facing set on an undeclared name is refused ──

    #[test]
    fn apply_rejects_an_undeclared_name_by_default() {
        let mut storage: ParameterStorage = ParameterStorage::new();
        let mut server = ParameterServer::new_in(storage.as_table());
        assert!(server.declare(NODE, "max_speed", ParameterValue::Double(1.0)));
        assert!(!server.allows_undeclared(NODE));

        // The typo from the issue: reported success and created a parameter
        // nobody reads.
        assert_eq!(
            server.check_apply(NODE, "max_speeed", &ParameterValue::Double(5.0)),
            Err(ParameterError::Undeclared)
        );
        assert_eq!(
            server.apply(NODE, "max_speeed", ParameterValue::Double(5.0)),
            Err(ParameterError::Undeclared)
        );
        assert!(!server.has(NODE, "max_speeed"));
        assert_eq!(server.len(NODE), 1);
        assert_eq!(server.get_double(NODE, "max_speed"), Some(1.0));

        // A declared name still sets.
        assert_eq!(
            server.apply(NODE, "max_speed", ParameterValue::Double(5.0)),
            Ok(())
        );
        assert_eq!(server.get_double(NODE, "max_speed"), Some(5.0));
    }

    #[test]
    fn apply_declares_an_unknown_name_under_allow_undeclared() {
        let mut storage: ParameterStorage = ParameterStorage::new();
        let mut server = ParameterServer::new_in(storage.as_table());
        server.set_allow_undeclared(NODE, true);
        assert!(server.allows_undeclared(NODE));

        assert_eq!(
            server.check_apply(NODE, "fresh", &ParameterValue::Integer(7)),
            Ok(())
        );
        assert_eq!(
            server.apply(NODE, "fresh", ParameterValue::Integer(7)),
            Ok(())
        );
        assert_eq!(server.get_integer(NODE, "fresh"), Some(7));
    }

    #[test]
    fn apply_under_allow_undeclared_still_reports_a_full_store() {
        let mut storage: ParameterStorage<1> = ParameterStorage::new();
        let mut server = ParameterServer::new_in(storage.as_table());
        server.set_allow_undeclared(NODE, true);
        assert!(server.declare(NODE, "only", ParameterValue::Bool(true)));
        assert_eq!(
            server.check_apply(NODE, "second", &ParameterValue::Bool(true)),
            Err(ParameterError::StorageFull)
        );
        assert_eq!(
            server.apply(NODE, "second", ParameterValue::Bool(true)),
            Err(ParameterError::StorageFull)
        );
    }

    #[test]
    fn apply_enforces_the_existing_entry_rules_through_the_same_predicate() {
        let mut storage: ParameterStorage = ParameterStorage::new();
        let mut server = ParameterServer::new_in(storage.as_table());
        let desc = ParameterDescriptor::new("rate", ParameterType::Integer)
            .unwrap()
            .with_integer_range(0, 100, 10);
        assert!(server.declare_with_descriptor(
            NODE,
            "rate",
            ParameterValue::Integer(20),
            Some(desc)
        ));
        assert_eq!(
            server.check_apply(NODE, "rate", &ParameterValue::Integer(35)),
            Err(ParameterError::OutOfRange)
        );
        assert_eq!(
            server.apply(NODE, "rate", ParameterValue::Integer(35)),
            Err(ParameterError::OutOfRange)
        );
        assert_eq!(
            server.apply(NODE, "rate", ParameterValue::Double(30.0)),
            Err(ParameterError::TypeMismatch)
        );
        assert_eq!(server.get_integer(NODE, "rate"), Some(20));
    }

    #[test]
    fn test_new_server() {
        let mut storage: ParameterStorage = ParameterStorage::new();
        let server = ParameterServer::new_in(storage.as_table());
        assert_eq!(server.len(NODE), 0);
        assert!(server.is_empty(NODE));
    }

    #[test]
    #[allow(clippy::approx_constant)]
    fn test_declare_and_get() {
        let mut storage: ParameterStorage = ParameterStorage::new();
        let mut server = ParameterServer::new_in(storage.as_table());

        assert!(server.declare(NODE, "my_bool", ParameterValue::Bool(true)));
        assert!(server.declare(NODE, "my_int", ParameterValue::Integer(42)));
        assert!(server.declare(NODE, "my_double", ParameterValue::Double(3.14)));

        assert_eq!(server.len(NODE), 3);
        assert!(!server.is_empty(NODE));

        assert_eq!(server.get_bool(NODE, "my_bool"), Some(true));
        assert_eq!(server.get_integer(NODE, "my_int"), Some(42));
        assert_eq!(server.get_double(NODE, "my_double"), Some(3.14));
    }

    #[test]
    fn test_unset_clears_the_value_and_reports_success() {
        // phase-359 W10 / issue 0080 — what survives of
        // ~~`test_dirty_tracks_value_changes_not_declarations`~~. That test existed
        // for the persistence flush's dirty flag, which is deleted with the
        // seam; but it carried the crate's ONLY coverage of `unset`, so the
        // assertion moves here rather than going out with the flag.
        let mut storage: ParameterStorage = ParameterStorage::new();
        let mut server = ParameterServer::new_in(storage.as_table());
        assert!(server.declare(NODE, "count", ParameterValue::Integer(0)));
        assert_eq!(server.set_integer(NODE, "count", 10), Ok(()));
        assert_eq!(server.get_integer(NODE, "count"), Some(10));

        assert_eq!(server.unset(NODE, "count"), Ok(()));
        assert_eq!(server.get_integer(NODE, "count"), None);
    }

    #[test]
    fn test_set_parameter() {
        let mut storage: ParameterStorage = ParameterStorage::new();
        let mut server = ParameterServer::new_in(storage.as_table());
        assert!(server.declare(NODE, "count", ParameterValue::Integer(0)));

        assert_eq!(server.set_integer(NODE, "count", 10), Ok(()));
        assert_eq!(server.get_integer(NODE, "count"), Some(10));
    }

    #[test]
    fn test_set_nonexistent() {
        let mut storage: ParameterStorage = ParameterStorage::new();
        let mut server = ParameterServer::new_in(storage.as_table());
        assert_eq!(
            server.set(NODE, "nonexistent", ParameterValue::Integer(1)),
            Err(ParameterError::NotFound)
        );
    }

    #[test]
    fn test_read_only_parameter() {
        let mut storage: ParameterStorage = ParameterStorage::new();
        let mut server = ParameterServer::new_in(storage.as_table());

        let desc = ParameterDescriptor::new("version", ParameterType::String)
            .unwrap()
            .with_read_only(true);

        assert!(server.declare_with_descriptor(
            NODE,
            "version",
            ParameterValue::from_string("1.0.0").unwrap(),
            Some(desc),
        ));

        assert_eq!(
            server.set_string(NODE, "version", "2.0.0"),
            Err(ParameterError::ReadOnly)
        );
        assert_eq!(server.get_string(NODE, "version"), Some("1.0.0"));
    }

    #[test]
    fn test_range_constraints() {
        let mut storage: ParameterStorage = ParameterStorage::new();
        let mut server = ParameterServer::new_in(storage.as_table());

        let desc = ParameterDescriptor::new("speed", ParameterType::Double)
            .unwrap()
            .with_float_range(0.0, 10.0, 0.0);

        assert!(server.declare_with_descriptor(
            NODE,
            "speed",
            ParameterValue::Double(5.0),
            Some(desc)
        ));

        // Valid value
        assert_eq!(server.set_double(NODE, "speed", 8.0), Ok(()));

        // Out of range
        assert_eq!(
            server.set_double(NODE, "speed", 15.0),
            Err(ParameterError::OutOfRange)
        );
        assert_eq!(server.get_double(NODE, "speed"), Some(8.0)); // Unchanged
    }

    #[test]
    fn test_remove_parameter() {
        let mut storage: ParameterStorage = ParameterStorage::new();
        let mut server = ParameterServer::new_in(storage.as_table());
        assert!(server.declare(NODE, "temp", ParameterValue::Double(25.0)));

        assert!(server.has(NODE, "temp"));
        assert!(server.remove(NODE, "temp"));
        assert!(!server.has(NODE, "temp"));
        assert!(!server.remove(NODE, "temp")); // Already removed
    }

    #[test]
    fn test_list_parameters() {
        let mut storage: ParameterStorage = ParameterStorage::new();
        let mut server = ParameterServer::new_in(storage.as_table());
        assert!(server.declare(NODE, "robot.speed", ParameterValue::Double(1.0)));
        assert!(server.declare(
            NODE,
            "robot.name",
            ParameterValue::from_string("bot1").unwrap(),
        ));
        assert!(server.declare(NODE, "sensor.range", ParameterValue::Double(10.0)));

        let robot_params: heapless::Vec<&str, 8> =
            server.list_with_prefix(NODE, "robot.").collect();
        assert_eq!(robot_params.len(), 2);
    }

    #[test]
    fn test_set_or_declare() {
        let mut storage: ParameterStorage = ParameterStorage::new();
        let mut server = ParameterServer::new_in(storage.as_table());

        // First call declares
        assert_eq!(
            server.set_or_declare(NODE, "new_param", ParameterValue::Integer(1)),
            Ok(())
        );
        assert_eq!(server.get_integer(NODE, "new_param"), Some(1));

        // Second call sets
        assert_eq!(
            server.set_or_declare(NODE, "new_param", ParameterValue::Integer(2)),
            Ok(())
        );
        assert_eq!(server.get_integer(NODE, "new_param"), Some(2));
    }

    #[test]
    fn test_duplicate_declare() {
        let mut storage: ParameterStorage = ParameterStorage::new();
        let mut server = ParameterServer::new_in(storage.as_table());
        assert!(server.declare(NODE, "param", ParameterValue::Integer(1)));
        assert!(!server.declare(NODE, "param", ParameterValue::Integer(2))); // Already exists
        assert_eq!(server.get_integer(NODE, "param"), Some(1)); // Unchanged
    }

    // ── phase-417 W4.a: descriptors, undeclare, and the on-set hook ──

    /// The four rclc descriptor verbs land on the ONE store and are visible
    /// through the ordinary descriptor read — a C or C++ wrapper forwarding to
    /// these adds no table of its own.
    #[test]
    // The comparison and the `min` below read as "always true" to clippy on a
    // build that leaves `NROS_MAX_PARAM_CONSTRAINTS_LEN` at its default of 0 --
    // and that is the point: the constant is a KNOB, the test is written
    // against it so it holds at 0, at 64 and at 256, and a version specialised
    // to whichever value this build resolved would read as coverage on the
    // other two.
    #[allow(clippy::absurd_extreme_comparisons, clippy::unnecessary_min_or_max)]
    fn descriptor_verbs_attach_to_a_declared_parameter() {
        let mut storage: ParameterStorage = ParameterStorage::new();
        let mut server = ParameterServer::new_in(storage.as_table());
        assert!(server.declare(NODE, "rate", ParameterValue::Double(2.0)));

        // A parameter declared with no descriptor still gets one on demand:
        // rclc's order is declare-then-describe.
        assert!(server.get_descriptor(NODE, "rate").is_none());
        assert!(server.set_parameter_description(NODE, "rate", "publish rate", "hz, positive"));
        assert!(server.set_parameter_float_range(NODE, "rate", 0.0, 10.0, 0.0));

        let desc = server.get_descriptor(NODE, "rate").expect("descriptor");
        assert_eq!(desc.description.as_str(), "publish rate");
        // Written against the CAPACITY, not a literal: `additional_constraints`
        // has its own knob (`NROS_MAX_PARAM_CONSTRAINTS_LEN`) whose default is
        // 0, so the default build stores a zero-length prefix. A test that
        // asserted the whole string would pass only on a build that states the
        // knob, and would read as coverage everywhere else.
        let constraints_fit = "hz, positive".len() <= crate::MAX_PARAM_CONSTRAINTS_LEN;
        assert_eq!(
            desc.additional_constraints.as_str(),
            &"hz, positive"[..crate::MAX_PARAM_CONSTRAINTS_LEN.min("hz, positive".len())]
        );
        assert_eq!(desc.param_type, ParameterType::Double);

        // And the range is ENFORCED by the same predicate every set uses.
        assert_eq!(
            server.apply(NODE, "rate", ParameterValue::Double(50.0)),
            Err(ParameterError::OutOfRange)
        );
        assert_eq!(
            server.apply(NODE, "rate", ParameterValue::Double(5.0)),
            Ok(())
        );

        // read-only is the same shape.
        assert!(server.set_parameter_read_only(NODE, "rate", true));
        assert_eq!(
            server.apply(NODE, "rate", ParameterValue::Double(6.0)),
            Err(ParameterError::ReadOnly)
        );

        // Text that did not fit is REPORTED, never dropped in silence — which
        // is what makes a capacity of 0 an honest default rather than a
        // trap. At a capacity that fits, nothing is reported.
        let mut reported = 0usize;
        server.take_truncated_descriptions(|_, _| reported += 1);
        assert_eq!(
            reported,
            usize::from(!constraints_fit),
            "a constraints string that did not fit the capacity went unreported"
        );

        // An undeclared name is refused rather than silently creating a
        // descriptor for a parameter that is not there.
        assert!(!server.set_parameter_description(NODE, "absent", "d", "c"));
        assert!(!server.set_parameter_read_only(NODE, "absent", true));
        assert!(!server.set_parameter_integer_range(NODE, "absent", 0, 1, 1));
    }

    /// A range the stored value already violates is REFUSED, and so is an
    /// ill-formed one. Attaching either would advertise a bound over a value
    /// no `set` could have produced.
    #[test]
    fn a_range_must_fit_the_value_it_is_attached_to() {
        let mut storage: ParameterStorage = ParameterStorage::new();
        let mut server = ParameterServer::new_in(storage.as_table());
        assert!(server.declare(NODE, "depth", ParameterValue::Integer(100)));

        assert!(!server.set_parameter_integer_range(NODE, "depth", 0, 10, 1));
        assert!(server.get_descriptor(NODE, "depth").is_none());

        // min > max
        assert!(!server.set_parameter_integer_range(NODE, "depth", 10, 0, 1));
        // negative step
        assert!(!server.set_parameter_integer_range(NODE, "depth", 0, 1000, -1));

        assert!(server.set_parameter_integer_range(NODE, "depth", 0, 1000, 1));
    }

    /// `remove` is the undeclare every language forwards to: the slot frees,
    /// the count drops, and a later declare of the same name succeeds.
    #[test]
    fn undeclare_frees_the_slot_for_that_node_only() {
        let mut storage: ParameterStorage = ParameterStorage::new();
        let mut server = ParameterServer::new_in(storage.as_table());
        let other = NodeKey::new(1);
        assert!(server.declare(NODE, "rate", ParameterValue::Integer(1)));
        assert!(server.declare(other, "rate", ParameterValue::Integer(2)));
        assert_eq!(server.total_len(), 2);

        assert!(server.remove(NODE, "rate"));
        assert!(!server.has(NODE, "rate"));
        assert!(server.has(other, "rate"));
        assert_eq!(server.total_len(), 1);
        assert!(!server.remove(NODE, "rate"));

        assert!(server.declare(NODE, "rate", ParameterValue::Integer(3)));
        assert_eq!(server.get_integer(NODE, "rate"), Some(3));
    }

    fn refuse_everything(
        _n: NodeKey,
        _name: &str,
        _old: Option<&ParameterValue>,
        _new: Option<&ParameterValue>,
        _ctx: OnSetContext,
    ) -> bool {
        false
    }

    fn refuse_if_ctx_matches(
        _n: NodeKey,
        _name: &str,
        _old: Option<&ParameterValue>,
        new: Option<&ParameterValue>,
        ctx: OnSetContext,
    ) -> bool {
        new.and_then(ParameterValue::as_integer) != Some(ctx.a as i64)
    }

    /// The hook runs on `apply` — the seam every facade and the six services
    /// funnel through — refuses with `Rejected`, leaves the value alone, and
    /// only for the node it was registered on.
    #[test]
    fn an_on_set_callback_vetoes_the_write_it_refuses() {
        let mut storage: ParameterStorage = ParameterStorage::new();
        let mut server = ParameterServer::new_in(storage.as_table());
        let other = NodeKey::new(1);
        assert!(server.declare(NODE, "rate", ParameterValue::Integer(1)));
        assert!(server.declare(other, "rate", ParameterValue::Integer(1)));

        let h = server
            .add_on_set_parameters_callback(NODE, refuse_everything, OnSetContext::default())
            .expect("a free slot");

        assert_eq!(
            server.apply(NODE, "rate", ParameterValue::Integer(9)),
            Err(ParameterError::Rejected)
        );
        assert_eq!(server.get_integer(NODE, "rate"), Some(1));
        // A sibling node's writes are untouched: the registration names a node.
        assert_eq!(
            server.apply(other, "rate", ParameterValue::Integer(9)),
            Ok(())
        );

        assert!(server.remove_on_set_parameters_callback(h));
        assert!(!server.remove_on_set_parameters_callback(h));
        assert_eq!(
            server.apply(NODE, "rate", ParameterValue::Integer(9)),
            Ok(())
        );
    }

    /// What the recording hook saw, reached through `OnSetContext::a`.
    #[derive(Default)]
    struct Seen {
        calls: u32,
        old: Option<i64>,
        new: Option<i64>,
        old_present: bool,
        new_present: bool,
    }

    fn record(
        _n: NodeKey,
        _name: &str,
        old: Option<&ParameterValue>,
        new: Option<&ParameterValue>,
        ctx: OnSetContext,
    ) -> bool {
        // SAFETY: every test passes a `&mut Seen` that outlives the server.
        let seen = unsafe { &mut *(ctx.a as *mut Seen) };
        seen.calls += 1;
        seen.old_present = old.is_some();
        seen.new_present = new.is_some();
        seen.old = old.and_then(ParameterValue::as_integer);
        seen.new = new.and_then(ParameterValue::as_integer);
        true
    }

    /// phase-491 W3 — the `/parameter_events` log: off by default, one entry
    /// per name (merged), and a burst past its bound is COUNTED, never blocked.
    #[test]
    fn the_event_log_merges_by_name_and_counts_overflow() {
        let mut storage: ParameterStorage = ParameterStorage::new();
        let mut server = ParameterServer::new_in(storage.as_table());
        assert!(server.declare(NODE, "quiet", ParameterValue::Integer(0)));
        assert!(server.pop_parameter_event().is_none(), "off by default");

        server.set_parameter_events(NODE, true);
        // Issue 2023 — the start-up event names what the node already holds.
        let e = server
            .pop_initial_parameter_event()
            .expect("the start-up event");
        assert_eq!(
            (e.name.as_str(), e.kind),
            ("quiet", ParameterEventKind::New)
        );
        assert!(server.pop_initial_parameter_event().is_none());
        assert!(server.declare(NODE, "a", ParameterValue::Integer(1)));
        assert_eq!(server.set(NODE, "a", ParameterValue::Integer(2)), Ok(()));
        assert!(server.remove(NODE, "a"));
        let e = server.pop_parameter_event().expect("one merged entry");
        assert_eq!(
            (e.name.as_str(), e.kind),
            ("a", ParameterEventKind::Deleted)
        );
        assert!(server.pop_parameter_event().is_none());

        for i in 0..PARAMETER_EVENT_LOG_LEN + 2 {
            let mut name = heapless::String::<8>::new();
            core::fmt::Write::write_fmt(&mut name, format_args!("p{i}")).unwrap();
            assert!(server.declare(NODE, &name, ParameterValue::Integer(0)));
        }
        assert_eq!(server.take_parameter_events_dropped(), 2);
        assert_eq!(
            server.take_parameter_events_dropped(),
            0,
            "the count resets"
        );
    }

    /// Issue 2023 — the start-up event: one `New` per parameter the node held
    /// when events went on, at its CURRENT value; an `Announce` for a node
    /// holding none; changes made while it is pending fold into it rather
    /// than doubling up in the log; and re-enabling is not a second start-up.
    #[test]
    fn the_start_up_event_reports_what_the_node_holds() {
        let mut storage: ParameterStorage = ParameterStorage::new();
        let mut server = ParameterServer::new_in(storage.as_table());
        let other = NodeKey::new(1);
        assert!(server.declare(NODE, "a", ParameterValue::Integer(1)));
        assert!(server.declare(NODE, "b", ParameterValue::Integer(2)));
        assert!(server.declare(other, "x", ParameterValue::Integer(9)));

        server.set_parameter_events(NODE, true);
        server.set_parameter_events(NodeKey::new(2), true);
        // Pending: a declaration and a removal fold into the drain.
        assert!(server.declare(NODE, "c", ParameterValue::Integer(3)));
        assert!(server.remove(NODE, "b"));
        assert!(server.pop_parameter_event().is_none(), "not logged twice");

        let mut seen = heapless::Vec::<(u8, ParameterEventKind, String<8>), 8>::new();
        while let Some(e) = server.pop_initial_parameter_event() {
            let mut n = String::new();
            n.push_str(e.name.as_str()).unwrap();
            seen.push((e.node.raw(), e.kind, n)).unwrap();
        }
        let names: heapless::Vec<(u8, ParameterEventKind, &str), 8> =
            seen.iter().map(|(n, k, s)| (*n, *k, s.as_str())).collect();
        assert_eq!(
            names.as_slice(),
            &[
                (0, ParameterEventKind::New, "a"),
                (0, ParameterEventKind::New, "c"),
                (2, ParameterEventKind::Announce, ""),
            ],
            "node 1 never enabled events; node 2 holds nothing"
        );

        // Already on: a second enable is not a second start-up.
        server.set_parameter_events(NODE, true);
        assert!(server.pop_initial_parameter_event().is_none());
        // After it drained, changes are logged as before.
        assert!(server.remove(NODE, "a"));
        let e = server.pop_parameter_event().expect("a logged deletion");
        assert_eq!(
            (e.name.as_str(), e.kind),
            ("a", ParameterEventKind::Deleted)
        );
    }

    /// phase-491 W2 — the hook sees BOTH sides: `(old, new)` on a write to an
    /// existing parameter, `(None, new)` on a remote create.
    #[test]
    fn the_on_set_hook_sees_old_and_new() {
        let mut storage: ParameterStorage = ParameterStorage::new();
        let mut server = ParameterServer::new_in(storage.as_table());
        let mut seen = Seen::default();
        assert!(server.declare(NODE, "rate", ParameterValue::Integer(1)));
        server
            .add_on_set_parameters_callback(NODE, record, OnSetContext::new(&raw mut seen as usize))
            .expect("a free slot");

        assert_eq!(
            server.apply(NODE, "rate", ParameterValue::Integer(2)),
            Ok(())
        );
        assert_eq!((seen.old, seen.new), (Some(1), Some(2)));

        server.set_allow_undeclared(NODE, true);
        assert_eq!(
            server.apply(NODE, "fresh", ParameterValue::Integer(7)),
            Ok(())
        );
        assert!(!seen.old_present, "a create has no old value");
        assert_eq!(seen.new, Some(7));
    }

    /// phase-491 W2 — rclc's LOCAL set: type, hook, write. Read-only and range
    /// do not bind the owner; a different type does.
    #[test]
    fn a_local_set_skips_read_only_and_range_but_not_type() {
        let mut storage: ParameterStorage = ParameterStorage::new();
        let mut server = ParameterServer::new_in(storage.as_table());
        let mut seen = Seen::default();
        assert!(server.declare(NODE, "rate", ParameterValue::Integer(1)));
        assert!(server.set_parameter_integer_range(NODE, "rate", 0, 10, 1));
        assert!(server.set_parameter_read_only(NODE, "rate", true));
        server
            .add_on_set_parameters_callback(NODE, record, OnSetContext::new(&raw mut seen as usize))
            .expect("a free slot");

        // The remote path refuses both.
        assert_eq!(
            server.apply(NODE, "rate", ParameterValue::Integer(50)),
            Err(ParameterError::ReadOnly)
        );
        assert_eq!(seen.calls, 0, "a refused remote write reached the hook");

        // The local path does not, and the hook sees (old, new).
        assert_eq!(
            server.set_local(NODE, "rate", ParameterValue::Integer(50)),
            Ok(())
        );
        assert_eq!((seen.old, seen.new), (Some(1), Some(50)));
        assert_eq!(server.get_integer(NODE, "rate"), Some(50));

        assert_eq!(
            server.set_local(NODE, "rate", ParameterValue::Double(1.0)),
            Err(ParameterError::TypeMismatch)
        );
        assert_eq!(
            server.set_local(NODE, "nope", ParameterValue::Integer(1)),
            Err(ParameterError::NotFound)
        );
        assert!(
            server
                .add_on_set_parameters_callback(NODE, refuse_everything, OnSetContext::default())
                .is_some()
        );
        assert_eq!(
            server.set_local(NODE, "rate", ParameterValue::Integer(3)),
            Err(ParameterError::Rejected)
        );
        assert_eq!(server.get_integer(NODE, "rate"), Some(50));
    }

    /// phase-491 W2 — rclc's set service DELETES on `NOT_SET`, after the hook
    /// accepts `(old, None)`; rclcpp's refuses. Per node.
    #[test]
    fn not_set_deletes_only_on_a_node_that_asked() {
        let mut storage: ParameterStorage = ParameterStorage::new();
        let mut server = ParameterServer::new_in(storage.as_table());
        let other = NodeKey::new(1);
        let mut seen = Seen::default();
        assert!(server.declare(NODE, "rate", ParameterValue::Integer(4)));
        assert!(server.declare(other, "rate", ParameterValue::Integer(4)));
        server.set_delete_on_not_set(NODE, true);
        server
            .add_on_set_parameters_callback(NODE, record, OnSetContext::new(&raw mut seen as usize))
            .expect("a free slot");

        assert_eq!(server.apply(NODE, "rate", ParameterValue::NotSet), Ok(()));
        assert!(!server.has(NODE, "rate"));
        assert_eq!((seen.old, seen.new_present), (Some(4), false));

        // The other node keeps rclcpp's semantics: no delete (with no descriptor
        // the store just records NOT_SET, as it always has).
        let _ = server.apply(other, "rate", ParameterValue::NotSet);
        assert!(
            server.has(other, "rate"),
            "the rclcpp node's parameter was deleted"
        );

        assert!(server.declare(NODE, "ro", ParameterValue::Integer(1)));
        assert!(server.set_parameter_read_only(NODE, "ro", true));
        assert_eq!(
            server.apply(NODE, "ro", ParameterValue::NotSet),
            Err(ParameterError::ReadOnly)
        );
        assert!(server.has(NODE, "ro"));
    }

    /// Issue 1812 / RFC-0105 D10.2 — `apply_staged` is `apply` with the value
    /// taken from a slot: the same verdict for every row `apply` decides, and
    /// the same store afterwards. Each row runs on two fresh, identically
    /// prepared servers, one per entry point.
    #[test]
    fn apply_staged_is_apply_from_a_slot() {
        fn reject(
            _: NodeKey,
            name: &str,
            _: Option<&ParameterValue>,
            _: Option<&ParameterValue>,
            _: OnSetContext,
        ) -> bool {
            name != "vetoed"
        }
        fn prepared(storage: &mut ParameterStorage<4>) -> ParameterServer<'_> {
            let mut server = ParameterServer::new_in(storage.as_table());
            assert!(server.declare(NODE, "rate", ParameterValue::Integer(1)));
            assert!(
                server.declare_with_descriptor(
                    NODE,
                    "ro",
                    ParameterValue::Integer(1),
                    Some(
                        ParameterDescriptor::new("ro", ParameterType::Integer)
                            .unwrap()
                            .with_read_only(true)
                    ),
                )
            );
            assert!(server.declare_with_descriptor(
                NODE,
                "typed",
                ParameterValue::Integer(1),
                Some(ParameterDescriptor::new("typed", ParameterType::Integer).unwrap()),
            ));
            assert!(server.declare(NODE, "vetoed", ParameterValue::Integer(1)));
            server
                .add_on_set_parameters_callback(NODE, reject, OnSetContext::new(0))
                .expect("a free slot");
            server
        }
        // The value is built per row (a `fn`), not held in the table: a
        // `ParameterValue` is ~8.5 KiB, which is the point of the slot.
        type Row = (&'static str, fn() -> ParameterValue, bool, bool);
        let rows: [Row; 7] = [
            ("rate", || ParameterValue::Integer(7), false, false),
            ("ro", || ParameterValue::Integer(7), false, false),
            ("typed", || ParameterValue::Bool(true), false, false),
            ("vetoed", || ParameterValue::Integer(7), false, false),
            ("fresh", || ParameterValue::Integer(7), false, false),
            // allow_undeclared: declares, but the 4-slot table is full.
            ("fresh", || ParameterValue::Integer(7), true, false),
            // delete_on_not_set: the rclc remote delete.
            ("rate", || ParameterValue::NotSet, false, true),
        ];
        for (name, value, allow, delete) in rows {
            let mut a_storage: ParameterStorage<4> = ParameterStorage::new();
            let mut b_storage: ParameterStorage<4> = ParameterStorage::new();
            let mut by_value = prepared(&mut a_storage);
            let mut by_slot = prepared(&mut b_storage);
            for s in [&mut by_value, &mut by_slot] {
                s.set_allow_undeclared(NODE, allow);
                s.set_delete_on_not_set(NODE, delete);
            }
            let want = by_value.apply(NODE, name, value());
            let mut slot = value();
            let got = by_slot.apply_staged(NODE, name, &mut slot);
            assert_eq!(got, want, "verdict for `{name}`");
            for probe in ["rate", "ro", "typed", "vetoed", "fresh"] {
                assert_eq!(
                    by_slot.get(NODE, probe).map(ParameterValue::param_type),
                    by_value.get(NODE, probe).map(ParameterValue::param_type),
                    "`{probe}` after setting `{name}`"
                );
                assert_eq!(
                    by_slot.get_integer(NODE, probe),
                    by_value.get_integer(NODE, probe),
                    "`{probe}` after setting `{name}`"
                );
            }
        }

        // The undeclared add, with room: declared, value swapped in.
        let mut storage: ParameterStorage<4> = ParameterStorage::new();
        let mut server = ParameterServer::new_in(storage.as_table());
        server.set_allow_undeclared(NODE, true);
        let mut slot = ParameterValue::Double(2.5);
        assert_eq!(server.apply_staged(NODE, "fresh", &mut slot), Ok(()));
        assert_eq!(server.get_double(NODE, "fresh"), Some(2.5));
        assert_eq!(server.len(NODE), 1);

        // A SWAP: the slot comes back holding the value it displaced.
        let mut slot = ParameterValue::Double(4.0);
        assert!(server.apply_staged(NODE, "fresh", &mut slot).is_ok());
        assert_eq!(slot.as_double(), Some(2.5));
        assert_eq!(server.get_double(NODE, "fresh"), Some(4.0));
    }

    /// The registry is BOUNDED and says so: the (N+1)th registration is `None`,
    /// never a silent eviction of somebody else's hook. The context reaches
    /// the callback.
    #[test]
    fn the_on_set_registry_refuses_past_its_stated_capacity() {
        let mut storage: ParameterStorage = ParameterStorage::new();
        let mut server = ParameterServer::new_in(storage.as_table());
        assert!(server.declare(NODE, "rate", ParameterValue::Integer(1)));

        assert_eq!(server.on_set_callbacks_free(), MAX_ON_SET_CALLBACKS);
        let mut handles = heapless::Vec::<_, MAX_ON_SET_CALLBACKS>::new();
        for i in 0..MAX_ON_SET_CALLBACKS {
            let h = server
                .add_on_set_parameters_callback(
                    NODE,
                    refuse_if_ctx_matches,
                    OnSetContext::new(40 + i),
                )
                .expect("within capacity");
            handles.push(h).expect("bounded by the same constant");
        }
        assert_eq!(server.on_set_callbacks_free(), 0);
        assert!(
            server
                .add_on_set_parameters_callback(NODE, refuse_everything, OnSetContext::default())
                .is_none()
        );

        // Every registered hook is consulted, and each got its own context.
        for i in 0..MAX_ON_SET_CALLBACKS {
            assert_eq!(
                server.apply(NODE, "rate", ParameterValue::Integer(40 + i as i64)),
                Err(ParameterError::Rejected)
            );
        }
        assert_eq!(
            server.apply(NODE, "rate", ParameterValue::Integer(7)),
            Ok(())
        );

        assert!(server.remove_on_set_parameters_callback(handles[0]));
        assert_eq!(server.on_set_callbacks_free(), 1);
    }

    /// The hook runs AFTER the store's own rules, so a callback never sees a
    /// write the descriptor already refused.
    #[test]
    fn the_store_rules_run_before_the_hook() {
        let mut storage: ParameterStorage = ParameterStorage::new();
        let mut server = ParameterServer::new_in(storage.as_table());
        let desc = ParameterDescriptor::new("rate", ParameterType::Integer)
            .expect("name fits")
            .with_read_only(true);
        assert!(server.declare_with_descriptor(
            NODE,
            "rate",
            ParameterValue::Integer(1),
            Some(desc)
        ));
        let _ = server
            .add_on_set_parameters_callback(NODE, refuse_everything, OnSetContext::default())
            .expect("a free slot");
        // ReadOnly, not Rejected: the callback was never consulted.
        assert_eq!(
            server.apply(NODE, "rate", ParameterValue::Integer(2)),
            Err(ParameterError::ReadOnly)
        );
    }

    /// The registry's cost is MEASURED, not asserted in prose: the doc on
    /// [`MAX_ON_SET_CALLBACKS`] names a byte count, and a slot that grows
    /// silently is how that number goes stale (the issue-1022 class).
    #[test]
    fn the_on_set_registry_costs_what_its_doc_says() {
        let slot = core::mem::size_of::<Option<OnSetCallbackSlot>>();
        let word = core::mem::size_of::<usize>();
        // node (1 byte, padded) + fn pointer + two context words, and the
        // `Option` rides in the padding or the null fn pointer.
        assert!(
            slot <= 4 * word,
            "an on-set slot is {slot} bytes, more than the four words its doc claims"
        );
        assert_eq!(
            core::mem::size_of::<[Option<OnSetCallbackSlot>; MAX_ON_SET_CALLBACKS]>(),
            MAX_ON_SET_CALLBACKS * slot
        );
    }

    /// `try_declare` says WHY, where `declare` said only `false`.
    #[test]
    fn try_declare_names_the_refusal() {
        let mut storage: ParameterStorage<2> = ParameterStorage::new();
        let mut server = ParameterServer::new_in(storage.as_table());

        assert_eq!(
            server.try_declare(NODE, "a", ParameterValue::Integer(1)),
            Ok(())
        );
        assert_eq!(
            server.try_declare(NODE, "a", ParameterValue::Integer(1)),
            Err(DeclarationError::AlreadyDeclared)
        );

        let bad_range = ParameterDescriptor::new("b", ParameterType::Integer)
            .expect("name fits")
            .with_integer_range(10, 0, 1);
        assert_eq!(
            server.try_declare_with_descriptor(
                NODE,
                "b",
                ParameterValue::Integer(1),
                Some(bad_range)
            ),
            Err(DeclarationError::InvalidRange)
        );

        let out_of_range = ParameterDescriptor::new("b", ParameterType::Integer)
            .expect("name fits")
            .with_integer_range(0, 10, 1);
        assert_eq!(
            server.try_declare_with_descriptor(
                NODE,
                "b",
                ParameterValue::Integer(99),
                Some(out_of_range)
            ),
            Err(DeclarationError::InitialValueOutOfRange)
        );

        assert_eq!(
            server.try_declare(NODE, "b", ParameterValue::Integer(1)),
            Ok(())
        );
        assert_eq!(
            server.try_declare(NODE, "c", ParameterValue::Integer(1)),
            Err(DeclarationError::StorageFull)
        );

        let long = "n".repeat(MAX_PARAM_NAME_LEN + 1);
        let mut roomy: ParameterStorage<2> = ParameterStorage::new();
        let mut server = ParameterServer::new_in(roomy.as_table());
        assert_eq!(
            server.try_declare(NODE, &long, ParameterValue::Integer(1)),
            Err(DeclarationError::NameTooLong)
        );
    }
}

// =============================================================================
// Ghost model validation
// =============================================================================

#[cfg(test)]
mod ghost_checks {
    use super::*;
    use nros_ghost_types::ParamServerGhost;

    /// phase-426 W1 — the node these single-node tests are about.
    const NODE: NodeKey = NodeKey::PRIMARY;

    /// Structural check: construct ParamServerGhost from ParameterServer private fields.
    /// If a field is renamed or retyped, this fails to compile.
    fn ghost_from_server(s: &ParameterServer<'_>) -> ParamServerGhost {
        ParamServerGhost {
            count: s.count,
            // phase-382 W2' — capacity is the borrowed TABLE's length. It used
            // to read `MAX_PARAMETERS`, which was the same number only because
            // the storage was the knob-sized array inside the server.
            max: s.table.capacity(),
        }
    }

    #[test]
    fn ghost_new_count() {
        let mut storage: ParameterStorage = ParameterStorage::new();
        let capacity = storage.capacity();
        let server = ParameterServer::new_in(storage.as_table());
        let ghost = ghost_from_server(&server);
        assert_eq!(ghost.count, 0);
        // phase-382 W2' — the ghost's `max` is the TABLE's length, and the
        // default-parameter `ParameterStorage` is `MAX_PARAMETERS` long. This
        // used to be a literal `32`, which held only while the knob was at its
        // default.
        assert_eq!(ghost.max, capacity);
        assert_eq!(ghost.max, MAX_PARAMETERS);
    }

    /// phase-382 W2' — capacity follows the TABLE, not `MAX_PARAMETERS`: a
    /// server built on a short table is full at that table's length.
    #[test]
    fn ghost_max_follows_a_shorter_table() {
        let mut storage = ParameterStorage::<2>::new();
        let mut server = ParameterServer::new_in(storage.as_table());
        assert_eq!(ghost_from_server(&server).max, 2);
        assert!(server.declare(NODE, "a", ParameterValue::Integer(1)));
        assert!(server.declare(NODE, "b", ParameterValue::Integer(2)));
        assert!(server.is_full());
        assert!(!server.declare(NODE, "c", ParameterValue::Integer(3)));
        let ghost = ghost_from_server(&server);
        assert!(ghost.count <= ghost.max);
    }

    /// phase-382 W2' — `new_in` ADOPTS the table it is handed, so a second
    /// server over the same storage sees the parameters the first declared
    /// and keeps its count in step with the slots.
    #[test]
    fn ghost_count_recovers_from_a_reused_table() {
        let mut storage = ParameterStorage::<4>::new();
        {
            let mut first = ParameterServer::new_in(storage.as_table());
            assert!(first.declare(NODE, "kept", ParameterValue::Integer(7)));
        }
        let second = ParameterServer::new_in(storage.as_table());
        assert_eq!(ghost_from_server(&second).count, 1);
        assert_eq!(second.get_integer(NODE, "kept"), Some(7));
    }

    #[test]
    fn ghost_declare_increments() {
        let mut storage: ParameterStorage = ParameterStorage::new();
        let mut server = ParameterServer::new_in(storage.as_table());
        let before = ghost_from_server(&server).count;
        assert!(server.declare(NODE, "test", ParameterValue::Integer(1)));
        let after = ghost_from_server(&server).count;
        assert_eq!(after, before + 1);
    }

    #[test]
    fn ghost_remove_decrements() {
        let mut storage: ParameterStorage = ParameterStorage::new();
        let mut server = ParameterServer::new_in(storage.as_table());
        assert!(server.declare(NODE, "test", ParameterValue::Integer(1)));
        let before = ghost_from_server(&server).count;
        assert!(server.remove(NODE, "test"));
        let after = ghost_from_server(&server).count;
        assert_eq!(after, before - 1);
    }

    #[test]
    fn ghost_count_bounded() {
        let mut storage: ParameterStorage = ParameterStorage::new();
        let mut server = ParameterServer::new_in(storage.as_table());
        for i in 0..MAX_PARAMETERS {
            let mut name = heapless::String::<64>::new();
            let _ = core::fmt::write(&mut name, format_args!("p{}", i));
            assert!(server.declare(NODE, name.as_str(), ParameterValue::Integer(i as i64)));
        }
        let ghost = ghost_from_server(&server);
        assert!(ghost.count <= ghost.max);
        assert_eq!(ghost.count, MAX_PARAMETERS);
        // Next declare should fail — count stays bounded
        assert!(!server.declare(NODE, "overflow", ParameterValue::Integer(0)));
        let ghost2 = ghost_from_server(&server);
        assert!(ghost2.count <= ghost2.max);
    }
}

// =============================================================================
// Kani bounded model checking proofs
// =============================================================================

#[cfg(kani)]
mod verification {
    use super::*;

    /// phase-426 W1 — the node these single-node tests are about.
    const NODE: NodeKey = NodeKey::PRIMARY;

    // phase-382 W2' — each proof used to build the whole store on the proof
    // stack (`ParameterServer::new()`); it now places a `ParameterStorage` and
    // lends it, which is the only shape a caller-owned store has. Kani still
    // sees the same `MAX_PARAMETERS` slots, so the bounds are unchanged.

    #[kani::proof]
    fn server_new_is_empty() {
        let mut storage: ParameterStorage = ParameterStorage::new();
        let server = ParameterServer::new_in(storage.as_table());
        assert!(server.is_empty(NODE));
        assert_eq!(server.len(NODE), 0);
        assert!(!server.is_full());
    }

    #[kani::proof]
    fn server_declare_get_roundtrip_integer() {
        let mut storage: ParameterStorage = ParameterStorage::new();
        let mut server = ParameterServer::new_in(storage.as_table());
        let val: i64 = kani::any();
        assert!(server.declare(NODE, "test", ParameterValue::Integer(val)));
        assert_eq!(server.get_integer(NODE, "test"), Some(val));
        assert!(server.has(NODE, "test"));
        assert_eq!(server.len(NODE), 1);
    }

    #[kani::proof]
    fn server_declare_get_roundtrip_bool() {
        let mut storage: ParameterStorage = ParameterStorage::new();
        let mut server = ParameterServer::new_in(storage.as_table());
        let val: bool = kani::any();
        assert!(server.declare(NODE, "test", ParameterValue::Bool(val)));
        assert_eq!(server.get_bool(NODE, "test"), Some(val));
    }

    #[kani::proof]
    fn server_set_requires_declare() {
        let mut storage: ParameterStorage = ParameterStorage::new();
        let mut server = ParameterServer::new_in(storage.as_table());
        let val: i64 = kani::any();
        // Set without declare should fail
        let result = server.set(NODE, "test", ParameterValue::Integer(val));
        assert_eq!(result, Err(ParameterError::NotFound));
    }

    #[kani::proof]
    fn server_duplicate_declare_fails() {
        let mut storage: ParameterStorage = ParameterStorage::new();
        let mut server = ParameterServer::new_in(storage.as_table());
        assert!(server.declare(NODE, "test", ParameterValue::Integer(1)));
        assert!(!server.declare(NODE, "test", ParameterValue::Integer(2)));
        // Original value preserved
        assert_eq!(server.get_integer(NODE, "test"), Some(1));
    }

    #[kani::proof]
    fn server_remove_clears() {
        let mut storage: ParameterStorage = ParameterStorage::new();
        let mut server = ParameterServer::new_in(storage.as_table());
        assert!(server.declare(NODE, "test", ParameterValue::Integer(42)));
        assert!(server.has(NODE, "test"));
        assert!(server.remove(NODE, "test"));
        assert!(!server.has(NODE, "test"));
        assert_eq!(server.len(NODE), 0);
    }

    #[kani::proof]
    fn server_get_nonexistent_returns_none() {
        let mut storage: ParameterStorage = ParameterStorage::new();
        let server = ParameterServer::new_in(storage.as_table());
        assert!(server.get(NODE, "nonexistent").is_none());
        assert!(server.get_bool(NODE, "nonexistent").is_none());
        assert!(server.get_integer(NODE, "nonexistent").is_none());
        assert!(server.get_double(NODE, "nonexistent").is_none());
        assert!(server.get_string(NODE, "nonexistent").is_none());
    }
}