esp-csi-rs 0.13.0

//! CSI delivery state machine.
//!
//! The WiFi-task callback (`capture_csi_info`) dispatches each captured CSI
//! report down exactly one user-facing path — inline callback, async queue, or
//! inline logging — selected by a single relaxed atomic load. This module owns
//! that dispatch state, the public registration functions, the lock-free CSI
//! queue + waker, the [`crate::CsiNodeClient`] consumer handle, and the
//! `WifiController` CSI wiring (`set_csi` / `build_csi_config`).

use embassy_futures::select::{Either3, select3};
use embassy_sync::blocking_mutex::raw::CriticalSectionRawMutex;
use embassy_sync::signal::Signal;
use embassy_sync::waitqueue::AtomicWaker;
#[cfg(feature = "statistics")]
use embassy_time::Instant;
use embassy_time::Timer;
use portable_atomic::{AtomicBool, Ordering};

use esp_radio::wifi::WifiController;
use esp_radio::wifi::csi::CsiConfig;

use super::CsiPacket;
use crate::STOP_SIGNAL;
#[cfg(feature = "statistics")]
use crate::stats::STATS;
#[cfg(feature = "statistics")]
use crate::stats::{RESET_SEQ_TRACKER, record_loss, seq_drop_detection_enabled};

/// Lock-free 32-slot MPMC ring used by the WiFi callback to deliver
/// captured `CsiPacket`s to user code via
/// [`CsiNodeClient::next_csi_packet`]. The producer is the WiFi-task
/// callback, the consumer is one async task, and the queue is
/// **lock-free** — no critical section on enqueue, so the WiFi-task
/// hot path is never delayed.
///
/// 32 × `sizeof(CsiPacket)` ≈ 23 KB BSS. What happens when it is full is the
/// [`OverflowPolicy`]; either way the drop is counted as `queue_full`.
static CSI_QUEUE: heapless::mpmc::Q32<CsiPacket> = heapless::mpmc::Q32::new();
/// What the async queue does with a new measurement when it is full.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[cfg_attr(feature = "defmt", derive(defmt::Format))]
#[non_exhaustive]
pub enum OverflowPolicy {
    /// Keep what is queued and drop the new measurement (the default). The consumer sees an
    /// unbroken run of older frames, then a gap.
    DropNewest,
    /// Drop the oldest queued measurement to make room. The consumer always sees the most recent
    /// frames, which is what a live display or a threshold detector wants.
    DropOldest,
}

static OVERFLOW_DROP_OLDEST: AtomicBool = AtomicBool::new(false);

/// Choose what the async queue ([`CsiNodeClient::next_csi_packet`]) does when it is full.
pub fn set_csi_queue_overflow(policy: OverflowPolicy) {
    OVERFLOW_DROP_OLDEST.store(policy == OverflowPolicy::DropOldest, Ordering::Relaxed);
}

/// The async queue's overflow policy.
pub fn csi_queue_overflow() -> OverflowPolicy {
    if OVERFLOW_DROP_OLDEST.load(Ordering::Relaxed) {
        OverflowPolicy::DropOldest
    } else {
        OverflowPolicy::DropNewest
    }
}

/// Single-slot waker for the CSI consumer. Registered by
/// [`CsiNodeClient::next_csi_packet`] and woken from the WiFi callback
/// after a successful `CSI_QUEUE.enqueue`.
static CSI_WAKER: AtomicWaker = AtomicWaker::new();

/// Whether the reporting currently in force reports at all (anything but `Reporting::None`).
///
/// Set from the configured mode at the start of every run, and moved mid-run in one case only: a
/// responder of the symmetric ESP-NOW exchange that hears an initiator reporting nothing promotes
/// itself to `Report { to: Host }`. Every other mode keeps its configured value for the whole run.
///
/// Deliberately NOT the same gate as `CSI_PUBLISH_ENABLED`. That one decides whether the WiFi
/// callback builds a packet at all; this one drives the ESP-NOW responder/initiator behaviour.
/// Conflating them silently breaks passive setups, where the user wants a node
/// that still reads CSI.
pub(crate) static REPORTS: AtomicBool = AtomicBool::new(false);

/// Signalled whenever [`set_runtime_reporting`] changes the gate, so an ESP-NOW task waiting
/// on a reporting change wakes immediately instead of polling.
pub(crate) static REPORTING_CHANGED: Signal<CriticalSectionRawMutex, ()> = Signal::new();

/// The user's runtime output preference, toggled by [`set_csi_output_enabled`] and, through it, by
/// `set-csi-output --enabled=` on the serial console and `POST /config/csi-output` on the HTTP API.
///
/// Defaults to enabled: a node that has been asked for no preference delivers.
pub(crate) static CSI_OUTPUT_ENABLED: AtomicBool = AtomicBool::new(true);

/// Whether any consumer is installed — a logger, a callback, or an async drainer. Set by the
/// functions that install one, and distinct from the two gates above so that switching a node to
/// `Reporting::None` does not forget that a callback was registered.
static USER_PUBLISH_INTENT: AtomicBool = AtomicBool::new(false);

// CSI publish gate. The WiFi callback checks this in a single relaxed load to decide whether to
// build and emit a CsiPacket. It is a cached AND of the three inputs above, recomputed by
// `apply_publish_gate` whenever one of them moves — the hot path must not pay for three loads and
// two branches per frame on `riscv32imc`.
static CSI_PUBLISH_ENABLED: AtomicBool = AtomicBool::new(false);
pub(crate) static CSI_OUTPUT_CHANGED: Signal<CriticalSectionRawMutex, ()> = Signal::new();

