rs_matter/

im.rs

/*
 *
 *    Copyright (c) 2022-2026 Project CHIP Authors
 *
 *    Licensed under the Apache License, Version 2.0 (the "License");
 *    you may not use this file except in compliance with the License.
 *    You may obtain a copy of the License at
 *
 *        http://www.apache.org/licenses/LICENSE-2.0
 *
 *    Unless required by applicable law or agreed to in writing, software
 *    distributed under the License is distributed on an "AS IS" BASIS,
 *    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *    See the License for the specific language governing permissions and
 *    limitations under the License.
 */

//! The Interaction Model as defined by the Matter Core spec: the interactions
//! (Read, Subscribe/Report, Write, Invoke, Timed) and their TLV-serde encoding
//! types, plus the engine - [`InteractionModel`] - that drives those
//! interactions against a [`crate::dm`] data model.
//!
//! It also contains a requestor-side IM client ([`client`]) and a very simple
//! responder - [`busy`] - which always returns a busy status code to all
//! incoming IM requests.

use core::cell::Cell;
use core::num::NonZeroU8;

use crate::utils::sync::blocking::Mutex;
use core::pin::pin;

use embassy_futures::select::{select3, select4};
use embassy_time::{Instant, Timer};

use crate::acl::Accessor;
use crate::crypto::Crypto;
use crate::dm::clusters::net_comm::{
    NetCtl, NetCtlStatus, NetworkType, Networks, NetworksAccess, SharedNetworks,
};
use crate::dm::clusters::wifi_diag::WirelessDiag;
use crate::dm::networks::eth::EthNetwork;
use crate::dm::networks::wireless::{NoopWirelessNetCtl, WirelessMgr, MAX_CREDS_SIZE};
use crate::dm::networks::NetChangeNotif;
use crate::dm::{
    AsyncHandler, AttrChangeNotifier, AttrDetails, Attribute, DataModel, EventEmitter,
    HandlerContext, LifecycleOp, MatchContextInstance, Metadata, ReportDataHandler,
};
use crate::error::{Error, ErrorCode};
use crate::im::events::{EventReader, EventTLVWrite, Events, DEFAULT_MAX_EVENTS_BUF_SIZE};
use crate::im::invoker::HandlerInvoker;
use crate::im::subscriptions::{
    ReportContext, Subscriptions, SubscriptionsBuffers, DEFAULT_MAX_SUBSCRIPTIONS,
};
use crate::persist::{KvBlobStoreAccess, NETWORKS_KEY};
use crate::respond::ExchangeHandler;
use crate::tlv::{get_root_node_struct, FromTLV, Nullable, TLVElement, TLVTag, TLVWrite, ToTLV};
use crate::transport::exchange::{Exchange, ExchangeId, MAX_EXCHANGE_TX_BUF_SIZE};
use crate::utils::init::{init, Init};
use crate::utils::select::Coalesce;
use crate::utils::storage::pooled::Buffers;
use crate::utils::storage::WriteBuf;
use crate::Matter;

pub use encoding::*;
pub use expand::{expand_invoke, expand_read, expand_write};

pub mod busy;
pub mod client;
pub mod encoding;
pub mod events;
pub mod expand;
pub mod invoker;
pub mod subscriptions;

/// Resource-utilisation metrics for the node, as reported by
/// `GeneralDiagnostics::DeviceLoadStatus`.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
#[cfg_attr(feature = "defmt", derive(defmt::Format))]
pub struct DeviceLoad {
    /// Subscriptions currently established on the node, across all fabrics.
    pub current_subscriptions: u16,
    /// Subscriptions currently established on the fabric of the reading
    /// subject. Zero when the figures were gathered without a fabric in scope.
    pub current_subscriptions_for_fabric: u16,
    /// Subscriptions accepted since boot, including those since torn down.
    pub total_subscriptions_established: u32,
    /// Interaction Model messages sent since boot.
    pub total_im_messages_sent: u32,
    /// Interaction Model messages received since boot.
    pub total_im_messages_received: u32,
}

/// Interaction-Model state threaded down to cluster handlers via
/// [`HandlerContext::im_stats`](crate::dm::HandlerContext::im_stats).
///
pub trait ImStats {
    /// The node's resource-utilisation metrics, for
    /// `GeneralDiagnostics::DeviceLoadStatus`.
    ///
    /// `fab_idx` is the fabric of the reading subject, for
    /// `CurrentSubscriptionsForFabric`; pass `None` when no fabric is in scope.
    fn device_load(&self, fab_idx: Option<NonZeroU8>) -> DeviceLoad;
}

impl<T> ImStats for &T
where
    T: ImStats,
{
    fn device_load(&self, fab_idx: Option<NonZeroU8>) -> DeviceLoad {
        (**self).device_load(fab_idx)
    }
}

/// An `ExchangeHandler` implementation capable of handling responder exchanges for the Interaction Model protocol.
/// The mutable, owned-together state a [`InteractionModel`] operates on: the
/// subscriptions table, the events queue and the network store.
///
/// Allocating these as a single value (rather than three separate locals wired
/// up by hand at every call site) is the whole point: construct one
/// `InteractionModelState`, then hand a reference to it to [`InteractionModel::new`]. Each of
/// the three pieces keeps its own lock internally (the proven multi-lock model);
/// folding them behind a single mutex is a possible later refinement.
///
/// `N` is the (raw) [`Networks`] implementation — e.g.
/// [`EthNetwork`](crate::dm::networks::eth::EthNetwork) or
/// [`WirelessNetworks`](crate::dm::networks::wireless::WirelessNetworks); it is
/// wrapped internally in a [`SharedNetworks`] so the data model and (later) the
/// wireless manager can share it. `NS`/`NE` bound the subscription table and the
/// event-buffer size and default to [`DEFAULT_MAX_SUBSCRIPTIONS`] /
/// [`DEFAULT_MAX_EVENTS_BUF_SIZE`].
pub struct InteractionModelState<
    N,
    const NS: usize = DEFAULT_MAX_SUBSCRIPTIONS,
    const NE: usize = DEFAULT_MAX_EVENTS_BUF_SIZE,
> {
    subscriptions: Subscriptions<NS>,
    events: Events<NE>,
    networks: SharedNetworks<N>,
    start_up_emitted: Mutex<Cell<bool>>,
}

impl<N, const NS: usize, const NE: usize> InteractionModelState<N, NS, NE> {
    /// Create a new state instance backed by the given (raw) [`Networks`] store.
    pub const fn new(networks: N) -> Self {
        Self {
            subscriptions: Subscriptions::new(),
            events: Events::new(),
            networks: SharedNetworks::new(networks),
            start_up_emitted: Mutex::new(Cell::new(false)),
        }
    }

    /// Return an in-place initializer for the state (for large `NE`, to
    /// avoid a big temporary on the stack).
    pub fn init(networks: impl Init<N>) -> impl Init<Self> {
        init!(Self {
            subscriptions <- Subscriptions::init(),
            events <- Events::init(),
            networks <- SharedNetworks::init(networks),
            start_up_emitted: Mutex::new(Cell::new(false)),
        })
    }

    /// Suppress the `BasicInformation::StartUp` event that
    /// [`InteractionModel::run`] would otherwise emit when it first starts.
    ///
    /// Call before `run` when starting the stack does *not* correspond to a
    /// device boot - e.g. a warm restart of the Matter stack within a running
    /// process, or a test fixture that needs a deterministic events queue.
    pub fn suppress_start_up_event(&self) {
        self.start_up_emitted.lock(|flag| flag.set(true));
    }

    /// Reset this state's persisted contents to factory defaults - the
    /// events-queue epoch, the network store and (if compiled in) the persisted
    /// subscriptions - removing them from `kv` using the scratch buffer
    /// provided by `kv`.
    ///
    /// Driven by [`InteractionModel::factory_reset`].
    fn reset_persist<K>(&self, kv: K) -> Result<(), Error>
    where
        K: KvBlobStoreAccess,
        N: Networks,
    {
        // The KV ops are sync, so do them all inside a single `access` closure.
        kv.access(|store, buf| {
            // The event-number epoch.
            self.events.reset_persist(&mut *store, buf)?;

            // The network store.
            self.networks.with_raw(|networks| networks.reset())?;
            store.remove(NETWORKS_KEY, buf)?;

            // Every persisted subscription record.
            #[cfg(feature = "persistent-subscriptions")]
            self.subscriptions.reset_persist(&mut *store, buf)?;

            Ok(())
        })
    }

    /// Re-hydrate this state's persisted contents - the events-queue epoch (so
    /// event numbers are not reused across reboots) and the network store - using
    /// the scratch buffer provided by `kv`.
    ///
    /// Driven by [`InteractionModel::startup`].
    fn load_persist<K>(&self, kv: K) -> Result<(), Error>
    where
        K: KvBlobStoreAccess,
        N: Networks,
    {
        // The KV ops are sync, so do them all inside a single `access` closure.
        kv.access(|store, buf| {
            // The event-number epoch.
            self.events.load_persist(&mut *store, buf)?;

            // The network store.
            self.networks.with_raw(|networks| {
                networks.reset()?;
                if let Some(data) = store.load(NETWORKS_KEY, buf)? {
                    networks.load(data)?;
                }

                Ok(())
            })
        })
    }

    /// The subscriptions table.
    pub const fn subscriptions(&self) -> &Subscriptions<NS> {
        &self.subscriptions
    }

    /// The events queue.
    pub const fn events(&self) -> &Events<NE> {
        &self.events
    }

    /// The network store (wrapped for shared, change-notifying access).
    pub const fn networks(&self) -> &SharedNetworks<N> {
        &self.networks
    }
}

/// The implementation needs a `DataModel` instance to interact with the underlying clusters of the data model.
///
/// `NC` is the network controller type driving the (optional) wireless connection
/// manager from [`InteractionModel::run`]. It defaults to [`NoopWirelessNetCtl`], which is
/// the right choice for Ethernet (and what the convenience [`InteractionModel::new`]
/// constructor wires up); wireless devices pass a real controller via
/// [`InteractionModel::new_with_net_ctl`].
pub struct InteractionModel<
    'a,
    C,
    B,
    T,
    K,
    N,
    NC = NoopWirelessNetCtl,
    R = (),
    const NS: usize = DEFAULT_MAX_SUBSCRIPTIONS,
    const NE: usize = DEFAULT_MAX_EVENTS_BUF_SIZE,
> where
    B: Buffers<IMBuffer>,
{
    matter: &'a Matter<'a>,
    crypto: C,
    buffers: &'a B,
    kv: K,
    net_ctl: NC,
    subscriptions_buffers: SubscriptionsBuffers<'a, B, NS>,
    state: &'a InteractionModelState<N, NS, NE>,
    handler: T,
    /// The controller-side `ReportData` consumer. Defaults to `()`, which
    /// disowns every inbound report (the accessory role). A controller injects a
    /// real one via [`InteractionModel::new_with_reports`].
    report_handler: R,
}

/// A [`InteractionModelState`] for an Ethernet device: its network store is a fixed
/// [`EthNetwork`], so call sites need only the (defaulted) subscription/event
/// sizes. Pairs with [`EthInteractionModel`].
pub type EthInteractionModelState<
    const NS: usize = DEFAULT_MAX_SUBSCRIPTIONS,
    const NE: usize = DEFAULT_MAX_EVENTS_BUF_SIZE,