/// CSI delivery mode — single-atomic dispatch in the WiFi callback.
///
/// Per-packet, the callback loads `CSI_DELIVERY_MODE` once and branches
/// on it. Exactly one of the user-facing delivery paths runs, so users
/// pay only for what they asked for:
/// - `Off`: nothing past the publish gate (apart from seq-drop tracking).
/// - `Callback`: dispatch to the `fn` stored in `CSI_CALLBACK` with a
///   `&CsiPacket` borrow. Lowest latency, runs on the WiFi-task
///   hot path. **Picked by [`set_csi_callback`].**
/// - `Async`: move the packet into the lock-free `CSI_QUEUE` and wake
///   the consumer registered via `CSI_WAKER`. Doesn't block the WiFi
///   task. **Picked lazily by the first
///   [`CsiNodeClient::next_csi_packet`].`**
///
/// The two are **mutually exclusive** so the WiFi callback never pays
/// for both a callback dispatch and a 640 B memcpy on the same packet.
/// Toggle explicitly with [`set_csi_delivery_mode`].
#[repr(u8)]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[cfg_attr(feature = "defmt", derive(defmt::Format))]
pub enum CsiDeliveryMode {
    /// No user delivery. Inline `log_csi` may still run if its gate is
    /// open (controlled by [`set_csi_logging_enabled`]).
    Off = 0,
    /// Dispatch to the `fn` registered with [`set_csi_callback`] inline
    /// in the WiFi callback context.
    Callback = 1,
    /// Move the packet into the lock-free `CSI_QUEUE` and wake the
    /// async consumer awaiting [`CsiNodeClient::next_csi_packet`].
    Async = 2,
}

/// Single-atomic dispatch select for the WiFi callback. Read once per
/// CSI event in `capture_csi_info`. See [`CsiDeliveryMode`] for the
/// branch semantics.
static CSI_DELIVERY_MODE: portable_atomic::AtomicU8 = portable_atomic::AtomicU8::new(0);

/// User CSI callback registered via [`set_csi_callback`]. Loaded only
/// when `CSI_DELIVERY_MODE == Callback`, so callers in `Off` / `Async`
/// modes don't pay for an extra atomic load.
static CSI_CALLBACK: core::sync::atomic::AtomicPtr<()> =
    core::sync::atomic::AtomicPtr::new(core::ptr::null_mut());

/// Raw CSI fast-path callback registered via [`set_csi_raw_callback`]. When
/// non-null, `capture_csi_info` invokes it and returns **before** building the
/// ~640 B [`CsiPacket`] — the cheapest possible delivery path, used by the
/// CPU-comparison DUT to match the ESP-IDF reference's do-nothing `csi_cb`.
/// Checked before [`CSI_DELIVERY_MODE`], so it overrides callback/async/off.
static CSI_RAW_CALLBACK: core::sync::atomic::AtomicPtr<()> =
    core::sync::atomic::AtomicPtr::new(core::ptr::null_mut());

/// Inline-logging gate. Independent of [`CSI_DELIVERY_MODE`] so the
/// per-packet UART/JTAG `log_csi` path is controlled separately
/// (toggle with [`set_csi_logging_enabled`]).
static CSI_INLINE_LOG_ENABLED: AtomicBool = AtomicBool::new(false);

/// Enable or disable inline CSI **logging** (per-packet UART/JTAG output).
///
/// This controls only the inline `log_csi` path inside the WiFi callback.
/// It does **not** disable a [`set_csi_callback`] hook — registering a
/// callback opens an independent publish gate, and that gate stays open
/// regardless of this flag. So a typical "process inline, no UART flood"
/// setup is:
///
/// ```ignore
/// init_logger(spawner, LogMode::Text);   // publish gate + log gate ON
/// set_csi_logging_enabled(false);        // log gate OFF (callback still fires)
/// set_csi_callback(on_csi);              // publish gate ON, log gate untouched
/// ```
///
/// Defaults / who flips this for you:
/// - `init_logger` enables it automatically in sync mode (the WiFi
///   callback writes CSI lines inline).
/// - In `async-print` mode `CsiNodeClient::get_csi_data` enables the
///   publish gate (separate from the log gate) lazily on first await.
/// - [`set_csi_callback`] enables only the publish gate — it does not
///   touch this log-output flag.
pub fn set_csi_logging_enabled(enabled: bool) {
    CSI_INLINE_LOG_ENABLED.store(enabled, Ordering::Release);
    // Keep the master publish gate paired with logging by default so existing callers that only
    // flip `set_csi_logging_enabled(true)` to get UART output still work.
    USER_PUBLISH_INTENT.store(enabled, Ordering::Release);
    apply_publish_gate();
}

/// Returns whether inline CSI logging is currently enabled (i.e. whether
/// the per-packet UART/JTAG `log_csi` path will run).
pub fn csi_logging_enabled() -> bool {
    CSI_INLINE_LOG_ENABLED.load(Ordering::Relaxed)
}

/// Set the active CSI delivery mode (callback / async / off).
///
/// The WiFi callback dispatches to **exactly one** path per packet —
/// callers pay no overhead for the path they didn't pick. Switching
/// with this fn is a single relaxed atomic store; the next CSI event
/// follows the new mode.
///
/// You normally don't call this directly:
/// - [`set_csi_callback`] sets the mode to [`CsiDeliveryMode::Callback`].
/// - First await of [`CsiNodeClient::next_csi_packet`] sets it to
///   [`CsiDeliveryMode::Async`].
/// - [`clear_csi_callback`] sets it to [`CsiDeliveryMode::Off`].
///
/// Use this fn when you want to **switch** between paths at runtime
/// without re-registering, or to fully disable user delivery while
/// leaving inline logging running.
pub fn set_csi_delivery_mode(mode: CsiDeliveryMode) {
    CSI_DELIVERY_MODE.store(mode as u8, Ordering::Release);
}

/// Returns the active CSI delivery mode.
pub fn csi_delivery_mode() -> CsiDeliveryMode {
    match CSI_DELIVERY_MODE.load(Ordering::Relaxed) {
        1 => CsiDeliveryMode::Callback,
        2 => CsiDeliveryMode::Async,
        _ => CsiDeliveryMode::Off,
    }
}

/// Register a user callback invoked inline for every captured CSI packet.
///
/// The callback runs in the WiFi task context (the same context that
/// formats and writes CSI lines), with a borrow of the [`CsiPacket`]
/// *before* it is consumed by the logging path. This is the supported
/// path for **on-device CSI processing** — zero channel hops, lowest
/// possible latency.
///
/// **Constraints**: the callback runs on the WiFi task hot path and MUST
/// be fast and non-blocking. Avoid heap allocation, locking, and long
/// format/write work. For heavier processing, copy what you need out of
/// the packet and post to your own task.
///
/// Registering opens the master publish gate and switches the
/// delivery mode to [`CsiDeliveryMode::Callback`]. Any prior async
/// drain mode is replaced — the WiFi callback only runs the inline
/// callback path from this point. The inline-logging gate
/// ([`set_csi_logging_enabled`]) is left untouched so `init_logger`'s
/// UART output (or its absence) is preserved.
///
/// Call [`clear_csi_callback`] to remove the hook and return to
/// [`CsiDeliveryMode::Off`].
pub fn set_csi_callback(cb: fn(&CsiPacket)) {
    // Store the fn pointer first so the WiFi callback never sees the
    // mode flipped to `Callback` while `CSI_CALLBACK` is still null.
    CSI_CALLBACK.store(cb as *mut (), core::sync::atomic::Ordering::Release);
    CSI_DELIVERY_MODE.store(CsiDeliveryMode::Callback as u8, Ordering::Release);
    USER_PUBLISH_INTENT.store(true, Ordering::Release);
    apply_publish_gate();
}

/// Remove the user CSI callback registered via [`set_csi_callback`]
/// and switch to [`CsiDeliveryMode::Off`].
///
/// The publish gate and inline-logging gate are left untouched — call
/// `set_csi_logging_enabled(false)` if you also want to suppress
/// logging output, or `set_csi_delivery_mode(CsiDeliveryMode::Async)`
/// to swap to async drain without re-driving lazy initialization.
pub fn clear_csi_callback() {
    CSI_DELIVERY_MODE.store(CsiDeliveryMode::Off as u8, Ordering::Release);
    CSI_CALLBACK.store(core::ptr::null_mut(), core::sync::atomic::Ordering::Release);
}

/// Register a **raw** CSI fast-path callback (CPU-benchmark use).
///
/// Unlike [`set_csi_callback`], the WiFi callback invokes `cb` and returns
/// immediately, **without** building a [`CsiPacket`] — so the per-frame
/// CSI cost is just the callback dispatch, matching the ESP-IDF reference's
/// self-timing `csi_cb`. Intended for the like-for-like CPU comparison DUT;
/// the callback receives no CSI data (it cannot, by design — that is the cost
/// being elided). Pass a `fn()` that does only minimal bookkeeping.
pub fn set_csi_raw_callback(cb: fn()) {
    CSI_RAW_CALLBACK.store(cb as *mut (), core::sync::atomic::Ordering::Release);
    USER_PUBLISH_INTENT.store(true, Ordering::Release);
    apply_publish_gate();
}

/// Toggle CSI output delivery at runtime, leaving capture running.
///
/// The radio keeps capturing and the RX path keeps its timing, but nothing is decoded, logged or
/// handed to a callback — which is what separates acquisition cost from delivery cost.
///
/// Before 0.11 this stored a flag that no CSI path read, so `set-csi-output --enabled=false` on the
/// console and `POST /config/csi-output` on the HTTP API reported success and changed nothing. It
/// now closes the publish gate.
///
/// This is only the delivery gate. Whether the node reports at all is its
/// [`Reporting`](crate::Reporting), set on the mode's config; a node reporting `None` delivers
/// nothing whatever this is set to.
pub fn set_csi_output_enabled(enabled: bool) {
    CSI_OUTPUT_ENABLED.store(enabled, Ordering::Relaxed);
    apply_publish_gate();
    CSI_OUTPUT_CHANGED.signal(());
}

/// Whether CSI output delivery is currently enabled.
pub fn csi_output_enabled() -> bool {
    CSI_OUTPUT_ENABLED.load(Ordering::Relaxed)
}

/// Source-MAC allowlist for captured CSI. All-zero (the default) means "accept any source".
///
/// A receiver is promiscuous by nature: it reports CSI for every frame its radio decodes, which on
/// a busy channel means the AP's own beacons and ACKs, the association exchange, and any THIRD-PARTY
/// device on the channel. That is correct behaviour and the reports are real, but it is not what
/// someone measuring their own link asked for, and it reads as corrupt data — the leading field of a
/// row is the frame's own 802.11 sequence number, which is per-transmitter and per-TID, so foreign
/// rows carry arbitrary sequence numbers and a legacy-rate PHY's shorter CSI length.
///
/// Filtering on the device rather than on the host is not just convenience. On a console-bound
/// receiver every rejected frame is console bandwidth handed back to the traffic the user actually
/// configured, and the rejection happens before the 612-byte packet copy and before any formatting.
/// Held as a packed `u64` (big-endian in the low 48 bits) rather than a mutex-wrapped `[u8; 6]`,
/// because this is read from inside the Wi-Fi CSI callback. On `riscv32imc` every atomic op takes a
/// critical section, and this file's own dispatch comments note that extra gate atomics in that ISR
/// path delay the Embassy timer ISR — a `CriticalSectionRawMutex` per captured frame is exactly the
/// cost the fast path is written to avoid. One relaxed 64-bit load has no such problem.
static CSI_PEER_FILTER: portable_atomic::AtomicU64 = portable_atomic::AtomicU64::new(0);

/// Minimum `sig_mode` (PHY class) to accept: `0` accepts everything, `1` accepts HT (802.11n) and
/// better, dropping the legacy-rate mgmt/control frames that produce the short L-LTF-only rows.
static CSI_MIN_SIG_MODE: portable_atomic::AtomicU8 = portable_atomic::AtomicU8::new(0);

/// Single gate so the DEFAULT (unfiltered) path costs one relaxed bool load rather than a 64-bit
/// load plus a `u8` load on every captured frame.
static CSI_FILTER_ACTIVE: AtomicBool = AtomicBool::new(false);

#[inline(always)]
fn pack_mac(mac: [u8; 6]) -> u64 {
    let mut v = 0u64;
    let mut i = 0;
    while i < 6 {
        v = (v << 8) | mac[i] as u64;
        i += 1;
    }
    v
}

fn refresh_filter_active() {
    let active = CSI_PEER_FILTER.load(Ordering::Relaxed) != 0
        || CSI_MIN_SIG_MODE.load(Ordering::Relaxed) != 0;
    CSI_FILTER_ACTIVE.store(active, Ordering::Relaxed);
}

/// Restrict captured CSI to one source MAC. All-zero clears the filter.
pub fn set_csi_peer_filter(mac: [u8; 6]) {
    CSI_PEER_FILTER.store(pack_mac(mac), Ordering::Relaxed);
    refresh_filter_active();
}