> = InteractionModelState<EthNetwork<'static>, NS, NE>;

/// A [`InteractionModel`] for an Ethernet device (network store fixed to [`EthNetwork`]),
/// so the `N` generic disappears from call sites. Pairs with [`EthInteractionModelState`].
pub type EthInteractionModel<
    'a,
    C,
    B,
    T,
    K,
    NC = NoopWirelessNetCtl,
    R = (),
    const NS: usize = DEFAULT_MAX_SUBSCRIPTIONS,
    const NE: usize = DEFAULT_MAX_EVENTS_BUF_SIZE,
> = InteractionModel<'a, C, B, T, K, EthNetwork<'static>, NC, R, NS, NE>;

/// A [`InteractionModelState`] for a wireless device, parameterized by the concrete
/// wireless network store `N` (e.g. `WifiNetworks<3>` or a Thread store). Pairs
/// with [`WirelessInteractionModel`].
pub type WirelessInteractionModelState<
    N,
    const NS: usize = DEFAULT_MAX_SUBSCRIPTIONS,
    const NE: usize = DEFAULT_MAX_EVENTS_BUF_SIZE,
> = InteractionModelState<N, NS, NE>;

/// A [`InteractionModel`] for a wireless device (network store `N`, network controller
/// `NC`). Pairs with [`WirelessInteractionModelState`].
pub type WirelessInteractionModel<
    'a,
    C,
    B,
    T,
    K,
    N,
    NC,
    R = (),
    const NS: usize = DEFAULT_MAX_SUBSCRIPTIONS,
    const NE: usize = DEFAULT_MAX_EVENTS_BUF_SIZE,
> = InteractionModel<'a, C, B, T, K, N, NC, R, NS, NE>;

impl<'a, C, B, T, K, N, const NS: usize, const NE: usize>
    InteractionModel<'a, C, B, T, K, N, NoopWirelessNetCtl, (), NS, NE>
where
    C: Crypto,
    B: Buffers<IMBuffer>,
    T: DataModel,
    K: KvBlobStoreAccess,
    N: Networks,
{
    /// Create the data model for a device that does not need an operational
    /// wireless connection manager (typically an Ethernet device).
    ///
    /// This is a convenience wrapper around [`InteractionModel::new_with_net_ctl`] that
    /// fixes the network controller to an inert [`NoopWirelessNetCtl`], so
    /// [`InteractionModel::run`]'s connection-management branch stays dormant.
    ///
    /// # Arguments
    /// - `matter` - a reference to the `Matter` instance
    /// - `buffers` - a reference to an implementation of `Buffers<IMBuffer>` which is used for allocating RX and TX buffers on the fly, when necessary
    /// - `handler` - an instance of type `T` which implements the `DataModel` trait. This instance is used for interacting with the underlying
    ///   clusters of the data model. Note that the expectations is for the user to provide a handler that handles the Matter system clusters
    ///   as well (Endpoint 0), possibly by decorating her own clusters with the `rs_matter::dm::root_endpoint::with_` methods
    /// - `kv` - an instance of type `K` which implements the `KvBlobStoreAccess` trait
    ///   (obtain one via [`Matter::kv`]). This instance is used for interacting with the key-value blob store.
    /// - `state` - a reference to the [`InteractionModelState`] holding the subscriptions table, the
    ///   events queue and the network store (the latter parameterized by the `Networks`
    ///   implementation `N`).
    #[inline(always)]
    pub fn new(
        matter: &'a Matter<'a>,
        crypto: C,
        buffers: &'a B,
        handler: T,
        kv: K,
        state: &'a InteractionModelState<N, NS, NE>,
    ) -> Self {
        Self::new_with_net_ctl(
            matter,
            crypto,
            buffers,
            handler,
            kv,
            NoopWirelessNetCtl::new(NetworkType::Ethernet),
            state,
        )
    }
}

impl<'a, C, B, T, K, N, NC, const NS: usize, const NE: usize>
    InteractionModel<'a, C, B, T, K, N, NC, (), NS, NE>
where
    C: Crypto,
    B: Buffers<IMBuffer>,
    T: DataModel,
    K: KvBlobStoreAccess,
    N: Networks,
{
    /// Create the data model with an explicit network controller `net_ctl`.
    ///
    /// Use this for wireless devices: `net_ctl` drives the operational connection
    /// manager run from [`InteractionModel::run`] (and is typically the same controller
    /// instance also wired into the `NetworkCommissioning` cluster handler). For
    /// Ethernet devices prefer the [`InteractionModel::new`] convenience constructor.
    ///
    /// # Arguments
    /// - `matter` - a reference to the `Matter` instance
    /// - `buffers` - a reference to an implementation of `Buffers<IMBuffer>` which is used for allocating RX and TX buffers on the fly, when necessary
    /// - `handler` - an instance of type `T` which implements the `DataModel` trait. This instance is used for interacting with the underlying
    ///   clusters of the data model. Note that the expectations is for the user to provide a handler that handles the Matter system clusters
    ///   as well (Endpoint 0), possibly by decorating her own clusters with the `rs_matter::dm::root_endpoint::with_` methods
    /// - `kv` - an instance of type `K` which implements the `KvBlobStoreAccess` trait
    ///   (obtain one via [`Matter::kv`]). This instance is used for interacting with the key-value blob store.
    /// - `net_ctl` - the network controller (`NetCtl` + `WirelessDiag` + `NetChangeNotif`) used by
    ///   the operational wireless connection manager driven from [`InteractionModel::run`].
    /// - `state` - a reference to the [`InteractionModelState`] holding the subscriptions table, the
    ///   events queue and the network store (the latter parameterized by the `Networks`
    ///   implementation `N`).
    #[inline(always)]
    pub fn new_with_net_ctl(
        matter: &'a Matter<'a>,
        crypto: C,
        buffers: &'a B,
        handler: T,
        kv: K,
        net_ctl: NC,
        state: &'a InteractionModelState<N, NS, NE>,
    ) -> Self {
        state.subscriptions.clear();

        Self {
            matter,
            crypto,
            buffers,
            kv,
            net_ctl,
            subscriptions_buffers: SubscriptionsBuffers::new(),
            state,
            handler,
            report_handler: (),
        }
    }
}

impl<'a, C, B, T, K, N, NC, R, const NS: usize, const NE: usize>
    InteractionModel<'a, C, B, T, K, N, NC, R, NS, NE>
where
    C: Crypto,
    B: Buffers<IMBuffer>,
    T: DataModel,
    K: KvBlobStoreAccess,
    N: Networks,
    R: ReportDataHandler,
{
    /// Create the data model with an explicit network controller `net_ctl` and a
    /// [`ReportDataHandler`] `report_handler` — the controller / subscriber role.
    ///
    /// This is [`InteractionModel::new_with_net_ctl`] plus a report consumer:
    /// after this node establishes subscriptions (via the IM client), the
    /// publishers push `ReportData` on fresh inbound exchanges, which the
    /// `InteractionModel` routes to `report_handler`. The default
    /// (`new`/`new_with_net_ctl`) constructors leave the report handler as `()`,
    /// which disowns every report — correct for a pure accessory.
    ///
    /// # Arguments
    /// Same as [`InteractionModel::new_with_net_ctl`], plus:
    /// - `report_handler` - an instance of type `R` implementing
    ///   [`ReportDataHandler`], invoked once per received `ReportData` chunk.
    #[inline(always)]
    #[allow(clippy::too_many_arguments)]
    pub fn new_with_reports(
        matter: &'a Matter<'a>,
        crypto: C,
        buffers: &'a B,
        handler: T,
        kv: K,
        net_ctl: NC,
        report_handler: R,
        state: &'a InteractionModelState<N, NS, NE>,
    ) -> Self {
        state.subscriptions.clear();

        Self {
            matter,
            crypto,
            buffers,
            kv,
            net_ctl,
            subscriptions_buffers: SubscriptionsBuffers::new(),
            state,
            handler,
            report_handler,
        }
    }

    /// Get a reference to the `Matter` instance this data model is associated with.
    pub const fn matter(&self) -> &'a Matter<'a> {
        self.matter
    }

    pub const fn crypto(&self) -> &C {
        &self.crypto
    }

    /// Open the basic commissioning window.
    ///
    /// Equivalent to [`Matter::open_basic_comm_window`] but additionally
    /// bumps the data version of the `AdministratorCommissioning`
    /// cluster on the root endpoint and routes the change to
    /// subscribers — both happen automatically because this `InteractionModel`
    /// is itself the [`AttrChangeNotifier`] passed down.
    ///
    /// Prefer this entry point over the `Matter` one for any code path
    /// that has a `InteractionModel` available; `Matter::open_basic_comm_window`
    /// is the building block we delegate to and does not bump dataver
    /// (see its docs).
    pub fn open_basic_comm_window(&self, timeout_secs: u16) -> Result<(), Error> {
        self.matter
            .open_basic_comm_window(timeout_secs, &self.crypto, self)
    }

    /// Close the active commissioning window.
    ///
    /// Equivalent to [`Matter::close_comm_window`] but additionally
    /// bumps the `AdministratorCommissioning` dataver and routes
    /// subscribers via this `InteractionModel`'s [`AttrChangeNotifier`]. See
    /// `open_basic_comm_window` for the rationale.
    pub fn close_comm_window(&self) -> Result<bool, Error> {
        self.matter.close_comm_window(self)
    }

    /// Bump `BasicInformation::ConfigurationVersion` by one, persist
    /// the new value, and notify subscribers (which also bumps the
    /// `BasicInformation` cluster's dataver via this `InteractionModel`'s
    /// [`AttrChangeNotifier`]).
    ///
    /// Per Matter Core Spec, callers MUST invoke this
    /// whenever the node's exposed fixed-quality surface changes —
    /// typically after a firmware update that adds or removes
    /// functionality, after an internal reconfiguration that changes
    /// any `F`-quality attribute (Descriptor::ServerList,
    /// PartsList, …), or (for bridges) after a bridged node is added
    /// or removed. It is not invoked automatically by `rs-matter`
    /// because the library has no way to know about an application's
    /// reconfiguration events.
    ///
    /// Returns the new `ConfigurationVersion` value.
    pub fn bump_configuration_version(&self) -> Result<u32, Error> {
        // Delegate to `Matter::bump_configuration_version` for the
        // in-memory bump + persist; pass `self` as the
        // `AttrChangeNotifier` so the cluster's `Dataver` is bumped
        // too (the `Matter`-level call by itself only routes
        // subscribers and persists).
        self.matter.bump_configuration_version(&self.kv, self)
    }

    /// Bring the Data Model to its operational state after a reboot.
    ///
    /// Call once, after constructing the `InteractionModel` and before running
    /// it or serving exchanges. In order:
    /// - Re-hydrates the [`InteractionModelState`] - the events-queue epoch (so
    ///   event numbers are not reused across reboots) and the network store;
    /// - Replays any persisted subscriptions into the reporter's table (if the
    ///   `persistent-subscriptions` feature is enabled), so a subscriber that
    ///   had a subscription before this reboot keeps receiving reports instead
    ///   of having to notice the loss and re-subscribe;
    /// - Broadcasts [`LifecycleOp::Startup`] to every cluster handler in the
    ///   data model, so handlers with a persistence story of their own (which
    ///   the Interaction Model otherwise treats as opaque) can re-hydrate their
    ///   state from [`HandlerContext::kv`].
    ///
    /// The `Matter`-level counterpart is [`Matter::startup`] - call that one
    /// first.
    pub async fn startup(&self) -> Result<(), Error> {
        self.state.load_persist(&self.kv)?;

        // Needs the events watermark loaded just above.
        self.resume_subscriptions()?;

        self.handler.lifecycle(self, LifecycleOp::Startup)
    }

    /// Factory-reset the Data Model persistable state.
    ///
    /// The counterpart of [`InteractionModel::startup`]. In order:
    /// - Resets the [`InteractionModelState`] persisted contents to factory
    ///   defaults - the events-queue epoch, the network store and (if compiled
    ///   in) the persisted subscriptions - removing them from the KV store;
    /// - Broadcasts [`LifecycleOp::FactoryReset`] to every cluster handler in
    ///   the data model, so handlers can reset their own state and remove their
    ///   persisted data from [`HandlerContext::kv`].
    ///
    /// Call when the node is factory-reset, alongside the `Matter`-level
    /// counterpart, [`Matter::factory_reset`].
    pub async fn factory_reset(&self) -> Result<(), Error> {
        self.state.reset_persist(&self.kv)?;

        self.handler.lifecycle(self, LifecycleOp::FactoryReset)
    }

    /// Run the Data Model instance.
    ///
    /// This drives the IM timeout checks, the data-model handler's own background
    /// job, the subscriptions reporting loop, and - for wireless devices - the
    /// operational connection manager (inert for Ethernet, where `net_ctl` is a
    /// [`NoopWirelessNetCtl`]).
    /// Emit the `BasicInformation::StartUp` event, once per process lifetime.
    ///
    /// Matter Core spec: the node emits `StartUp` "upon completion of a boot
    /// or reboot process"; the closest stack-observable moment is the start of
    /// [`Self::run`], i.e. when the node becomes operational. The stack owns
    /// this (rather than a public emit API the application must remember to
    /// call) so every device is conformant by default. Event numbers stay
    /// monotonic across reboots via the events store's epoch persistence, so
    /// each boot's `StartUp` is a fresh, correctly-numbered event.
    ///
    /// Failure to emit (e.g. events buffer exhaustion) is logged, not fatal:
    /// a missing diagnostic event must not take the node down.
    fn emit_start_up_once(&self) {
        let emit = self.state.start_up_emitted.lock(|flag| !flag.replace(true));

        if emit {
            let result = crate::dm::clusters::decl::basic_information::StartUp::emit_for(
                self,
                crate::dm::endpoints::ROOT_ENDPOINT_ID,
                |event| event.software_version(self.matter.dev_det().sw_ver)?.end(),
            );

            if let Err(e) = result {
                warn!("Failed to emit the StartUp event: {:?}", e);
            }
        }
    }

    pub async fn run(&self) -> Result<(), Error>
    where
        NC: NetCtl + NetCtlStatus + WirelessDiag + NetChangeNotif,
    {
        self.emit_start_up_once();

        let mut timeouts = pin!(self.run_timeout_checks());
        let mut handler = pin!(self.handler.run(self));
        let mut subs = pin!(self.process_subscriptions(self.matter));
        let mut net = pin!(self.run_net_mgr());

        select4(&mut timeouts, &mut handler, &mut subs, &mut net)
            .coalesce()
            .await
    }

    /// Drive the operational wireless connection manager.
    ///
    /// For Ethernet devices (`net_ctl.net_type() == NetworkType::Ethernet`) there
    /// is nothing to manage, so this future simply pends forever. For wireless
    /// devices it runs a [`WirelessMgr`] over the network store, cycling through
    /// the registered networks and (re)connecting as needed once commissioned.
    async fn run_net_mgr(&self) -> Result<(), Error>
    where
        NC: NetCtl + NetCtlStatus + WirelessDiag + NetChangeNotif,
    {
        if self.net_ctl.net_type() == NetworkType::Ethernet {
            // Nothing to manage for a wired device - just pend forever.
            return core::future::pending().await;
        }

        let mut buf = [0u8; MAX_CREDS_SIZE];
        let mut mgr = WirelessMgr::new(self.state.networks(), &self.net_ctl, &mut buf);

        mgr.run().await
    }

    /// Perform a single, one-shot connect to the wireless network with the given
    /// ID, immediately and regardless of the commissioning status.
    ///
    /// This drives the same [`WirelessMgr`] used by [`InteractionModel::run`]
    /// (over this model's network controller and the network store owned by its
    /// [`InteractionModelState`]), but calls [`WirelessMgr::connect_once`] rather
    /// than the operational loop. It exists so a stack performing **non-concurrent**
    /// (BLE-only) commissioning can replay the deferred `ConnectNetwork` once the
    /// operational radio is up but before commissioning completes - without having
    /// to own a `WirelessMgr` (or the networks) itself.
    pub async fn connect_once(&self, network_id: &[u8]) -> Result<(), Error>
    where
        NC: NetCtl + NetCtlStatus + WirelessDiag + NetChangeNotif,
    {
        let mut buf = [0u8; MAX_CREDS_SIZE];
        let mut mgr = WirelessMgr::new(self.state.networks(), &self.net_ctl, &mut buf);

        mgr.connect_once(network_id).await
    }

    async fn run_timeout_checks(&self) -> Result<(), Error> {
        const CHECK_INTERVAL_SECS: u64 = 1;

        loop {
            Timer::after_secs(CHECK_INTERVAL_SECS).await;

            self.check_timeouts(None)?;
        }
    }

    fn check_timeouts(&self, exch_id: Option<ExchangeId>) -> Result<(), Error> {
        let mut notify_mdns = || self.matter.transport().notify_mdns_changed();
        let mut notify_change =
            |endpt_id, clust_id| self.notify_cluster_changed(endpt_id, clust_id);

        let removed_fabric = self.matter.with_state(|state| {
            let expire_sess_id = exch_id.and_then(|exch_id| {
                state
                    .sessions
                    .get(exch_id.session_id())
                    .map(|sess| sess.id())
            });

            // Disarm the failsafe on timeout
            let removed_fabric = state.failsafe.check_failsafe_timeout(
                &mut state.fabrics,
                &mut state.sessions,
                &self.state.networks,
                &self.kv,
                expire_sess_id,
                &mut notify_mdns,
                &mut notify_change,
            )?;

            // Close the commissioning window on timeout
            state
                .pase
                .check_comm_window_timeout(&mut notify_mdns, &mut notify_change)?;

            Ok::<_, Error>(removed_fabric)
        })?;

        // Outside `with_state`, since the broadcast runs the handlers inline
        // and they are free to access the Matter state themselves
        if let Some(fab_idx) = removed_fabric {
            self.notify_fabric_removed(fab_idx);
        }

        Ok(())
    }

    /// Answer a responding exchange using the `DataModel` instance wrapped by this exchange handler.
    pub async fn handle(&self, exchange: &mut Exchange<'_>) -> Result<(), Error> {
        let fetch_meta = |exchange: &mut Exchange| {
            let meta = exchange.rx()?.meta();
            if meta.proto_id != PROTO_ID_INTERACTION_MODEL {
                Err(ErrorCode::InvalidProto)?;
            }

            Result::<_, Error>::Ok(meta)
        };

        if exchange.rx().is_err() {
            exchange.recv_fetch().await?;
        }

        let is_groupcast = exchange.is_groupcast()?;

        let mut meta = fetch_meta(exchange)?;

        let timeout_instant = if !is_groupcast && meta.opcode::<OpCode>()? == OpCode::TimedRequest {
            let timeout = self.timed(exchange).await?;

            exchange.recv_fetch().await?;
            meta = fetch_meta(exchange)?;

            Some(timeout)
        } else {
            None
        };

        self.check_timeouts(Some(exchange.id()))?;

        // TODO: Handle the cases where we receive a timeout request
        // before read and subscribe. This is probably not allowed.

        match meta.opcode::<OpCode>()? {
            OpCode::ReadRequest if is_groupcast => {
                error!("Received a groupcast message for opcode: ReadRequest")
            }
            OpCode::ReadRequest if !is_groupcast => self.read(exchange).await?,
            OpCode::WriteRequest => self.write(exchange, timeout_instant, is_groupcast).await?,
            OpCode::InvokeRequest => self.invoke(exchange, timeout_instant, is_groupcast).await?,
            OpCode::SubscribeRequest if is_groupcast => {
                error!("Received a groupcast message for opcode: SubscribeRequest")
            }
            OpCode::SubscribeRequest if !is_groupcast => self.subscribe(exchange).await?,
            OpCode::ReportData if is_groupcast => {
                error!("Received a groupcast message for opcode: ReportData")
            }
            OpCode::ReportData if !is_groupcast => self.handle_report_data(exchange).await?,
            OpCode::TimedRequest if !is_groupcast => {
                Self::send_status(exchange, IMStatusCode::InvalidAction).await?
            }
            _ if is_groupcast => {
                // Silently drop unsupported opcodes for group messages
            }
            opcode => {
                error!("Invalid opcode: {:?}", opcode);
                Err(ErrorCode::InvalidOpcode)?
            }
        }

        if !is_groupcast {
            exchange.acknowledge().await?;
        }

        Ok(())
    }

    /// Respond to a `ReadReq` request.
    async fn read(&self, exchange: &mut Exchange<'_>) -> Result<(), Error> {
        let Some((mut tx, rx)) = self.buffers(exchange).await? else {
            return Ok(());
        };

        let read_req = ReadReq::new(TLVElement::new(&rx));
        debug!("IM: Read request: {:?}", read_req);

        if let Err(err) = Self::validate_read(&read_req) {
            error!("Invalid read request: {:?}", err);
            return Self::send_status(exchange, err.code().into()).await;
        }

        let req = ReportDataReq::Read(&read_req);

        let mut wb = WriteBuf::new(&mut tx);

        // Honor the `fabricFiltered` flag on the originating Read request.
        // When set, fabric-sensitive events emitted on other fabrics are
        // dropped before they reach the wire (Matter Core spec).
        let fabric_filtered = req.fabric_filtered().unwrap_or(true);

        let mut resp = ReportDataResponder::new(
            &req,
            None,
            HandlerInvoker::new(exchange, self),
            EventReader::new(0, u64::MAX, fabric_filtered),
            &self.state.events,
        );

        resp.respond(&mut wb, true, true, &self.handler, |_, _, _| true)
            .await?;

        Ok(())
    }

    /// Validate a `ReadReq` request prior to processing.
    fn validate_read(req: &ReadReq<'_>) -> Result<(), Error> {
        if let Some(attr_requests) = req.attr_requests()? {
            for attr_req in attr_requests {
                Self::validate_attr_wildcard_path(&attr_req?)?;
            }
        }

        Ok(())
    }

    /// Per-spec validation of an `AttrPath` that may contain wildcards.
    ///
    /// Per Matter spec, when a path uses a wildcard cluster
    /// but specifies a concrete attribute id, that attribute id must be a
    /// global (system) attribute. Any other combination must be rejected with
    /// `INVALID_ACTION`.
    fn validate_attr_wildcard_path(path: &AttrPath) -> Result<(), Error> {
        if path.cluster.is_none() {
            if let Some(attr_id) = path.attr {
                if !Attribute::is_system_attr(attr_id) {
                    return Err(ErrorCode::InvalidAction.into());
                }
            }
        }

        Ok(())
    }

    /// Respond to a `WriteReq` request.
    ///
    /// Arguments:
    /// - `exchange` - the exchange to respond to
    /// - `timeout_instant` - an optional timeout instant, if the request is a timed request
    async fn write(
        &self,
        exchange: &mut Exchange<'_>,
        timeout_instant: Option<Instant>,
        is_groupcast: bool,
    ) -> Result<(), Error> {
        while exchange.rx().is_ok() {
            // Loop while there are more write request chunks to process

            let Some((mut tx, rx)) = self.buffers(exchange).await? else {
                break;
            };

            let req = WriteReq::new(TLVElement::new(&rx));
            debug!("IM: Write request: {:?}", req);

            let timed = req.timed_request()?;

            if self.timed_out(exchange, timeout_instant, timed).await? {
                break;
            }

            let mut wb = WriteBuf::new(&mut tx);

            let mut resp = WriteResponder::new(&req, HandlerInvoker::new(exchange, self));

            resp.respond(&mut wb, &self.handler, is_groupcast).await?;

            if req.more_chunks()? {
                // This write request is just one of the chunks, so we need to wait and process
                // the next chunk as well
                exchange.recv_fetch().await?;
            }
        }

        Ok(())
    }

    /// Respond to an `InvokeReq` request.
    ///
    /// Arguments:
    /// - `exchange` - the exchange to respond to
    /// - `timeout_instant` - an optional timeout instant, if the request is a timed request
    async fn invoke(
        &self,
        exchange: &mut Exchange<'_>,
        timeout_instant: Option<Instant>,
        is_groupcast: bool,
    ) -> Result<(), Error> {
        let Some((mut tx, rx)) = self.buffers(exchange).await? else {
            return Ok(());
        };

        let req = InvReq::new(TLVElement::new(&rx));
        debug!("IM: Invoke request: {:?}", req);

        let timed = req.timed_request()?;

        if self.timed_out(exchange, timeout_instant, timed).await? {
            return Ok(());
        }

        let max_paths = exchange.matter().dev_det().max_paths_per_invoke as usize;

        if let Some(reqs) = req.inv_requests()? {
            let mut count = 0;
            for r in &reqs {
                let _ = r?;
                count += 1;
            }

            if count > max_paths {
                return Self::send_status(exchange, IMStatusCode::InvalidAction).await;
            }

            // Per Matter Core spec: when an `InvokeRequestMessage`
            // carries multiple `CommandDataIB` entries, each MUST include a unique
            // `CommandRef` and the request paths SHALL be unique. `count` is bounded
            // by `max_paths_per_invoke` (typically a single-digit number), so the
            // O(n²) pairwise check below is cheaper than allocating buffers.
            if count > 1 {
                for (i, req_i) in reqs.iter().enumerate() {
                    let req_i = req_i?;
                    if req_i.command_ref.is_none() {
                        return Self::send_status(exchange, IMStatusCode::InvalidAction).await;
                    }
                    for req_j in reqs.iter().skip(i + 1) {
                        let req_j = req_j?;
                        if req_i.path == req_j.path || req_i.command_ref == req_j.command_ref {
                            return Self::send_status(exchange, IMStatusCode::InvalidAction).await;
                        }
                    }
                }
            }
        }

        let mut wb = WriteBuf::new(&mut tx);

        let mut resp = InvokeResponder::new(&req, HandlerInvoker::new(exchange, self));

        resp.respond(&mut wb, &self.handler, is_groupcast).await
    }

    /// Respond to a `SubscribeReq` request by priming the subscription (i.e. doing an initial data report)
    /// and if the priming is successful, sending a `SubscribeResp` response to the peer and registering
    /// the subscription details in the `Subscriptions` instance.
    async fn subscribe(&self, exchange: &mut Exchange<'_>) -> Result<(), Error> {
        let Some((mut tx, rx)) = self.buffers(exchange).await? else {
            return Ok(());
        };

        let req = SubscribeReq::new(TLVElement::new(&rx));
        debug!("IM: Subscribe request: {:?}", req);

        let accessor = exchange.accessor(&self.handler)?;

        if let Err(err) = self.validate_subscribe(&req, &accessor) {
            error!("Invalid subscribe request: {:?}", err);
            return Self::send_status(exchange, err.code().into()).await;
        }

        let (fab_idx, peer_node_id) = exchange.with_state(|state| {
            let sess = exchange.id().session(&mut state.sessions);

            let fab_idx = NonZeroU8::new(sess.get_local_fabric_idx()).ok_or(ErrorCode::Invalid)?;
            let peer_node_id = sess.get_peer_node_id().ok_or(ErrorCode::Invalid)?;

            Ok((fab_idx, peer_node_id))
        })?;

        if !req.keep_subs()? {
            self.state
                .subscriptions
                .remove(&self.subscriptions_buffers, |sub| {
                    (sub.ids().fab_idx == fab_idx && sub.ids().peer_node_id == peer_node_id)
                        .then_some("new subscription request")
                });
        }

        let max_int_secs = core::cmp::max(req.max_int_ceil()?, 40); // Say we need at least 4 secs for potential latencies
        let min_int_secs = req.min_int_floor()?;

        let now = Instant::now();

        let Some(mut rctx) = self.state.subscriptions.add(
            now,
            fab_idx,
            peer_node_id,
            min_int_secs,
            max_int_secs,
            self.state.events.watermark(),
            rx,
            &self.subscriptions_buffers,
        ) else {
            return Self::send_status(exchange, IMStatusCode::ResourceExhausted).await;
        };

        let primed = self.report_data(&mut rctx, &mut tx, exchange, true).await?;

        if primed {
            exchange
                .send_with(|_, wb| {
                    SubscribeResp::write(wb, rctx.subscription().ids().id, max_int_secs)?;
                    Ok(Some(OpCode::SubscribeResponse.into()))
                })
                .await?;

            rctx.set_keep();

            info!("Subscription {:?} primed", rctx.subscription().ids());

            // Commit the subscription into the table now (its `report_complete`
            // runs on `Drop`) and then wake the reporter so it can account for
            // the new subscription's deadline.
            drop(rctx);

            // Persist the (now-committed) table so this subscription can be
            // resumed across a reboot.
            self.persist_subscriptions();

            self.state.subscriptions.notification.notify();
        }

        Ok(())
    }

    /// Handle a server-initiated `ReportData` message — the controller /
    /// subscriber side of a subscription.
    ///
    /// After a controller establishes a subscription (via the IM client), the
    /// publisher pushes `ReportData` messages on fresh inbound exchanges for the
    /// life of the subscription. Those exchanges land here (the `InteractionModel`
    /// owns all inbound IM traffic); we parse each chunk, hand it to the
    /// `DataModel`'s [`ReportDataHandler`](crate::dm::ReportDataHandler) side, and
    /// reply `StatusResponse` — `Success` when the handler accepts, or the
    /// handler's chosen status (e.g. `InvalidSubscription`) when it disowns the
    /// report. A pure accessory's handler always disowns reports, so this is inert
    /// for the accessory role.
    ///
    /// Mirrors the priming-chunk loop the IM client walks during subscribe
    /// establishment: for each chunk we ack with `StatusResponse(Success)` and
    /// fetch the next until `more_chunks` clears. The terminal MRP ack is left to
    /// the caller's `exchange.acknowledge()` (idempotent after our `send_status`).
    async fn handle_report_data(&self, exchange: &mut Exchange<'_>) -> Result<(), Error> {
        // The peer identity is a property of the (secure) session, constant
        // across every chunk of this report.
        let (fabric_idx, peer_node_id) = exchange.with_state(|state| {
            let sess = exchange.id().session(&mut state.sessions);

            let fabric_idx =
                NonZeroU8::new(sess.get_local_fabric_idx()).ok_or(ErrorCode::Invalid)?;
            let peer_node_id = sess.get_peer_node_id().ok_or(ErrorCode::Invalid)?;

            Ok((fabric_idx, peer_node_id))
        })?;

        loop {
            let (result, more_chunks, suppress_response) = {
                let rx = exchange.rx()?;
                let report = ReportDataResp::from_tlv(&TLVElement::new(rx.payload()))?;

                let subscription = crate::dm::SubscriptionCtx {
                    fabric_idx,
                    peer_node_id,
                    subscription_id: report.subscription_id,
                };

                // The report context is a `HandlerContext` (matter/crypto/kv/…)
                // over a shared borrow of the exchange, valid only for the
                // duration of this handler call — after which we resume the
                // mutable ack/fetch loop below.
                let ctx = crate::dm::ReportContextInstance::new(&*exchange, self, subscription);

                let result = self.report_handler.handle_report(ctx, &report).await;

                (
                    result,
                    report.more_chunks.unwrap_or(false),
                    report.suppress_response.unwrap_or(false),
                )
            };

            // Spec forbids `suppress_response=true` alongside `more_chunks=true`.
            if more_chunks && suppress_response {
                return Self::send_status(exchange, IMStatusCode::InvalidAction).await;
            }

            // A non-Success handler result disowns the (sub)report; report the
            // status and stop — no point ack-ing further chunks of a report we
            // rejected.
            //
            // `SuppressResponse` only suppresses the *Success* StatusResponse: a
            // failure status MUST always be sent (Matter Core spec), so the
            // publisher can tear down a subscription we no longer track (e.g. our
            // `InvalidSubscription`). Sending it regardless of `suppress_response`.
            if let Err(status) = result {
                return Self::send_status(exchange, status).await;
            }

            if !suppress_response {
                Self::send_status(exchange, IMStatusCode::Success).await?;
            }

            if !more_chunks {
                break;
            }

            // Next chunk rides in on the same exchange.
            exchange.recv_fetch().await?;
        }

        Ok(())
    }

    /// Validates the subscription request
    fn validate_subscribe(
        &self,
        req: &SubscribeReq<'_>,
        accessor: &Accessor<'_>,
    ) -> Result<(), Error> {
        // As per spec, we need to validate that the subscription request
        // contains existing endpoints, clusters and attributes, and if not
        // we should (a bit surprisingly) return InvalidAction

        self.handler.access(|node| {
            let mut has_attrs = false;
            let mut has_events = false;

            if let Some(attr_requests) = req.attr_requests()? {
                has_attrs = true;

                for attr_req in attr_requests {
                    let path = attr_req?;

                    if path.is_wildcard() {
                        Self::validate_attr_wildcard_path(&path)?;

                        if !node.has_accessible_attr(&path, accessor) {
                            return Err(ErrorCode::InvalidAction.into());
                        }
                    } else {
                        node.validate_attr_path(&path, false, false, accessor)
                            .map_err(|_| ErrorCode::InvalidAction)?;
                    }
                }
            }

            if let Some(event_reqs) = req.event_requests()? {
                has_events = true;

                for event_req in event_reqs {
                    let path = event_req?;

                    if !path.is_wildcard() {
                        node.validate_event_path(&path, accessor)
                            .map_err(|_| ErrorCode::InvalidAction)?;
                    }
                }
            }

            if !has_attrs && !has_events {
                // Empty subscribe requests are not allowed either
                return Err(ErrorCode::InvalidAction.into());
            }

            Ok(())
        })
    }

    /// Persist the current subscription table to `kv`, one record per key.
    ///
    /// A no-op when the store is a dummy (no scratch buffer). Best-effort: a
    /// persistence failure is logged but never propagated, so it cannot break
    /// reporting — persisting subscriptions is spec-optional.
    ///
    /// Compiled to a no-op unless the `persistent-subscriptions` feature is
    /// enabled, so the whole persistence path (TLV serialization, the key-value
    /// store writes) is dropped from a device that does not want it.
    #[cfg(feature = "persistent-subscriptions")]
    fn persist_subscriptions(&self) {
        let result = self.kv.access(|store, buf| {
            self.state
                .subscriptions
                .persist_all(&self.subscriptions_buffers, store, buf)
        });

        if let Err(e) = result {
            warn!("Failed to persist subscriptions: {:?}", e);
        }
    }

    #[cfg(not(feature = "persistent-subscriptions"))]
    #[inline(always)]
    fn persist_subscriptions(&self) {}

    /// Re-hydrate the subscription table from `kv` and re-arm the reporter.
    ///
    /// Driven by [`InteractionModel::startup`], after the events state has been
    /// loaded (the replayed subscriptions are primed against the loaded events
    /// watermark). Each persisted subscription is replayed into the table as a
    /// live (already primed) subscription; the reporter then reaches the
    /// subscriber on demand by establishing a session when the next report is
    /// due. No session is opened proactively here.
    ///
    /// Compiled to a no-op (returning `Ok`) unless the `persistent-subscriptions`
    /// feature is enabled.
    #[cfg(feature = "persistent-subscriptions")]
    fn resume_subscriptions(&self) -> Result<(), Error> {
        let now = Instant::now();
        let watermark = self.state.events.watermark();

        self.kv.access(|store, buf| {
            self.state.subscriptions.load_persist(
                self.buffers,
                &self.subscriptions_buffers,
                store,
                buf,
                now,
                watermark,
            )
        })?;

        // Wake the reporter so it accounts for the resumed subscriptions'
        // liveness deadlines.
        self.state.subscriptions.notification.notify();

        Ok(())
    }

    /// See the `persistent-subscriptions` variant above; a no-op when that
    /// feature is off.
    #[cfg(not(feature = "persistent-subscriptions"))]
    #[inline(always)]
    fn resume_subscriptions(&self) -> Result<(), Error> {
        Ok(())
    }

    /// Process all valid subscriptions in an endless loop, checking for changes
    /// and reporting them to the peers.
    async fn process_subscriptions(&self, matter: &Matter<'_>) -> Result<(), Error> {
        loop {
            // Sleep until the soonest subscription deadline: the end of a
            // `min_int` quiet period for a subscription holding back a change,
            // or the chosen liveness wake point. A change, event, removal, a torn
            // down session, or a newly accepted subscription (the accept path
            // notifies) all wake the loop early. With no subscription there is
            // no deadline, so just wait to be notified.
            let mut notification = pin!(self.state.subscriptions.notification.wait());
            let mut session_removed = pin!(matter.transport().wait_session_removed());

            // With no subscription (or none primed) the deadline is `Instant::MAX`,
            // so the timer effectively never fires and the loop just waits to be
            // notified.
            let deadline = self
                .state
                .subscriptions
                .next_report_at(self.state.events.watermark(), &self.subscriptions_buffers);
            let mut timeout = pin!(Timer::at(deadline));

            select3(&mut notification, &mut timeout, &mut session_removed).await;

            let now = Instant::now();

            // First remove all expired or no-longer valid subscriptions

            let mut removed_any = false;
            loop {
                let removed = self
                    .state
                    .subscriptions
                    .remove(&self.subscriptions_buffers, |sub| {
                        if sub.is_expired(now) {
                            return Some("expired");
                        }

                        matter.with_state(|state| {
                            if state.fabrics.get(sub.ids().fab_idx).is_none() {
                                return Some("fabric removed");
                            }

                            // A subscription is NOT dropped merely because the
                            // session it was accepted on is gone (eviction,
                            // peer-side re-handshake, unreachable peer, a received
                            // Close, ...): reports route by `(fabric, node)` and a
                            // fresh session is established on demand. A session
                            // ending is a transport event, not a subscription
                            // teardown. It ends only through its own lifecycle —
                            // `max_int` liveness timeout (handled above as
                            // "expired"), or the subscriber answering a report with
                            // a non-success status (handled in the report loop, which
                            // then purges the persisted record).
                            None
                        })
                    });

                removed_any |= removed;

                if !removed {
                    break;
                }
            }

            // Keep the persisted set an exact mirror of the (now-smaller) table:
            // any dropped subscription's record is removed so it will not be
            // resumed on the next reboot.
            if removed_any {
                self.persist_subscriptions();
            }

            // Now report while there are subscriptions which are due for reporting

            let event_numbers_watermark = self.state.events.watermark();

            // Track whether any subscription left the table during reporting (the
            // subscriber answered a report with a non-success status, i.e. a
            // deliberate unsubscribe), so its persisted record can be purged.
            let mut dropped_any = false;

            loop {
                let Some(mut rctx) = self.state.subscriptions.report(
                    now,
                    event_numbers_watermark,
                    &self.subscriptions_buffers,
                ) else {
                    break;
                };

                let result = self.process_subscription(matter, &mut rctx).await;

                match result {
                    Ok(true) => rctx.set_keep(),
                    // Not kept: the subscriber tore the subscription down (or we
                    // could not report). Dropping it from the table on `rctx`
                    // drop means its persisted record must be purged too.
                    Ok(false) => dropped_any = true,
                    Err(e) => {
                        // Reporting failed — typically because the session to the
                        // subscriber died (peer unreachable, MRP retransmissions
                        // exhausted). Drop that session so the next report to this
                        // peer establishes a fresh one, and keep the subscription
                        // so it retries rather than being torn down.
                        let (fab_idx, peer_node_id) = {
                            let ids = rctx.subscription().ids();
                            (ids.fab_idx, ids.peer_node_id)
                        };

                        warn!(
                            "Error processing subscription (fab {}, node {:x}): {:?}; dropping its session, will retry",
                            fab_idx.get(),
                            peer_node_id,
                            e
                        );

                        matter.with_state(|state| {
                            if let Some(id) = state
                                .sessions
                                .get_for_node(fab_idx, peer_node_id)
                                .map(|s| s.id)
                            {
                                state.sessions.remove(id);
                            }
                        });

                        // Keep the subscription to retry, but do NOT advance its
                        // watermarks: the changes/events this report was carrying
                        // never reached the subscriber and must be re-sent.
                        rctx.set_keep_retry();
                    }
                }
            }

            // A subscription that was torn down during reporting is now gone from
            // the table; re-persist so the on-disk set stays an exact mirror and
            // the torn-down subscription is not resumed on the next reboot.
            if dropped_any {
                self.persist_subscriptions();
            }

            // Periodically trim changed-attr entries that have been reported by every
            // subscription, so the table does not accumulate stale promoted wildcards.
            self.state.subscriptions.purge_reported_changes();
        }
    }

    /// Process one valid subscription, reporting the data to the peer.
    async fn process_subscription(
        &self,
        matter: &Matter<'_>,
        rctx: &mut ReportContext<'_, '_, B, NS>,
    ) -> Result<bool, Error> {
        // Route the report by the subscriber's `(fabric, node)`: reuse the best
        // live session to that peer, or (with the `case-responder-only` feature
        // off) establish a fresh one on demand. A subscription is identified by
        // its id, not bound to the session it was accepted on.
        let ids = rctx.subscription().ids();
        let mut exchange =
            Exchange::initiate(matter, self.crypto(), ids.fab_idx, ids.peer_node_id).await?;

        if let Some(mut tx) = self.buffers.get().await {
            // Always safe as `IMBuffer` is defined to be `MAX_EXCHANGE_RX_BUF_SIZE`, which is bigger than `MAX_EXCHANGE_TX_BUF_SIZE`
            unwrap!(tx.resize_default(MAX_EXCHANGE_TX_BUF_SIZE));

            let primed = self
                .report_data(rctx, &mut tx, &mut exchange, false)
                .await?;

            exchange.acknowledge().await?;

            Ok(primed)
        } else {
            error!(
                "No TX buffer available for processing subscription {:?}",
                rctx.subscription().ids(),
            );

            Ok(false)
        }
    }

    /// Process a `TimedReq` request, which is used to set a timeout for the following Write/Invoke request.
    async fn timed(&self, exchange: &mut Exchange<'_>) -> Result<Instant, Error> {
        let req = TimedReq::from_tlv(&get_root_node_struct(exchange.rx()?.payload())?)?;
        debug!("IM: Timed request: {:?}", req);

        let timeout_instant = req.timeout_instant();

        Self::send_status(exchange, IMStatusCode::Success).await?;

        Ok(timeout_instant)
    }

    /// A utility to check whether a timed request has timed out, and if so, send a timeout status response
    async fn timed_out(
        &self,
        exchange: &mut Exchange<'_>,
        timeout_instant: Option<Instant>,
        timed_req: bool,
    ) -> Result<bool, Error> {
        let status = {
            if timed_req != timeout_instant.is_some() {
                Some(IMStatusCode::TimedRequestMisMatch)
            } else if timeout_instant
                .map(|timeout_instant| Instant::now() > timeout_instant)
                .unwrap_or(false)
            {
                Some(IMStatusCode::Timeout)
            } else {
                None
            }
        };

        if let Some(status) = status {
            Self::send_status(exchange, status).await?;

            Ok(true)
        } else {
            Ok(false)
        }
    }

    /// A utility to respond with a `ReportData` response to a subscription request, which is used to report data to the peer.
    ///
    /// Arguments:
    /// - `id` - the subscription ID
    /// - `fabric_idx` - the fabric index of the peer
    /// - `peer_node_id` - the node ID of the peer
    /// - `min_event_number` - the minimum event number to report
    /// - `rx` - the received data for the subscription, when the subscription was primed
    /// - `tx` - the TX buffer to write the response to
    /// - `exchange` - the exchange to respond to
    /// - `with_dataver` - whether to include the data version in the response
    #[allow(clippy::too_many_arguments)]
    async fn report_data(
        &self,
        rctx: &mut ReportContext<'_, '_, B, NS>,
        tx: &mut [u8],
        exchange: &mut Exchange<'_>,
        with_dataver: bool,
    ) -> Result<bool, Error>
    where
        T: DataModel,
    {
        let mut wb = WriteBuf::new(tx);

        let sub_req = SubscribeReq::new(TLVElement::new(rctx.rx()));
        let req = if with_dataver {
            ReportDataReq::Subscribe(&sub_req)
        } else {
            ReportDataReq::SubscribeReport(&sub_req)
        };

        // Honor the `fabricFiltered` flag on the originating Subscribe request.
        // When set, fabric-sensitive events emitted on other fabrics are
        // dropped before they reach the wire (Matter Core spec).
        let fabric_filtered = req.fabric_filtered().unwrap_or(true);

        let mut resp = ReportDataResponder::new(
            &req,
            Some(rctx.subscription().ids().id),
            HandlerInvoker::new(exchange, self),
            EventReader::new(
                rctx.max_seen_event_number(),
                rctx.next_max_seen_event_number(),
                fabric_filtered,
            ),
            &self.state.events,
        );

        let sub_valid = resp
            .respond(
                &mut wb,
                false,
                rctx.should_send_if_empty(),
                &self.handler,
                |e, c, a| rctx.should_report_attr(e, c, a),
            )
            .await?;

        if !sub_valid {
            warn!(
                "Subscription {:?} removed during reporting",
                rctx.subscription().ids()
            );
        }

        Ok(sub_valid)
    }

    /// A utility to fetch a pair of TX/RX buffers for processing an Interaction Model request.
    ///
    /// If there are no free buffers available, this method will send a `Busy` status response to the peer.
    ///
    /// Upon returning:
    /// - The RX buffer will contain the payload of the received Interaction Model request
    /// - The TX buffer will be resized to `MAX_EXCHANGE_TX_BUF_SIZE` and will be ready to be written to
    ///
    /// Returns:
    /// - `Ok(Some((tx, rx)))` - if both TX and RX buffers are available
    /// - `Ok(None)` - if no buffers are available, and a `Busy` status response has been sent
    /// - `Err(Error)` - if an error occurred while fetching the buffers or sending the status response
    async fn buffers(
        &self,
        exchange: &mut Exchange<'_>,
    ) -> Result<Option<(B::Buffer<'a>, B::Buffer<'a>)>, Error> {
        if let Some(tx) = self.tx_buffer(exchange).await? {
            if let Some(rx) = self.rx_buffer(exchange).await? {
                return Ok(Some((tx, rx)));
            }
        }

        Ok(None)
    }

    /// A utility to fetch a RX buffer for processing an Interaction Model request.
    ///
    /// If there are no free buffers available, this method will send a `Busy` status response to the peer.
    ///
    /// Upon returning, the RX buffer will contain the payload of the received Interaction Model request.
    ///
    /// Returns:
    /// - `Ok(Some(rx))` - if a RX buffer is available
    /// - `Ok(None)` - if no RX buffer is available, and a `Busy` status response has been sent
    /// - `Err(Error)` - if an error occurred while fetching the buffer or sending the status response
    async fn rx_buffer(&self, exchange: &mut Exchange<'_>) -> Result<Option<B::Buffer<'a>>, Error> {
        if let Some(mut buffer) = self.buffer(exchange).await? {
            let rx = exchange.rx()?;

            buffer.clear();

            // Safe to unwrap, as `IMBuffer` is defined to be `MAX_EXCHANGE_RX_BUF_SIZE`, i.e. it cannot be overflown
            // by the payload of the received exchange.
            unwrap!(buffer.extend_from_slice(rx.payload()));

            exchange.rx_done()?;

            Ok(Some(buffer))
        } else {
            Ok(None)
        }
    }

    /// A utility to fetch a TX buffer for processing an Interaction Model request.
    ///
    /// If there are no free buffers available, this method will send a `Busy` status response to the peer.
    ///
    /// Upon returning, the TX buffer will be resized to `MAX_EXCHANGE_TX_BUF_SIZE` and will be ready to be written to.
    ///
    /// Returns:
    /// - `Ok(Some(tx))` - if a TX buffer is available
    /// - `Ok(None)` - if no TX buffer is available, and a `Busy` status response has been sent
    /// - `Err(Error)` - if an error occurred while fetching the buffer or sending the status response
    async fn tx_buffer(&self, exchange: &mut Exchange<'_>) -> Result<Option<B::Buffer<'a>>, Error> {
        if let Some(mut buffer) = self.buffer(exchange).await? {
            // Always safe as `IMBuffer` is defined to be `MAX_EXCHANGE_RX_BUF_SIZE`, which is bigger than `MAX_EXCHANGE_TX_BUF_SIZE`
            unwrap!(buffer.resize_default(MAX_EXCHANGE_TX_BUF_SIZE));

            Ok(Some(buffer))
        } else {
            Ok(None)
        }
    }

    /// A utility to fetch a buffer for processing an Interaction Model request.
    ///
    /// If there are no free buffers available, this method will send a `Busy` status response to the peer.
    ///
    /// Upon returning, the buffer will be UNINITIALIZED. I.e. it is up to the user to resize it appropriately
    /// if it is to be used for sending a response, or to fill it with data, if it is to be used for receiving data.
    ///
    /// Returns:
    /// - `Ok(Some(buffer))` - if a buffer is available
    /// - `Ok(None)` - if no buffer is available, and a `Busy` status response has been sent
    /// - `Err(Error)` - if an error occurred while fetching the buffer or sending the status response
    async fn buffer(&self, exchange: &mut Exchange<'_>) -> Result<Option<B::Buffer<'a>>, Error> {
        if let Some(buffer) = self.buffers.get().await {
            Ok(Some(buffer))
        } else {
            Self::send_status(exchange, IMStatusCode::Busy).await?;

            Ok(None)
        }
    }

    /// A utility to send a status response to the peer.
    async fn send_status(exchange: &mut Exchange<'_>, status: IMStatusCode) -> Result<(), Error> {
        exchange
            .send_with(|_, wb| {
                StatusResp::write(wb, status)?;

                Ok(Some(OpCode::StatusResponse.into()))
            })
            .await
    }
}

impl<C, B, T, K, N, NC, R, const NS: usize, const NE: usize> ExchangeHandler
    for InteractionModel<'_, C, B, T, K, N, NC, R, NS, NE>
where
    C: Crypto,
    B: Buffers<IMBuffer>,
    T: DataModel,
    K: KvBlobStoreAccess,
    N: Networks,
    R: ReportDataHandler,
{
    async fn handle(&self, mut exchange: Exchange<'_>) -> Result<(), Error> {
        InteractionModel::handle(self, &mut exchange).await
    }
}

impl<C, B, T, K, N, NC, R, const NS: usize, const NE: usize>
    InteractionModel<'_, C, B, T, K, N, NC, R, NS, NE>
where
    C: Crypto,
    B: Buffers<IMBuffer>,
    T: DataModel,
    K: KvBlobStoreAccess,
    N: Networks,
{
    /// The node's resource-utilisation metrics, for the `GeneralDiagnostics`
    /// cluster's `DeviceLoadStatus` attribute.
    ///
    /// `fab_idx` is the fabric of the reading subject, for
    /// `CurrentSubscriptionsForFabric`; pass `None` when no fabric is in scope.
    ///
    /// Handlers reach this through [`ImStats::device_load`] rather than calling
    /// it directly, but it is public so a controller can query its own load.
    ///
    /// Deliberately free of the `R: ReportDataHandler` bound the other
    /// `InteractionModel` methods carry: the figures come from the transport and
    /// the subscriptions table, so both the accessory and controller roles can
    /// report them.
    pub fn load_stats(&self, fab_idx: Option<NonZeroU8>) -> DeviceLoad {
        let message_counters = self.matter.transport().counters();

        DeviceLoad {
            total_im_messages_sent: message_counters.im_sent,
            total_im_messages_received: message_counters.im_received,
            ..self.state.subscriptions().load_stats(fab_idx)
        }
    }
}

impl<C, B, T, K, N, NC, R, const NS: usize, const NE: usize> ImStats
    for InteractionModel<'_, C, B, T, K, N, NC, R, NS, NE>
where
    C: Crypto,
    B: Buffers<IMBuffer>,
    T: DataModel,
    K: KvBlobStoreAccess,
    N: Networks,
{
    fn device_load(&self, fab_idx: Option<NonZeroU8>) -> DeviceLoad {
        self.load_stats(fab_idx)
    }
}

impl<C, B, T, K, N, NC, R, const NS: usize, const NE: usize> HandlerContext
    for InteractionModel<'_, C, B, T, K, N, NC, R, NS, NE>
where
    C: Crypto,
    B: Buffers<IMBuffer>,
    T: DataModel,
    K: KvBlobStoreAccess,
    N: Networks,
{
    fn matter(&self) -> &Matter<'_> {
        self.matter
    }

    fn crypto(&self) -> impl Crypto + '_ {
        &self.crypto
    }

    fn kv(&self) -> impl KvBlobStoreAccess + '_ {
        &self.kv
    }

    fn networks(&self) -> impl NetworksAccess + '_ {
        &self.state.networks
    }

    fn metadata(&self) -> impl Metadata + '_ {
        &self.handler
    }

    fn handler(&self) -> impl AsyncHandler + '_ {
        &self.handler
    }

    fn buffers(&self) -> impl Buffers<IMBuffer> + '_ {
        self.buffers
    }

    fn im_stats(&self) -> impl ImStats + '_ {
        self
    }

    fn notify_fabric_removed(&self, fab_idx: NonZeroU8) {
        if let Err(e) = self
            .handler
            .lifecycle(self, LifecycleOp::FabricRemoval { fab_idx })
        {
            warn!(
                "Failed to broadcast the removal of fabric {}: {:?}",
                fab_idx, e
            );
        }

        #[cfg(feature = "groups")]
        self.matter.transport().notify_groups_changed();
    }
}