/// The current source-MAC filter (all-zero = no filter).
pub fn csi_peer_filter() -> [u8; 6] {
    let v = CSI_PEER_FILTER.load(Ordering::Relaxed);
    let mut mac = [0u8; 6];
    let mut i = 0;
    while i < 6 {
        mac[i] = (v >> (8 * (5 - i))) as u8;
        i += 1;
    }
    mac
}

/// Restrict captured CSI to frames whose `sig_mode` is at least `min`.
///
/// `0` accepts every PHY (default); `1` keeps HT and better.
pub fn set_csi_min_sig_mode(min: u8) {
    CSI_MIN_SIG_MODE.store(min, Ordering::Relaxed);
    refresh_filter_active();
}

/// The current minimum-`sig_mode` filter.
pub fn csi_min_sig_mode() -> u8 {
    CSI_MIN_SIG_MODE.load(Ordering::Relaxed)
}

/// Whether a frame passes the attribution filters, judged from the raw `WifiCsiInfo`.
///
/// Deliberately reads `info` rather than a built `CsiPacket`: both fields are available on the
/// raw callback argument, so a rejected frame costs one MAC compare instead of a 612-byte copy plus
/// a format pass. On the classic MACs `packet_mode()` is the `sig_mode` field; on C5/C6 there is no
/// such field, so the PHY half of the filter is a no-op there and only the MAC filter applies.
#[inline(always)]
fn csi_passes_filter(info: &esp_radio::wifi::csi::WifiCsiInfo<'_>) -> bool {
    // One relaxed load on the unfiltered default, which is what almost every run uses.
    if !CSI_FILTER_ACTIVE.load(Ordering::Relaxed) {
        return true;
    }
    let want = CSI_PEER_FILTER.load(Ordering::Relaxed);
    if want != 0 && pack_mac(*info.mac()) != want {
        return false;
    }
    #[cfg(not(any(feature = "esp32c5", feature = "esp32c6")))]
    {
        let min = CSI_MIN_SIG_MODE.load(Ordering::Relaxed);
        if min != 0 && (info.packet_mode() as u8) < min {
            return false;
        }
    }
    true
}

/// The reporting currently **in force**, which is not always the one that was configured.
///
/// A responder that hears an initiator reporting [`None`](crate::Reporting::None) promotes itself
/// to [`Report { to: Host }`](crate::Reporting::Report) mid-run so the pair still produces a
/// dataset. That promotion moves this value and deliberately does not rewrite the node's
/// configuration, so [`CsiNode::reporting`](crate::CsiNode::reporting) keeps reporting what you
/// asked for while this reports what is happening.
///
/// The policy is process-wide and is reset between runs, so before the first
/// [`CsiNode::run`](crate::CsiNode::run) it reads `None` regardless of how the node was built.
/// Ask the node, not this function, if you want the configured value.
pub fn runtime_reporting() -> crate::Reporting {
    match POLICY_KIND.load(Ordering::Relaxed) {
        POLICY_REPORT => crate::Reporting::Report {
            to: if REPORT_LOCAL.load(Ordering::Relaxed) {
                crate::ReportTo::Local
            } else {
                crate::ReportTo::Host
            },
        },
        POLICY_THRESHOLD => crate::Reporting::ThresholdBased(crate::Threshold {
            level: POLICY_A.load(Ordering::Relaxed),
            hold_ms: POLICY_B.load(Ordering::Relaxed),
            variation_only: POLICY_VARIATION.load(Ordering::Relaxed),
        }),
        POLICY_DECIMATE => crate::Reporting::Decimate(POLICY_A.load(Ordering::Relaxed)),
        _ => crate::Reporting::None,
    }
}

/// Recompute the publish gate from its three inputs: a consumer is installed, the node reports,
/// and output has not been switched off at runtime.
///
/// Called from every writer of those three rather than read three times per frame, because this is
/// the gate the Wi-Fi callback loads on the hot path for every captured frame.
fn apply_publish_gate() {
    let open = USER_PUBLISH_INTENT.load(Ordering::Relaxed)
        && REPORTS.load(Ordering::Relaxed)
        && CSI_OUTPUT_ENABLED.load(Ordering::Relaxed);
    CSI_PUBLISH_ENABLED.store(open, Ordering::Release);
}

const POLICY_NONE: u8 = 0;
const POLICY_REPORT: u8 = 1;
const POLICY_THRESHOLD: u8 = 2;
const POLICY_DECIMATE: u8 = 3;

/// The policy in force, decomposed into atomics the CSI callback can read without a lock:
/// its kind, `level` / decimation factor, `hold_ms`, and `variation_only`.
static POLICY_KIND: portable_atomic::AtomicU8 = portable_atomic::AtomicU8::new(POLICY_NONE);
static POLICY_A: portable_atomic::AtomicU16 = portable_atomic::AtomicU16::new(0);
static POLICY_B: portable_atomic::AtomicU16 = portable_atomic::AtomicU16::new(0);
static POLICY_VARIATION: AtomicBool = AtomicBool::new(false);
/// `Report { to: Local }` is in force: deliver to the on-device consumers (callback, async queue)
/// but never to the host output. Read only on the inline-log path, after a consumer was chosen.
static REPORT_LOCAL: AtomicBool = AtomicBool::new(false);
/// Set by every policy change; the callback resets its threshold/decimation state on seeing it.
static POLICY_RESET: AtomicBool = AtomicBool::new(false);

/// Change the reporting at runtime — e.g. an initiator signalling a responder to start or stop.
///
/// Reopening the gate here is not optional. A responder paired with a silent initiator promotes
/// itself mid-run; without this the promoted node would capture in silence for the rest of the
/// run, which is the failure a `None` gate is most likely to introduce.
pub(crate) fn set_runtime_reporting(policy: crate::Reporting) {
    let (kind, a, b, var) = match policy {
        crate::Reporting::Report { .. } => (POLICY_REPORT, 0, 0, false),
        crate::Reporting::None => (POLICY_NONE, 0, 0, false),
        crate::Reporting::ThresholdBased(t) => {
            (POLICY_THRESHOLD, t.level, t.hold_ms, t.variation_only)
        }
        crate::Reporting::Decimate(n) => (POLICY_DECIMATE, n.max(1), 0, false),
    };
    REPORT_LOCAL.store(
        matches!(policy, crate::Reporting::Report { to: crate::ReportTo::Local }),
        Ordering::Relaxed,
    );
    POLICY_A.store(a, Ordering::Relaxed);
    POLICY_B.store(b, Ordering::Relaxed);
    POLICY_VARIATION.store(var, Ordering::Relaxed);
    POLICY_RESET.store(true, Ordering::Relaxed);
    POLICY_KIND.store(kind, Ordering::Release);
    REPORTS.store(policy.reports(), Ordering::Relaxed);
    apply_publish_gate();
    REPORTING_CHANGED.signal(());
}