impl<C, B, T, K, N, NC, R, const NS: usize, const NE: usize> AttrChangeNotifier
    for InteractionModel<'_, C, B, T, K, N, NC, R, NS, NE>
where
    C: Crypto,
    B: Buffers<IMBuffer>,
    T: DataModel,
    K: KvBlobStoreAccess,
    N: Networks,
{
    fn notify_attr_changed(&self, endpoint_id: EndptId, cluster_id: ClusterId, attr_id: AttrId) {
        self.handler.bump_dataver(MatchContextInstance::new(
            Some(endpoint_id),
            Some(cluster_id),
        ));
        self.state
            .subscriptions
            .notify_attr_changed(endpoint_id, cluster_id, attr_id);
    }

    fn notify_cluster_changed(&self, endpoint_id: EndptId, cluster_id: ClusterId) {
        self.handler.bump_dataver(MatchContextInstance::new(
            Some(endpoint_id),
            Some(cluster_id),
        ));
        self.state
            .subscriptions
            .notify_cluster_changed(endpoint_id, cluster_id);
    }

    fn notify_endpoint_changed(&self, endpoint_id: EndptId) {
        self.handler
            .bump_dataver(MatchContextInstance::new(Some(endpoint_id), None));
        self.state
            .subscriptions
            .notify_endpoint_changed(endpoint_id)
    }

    fn notify_all_changed(&self) {
        self.handler
            .bump_dataver(MatchContextInstance::new(None, None));
        self.state.subscriptions.notify_all_changed()
    }
}

impl<C, B, T, K, N, NC, R, const NS: usize, const NE: usize> EventEmitter
    for InteractionModel<'_, C, B, T, K, N, NC, R, NS, NE>
where
    C: Crypto,
    B: Buffers<IMBuffer>,
    T: DataModel,
    K: KvBlobStoreAccess,
    N: Networks,
{
    fn emit_event<F>(
        &self,
        endpoint_id: EndptId,
        cluster_id: ClusterId,
        event_id: EventId,
        priority: EventPriority,
        f: F,
    ) -> Result<u64, Error>
    where
        F: FnOnce(EventTLVWrite<'_>) -> Result<(), Error>,
    {
        let event_number =
            self.state
                .events
                .push(endpoint_id, cluster_id, event_id, priority, &self.kv, f)?;

        self.state
            .subscriptions
            .notify_event_emitted(endpoint_id, cluster_id, event_id);

        Ok(event_number)
    }
}

pub enum RespondOutcome {
    Accepted,
    Rejected,
    Empty,
}

/// This type responds with a `ReportData` response to all of:
/// - A `ReadReq`
/// - A `SubscribeReq`
/// - A `SubscribeReportReq` (i.e. once a valid recorded subscription is detected as in a need to be reported on)
///
/// The responder handles chunking as needed. I.e. if reported data is too large to fit into a single
/// Matter message, it will send the data in multiple chunks (i.e. with multiple Matter messages), waiting for
/// a `Success` response from the peer after each chunk, and then continuing to send the next chunk until all data is sent.
struct ReportDataResponder<'a, 'b, 'c, const NE: usize, C> {
    req: &'a ReportDataReq<'a>,
    subscription_id: Option<u32>,
    invoker: HandlerInvoker<'b, 'c, C>,
    event_reader: EventReader,
    events: &'a Events<NE>,
}