/// Set the reporting from outside the crate, for a node that owns the radio directly.
///
/// `CsiNode::run` does this for every mode it dispatches, but an out-of-tree node that drives the
/// radio itself never passes through it, and the policy it sets is process-wide. A board left
/// reporting `None` by one run therefore stayed silent across every later run that did not set
/// it, capturing normally and delivering nothing until a reboot. Nothing upstream reports that: the
/// capture counters climb, and the silence looks like a radio problem.
///
/// So any node that bypasses `CsiNode::run` must state its reporting here. A mode that fixes the
/// attribute states the fixed value — a node that only measures reports, an emitter does not —
/// rather than leaving the previous run's choice in place.
///
/// This is the reporting, not the runtime output toggle: [`set_csi_output_enabled`] is the
/// separate user override, and delivery needs both. Nodes built with [`CsiNode`](crate::CsiNode)
/// set the reporting on their mode's config instead.
pub fn set_reporting(reporting: crate::Reporting) {
    set_runtime_reporting(reporting);
}

/// Whether delivery is open at all: a consumer is installed, the node reports, output is on.
pub(crate) fn publish_open() -> bool {
    CSI_PUBLISH_ENABLED.load(Ordering::Relaxed)
}

/// Apply the reporting policy to a measurement from any source. Threshold scoring needs a raw
/// estimate, so a payload without one passes a threshold policy unscored.
pub(crate) fn policy_admits_frame(frame: &mut crate::wire::CsiFrame) -> bool {
    let kind = POLICY_KIND.load(Ordering::Relaxed);
    if kind == POLICY_REPORT {
        return true;
    }
    let buf: &[i8] = match &frame.payload {
        crate::wire::CsiPayload::EspRaw { bytes, .. } => bytes,
        _ if kind == POLICY_THRESHOLD => return true,
        _ => &[],
    };
    let (keep, score) = policy_admits(buf, embassy_time::Instant::now().as_micros());
    if !keep {
        #[cfg(feature = "statistics")]
        STATS.rx_policy_suppressed.fetch_add(1, Ordering::Relaxed);
        return false;
    }
    if kind == POLICY_THRESHOLD && POLICY_VARIATION.load(Ordering::Relaxed) {
        frame.payload = crate::wire::CsiPayload::Variation { value: score };
    }
    true
}

/// Apply the threshold / decimation policy to a captured buffer. `true` means deliver; for a
/// threshold policy the score is returned too, for a variation-only report.
///
/// Runs in the CSI callback, its only caller, which is what makes the `static mut` state safe.
#[inline]
fn policy_admits(buf: &[i8], now_us: u64) -> (bool, u16) {
    use super::policy::{Decimator, ThresholdState};
    static mut THRESHOLD: ThresholdState = ThresholdState::new();
    static mut DECIMATOR: Decimator = Decimator::new();
    #[allow(static_mut_refs)]
    // SAFETY: only the CSI callback calls this, and it does not re-enter.
    unsafe {
        if POLICY_RESET.swap(false, Ordering::Relaxed) {
            THRESHOLD.reset();
            DECIMATOR.reset();
        }
        match POLICY_KIND.load(Ordering::Relaxed) {
            POLICY_THRESHOLD => {
                let score = THRESHOLD.score(buf);
                let keep = THRESHOLD.decide(
                    score,
                    POLICY_A.load(Ordering::Relaxed),
                    POLICY_B.load(Ordering::Relaxed),
                    now_us,
                );
                (keep, score)
            }
            POLICY_DECIMATE => (DECIMATOR.tick(POLICY_A.load(Ordering::Relaxed)), 0),
            _ => (true, 0),
        }
    }
}

/// Reset the CSI delivery gates. Called by `reset_globals` between runs.
///
/// Closes all CSI delivery gates so any late-firing WiFi callback runs
/// are no-ops. The CSI callback stays registered with esp-radio after
/// stop (the radio itself is still up), but with the gates closed the
/// callback short-circuits before it touches the log channel or the
/// user's callback.
pub(crate) fn reset() {
    CSI_INLINE_LOG_ENABLED.store(false, Ordering::Release);
    USER_PUBLISH_INTENT.store(false, Ordering::Release);
    CSI_PUBLISH_ENABLED.store(false, Ordering::Release);
    // A user preference, not run state: restored rather than carried into the next run, so a node
    // left with output off does not come back deaf.
    CSI_OUTPUT_ENABLED.store(true, Ordering::Relaxed);
    CSI_DELIVERY_MODE.store(CsiDeliveryMode::Off as u8, Ordering::Release);
    CSI_CALLBACK.store(core::ptr::null_mut(), core::sync::atomic::Ordering::Release);
    // The attribution filters are deliberately NOT cleared: they are user configuration set from
    // the CLI before `start`, exactly like the log mode, not per-run state. Clearing them here
    // would silently un-filter the next run.
}

/// Handle for controlling a running [`CsiNode`](crate::CsiNode) from user code.
///
/// CSI packets are delivered to user code via [`set_csi_callback`] (the
/// preferred path: zero channel hops, lowest latency) or by awaiting
/// [`Self::next_csi_packet`] / [`Self::print_csi_w_metadata`], which work with
/// either logging backend (sync or async). The client also signals the running
/// node to stop early via [`Self::send_stop`].
pub struct CsiNodeClient {
    _private: (),
}

impl CsiNodeClient {
    /// Create a new CSI node client.
    ///
    /// Constructing a client does not by itself open the publish gate.
    /// In async-print mode the gate is opened lazily on the first
    /// `get_csi_data()` await; in sync mode it is opened by
    /// `init_logger` or `set_csi_callback`. Use `set_csi_logging_enabled`
    /// to override.
    pub fn new() -> Self {
        Self { _private: () }
    }

    /// Await the next CSI packet captured by the WiFi callback.
    ///
    /// Drains the lock-free `CSI_QUEUE`. Available in **both** sync and
    /// `async-print` modes — same API, same delivery path. Dequeue →
    /// register waker → re-check, which closes the lost-wakeup window.
    ///
    /// The first call lazily switches [`CsiDeliveryMode`] to
    /// [`CsiDeliveryMode::Async`] and opens the master publish gate so
    /// the WiFi callback starts enqueueing. **This replaces any prior
    /// `set_csi_callback`** — the two delivery paths are mutually
    /// exclusive so the WiFi callback only ever runs one of them per
    /// packet (no double-dispatch overhead).
    ///
    /// **Single consumer**: the underlying `AtomicWaker` is single-slot.
    /// Awaiting `next_csi_packet` from two different tasks at once will
    /// cause one of them to miss wake-ups — register exactly one
    /// drainer task per node.
    pub async fn next_csi_packet(&mut self) -> CsiPacket {
        // Conservative lazy init: only flip into `Async` mode if no
        // delivery path is currently active (`Off`). If the user has
        // already set `Callback` mode, we don't disrupt it — the
        // drainer just parks on the waker until the user explicitly
        // switches via `set_csi_delivery_mode(CsiDeliveryMode::Async)`.
        // This lets the two APIs coexist as runtime-toggleable choices
        // without one clobbering the other.
        if CSI_DELIVERY_MODE.load(Ordering::Relaxed) == CsiDeliveryMode::Off as u8 {
            CSI_DELIVERY_MODE.store(CsiDeliveryMode::Async as u8, Ordering::Release);
            USER_PUBLISH_INTENT.store(true, Ordering::Release);
            apply_publish_gate();
        }
        core::future::poll_fn(|cx| {
            if let Some(p) = CSI_QUEUE.dequeue() {
                return core::task::Poll::Ready(p);
            }
            CSI_WAKER.register(cx.waker());
            // Re-check after register to close the lost-wakeup window:
            // the WiFi callback could have enqueued + woken between our
            // first dequeue and `register` if we hadn't checked again.
            if let Some(p) = CSI_QUEUE.dequeue() {
                core::task::Poll::Ready(p)
            } else {
                core::task::Poll::Pending
            }
        })
        .await
    }

    /// Back-compat alias for [`Self::next_csi_packet`]. Older code paths
    /// referred to this name.
    pub async fn get_csi_data(&mut self) -> CsiPacket {
        self.next_csi_packet().await
    }

    /// Receive the next CSI packet and emit it via the crate logging
    /// backend (`log_csi`). Convenience wrapper for "drain + log to
    /// UART/JTAG" loops:
    /// ```ignore
    /// loop { client.print_csi_w_metadata().await; }
    /// ```
    pub async fn print_csi_w_metadata(&mut self) {
        let packet = self.next_csi_packet().await;
        crate::logging::logging::log_csi(packet);
        embassy_futures::yield_now().await;
    }

    /// Signal the running node to stop.
    pub async fn send_stop(&self) {
        STOP_SIGNAL.signal(());
    }
}

impl Default for CsiNodeClient {
    fn default() -> Self {
        Self::new()
    }
}

#[cfg(feature = "esp32c5")]
pub(crate) fn build_csi_config(csi_config: &crate::CsiConfig) -> CsiConfig {
    CsiConfig {
        enable: csi_config.enable,
        acquire_csi_legacy: csi_config.acquire_csi_legacy,
        acquire_csi_force_lltf: csi_config.acquire_csi_force_lltf,
        acquire_csi_ht20: csi_config.acquire_csi_ht20,
        acquire_csi_ht40: csi_config.acquire_csi_ht40,
        acquire_csi_vht: csi_config.acquire_csi_vht,
        acquire_csi_su: csi_config.acquire_csi_su,
        acquire_csi_mu: csi_config.acquire_csi_mu,
        acquire_csi_dcm: csi_config.acquire_csi_dcm,
        acquire_csi_beamformed: csi_config.acquire_csi_beamformed,
        acquire_csi_he_stbc: csi_config.acquire_csi_he_stbc,
        val_scale_cfg: csi_config.val_scale_cfg,
        dump_ack_en: csi_config.dump_ack_en,
        reserved: csi_config.reserved,
    }
}

#[cfg(feature = "esp32c6")]
pub(crate) fn build_csi_config(csi_config: &crate::CsiConfig) -> CsiConfig {
    CsiConfig {
        enable: csi_config.enable,
        acquire_csi_legacy: csi_config.acquire_csi_legacy,
        acquire_csi_ht20: csi_config.acquire_csi_ht20,
        acquire_csi_ht40: csi_config.acquire_csi_ht40,
        acquire_csi_su: csi_config.acquire_csi_su,
        acquire_csi_mu: csi_config.acquire_csi_mu,
        acquire_csi_dcm: csi_config.acquire_csi_dcm,
        acquire_csi_beamformed: csi_config.acquire_csi_beamformed,
        acquire_csi_he_stbc: csi_config.acquire_csi_he_stbc,
        val_scale_cfg: csi_config.val_scale_cfg,
        dump_ack_en: csi_config.dump_ack_en,
        reserved: csi_config.reserved,
    }
}

#[cfg(not(any(feature = "esp32c5", feature = "esp32c6")))]
pub(crate) fn build_csi_config(csi_config: &crate::CsiConfig) -> CsiConfig {
    CsiConfig {
        lltf_en: csi_config.lltf_en,
        htltf_en: csi_config.htltf_en,
        stbc_htltf2_en: csi_config.stbc_htltf2_en,
        ltf_merge_en: csi_config.ltf_merge_en,
        channel_filter_en: csi_config.channel_filter_en,
        manu_scale: csi_config.manu_scale,
        shift: csi_config.shift,
        dump_ack_en: csi_config.dump_ack_en,
    }
}