impl<'a, 'b, 'c, const NE: usize, C> ReportDataResponder<'a, 'b, 'c, NE, C>
where
    C: HandlerContext,
{
    // This is the amount of space we reserve for the structure/array closing TLVs
    // to be attached towards the end of long reads
    const LONG_READS_TLV_RESERVE_SIZE: usize = 24;

    /// Create a new `ReportDataResponder`.
    const fn new(
        req: &'a ReportDataReq<'a>,
        subscription_id: Option<u32>,
        invoker: HandlerInvoker<'b, 'c, C>,
        event_reader: EventReader,
        events: &'a Events<NE>,
    ) -> Self {
        Self {
            req,
            subscription_id,
            invoker,
            event_reader,
            events,
        }
    }

    /// Respond to the request with a `ReportData` response, possibly with more than one
    /// chunk if the data is too large to fit into a single Matter message.
    ///
    /// Arguments:
    /// - `wb` - the buffer to use while sending the response
    /// - `suppress_last_resp` - whether to suppress the response from the peer. When multiple Matter messages are
    ///   being sent due to chunking, this is valid for the last chunk only, as the others - by necessity need to have a
    ///   status response by the other peer
    async fn respond<M, F>(
        &mut self,
        wb: &mut WriteBuf<'_>,
        suppress_last_resp: bool,
        send_if_empty: bool,
        metadata: M,
        mut filter: F,
    ) -> Result<bool, Error>
    where
        M: Metadata,
        F: FnMut(EndptId, ClusterId, u32) -> bool,
    {
        let mut empty = true;

        self.start_reply(wb)?;

        if !self
            .report_attributes(wb, &mut empty, &metadata, &mut filter)
            .await?
        {
            return Ok(false);
        }

        if !self.report_events(wb, &mut empty, &metadata).await? {
            return Ok(false);
        }

        if send_if_empty || !empty {
            self.send(ReportDataChunkState::Done, suppress_last_resp, wb)
                .await
        } else {
            debug!("No data to report, skipping sending ReportData response");

            Ok(true)
        }
    }

    async fn report_attributes<M, F>(
        &mut self,
        wb: &mut WriteBuf<'_>,
        empty: &mut bool,
        metadata: M,
        mut filter: F,
    ) -> Result<bool, Error>
    where
        M: Metadata,
        F: FnMut(EndptId, ClusterId, u32) -> bool,
    {
        let accessor = self.invoker.exchange().accessor(&metadata)?;

        if self.req.attr_requests()?.is_some() {
            wb.start_array(&TLVTag::Context(ReportDataRespTag::AttributeReports as u8))?;

            for item in expand_read(&metadata, self.req, &accessor, &mut filter)? {
                let item = item?;

                *empty = false;

                loop {
                    let result = self.invoker.process_read(&item, &mut *wb).await;

                    match result {
                        Ok(()) => break,
                        Err(err) if err.code() == ErrorCode::NoSpace => {
                            let array_attr = item.as_ref().ok().filter(|attr| {
                                attr.list_index.is_none()
                                    // The whole attribute is requested
                                    // Check if it is an array, and if so, send it as individual items instead
                                    && attr.array
                            });

                            if let Some(array_attr) = array_attr {
                                if self.send_array_items(array_attr, wb).await? {
                                    break;
                                } else {
                                    return Ok(false);
                                }
                            } else {
                                debug!("<<< No TX space, chunking >>>");
                                if !self
                                    .send(ReportDataChunkState::ChunkingAttributes, false, wb)
                                    .await?
                                {
                                    return Ok(false);
                                }
                            }
                        }
                        Err(err) => Err(err)?,
                    }
                }
            }

            wb.end_container()?;
        }

        Ok(true)
    }

    async fn report_events<M>(
        &mut self,
        wb: &mut WriteBuf<'_>,
        empty: &mut bool,
        metadata: M,
    ) -> Result<bool, Error>
    where
        M: Metadata,
    {
        let accessor = self.invoker.exchange().accessor(&metadata)?;

        if let Some(event_reqs) = self.req.event_requests()? {
            wb.start_array(&TLVTag::Context(ReportDataRespTag::EventReports as _))?;

            // Validate concrete event paths against node metadata
            // and emit EventStatusIB for non-wildcard paths that don't match
            for event_req in event_reqs.iter() {
                let path = event_req?;

                if !path.is_wildcard() {
                    if let Err(status) =
                        metadata.access(|node| node.validate_event_path(&path, &accessor))
                    {
                        if matches!(status, IMStatusCode::UnsupportedEvent) {
                            // Event does not exist on this endpoint
                            // TODO: Look at TestEventsById.yaml
                            // Seems we should not error out in that case?
                            continue;
                        }

                        *empty = false;

                        let resp = EventResp::Status(EventStatus::new(path, status, None));

                        let mut result = resp.to_tlv(&TLVTag::Anonymous, &mut *wb);

                        if let Err(e) = &result {
                            if e.code() == ErrorCode::NoSpace {
                                debug!("<<< No TX space, chunking >>>");
                                if !self
                                    .send(ReportDataChunkState::ChunkingEvents, false, &mut *wb)
                                    .await?
                                {
                                    return Ok(false);
                                }

                                result = resp.to_tlv(&TLVTag::Anonymous, &mut *wb);
                            }
                        }

                        result?;
                    }
                }
            }

            let event_filters = self.req.event_filters()?;

            loop {
                let finished = self.events.fetch(|events| {
                    metadata.access(|node| {
                        for event in events {
                            let result = self.event_reader.process_read(
                                event,
                                &event_reqs,
                                &event_filters,
                                node,
                                &accessor,
                                &mut *wb,
                            );

                            if let Err(e) = &result {
                                if e.code() == ErrorCode::NoSpace {
                                    return Ok::<_, Error>(false);
                                }
                            }

                            if result? {
                                *empty = false;
                            }
                        }

                        Ok(true)
                    })
                })?;

                if finished {
                    break;
                }

                debug!("<<< No TX space, chunking >>>");
                if !self
                    .send(ReportDataChunkState::ChunkingEvents, false, wb)
                    .await?
                {
                    return Ok(false);
                }
            }

            wb.end_container()?;
        }

        Ok(true)
    }

    /// Send the items of an array attribute one by one, until the end of the array is reached.
    ///
    /// The data is potentially sent in multiple chunks if it cannot fit into a single Matter message.
    ///
    /// Arguments:
    /// - `attr` - the array attribute to send the items of
    /// - `wb` - the buffer to use while sending the items
    async fn send_array_items(
        &mut self,
        attr: &AttrDetails,
        wb: &mut WriteBuf<'_>,
    ) -> Result<bool, Error> {
        let mut attr = attr.clone();

        // First generate an empty array
        let mut list_index = None;
        attr.list_chunked = true;
        attr.list_index = Some(Nullable::new(list_index));

        loop {
            let pos = wb.get_tail();

            let result = self.invoker.read(&attr, &mut *wb).await;

            if result.is_err() {
                // If we got an error, we rewind to the position before the read
                // and handle it accordingly
                wb.rewind_to(pos);
            }

            match result {
                Ok(()) => {
                    // The empty array payload was sent
                    // Now iterate over the array and send each item one by one as separate payload

                    let new_list_index = if let Some(list_index) = list_index {
                        list_index + 1
                    } else {
                        0
                    };

                    list_index = Some(new_list_index);
                    attr.list_index = Some(Nullable::some(new_list_index));
                }
                Err(err) if err.code() == ErrorCode::NoSpace => {
                    debug!("<<< No TX space, chunking >>>");
                    if !self
                        .send(ReportDataChunkState::ChunkingAttributes, false, wb)
                        .await?
                    {
                        return Ok(false);
                    }
                }
                Err(err) if err.code() == ErrorCode::ConstraintError => break, // Got to the end of the array
                Err(err) => Err(err)?,
            }
        }

        Ok(true)
    }

    /// Send the reply to the peer, potentially opening another reply.
    ///
    /// Arguments:
    /// - `state`: tracks chunking state - are we just sending a chunk packet or are we done and wrapping up?
    /// - `suppress_last_resp`: whether to suppress the response from the peer, this is ignored if state is != Done
    /// - `wb`: the buffer containing the reply. Once the reply is sent, the buffer is re-initialized for a new reply if `more_chunks` is `true`
    async fn send(
        &mut self,
        state: ReportDataChunkState,
        suppress_last_resp: bool,
        wb: &mut WriteBuf<'_>,
    ) -> Result<bool, Error> {
        self.end_reply(state, suppress_last_resp, wb)?;

        self.invoker
            .exchange()
            .send(OpCode::ReportData, wb.as_slice())
            .await?;

        let cont = match state {
            ReportDataChunkState::ChunkingAttributes => {
                let cont = self.recv_status_success().await?;
                self.start_reply(wb)?;
                wb.start_array(&TLVTag::Context(ReportDataRespTag::AttributeReports as u8))?;
                cont
            }
            ReportDataChunkState::ChunkingEvents => {
                let cont = self.recv_status_success().await?;
                self.start_reply(wb)?;
                wb.start_array(&TLVTag::Context(ReportDataRespTag::EventReports as u8))?;
                cont
            }
            ReportDataChunkState::Done => {
                if !suppress_last_resp {
                    self.recv_status_success().await?
                } else {
                    false
                }
            }
        };

        Ok(cont)
    }

    /// Receive a status response from the peer
    ///
    /// If the response is not a status response, the method will fail with an `Invalid` error.
    ///
    /// Return `Ok(true)` if the response is a success response, `Ok(false)` if the response is not a success response.
    async fn recv_status_success(&mut self) -> Result<bool, Error> {
        let rx = self.invoker.exchange().recv().await?;
        let opcode = rx.meta().proto_opcode;

        if opcode != OpCode::StatusResponse as u8 {
            warn!(
                "Got opcode {:02x}, while expecting status code {:02x}",
                opcode,
                OpCode::StatusResponse as u8
            );

            return Err(ErrorCode::Invalid.into());
        }

        let resp = StatusResp::from_tlv(&get_root_node_struct(rx.payload())?)?;

        if resp.status == IMStatusCode::Success {
            Ok(true)
        } else {
            warn!(
                "Got status response {:?}, aborting interaction",
                resp.status
            );

            drop(rx);

            self.invoker.exchange().acknowledge().await?;

            Ok(false)
        }
    }

    /// Start a reply by initializing the `WriteBuf` and writing the initial TLVs.
    fn start_reply(&self, wb: &mut WriteBuf<'_>) -> Result<(), Error> {
        wb.reset();
        wb.shrink(Self::LONG_READS_TLV_RESERVE_SIZE)?;

        wb.start_struct(&TLVTag::Anonymous)?;

        if let Some(subscription_id) = self.subscription_id {
            assert!(matches!(
                self.req,
                ReportDataReq::Subscribe(_) | ReportDataReq::SubscribeReport(_)
            ));
            wb.u32(
                &TLVTag::Context(ReportDataRespTag::SubscriptionId as u8),
                subscription_id,
            )?;
        } else {
            assert!(matches!(self.req, ReportDataReq::Read(_)));
        }

        Ok(())
    }

    /// End a reply by writing the closing TLVs and potentially indicating that there are more chunks to send.
    fn end_reply(
        &self,
        state: ReportDataChunkState,
        suppress_resp: bool,
        wb: &mut WriteBuf<'_>,
    ) -> Result<(), Error> {
        wb.expand(Self::LONG_READS_TLV_RESERVE_SIZE)?;

        match state {
            ReportDataChunkState::ChunkingAttributes | ReportDataChunkState::ChunkingEvents => {
                wb.end_container()?;
                wb.bool(
                    &TLVTag::Context(ReportDataRespTag::MoreChunkedMsgs as u8),
                    true,
                )?;
            }
            ReportDataChunkState::Done => {
                if suppress_resp {
                    wb.bool(
                        &TLVTag::Context(ReportDataRespTag::SupressResponse as u8),
                        true,
                    )?;
                }
            }
        };

        // InteractionModelRevision is mandatory in all IM messages from
        // Matter 1.0 onward (TLV tag 0xFF). matter.js validates this
        // strictly and refuses to commission devices that omit it; the
        // reference chip-tool happens to tolerate the absence.
        wb.u8(
            &TLVTag::Context(crate::im::encoding::IM_REVISION_TAG),
            IM_REVISION,
        )?;

        wb.end_container()?;

        Ok(())
    }
}