/// Compensate for esp-radio 1.0.0-beta.1 packing the ESP32-C5's CSI acquisition config in the
/// ESP32-C6's field order.
///
/// The C5's `wifi_csi_acquire_config_t` has 15 bitfield slots — `enable, legacy, force_lltf, ht20,
/// ht40, vht, su, mu, dcm, beamformed, he_stbc_mode, val_scale_cfg, dump_ack_en, lltf_bit_mode,
/// reserved` — but esp-radio's `From<CsiConfig>` passes the C6's 12 fields, in the C6's order:
/// `enable, legacy, ht20, ht40, su, mu, dcm, beamformed, he_stbc, val_scale_cfg, dump_ack_en,
/// reserved`, then three zeros. Every flag after `legacy` lands one or two slots late, and the
/// config's own `acquire_csi_force_lltf` / `acquire_csi_vht` are never read. Uncorrected, a default
/// config forces L-LTF because `ht20` lands in `force_lltf`, an HT-only config falls back to L-LTF
/// for the same reason, and `val_scale_cfg` is always 0.
///
/// So each field is pre-placed where esp-radio will move it to the slot it belongs in. Two slots
/// cannot be reached: `dump_ack_en` and `lltf_bit_mode` always arrive as 0, so ACK frames are not
/// dumped on the C5. Applied last, after any [`RadioProfile`](crate::RadioProfile) has tuned the
/// config by field name. Remove once esp-radio packs the C5 layout correctly.
#[cfg(feature = "esp32c5")]
fn c5_bitfield_workaround(want: CsiConfig) -> CsiConfig {
    CsiConfig {
        enable: want.enable,
        acquire_csi_legacy: want.acquire_csi_legacy,
        acquire_csi_ht20: want.acquire_csi_force_lltf as u32,
        acquire_csi_ht40: want.acquire_csi_ht20,
        acquire_csi_su: want.acquire_csi_ht40,
        acquire_csi_mu: want.acquire_csi_vht as u32,
        acquire_csi_dcm: want.acquire_csi_su,
        acquire_csi_beamformed: want.acquire_csi_mu,
        acquire_csi_he_stbc: want.acquire_csi_dcm,
        val_scale_cfg: want.acquire_csi_beamformed,
        dump_ack_en: want.acquire_csi_he_stbc,
        reserved: want.val_scale_cfg,
        // Not read by esp-radio's C5 conversion.
        acquire_csi_force_lltf: false,
        acquire_csi_vht: false,
    }
}

/// Sets CSI Configuration.
pub(crate) fn set_csi(controller: &mut WifiController, config: CsiConfig) {
    #[cfg(feature = "esp32c5")]
    let config = c5_bitfield_workaround(config);
    // Set CSI Configuration with callback
    controller
        .set_csi(config, |info: esp_radio::wifi::csi::WifiCsiInfo<'_>| {
            capture_csi_info(info);
        })
        .unwrap();
}

// Function to capture CSI info from callback and publish to channel
fn capture_csi_info(info: esp_radio::wifi::csi::WifiCsiInfo<'_>) {
    // Count every CSI report regardless of mode so `rx_count` / `rx_rate_hz`
    // / `pps_rx` reflect actual radio CSI throughput. This is the only path
    // that fires for every receiver capture path — counting here keeps the
    // metric consistent across all of them.
    #[cfg(feature = "statistics")]
    STATS.rx_count.fetch_add(1, Ordering::Relaxed);
    // `cur_bb_format()` only exists on the newer MAC (C5/C6). The classic
    // esp32 / C3 / S3 (`wifi_mac_version = "1"`) radios have no baseband-format
    // field, so the histogram simply stays empty there.
    #[cfg(all(feature = "statistics", any(feature = "esp32c5", feature = "esp32c6")))]
    crate::stats::record_cur_bb_format(info.cur_bb_format() as u32);

    // Raw fast-path (CPU-benchmark DUT): if a raw callback is registered,
    // invoke it and return before the ~640 B CsiPacket build. This makes
    // the per-frame CSI cost just the callback dispatch — the fair match to
    // the ESP-IDF reference's do-nothing `csi_cb`. See `set_csi_raw_callback`.
    let raw_cb_ptr = CSI_RAW_CALLBACK.load(Ordering::Relaxed);
    if !raw_cb_ptr.is_null() {
        let raw_cb: fn() = unsafe { core::mem::transmute::<*mut (), fn()>(raw_cb_ptr) };
        raw_cb();
        return;
    }

    // Single-atomic fast path: returns immediately when no consumer is
    // installed (no logger, callback, or CsiNodeClient subscriber). Building the
    // CsiPacket and calling publish_immediate acquires CriticalSectionRawMutex
    // and on `riscv32imc` every other atomic op also takes a critical section,
    // so additional gate atomics in the hot ISR path delay the Embassy timer ISR.
    if !CSI_PUBLISH_ENABLED.load(Ordering::Relaxed) {
        return;
    }

    // Attribution filter, applied BEFORE the 612-byte packet build and before any formatting, so a
    // rejected frame costs one relaxed load (see `csi_passes_filter`). Counted as an rx drop: the
    // radio did capture it, we chose not to deliver it, and a user comparing `RX Total Packets`
    // against delivered rows needs that difference to be visible rather than unexplained.
    if !csi_passes_filter(&info) {
        #[cfg(feature = "statistics")]
        STATS.rx_filtered.fetch_add(1, Ordering::Relaxed);
        return;
    }

    // Reporting policy (threshold / decimation), judged on the raw buffer so a suppressed frame
    // costs no packet build. `Report` returns at the first branch.
    let (admitted, score) = if POLICY_KIND.load(Ordering::Relaxed) == POLICY_REPORT {
        (true, 0)
    } else {
        policy_admits(info.buf(), embassy_time::Instant::now().as_micros())
    };
    if !admitted {
        #[cfg(feature = "statistics")]
        STATS.rx_policy_suppressed.fetch_add(1, Ordering::Relaxed);
        return;
    }

    // No CS-locked early-drop pre-check: the lock-free `CSI_QUEUE`
    // returns `Err` from `enqueue` when full, so we do drop accounting at
    // the enqueue site below. The 640 B `CsiPacket` build still has
    // to run unconditionally — there's no cheaper way to know if the
    // packet is interesting until it's parsed.

    // Build the packet (copies the CSI buffer, classifies the format). `None`
    // means the payload overflowed the 612-byte cap — do drop accounting here.
    #[allow(unused_mut)]
    let mut csi_packet = match CsiPacket::from_wifi_csi_info(&info) {
        Some(p) => p,
        None => {
            #[cfg(feature = "statistics")]
            record_loss(&STATS.rx_oversize, 1);
            return;
        }
    };
    if POLICY_VARIATION.load(Ordering::Relaxed)
        && POLICY_KIND.load(Ordering::Relaxed) == POLICY_THRESHOLD
    {
        csi_packet.frame.payload = crate::wire::CsiPayload::Variation { value: score };
    }

    deliver(csi_packet);
}

/// The source-independent half of delivery: per-transmitter statistics, then exactly one of the
/// callback / async queue / inline-log paths. Every measurement source ends here — the ESP vendor
/// callback above, and anything pushed through [`emit`](crate::csi::source::emit).
pub(crate) fn deliver(csi_packet: CsiPacket) {
    // Per-transmitter accounting and on-air loss. The C5's receive sequence number is not usable
    // for gap counting, so it is recorded without one there.
    #[cfg(feature = "statistics")]
    if seq_drop_detection_enabled() {
        #[cfg(not(feature = "esp32c5"))]
        let seq = csi_packet.frame.meta.frame_seq;
        #[cfg(feature = "esp32c5")]
        let seq = None;
        let retry = csi_packet.frame.header.map(|h| h.is_retry());
        crate::stats::record_tx_frame(csi_packet.mac(), seq, retry);
    }

    // Single-atomic delivery dispatch. One relaxed load, one branch.
    // Exactly one of Callback / Async / Off runs — the WiFi callback
    // never pays for both a fn-pointer dispatch and a 640 B memcpy on
    // the same packet. See `CsiDeliveryMode` for semantics.
    match CSI_DELIVERY_MODE.load(Ordering::Relaxed) {
        m if m == CsiDeliveryMode::Callback as u8 => {
            // Inline callback: zero-copy `&CsiPacket` borrow.
            let cb_ptr = CSI_CALLBACK.load(core::sync::atomic::Ordering::Relaxed);
            if !cb_ptr.is_null() {
                let cb: fn(&CsiPacket) =
                    unsafe { core::mem::transmute::<*mut (), fn(&CsiPacket)>(cb_ptr) };
                cb(&csi_packet);
            }
            return;
        }
        m if m == CsiDeliveryMode::Async as u8 => {
            // Lock-free MPMC enqueue + wake. No critical section, no
            // IRQ disable — the WiFi-task hot path is never blocked by
            // the user's async drainer.
            match CSI_QUEUE.enqueue(csi_packet) {
                Ok(()) => CSI_WAKER.wake(),
                Err(packet) => {
                    #[cfg(feature = "statistics")]
                    record_loss(&STATS.rx_queue_full, 1);
                    // Make room by discarding the oldest frame. If the consumer drained a slot in
                    // the meantime the retry simply succeeds; if it still fails the new frame is
                    // the one lost, and it was already counted.
                    if OVERFLOW_DROP_OLDEST.load(Ordering::Relaxed) {
                        let _ = CSI_QUEUE.dequeue();
                        if CSI_QUEUE.enqueue(packet).is_ok() {
                            CSI_WAKER.wake();
                        }
                    }
                }
            }
            return;
        }
        _ => {}
    }

    // Off mode: fall through to the inline-log path. In sync mode
    // `log_csi` writes the CSI line directly to UART/JTAG here in the
    // WiFi callback (matches ESP32-CSI-Tool's `_wifi_csi_cb`); in
    // async-print mode it enqueues to the logger backend's own channel
    // (`logging::logging::CSI_CHANNEL`, drained by `logger_backend`).
    // Either way the packet is consumed.
    //
    // `Report { to: Local }` stops here: the measurement was for on-device consumers only, and
    // there is none on this path.
    if CSI_INLINE_LOG_ENABLED.load(Ordering::Relaxed) && !REPORT_LOCAL.load(Ordering::Relaxed) {
        crate::logging::logging::log_csi(csi_packet);
    }
}

/// Internal task that maintains the rate statistics and restarts them when CSI output is toggled.
///
/// Runs for the whole of a capturing run and exits on the stop signal. Seq drop detection runs
/// inside `capture_csi_info` (ISR context), so this task never drains `CSI_QUEUE`, leaving the
/// queue exclusively for [`CsiNodeClient`].
pub async fn run_process_csi_packet() {
    #[cfg(feature = "statistics")]
    STATS
        .capture_start_time
        .store(Instant::now().as_ticks(), Ordering::Relaxed);
    #[cfg(feature = "statistics")]
    let mut last_rate_update = Instant::now();
    #[cfg(feature = "statistics")]
    let mut last_rx_count = STATS.rx_count.load(Ordering::Relaxed);
    #[cfg(feature = "statistics")]
    let mut last_tx_count = STATS.tx_count.load(Ordering::Relaxed);

    loop {
        match select3(
            STOP_SIGNAL.wait(),
            CSI_OUTPUT_CHANGED.wait(),
            Timer::after_millis(500),
        )
        .await
        {
            Either3::First(_) => {
                STOP_SIGNAL.signal(());
                break;
            }
            Either3::Second(_) => {
                CSI_OUTPUT_CHANGED.reset();
                // A runtime output toggle is not a run teardown. Keep
                // CSI delivery gates and callbacks intact; closing them here
                // disables output mid-run until the next CLI `start`.
                #[cfg(feature = "statistics")]
                {
                    STATS
                        .capture_start_time
                        .store(Instant::now().as_ticks(), Ordering::Relaxed);
                    last_rate_update = Instant::now();
                    last_rx_count = STATS.rx_count.load(Ordering::Relaxed);
                    last_tx_count = STATS.tx_count.load(Ordering::Relaxed);
                    RESET_SEQ_TRACKER.store(true, Ordering::Relaxed);
                }
            }
            Either3::Third(_) => {
                #[cfg(feature = "statistics")]
                {
                    let elapsed_secs = last_rate_update.elapsed().as_secs();
                    if elapsed_secs >= 1 {
                        let current_rx = STATS.rx_count.load(Ordering::Relaxed);
                        let current_tx = STATS.tx_count.load(Ordering::Relaxed);

                        let rx_rate =
                            ((current_rx.saturating_sub(last_rx_count)) / elapsed_secs) as u32;
                        let tx_rate =
                            ((current_tx.saturating_sub(last_tx_count)) / elapsed_secs) as u32;

                        STATS.rx_rate_hz.store(rx_rate, Ordering::Relaxed);
                        STATS.tx_rate_hz.store(tx_rate, Ordering::Relaxed);

                        last_rx_count = current_rx;
                        last_tx_count = current_tx;
                        last_rate_update = Instant::now();
                    }
                }
            }
        }
    }
}