/// Used to avoid duplicating the chunking logic for events and attributes; they both
/// share the same write path when the current packet fills up, and use this to determine
/// which field they should be setting up an array in for more output in the next packet
#[derive(Clone, Copy)]
enum ReportDataChunkState {
    ChunkingAttributes,
    ChunkingEvents,
    Done,
}

/// This type responds to a `WriteReq` by invoking the
/// corresponding handlers for each write attribute in the request.
///
/// The responser assumes that all response data can fit in a single Matter message,
/// which is a fair assumption and as per the Matter spec, in that the response of a
/// write request is always shorter than the write request itself, so given that the
/// write request fits in a single Matter message, the write reponse should as well.
///
/// With that said, the write request might itself be just one out of many chunks that
/// the other peers is sending, but processing all of those chunks is not done here,
/// but is rather - a responsibility of the caller who should call in a loop `WriteResponder::respond`
/// for all the chunks of the write request, until the `WriteReq::more_chunks()` returns `false`.
struct WriteResponder<'a, 'b, 'c, C> {
    req: &'a WriteReq<'a>,
    invoker: HandlerInvoker<'b, 'c, C>,
}

impl<'a, 'b, 'c, C> WriteResponder<'a, 'b, 'c, C>
where
    C: HandlerContext,
{
    /// Create a new `WriteResponder`.
    const fn new(req: &'a WriteReq<'a>, invoker: HandlerInvoker<'b, 'c, C>) -> Self {
        Self { req, invoker }
    }

    /// Respond to the write request by processing each write attribute in the request
    /// and sending a response back.
    async fn respond<M>(
        &mut self,
        wb: &mut WriteBuf<'_>,
        metadata: M,
        suppress_resp: bool,
    ) -> Result<(), Error>
    where
        M: Metadata,
    {
        let accessor = self.invoker.exchange().accessor(&metadata)?;

        wb.reset();

        wb.start_struct(&TLVTag::Anonymous)?;
        wb.start_array(&TLVTag::Context(WriteRespTag::WriteResponses as u8))?;

        for item in expand_write(metadata, self.req, &accessor)? {
            self.invoker.process_write(&item?, &mut *wb).await?;
        }

        if suppress_resp {
            return Ok(());
        }

        wb.end_container()?;
        // Mandatory `interactionModelRevision` (tag 0xFF); see note in
        // the ReportData emitter above.
        wb.u8(
            &TLVTag::Context(crate::im::encoding::IM_REVISION_TAG),
            IM_REVISION,
        )?;
        wb.end_container()?;

        self.invoker
            .exchange()
            .send(OpCode::WriteResponse, wb.as_slice())
            .await
    }
}

/// This type responds to an `InvRequest` by invoking the
/// corresponding handlers for each command in the invoke request.
///
/// NOTE: In future, this responder should support chunking in that
/// if the reply to all the commands in the invoke request is too large to fit
/// into a single Matter message, it should send the response in multiple chunks.
///
/// The simplest strategy for chunking would be to simply - and unconditionally - send each individual
/// command response in a separate Matter message, i.e. if the invoke request contains 3 commands,
/// the responder will send 3 Matter messages, each containing a single command response.
struct InvokeResponder<'a, 'b, 'c, C> {
    req: &'a InvReq<'a>,
    invoker: HandlerInvoker<'b, 'c, C>,
}

impl<'a, 'b, 'c, C> InvokeResponder<'a, 'b, 'c, C>
where
    C: HandlerContext,
{
    /// Create a new `InvokeResponder`.
    const fn new(req: &'a InvReq<'a>, invoker: HandlerInvoker<'b, 'c, C>) -> Self {
        Self { req, invoker }
    }

    /// Respond to the invoke request by processing each command in the request
    /// and sending one or more reponses back.
    async fn respond<M>(
        &mut self,
        wb: &mut WriteBuf<'_>,
        metadata: M,
        suppress_resp: bool,
    ) -> Result<(), Error>
    where
        M: Metadata,
    {
        wb.reset();

        wb.start_struct(&TLVTag::Anonymous)?;

        // Suppress Response -> TODO: Need to revisit this for cases where we send a command back
        wb.bool(
            &TLVTag::Context(InvRespTag::SupressResponse as u8),
            suppress_resp,
        )?;

        let has_requests = self.req.inv_requests()?.is_some();

        if has_requests {
            wb.start_array(&TLVTag::Context(InvRespTag::InvokeResponses as u8))?;
        }

        let accessor = self.invoker.exchange().accessor(&metadata)?;

        // When the Groupcast testing mode is armed for listener testing by
        // this group session's fabric, record an observation per processed
        // path (turned into `GroupcastTesting` events by the Groupcast
        // handler): a success per invoked concrete path, or - when nothing
        // was invocable at all (typically: access denied on every group
        // endpoint) - a single failed-auth observation.
        // (For invokes, `suppress_resp` is set exactly for group-addressed
        // requests - see the `respond` callers.)
        #[cfg(feature = "groups")]
        let group_testing = if suppress_resp {
            let exchange = self.invoker.exchange();

            exchange
                .matter()
                .groupcast_testing()
                .armed(crate::dm::clusters::groupcast::GroupcastTestingEnum::EnableListenerTesting)
                .and_then(|mode| {
                    exchange
                        .with_state(|state| {
                            let sess = exchange.id().session(&mut state.sessions);

                            let crate::transport::session::SessionMode::Group {
                                fab_idx,
                                group_id,
                            } = sess.get_session_mode()
                            else {
                                return Ok(None);
                            };

                            if *fab_idx != mode.fab_idx {
                                return Ok(None);
                            }

                            let src_ip = crate::dm::clusters::groupcast::TestingObservation::addr_ip(&sess.get_peer_addr());
                            let group_id = *group_id;
                            let fab_idx = *fab_idx;
                            let dst_ip = crate::dm::clusters::groupcast::TestingObservation::group_dst_ip(
                                &state.fabrics,
                                fab_idx,
                                group_id,
                            );

                            Ok(Some((group_id, src_ip, dst_ip)))
                        })
                        .unwrap_or(None)
                })
        } else {
            None
        };

        #[cfg(feature = "groups")]
        let mut any_invoked = false;

        for item in expand_invoke(metadata, self.req, &accessor)? {
            let item = item?;
            #[cfg_attr(not(feature = "groups"), allow(unused_variables))]
            let invoked = self.invoker.process_invoke(&item, &mut *wb).await?;

            #[cfg(feature = "groups")]
            if invoked {
                any_invoked = true;

                if let (Some((group_id, src_ip, dst_ip)), Ok((cmd, _))) = (&group_testing, &item) {
                    self.invoker
                        .exchange()
                        .matter()
                        .groupcast_testing()
                        .observe(crate::dm::clusters::groupcast::TestingObservation {
                            src_ip: *src_ip,
                            dst_ip: Some(*dst_ip),
                            group_id: Some(*group_id),
                            endpoint_id: Some(cmd.endpoint_id),
                            cluster_id: Some(cmd.cluster_id),
                            element_id: Some(cmd.cmd_id),
                            access_allowed: Some(true),
                            result:
                                crate::dm::clusters::groupcast::GroupcastTestResultEnum::Success,
                        });
                }
            }
        }

        #[cfg(feature = "groups")]
        if let Some((group_id, src_ip, dst_ip)) = group_testing {
            if !any_invoked {
                // Nothing was invokable for this group message - report it
                // as access-denied (the dominant cause: no ACL grant covers
                // the group's endpoints)
                self.invoker
                    .exchange()
                    .matter()
                    .groupcast_testing()
                    .observe(crate::dm::clusters::groupcast::TestingObservation {
                        src_ip,
                        dst_ip: Some(dst_ip),
                        group_id: Some(group_id),
                        endpoint_id: None,
                        cluster_id: None,
                        element_id: None,
                        access_allowed: Some(false),
                        result: crate::dm::clusters::groupcast::GroupcastTestResultEnum::FailedAuth,
                    });
            }
        }

        if suppress_resp {
            return Ok(());
        }

        if has_requests {
            wb.end_container()?;
        }

        // Mandatory `interactionModelRevision` (tag 0xFF) at the end of
        // every IM message — see the matching note in the ReportData
        // emitter above.
        wb.u8(
            &TLVTag::Context(crate::im::encoding::IM_REVISION_TAG),
            IM_REVISION,
        )?;
        wb.end_container()?;

        self.invoker
            .exchange()
            .send(OpCode::InvokeResponse, wb.as_slice())
            .await?;

        Ok(())
    }
}