ad-core-rs 0.28.1

//! Plugin runtime: control plane (PortActor) + data plane (processing thread).
//!
//! # Single-threaded data plane (intentional, G4)
//!
//! C++ `NDPluginDriver` runs `numThreads` worker threads sharing one input
//! queue (`createCallbackThreads`). The Rust port deliberately runs **exactly
//! one** per-plugin data thread driving a `tokio::select!` loop. This is an
//! intentional design choice: a single owner of the processing state removes
//! the C++ worker-pool races (shared `prevUniqueId_`, sort-buffer contention)
//! and keeps array ordering trivially correct. The `NUM_THREADS` / `MAX_THREADS`
//! PVs are therefore not backed by a real worker pool — instead `NumThreads`
//! is validated and clamped to `[1, MaxThreads]` on write and the clamped
//! value is written back, so the PV is honest about the accepted value rather
//! than silently inert.

// RTEMS-EXEC-MODEL-ALLOW(7): checked, not waived — all 7 ran and passed
// on the exec backend (measured on this tree:
// `EPICS_RS_BUILD_EXEC_BACKEND=thread cargo nextest run -p ad-core-rs
// --all-features`, 345/345). ad-core-rs became a census subject when its
// `build.rs` began deriving `tokio_backend`; nothing here builds a CA
// server, and the reactor these obtain comes from `#[tokio::test]`
// itself, which the backend does not remove.

use std::collections::BTreeMap;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::thread;

use asyn_rs::error::AsynResult;
use asyn_rs::port::{PortDriver, PortDriverBase, PortFlags};
use asyn_rs::runtime::config::RuntimeConfig;
use asyn_rs::runtime::port::{PortRuntimeHandle, create_port_runtime, port_runtime_unavailable};
use asyn_rs::user::AsynUser;
use epics_libcom_rs::runtime::task::{MandatoryThread, StackSizeClass, ThreadPriority};

use asyn_rs::port_handle::PortHandle;

use crate::ndarray::NDArray;
use crate::ndarray_pool::NDArrayPool;
use crate::params::ndarray_driver::NDArrayDriverParams;
use asyn_rs::param::ParamValue;

use super::channel::{
    Admission, ArrayMessage, NDArrayOutput, NDArrayReceiver, NDArraySender, PublishOutcome,
    ndarray_channel,
};
use super::params::PluginBaseParams;
use super::wiring::{WiringRegistry, upstream_key};

/// Message sent through the param channel from control plane to data plane.
///
/// The channel is FIFO, which is what gives [`PluginParamMsg::Barrier`] its
/// meaning: when the data thread acknowledges a barrier, every `Change`
/// enqueued before it has been fully applied (enable flips, wiring rewires,
/// processor param updates).
#[derive(Debug)]
enum PluginParamMsg {
    /// A param write to apply.
    Change(usize, i32, ParamChangeValue),
    /// Sync barrier — acknowledged (best-effort send of `()`) at full
    /// quiescence: every `Change` enqueued before it has been applied (FIFO
    /// channel) AND the array queue has drained. The second condition exists
    /// because arrays travel a separate channel: without it, a param applied
    /// while an older array still waits in the queue would retroactively
    /// change how that array is processed.
    Barrier(std::sync::mpsc::SyncSender<()>),
}

/// Value sent through the param change channel from control plane to data plane.
#[derive(Debug, Clone)]
pub enum ParamChangeValue {
    Int32(i32),
    Float64(f64),
    Octet(String),
}

impl ParamChangeValue {
    pub fn as_i32(&self) -> i32 {
        match self {
            ParamChangeValue::Int32(v) => *v,
            ParamChangeValue::Float64(v) => *v as i32,
            ParamChangeValue::Octet(_) => 0,
        }
    }

    pub fn as_f64(&self) -> f64 {
        match self {
            ParamChangeValue::Int32(v) => *v as f64,
            ParamChangeValue::Float64(v) => *v,
            ParamChangeValue::Octet(_) => 0.0,
        }
    }

    pub fn as_string(&self) -> Option<&str> {
        match self {
            ParamChangeValue::Octet(s) => Some(s),
            _ => None,
        }
    }
}

/// A single parameter update produced by a plugin's process_array.
pub enum ParamUpdate {
    Int32 {
        reason: usize,
        addr: i32,
        value: i32,
    },
    Float64 {
        reason: usize,
        addr: i32,
        value: f64,
    },
    Octet {
        reason: usize,
        addr: i32,
        value: String,
    },
    Float64Array {
        reason: usize,
        addr: i32,
        value: Vec<f64>,
    },
}

impl ParamUpdate {
    /// Create an Int32 update at addr 0.
    pub fn int32(reason: usize, value: i32) -> Self {
        Self::Int32 {
            reason,
            addr: 0,
            value,
        }
    }
    /// Create a Float64 update at addr 0.
    pub fn float64(reason: usize, value: f64) -> Self {
        Self::Float64 {
            reason,
            addr: 0,
            value,
        }
    }
    /// Create an Int32 update at a specific addr.
    pub fn int32_addr(reason: usize, addr: i32, value: i32) -> Self {
        Self::Int32 {
            reason,
            addr,
            value,
        }
    }
    /// Create a Float64 update at a specific addr.
    pub fn float64_addr(reason: usize, addr: i32, value: f64) -> Self {
        Self::Float64 {
            reason,
            addr,
            value,
        }
    }
    /// Create a Float64Array update at addr 0.
    pub fn float64_array(reason: usize, value: Vec<f64>) -> Self {
        Self::Float64Array {
            reason,
            addr: 0,
            value,
        }
    }
    /// Create a Float64Array update at a specific addr.
    pub fn float64_array_addr(reason: usize, addr: i32, value: Vec<f64>) -> Self {
        Self::Float64Array {
            reason,
            addr,
            value,
        }
    }
    /// Create an Octet (string) update at addr 0.
    pub fn octet(reason: usize, value: String) -> Self {
        Self::Octet {
            reason,
            addr: 0,
            value,
        }
    }
    /// Create an Octet (string) update at a specific addr.
    pub fn octet_addr(reason: usize, addr: i32, value: String) -> Self {
        Self::Octet {
            reason,
            addr,
            value,
        }
    }
}

/// Result of processing one array: output arrays + param updates to write back.
pub struct ProcessResult {
    pub output_arrays: Vec<Arc<NDArray>>,
    pub param_updates: Vec<ParamUpdate>,
    /// When `true`, the output arrays are *scattered* — delivered to a single
    /// downstream consumer in round-robin order rather than broadcast to all.
    /// The target consumer (and reroute-past-full / drop-on-last decisions) is
    /// owned by the runtime delivery path, which holds the persistent cursor
    /// (C++ `NDPluginScatter::nextClient_`); the processor only marks the frame
    /// as a scatter frame.
    pub scatter: bool,
}

impl ProcessResult {
    /// Convenience: sink plugin with only param updates, no output arrays.
    pub fn sink(param_updates: Vec<ParamUpdate>) -> Self {
        Self {
            output_arrays: vec![],
            param_updates,
            scatter: false,
        }
    }

    /// Convenience: passthrough/transform plugin with output arrays but no param updates.
    pub fn arrays(output_arrays: Vec<Arc<NDArray>>) -> Self {
        Self {
            output_arrays,
            param_updates: vec![],
            scatter: false,
        }
    }

    /// Convenience: no outputs, no param updates.
    pub fn empty() -> Self {
        Self {
            output_arrays: vec![],
            param_updates: vec![],
            scatter: false,
        }
    }

    /// Convenience: scatter output — deliver to the next downstream consumer in
    /// round-robin order (the runtime owns the cursor and reroute logic).
    pub fn scatter(output_arrays: Vec<Arc<NDArray>>) -> Self {
        Self {
            output_arrays,
            param_updates: vec![],
            scatter: true,
        }
    }
}

/// Result of handling a control-plane param change.
pub struct ParamChangeResult {
    pub output_arrays: Vec<Arc<NDArray>>,
    pub param_updates: Vec<ParamUpdate>,
}

impl ParamChangeResult {
    pub fn updates(param_updates: Vec<ParamUpdate>) -> Self {
        Self {
            output_arrays: vec![],
            param_updates,
        }
    }

    pub fn arrays(output_arrays: Vec<Arc<NDArray>>) -> Self {
        Self {
            output_arrays,
            param_updates: vec![],
        }
    }

    pub fn combined(output_arrays: Vec<Arc<NDArray>>, param_updates: Vec<ParamUpdate>) -> Self {
        Self {
            output_arrays,
            param_updates,
        }
    }

    pub fn empty() -> Self {
        Self {
            output_arrays: vec![],
            param_updates: vec![],
        }
    }
}

/// Pure processing logic.
///
/// `process_array` and `on_param_change` take `&self` and the trait requires
/// `Sync` so that C's `NumThreads` worth of callback threads
/// (`NDPluginDriver.cpp:996-1001`) can run one processor concurrently. A
/// plugin that carries state owns the interior mutability for it — one lock
/// over the whole per-frame state, not one lock per field, so a frame's
/// reads and writes stay atomic against a concurrent frame.
pub trait NDPluginProcess: Send + Sync + 'static {
    /// Process one array. Return output arrays and param updates.
    ///
    /// May run concurrently with itself on `NumThreads` worker threads.
    fn process_array(&self, array: &NDArray, pool: &NDArrayPool) -> ProcessResult;

    /// Plugin type name for PLUGIN_TYPE param.
    fn plugin_type(&self) -> &str;

    /// Whether this plugin can process compressed (`codec != None`) arrays
    /// (C++ `compressionAware_`, G3). Defaults to `false`: a plugin that
    /// operates on raw pixels must not be handed compressed bytes — the
    /// runtime drops compressed input and counts it into DroppedArrays.
    /// A codec/file plugin that understands compressed data overrides this.
    fn compression_aware(&self) -> bool {
        false
    }

    /// Whether this plugin delivers arrays to downstream plugins, i.e. the
    /// initial `NDArrayCallbacks` param value. Defaults to `true`: most plugins
    /// do array callbacks. Terminal plugins that never deliver downstream
    /// (`NDPluginStdArrays`, `NDPluginAttribute`, every `NDPluginFile` writer)
    /// override this to `false` so the param reflects the behaviour, matching C
    /// (e.g. `NDPluginFile.cpp:948` `setIntegerParam(NDArrayCallbacks, 0)`).
    fn does_array_callbacks(&self) -> bool {
        true
    }

    /// Register plugin-specific params on the base. Called once during construction.
    fn register_params(
        &mut self,
        _base: &mut PortDriverBase,
    ) -> Result<(), asyn_rs::error::AsynError> {
        Ok(())
    }

    /// Called when a param changes. Reason is the param index.
    /// Return param updates to be written back to the port driver.
    fn on_param_change(&self, _reason: usize, _params: &PluginParamSnapshot) -> ParamChangeResult {
        ParamChangeResult::empty()
    }

    /// Return a handle to the latest NDArray data for array reads.
    /// Override this in plugins like NDPluginStdArrays that serve pixel data
    /// via readInt8Array/readInt16Array/etc.
    fn array_data_handle(&self) -> Option<Arc<parking_lot::Mutex<Option<Arc<NDArray>>>>> {
        None
    }
}

/// Read-only snapshot of param values available to the processing thread.
pub struct PluginParamSnapshot {
    pub enable_callbacks: bool,
    /// The param reason that changed.
    pub reason: usize,
    /// The address (sub-device) that changed.
    pub addr: i32,
    /// The new value.
    pub value: ParamChangeValue,
}

/// One buffered entry in the sort buffer: the output arrays for a uniqueId
/// plus the instant they were inserted (for the per-element staleness
/// deadline — C++ `sortedListElement::insertionTime_`).
struct SortEntry {
    arrays: Vec<Arc<NDArray>>,
    inserted: std::time::Instant,
}

/// Sort buffer for reordering out-of-order output arrays by uniqueId.
///
/// Port of C++ `sortedNDArrayList_` semantics (NDPluginDriver.cpp).
/// Only arrays that arrive *out of order* are buffered here — in-order
/// arrays are emitted immediately by the caller (B2). The drain logic
/// (`drain_ready`) releases the head while the next-expected uniqueId is
/// contiguous OR the head has been buffered longer than `sort_time` (B3).
struct SortBuffer {
    /// Buffered out-of-order arrays keyed by uniqueId.
    entries: BTreeMap<i32, SortEntry>,
    /// uniqueId of the last array emitted downstream (C++ `prevUniqueId_`).
    ///
    /// Starts at C's sentinel so that "nothing emitted yet" is expressed here
    /// and nowhere else: no real uniqueId is adjacent to it, so the first
    /// output array is out of order by the same rule every later one is
    /// judged by, and no caller needs a separate "is this the first?" branch.
    prev_unique_id: i32,
    /// Whether any array has been emitted yet (C++ `firstOutputArray_`).
    first_output: bool,
    /// Cumulative count of arrays emitted out of order (C++ DisorderedArrays).
    disordered_arrays: i32,
    /// Cumulative count of arrays dropped because the buffer was full
    /// (C++ DroppedOutputArrays — sort-buffer-overflow portion).
    dropped_output_arrays: i32,
}

impl SortBuffer {
    fn new() -> Self {
        Self {
            entries: BTreeMap::new(),
            prev_unique_id: -1000,
            first_output: true,
            disordered_arrays: 0,
            dropped_output_arrays: 0,
        }
    }

    /// True if `unique_id` follows `prev_unique_id` in order (C++ `orderOK`).
    fn order_ok(&self, unique_id: i32) -> bool {
        unique_id == self.prev_unique_id || unique_id == self.prev_unique_id + 1
    }

    /// Record that an array with `unique_id` was emitted downstream.
    /// Updates `prev_unique_id` and counts a disorder if it was out of order.
    fn note_emitted(&mut self, unique_id: i32) {
        if !self.first_output && !self.order_ok(unique_id) {
            self.disordered_arrays += 1;
        }
        self.first_output = false;
        self.prev_unique_id = unique_id;
    }

    /// Insert an out-of-order array into the sort buffer.
    ///
    /// Returns `false` if the buffer was full and the array was dropped
    /// (C++ NDPluginDriver.cpp:307-316), `true` if buffered.
    fn insert(&mut self, unique_id: i32, arrays: Vec<Arc<NDArray>>, sort_size: i32) -> bool {
        if sort_size > 0 && self.entries.len() as i32 >= sort_size {
            self.dropped_output_arrays += 1;
            return false;
        }
        self.entries
            .entry(unique_id)
            .or_insert_with(|| SortEntry {
                arrays: Vec::new(),
                inserted: std::time::Instant::now(),
            })
            .arrays
            .extend(arrays);
        true
    }

    /// Drain the buffer head-first while either the next expected uniqueId is
    /// contiguous OR the head element has aged past `sort_time` seconds.
    /// Port of C++ `sortingTask` loop (NDPluginDriver.cpp:619-670).
    fn drain_ready(&mut self, sort_time: f64) -> Vec<(i32, Vec<Arc<NDArray>>)> {
        let now = std::time::Instant::now();
        let mut out = Vec::new();
        while let Some((&head_id, entry)) = self.entries.iter().next() {
            let delta = now.duration_since(entry.inserted).as_secs_f64();
            let order_ok = self.order_ok(head_id);
            if (!self.first_output && order_ok) || delta > sort_time {
                let entry = self.entries.remove(&head_id).unwrap();
                self.note_emitted(head_id);
                out.push((head_id, entry.arrays));
            } else {
                break;
            }
        }
        out
    }

    /// Drain every buffered array in uniqueId order, regardless of contiguity
    /// or age. Used when sort mode is turned off.
    fn drain_all(&mut self) -> Vec<(i32, Vec<Arc<NDArray>>)> {
        let entries = std::mem::take(&mut self.entries);
        let mut out = Vec::with_capacity(entries.len());
        for (id, entry) in entries {
            self.note_emitted(id);
            out.push((id, entry.arrays));
        }
        out
    }

    /// Number of uniqueId entries currently buffered.
    fn len(&self) -> i32 {
        self.entries.len() as i32
    }
}

/// Shared processor state protected by a mutex, accessible from both
/// the data thread (non-blocking mode) and the caller thread (blocking mode).
struct SharedProcessorInner {
    output: Arc<parking_lot::Mutex<NDArrayOutput>>,
    pool: Arc<NDArrayPool>,
    ndarray_params: NDArrayDriverParams,
    plugin_params: PluginBaseParams,
    port_handle: PortHandle,
    /// ArrayCounter — owned in the param library (C++ `NDArrayCounter`), held
    /// here only as a working copy that is kept in sync with the param so a
    /// control-plane write of `ARRAY_COUNTER` resets it (B12).
    array_counter: i32,
    /// Param index for STD_ARRAY_DATA (if this is a StdArrays plugin).
    std_array_data_param: Option<usize>,
    /// NDArrayCallbacks (C++ `NDArrayCallbacks`): when `false`, the plugin
    /// still processes and updates its metadata params but does NOT deliver the
    /// output array downstream — `endProcessCallbacks` (NDPluginDriver.cpp:
    /// 257-265) returns before the sort/throttle/`doCallbacksGenericPointer`
    /// path. Distinct from `enabled` (`EnableCallbacks`), which gates whether
    /// the plugin processes the input at all.
    array_callbacks: bool,
    /// Sort mode: 0 = disabled, 1 = sorted output.
    sort_mode: i32,
    /// Sort time: seconds — per-element staleness deadline for the sort buffer.
    sort_time: f64,
    /// Sort size: maximum number of uniqueId entries in the sort buffer.
    sort_size: i32,
    /// Sort buffer for reordering output arrays by uniqueId.
    sort_buffer: SortBuffer,
    /// Cumulative count of dropped *input* arrays: a full queue or a
    /// compressed array on a non-aware plugin. NOT a MinCallbackTime throttle,
    /// which C counts nowhere. Shared with every upstream sender and with the
    /// admission gate, which is where both drops are decided.
    dropped_arrays: Arc<std::sync::atomic::AtomicI32>,
    /// Output byte-rate limit (C++ `MaxByteRate`); 0 disables throttling.
    max_byte_rate: f64,
    /// Token-bucket throttler enforcing `max_byte_rate` on the output path (G7).
    throttler: super::throttler::Throttler,
    /// Last *input* array, cached for ProcessPlugin re-injection
    /// (C++ `pPrevInputArray_`, G5). Released on `EnableCallbacks=0` (B6).
    prev_input_array: Option<Arc<NDArray>>,
    /// Last array the plugin handed to `endProcessCallbacks` (C++
    /// `pArrays[0]`), cached whatever NDArrayCallbacks, the MaxByteRate
    /// throttle or the sort buffer then decide about it
    /// (NDPluginDriver.cpp:259-264, :275-278). It is the dimensions template
    /// `NDPoolPreAllocBuffers` allocates from.
    last_output_array: Option<Arc<NDArray>>,
    /// Shadow of `POOL_NUM_PRE_ALLOC_BUFFERS`, the count C reads back inside
    /// `preAllocateBuffers` (asynNDArrayDriver.cpp:723).
    num_pre_alloc_buffers: usize,
    /// Previous array dimensions, for firing an NDDimensions int32-array
    /// callback when dimensions change (C++ `dimsPrev_`, G8).
    dims_prev: Vec<i32>,
    /// Source address selected via the NDArrayAddr PV (C++ `NDArrayAddr`, G6).
    nd_array_addr: i32,
    /// MaxThreads — the clamp ceiling for NumThreads (C++ `MaxThreads`).
    max_threads: i32,
    /// NumThreads — validated/clamped to [1, MaxThreads] on write (G4).
    num_threads: i32,
}

impl SharedProcessorInner {
    /// Byte cost of an array for throttling (C++ `NDPluginDriver::throttled`):
    /// compressed size when a codec is present, else total raw bytes.
    fn array_byte_cost(array: &NDArray) -> f64 {
        match &array.codec {
            Some(c) => c.compressed_size as f64,
            None => array.info().total_bytes as f64,
        }
    }

    /// Apply the output throttle to one array. Returns `true` if the array
    /// should be emitted, `false` if it was dropped (and counts the drop).
    fn throttle_ok(&mut self, array: &NDArray) -> bool {
        if self.max_byte_rate == 0.0 {
            return true;
        }
        let cost = Self::array_byte_cost(array);
        if self.throttler.try_take(cost) {
            true
        } else {
            self.sort_buffer.dropped_output_arrays += 1;
            false
        }
    }

    /// Route output arrays through the throttle, the in-order fast path, and
    /// the sort buffer. Returns arrays ready to emit *now*, in order.
    ///
    /// Port of C++ `endProcessCallbacks` (NDPluginDriver.cpp:295-328): an
    /// array whose uniqueId is contiguous with `prevUniqueId_` is emitted
    /// immediately (B2); only out-of-order arrays enter the sort buffer.
    /// Disordered arrays are counted at emission time in both modes (B4).
    fn route_output_arrays(&mut self, arrays: Vec<Arc<NDArray>>) -> Vec<Arc<NDArray>> {
        let mut ready = Vec::new();
        for arr in arrays {
            if !self.throttle_ok(&arr) {
                continue; // G7: dropped by MaxByteRate throttle
            }
            let uid = arr.unique_id;
            if self.sort_mode != 0 && !self.sort_buffer.order_ok(uid) {
                // Out of order with sort mode on: buffer it (B2/B3).
                self.sort_buffer.insert(uid, vec![arr], self.sort_size);
            } else {
                // In order (or sort mode off): emit immediately, count disorder.
                self.sort_buffer.note_emitted(uid);
                ready.push(arr);
            }
        }
        // After emitting in-order arrays, the sort buffer head may now be
        // contiguous — release any newly-ready run (C++ sortingTask).
        if self.sort_mode != 0 {
            for (_id, mut bucket) in self.sort_buffer.drain_ready(self.sort_time) {
                ready.append(&mut bucket);
            }
        }
        ready
    }

    /// C `pPrevInputArray_`, set in `beginProcessCallbacks`
    /// (NDPluginDriver.cpp:231-234) — so only an array that got past the
    /// admission gate and is about to be processed is ever cached, and
    /// `ProcessPlugin` re-injects the last PROCESSED input rather than the
    /// last one offered.
    ///
    /// The rule for callers: whichever thread is about to run the work calls
    /// this first. That is the callback thread for a queued frame and the
    /// writer's own thread for `blockingCallbacks`, which is exactly the pair
    /// C reaches `processCallbacks` from (`:419-422` vs `processTask`).
    /// Nothing else may write it — an enqueue-time write would name a frame
    /// still waiting in the queue.
    fn cache_input_array(&mut self, array: &Arc<NDArray>) {
        self.prev_input_array = Some(Arc::clone(array));
    }

    /// Process array and return a `ProcessOutput`. Does NOT send to actor.
    /// Direct interrupts (std_array_data_param) happen here (sync).
    /// The returned output must be published and flushed by the caller in async context.
    fn process_and_publish<P: NDPluginProcess>(
        &mut self,
        processor: &P,
        array: &Arc<NDArray>,
    ) -> ProcessOutput {
        self.cache_input_array(array);
        let t0 = std::time::Instant::now();
        let result = processor.process_array(array, &self.pool);
        let elapsed_ms = t0.elapsed().as_secs_f64() * 1000.0;
        self.post_process(array, result, elapsed_ms)
    }

    /// The completion half: route the produced arrays and build the param
    /// batch. Runs on the data loop, never on a callback worker, so the
    /// counters and the sort buffer keep a single owner however many workers
    /// are running.
    fn post_process(
        &mut self,
        array: &Arc<NDArray>,
        result: ProcessResult,
        elapsed_ms: f64,
    ) -> ProcessOutput {
        // C++ endProcessCallbacks (NDPluginDriver.cpp:257-265): when
        // NDArrayCallbacks==0 the method caches the array and returns BEFORE the
        // throttle / sort-admission / `doCallbacksGenericPointer` path. So a
        // non-delivering frame must not enter the MaxByteRate throttle or the
        // sort buffer — only the metadata params (beginProcessCallbacks) are
        // published. Route (throttle + sort) only when delivering.
        //
        // NDPluginStdArrays is the exception: it sets NDArrayCallbacks=0 yet
        // still serves its typed-array waveforms (STD_ARRAY_DATA). Those are
        // NOT the `doCallbacksGenericPointer` downstream path — they fire
        // regardless of NDArrayCallbacks and ARE subject to the MaxByteRate
        // throttle (NDPluginStdArrays.cpp:58 per-interface `throttled()`). So
        // route whenever we deliver downstream OR serve the StdArray waveforms.
        let produced = result.output_arrays.len();
        // C caches `pArrays[0]` on both branches of `endProcessCallbacks` and
        // before it consults the throttle or the sort buffer, so what a client
        // pre-allocates against is the last array the plugin produced, not the
        // last one it delivered.
        if let Some(last) = result.output_arrays.last() {
            self.last_output_array = Some(Arc::clone(last));
        }
        let ready = if self.array_callbacks || self.std_array_data_param.is_some() {
            self.route_output_arrays(result.output_arrays)
        } else {
            Vec::new()
        };
        // A StdArrays frame that produced a waveform which the MaxByteRate
        // throttle then dropped (`produced > 0` but `ready` empty) must not
        // advance ArrayCounter — C nets it back out
        // (NDPluginStdArrays.cpp:202-211).
        let count_frame =
            !(self.std_array_data_param.is_some() && produced > 0 && ready.is_empty());
        let mut output = self.build_publish_batch(
            ready,
            result.param_updates,
            result.scatter,
            Some(array.as_ref()),
            elapsed_ms,
            self.array_callbacks,
            count_frame,
        );
        output.batch.merge(self.build_status_params_batch());
        output
    }

    /// The pool branch of C `asynNDArrayDriver::writeInt32`
    /// (asynNDArrayDriver.cpp:684-694). A plugin port answers it exactly as a
    /// driver port does, because `NDPluginDriver::writeInt32` hands every
    /// parameter below `FIRST_NDPLUGIN_PARAM` straight to the base class
    /// (NDPluginDriver.cpp:690-692) — so `PoolPollStats`, `PoolEmptyFreeList`
    /// and `PoolPreAllocBuffers` are live PVs on a plugin, not decoration.
    ///
    /// Returns the readback batch when `reason` was one of the three, `None`
    /// otherwise so the caller can go on matching.
    fn pool_control_write(&mut self, reason: usize, value: i32) -> Option<ProcessOutput> {
        let p = &self.ndarray_params;
        let (empty, poll, pre_alloc, num_pre_alloc) = (
            p.pool_empty_free_list,
            p.pool_poll_stats,
            p.pool_pre_alloc,
            p.pool_num_pre_alloc_buffers,
        );
        let mut batch = ParamBatch::empty();
        if reason == num_pre_alloc {
            // Shadow the count so `PoolPreAllocBuffers` can read it without a
            // round trip to the param table; not itself a control write.
            self.num_pre_alloc_buffers = value.max(0) as usize;
            return None;
        } else if reason == empty {
            self.pool.empty_free_list();
        } else if reason == pre_alloc {
            // C refuses with asynError when no array has been produced yet —
            // there are no dimensions to allocate against
            // (asynNDArrayDriver.cpp:715-721).
            if let Some(template) = self.last_output_array.clone() {
                let _ = self
                    .pool
                    .pre_allocate_buffers(&template, self.num_pre_alloc_buffers);
            }
            // C clears the command PV once it has run (`:688`).
            batch.addr0.push(asyn_rs::request::ParamSetValue::new(
                pre_alloc,
                0,
                ParamValue::Int32(0),
            ));
        } else if reason != poll {
            return None;
        }
        batch
            .addr0
            .extend(crate::driver::ndarray_driver::pool_stats_values(
                &self.ndarray_params,
                &self.pool,
            ));
        Some(ProcessOutput {
            arrays: vec![],
            scatter: false,
            batch,
        })
    }

    /// A param batch carrying only the current DroppedArrays / queue counters,
    /// used when an array is dropped before processing (B5).
    fn dropped_arrays_only_batch(&self) -> ProcessOutput {
        ProcessOutput {
            arrays: vec![],
            scatter: false,
            batch: self.build_status_params_batch(),
        }
    }

    /// Flush the sort buffer head-first while contiguous or stale (C++
    /// sortingTask periodic tick). Does NOT drain non-contiguous fresh arrays.
    fn tick_sort_buffer(&mut self) -> ProcessOutput {
        let entries = self.sort_buffer.drain_ready(self.sort_time);
        self.emit_drained(entries)
    }

    /// Drain the entire sort buffer in uniqueId order (sort mode turned off).
    fn flush_sort_buffer(&mut self) -> ProcessOutput {
        let entries = self.sort_buffer.drain_all();
        self.emit_drained(entries)
    }

    fn emit_drained(&mut self, entries: Vec<(i32, Vec<Arc<NDArray>>)>) -> ProcessOutput {
        let mut all_arrays = Vec::new();
        let mut combined = ParamBatch::empty();
        for (_unique_id, arrays) in entries {
            // Sort-buffer entries were admitted only while NDArrayCallbacks was
            // on (route_output_arrays runs past the delivery gate); the C++
            // sort thread delivers them regardless of the *current* flag, so
            // they always deliver here.
            let output = self.build_publish_batch(arrays, vec![], false, None, 0.0, true, true);
            all_arrays.extend(output.arrays);
            combined.merge(output.batch);
        }
        combined.merge(self.build_sort_params_batch());
        ProcessOutput {
            arrays: all_arrays,
            scatter: false,
            batch: combined,
        }
    }

    fn build_sort_params_batch(&self) -> ParamBatch {
        use asyn_rs::request::ParamSetValue;
        let sort_free = self.sort_size - self.sort_buffer.len();
        ParamBatch {
            addr0: vec![
                ParamSetValue::new(
                    self.plugin_params.sort_free,
                    0,
                    ParamValue::Int32(sort_free),
                ),
                ParamSetValue::new(
                    self.plugin_params.disordered_arrays,
                    0,
                    ParamValue::Int32(self.sort_buffer.disordered_arrays),
                ),
                ParamSetValue::new(
                    self.plugin_params.dropped_output_arrays,
                    0,
                    ParamValue::Int32(self.sort_buffer.dropped_output_arrays),
                ),
            ],
            extra: std::collections::HashMap::new(),
        }
    }

    /// Build a param batch carrying the runtime status counters:
    /// DroppedArrays (G1) plus the sort/disorder counters.
    /// Put a control-plane write of a mirrored counter back into the working
    /// copy the data path increments.
    ///
    /// C++ holds all four of these in the param library and increments each
    /// with a read-modify-write — `NDArrayCounter` at
    /// `NDPluginDriver.cpp:206-208`, `DroppedArrays` at `:384-389` and
    /// `:436-441`, `DroppedOutputArrays` read at `:268` and stored at
    /// `:291-292` and `:306-307`, `DisorderedArrays` at `:319-321` and
    /// `:650-652` — so a PV write to any of them resets the count and the next
    /// increment continues from the written value. Rust keeps a working copy
    /// per counter for the data path, which makes the param the source of
    /// truth only if every write lands back here.
    ///
    /// This is the one place that does that, so a mirrored counter cannot be
    /// published without its reset path: adding a counter means adding an arm
    /// here (B12 for `ArrayCounter`, B15 for the other three).
    ///
    /// Returns `true` if `reason` named a mirrored counter. `QueueFree` is
    /// deliberately absent — it is recomputed from the queue depth on every
    /// tick in C and in Rust alike, so a write to it is overwritten rather
    /// than retained.
    fn resync_counter(&mut self, reason: usize, value: i32) -> bool {
        if reason == self.ndarray_params.array_counter {
            self.array_counter = value;
        } else if reason == self.plugin_params.dropped_arrays {
            self.dropped_arrays
                .store(value, std::sync::atomic::Ordering::Release);
        } else if reason == self.plugin_params.dropped_output_arrays {
            self.sort_buffer.dropped_output_arrays = value;
        } else if reason == self.plugin_params.disordered_arrays {
            self.sort_buffer.disordered_arrays = value;
        } else {
            return false;
        }
        true
    }

    fn build_status_params_batch(&self) -> ParamBatch {
        use asyn_rs::request::ParamSetValue;
        let mut batch = self.build_sort_params_batch();
        batch.addr0.push(ParamSetValue::new(
            self.plugin_params.dropped_arrays,
            0,
            ParamValue::Int32(
                self.dropped_arrays
                    .load(std::sync::atomic::Ordering::Acquire),
            ),
        ));
        batch
    }

    /// Build a ProcessOutput: fires direct interrupts (sync) and collects
    /// param updates into a batch. Does NOT publish arrays — the caller
    /// must publish them in async context.
    ///
    /// `deliver` is the NDArrayCallbacks gate (C++ `endProcessCallbacks`,
    /// NDPluginDriver.cpp:257-265): when `false`, the downstream array
    /// delivery — the STD_ARRAY_DATA generic-pointer interrupt and the returned
    /// `ProcessOutput.arrays` — is suppressed, while the metadata params from
    /// `beginProcessCallbacks` (counter, dims, datatype, …) are still set.
    fn build_publish_batch(
        &mut self,
        output_arrays: Vec<Arc<NDArray>>,
        param_updates: Vec<ParamUpdate>,
        scatter: bool,
        fallback_array: Option<&NDArray>,
        elapsed_ms: f64,
        deliver: bool,
        count_frame: bool,
    ) -> ProcessOutput {
        use asyn_rs::request::ParamSetValue;

        let mut addr0: Vec<ParamSetValue> = Vec::new();
        let mut extra: std::collections::HashMap<i32, Vec<ParamSetValue>> =
            std::collections::HashMap::new();

        if let Some(report_arr) = output_arrays.first().map(|a| a.as_ref()).or(fallback_array) {
            // A StdArrays frame whose waveform output the MaxByteRate throttle
            // dropped (`count_frame == false`) must not bump ArrayCounter — C
            // decrements it back so clients monitoring ArrayCounter see no new
            // data (NDPluginStdArrays.cpp:202-211).
            if count_frame {
                self.array_counter += 1;
            }

            // Fire the StdArray waveform interrupt directly (C EPICS pattern).
            // This is NDPluginStdArrays' typed-array callback
            // (NDPluginStdArrays.cpp:71-73 `arrayInterruptCallback`), NOT the
            // `doCallbacksGenericPointer` downstream path: C fires it whether or
            // not NDArrayCallbacks is set (StdArrays defaults NDArrayCallbacks=0),
            // so it must NOT be gated by `deliver`. It fires only with the
            // routed/served output (`output_arrays.first()`), never the
            // `fallback_array`: C skips the interface callback on throttle
            // (NDPluginStdArrays.cpp:58), so a throttled frame leaves
            // `output_arrays` empty and serves nothing.
            if let (Some(param), Some(served)) = (
                self.std_array_data_param,
                output_arrays.first().map(|a| a.as_ref()),
            ) {
                use crate::ndarray::NDDataBuffer;
                use asyn_rs::param::ParamValue;
                let value = match &served.data {
                    NDDataBuffer::I8(v) => {
                        Some(ParamValue::Int8Array(std::sync::Arc::from(v.as_slice())))
                    }
                    NDDataBuffer::U8(v) => Some(ParamValue::Int8Array(std::sync::Arc::from(
                        v.iter().map(|&x| x as i8).collect::<Vec<_>>().as_slice(),
                    ))),
                    NDDataBuffer::I16(v) => {
                        Some(ParamValue::Int16Array(std::sync::Arc::from(v.as_slice())))
                    }
                    NDDataBuffer::U16(v) => Some(ParamValue::Int16Array(std::sync::Arc::from(
                        v.iter().map(|&x| x as i16).collect::<Vec<_>>().as_slice(),
                    ))),
                    NDDataBuffer::I32(v) => {
                        Some(ParamValue::Int32Array(std::sync::Arc::from(v.as_slice())))
                    }
                    NDDataBuffer::U32(v) => Some(ParamValue::Int32Array(std::sync::Arc::from(
                        v.iter().map(|&x| x as i32).collect::<Vec<_>>().as_slice(),
                    ))),
                    NDDataBuffer::I64(v) => {
                        Some(ParamValue::Int64Array(std::sync::Arc::from(v.as_slice())))
                    }
                    NDDataBuffer::U64(v) => Some(ParamValue::Int64Array(std::sync::Arc::from(
                        v.iter().map(|&x| x as i64).collect::<Vec<_>>().as_slice(),
                    ))),
                    NDDataBuffer::F32(v) => {
                        Some(ParamValue::Float32Array(std::sync::Arc::from(v.as_slice())))
                    }
                    NDDataBuffer::F64(v) => {
                        Some(ParamValue::Float64Array(std::sync::Arc::from(v.as_slice())))
                    }
                };
                if let Some(value) = value {
                    let ts = served.timestamp.to_system_time();
                    self.port_handle
                        .interrupts()
                        .notify(asyn_rs::interrupt::InterruptValue {
                            reason: param,
                            addr: 0,
                            value,
                            timestamp: ts,
                            uint32_changed_mask: 0,
                            ..Default::default()
                        });
                }
            }

            let info = report_arr.info();
            // B11: read ColorMode / BayerPattern from the NDArray attributes
            // (C++ beginProcessCallbacks). `info()` already resolves the
            // ColorMode attribute when present; fall back to it for the param.
            let color_mode = report_arr
                .attributes
                .get("ColorMode")
                .and_then(|a| a.value.as_i64())
                .map(|v| v as i32)
                .unwrap_or(info.color_mode as i32);
            let bayer_pattern = report_arr
                .attributes
                .get("BayerPattern")
                .and_then(|a| a.value.as_i64())
                .map(|v| v as i32)
                .unwrap_or(0);

            // G8: fire an int32-array callback on NDDimensions when the array
            // dimensions change (C++ beginProcessCallbacks dimsPrev_). C++ keeps
            // a fixed `dimsPrev_[ND_ARRAY_MAX_DIMS]` zero-filled beyond `ndims`,
            // compares element-wise over all 10 slots, and posts the full
            // 10-element array (NDPluginDriver.cpp:220-231) — so a caget reads
            // NORD=10 with trailing zeros, not `ndims`.
            let mut cur_dims = vec![0i32; crate::ndarray::ND_ARRAY_MAX_DIMS];
            for (slot, d) in cur_dims.iter_mut().zip(
                report_arr
                    .dims
                    .iter()
                    .take(crate::ndarray::ND_ARRAY_MAX_DIMS),
            ) {
                *slot = d.size as i32;
            }
            if cur_dims != self.dims_prev {
                self.dims_prev = cur_dims.clone();
                self.port_handle
                    .interrupts()
                    .notify(asyn_rs::interrupt::InterruptValue {
                        reason: self.ndarray_params.array_dimensions,
                        addr: 0,
                        value: asyn_rs::param::ParamValue::Int32Array(std::sync::Arc::from(
                            cur_dims.as_slice(),
                        )),
                        timestamp: report_arr.timestamp.to_system_time(),
                        uint32_changed_mask: 0,
                        ..Default::default()
                    });
            }

            addr0.extend([
                ParamSetValue::new(
                    self.ndarray_params.array_counter,
                    0,
                    ParamValue::Int32(self.array_counter),
                ),
                ParamSetValue::new(
                    self.ndarray_params.unique_id,
                    0,
                    ParamValue::Int32(report_arr.unique_id),
                ),
                ParamSetValue::new(
                    self.ndarray_params.n_dimensions,
                    0,
                    ParamValue::Int32(report_arr.dims.len() as i32),
                ),
                ParamSetValue::new(
                    self.ndarray_params.array_size_x,
                    0,
                    ParamValue::Int32(info.x_size as i32),
                ),
                ParamSetValue::new(
                    self.ndarray_params.array_size_y,
                    0,
                    ParamValue::Int32(info.y_size as i32),
                ),
                ParamSetValue::new(
                    self.ndarray_params.array_size_z,
                    0,
                    ParamValue::Int32(info.color_size as i32),
                ),
                ParamSetValue::new(
                    self.ndarray_params.array_size,
                    0,
                    ParamValue::Int32(info.total_bytes as i32),
                ),
                ParamSetValue::new(
                    self.ndarray_params.data_type,
                    0,
                    ParamValue::Int32(report_arr.data.data_type() as i32),
                ),
                ParamSetValue::new(
                    self.ndarray_params.color_mode,
                    0,
                    ParamValue::Int32(color_mode),
                ),
                ParamSetValue::new(
                    self.ndarray_params.bayer_pattern,
                    0,
                    ParamValue::Int32(bayer_pattern),
                ),
                ParamSetValue::new(
                    self.ndarray_params.timestamp_rbv,
                    0,
                    // C `setDoubleParam(NDTimeStamp, pArray->timeStamp)`
                    // (NDPluginDriver.cpp:217) — the standalone double, which a
                    // driver may set from its own clock; NDEpicsTSSec/nSec below
                    // carry epicsTS (`:218-219`).
                    ParamValue::Float64(report_arr.time_stamp),
                ),
                ParamSetValue::new(
                    self.ndarray_params.epics_ts_sec,
                    0,
                    ParamValue::Int32(report_arr.timestamp.sec as i32),
                ),
                ParamSetValue::new(
                    self.ndarray_params.epics_ts_nsec,
                    0,
                    ParamValue::Int32(report_arr.timestamp.nsec as i32),
                ),
            ]);

            // NDCodec / NDCompressedSize — C++ beginProcessCallbacks
            // (NDPluginDriver.cpp:213-214) sets these on every array. An
            // uncompressed array carries an empty codec name and a
            // compressedSize equal to the raw byte count (matching the
            // driver-base path in ndarray_driver::prepare_array).
            match &report_arr.codec {
                Some(codec) => {
                    addr0.push(ParamSetValue::new(
                        self.ndarray_params.codec,
                        0,
                        ParamValue::Octet(codec.name.as_str().as_bytes().to_vec()),
                    ));
                    addr0.push(ParamSetValue::new(
                        self.ndarray_params.compressed_size,
                        0,
                        ParamValue::Int32(codec.compressed_size as i32),
                    ));
                }
                None => {
                    addr0.push(ParamSetValue::new(
                        self.ndarray_params.codec,
                        0,
                        ParamValue::Octet(Vec::new()),
                    ));
                    addr0.push(ParamSetValue::new(
                        self.ndarray_params.compressed_size,
                        0,
                        ParamValue::Int32(info.total_bytes as i32),
                    ));
                }
            }
        }

        addr0.push(ParamSetValue::new(
            self.plugin_params.execution_time,
            0,
            ParamValue::Float64(elapsed_ms),
        ));

        // ArrayRate_RBV is computed by a calc record in the DB template
        // (SCAN "1 second", reading ArrayCounter_RBV delta), not in Rust.

        // Plugin-specific param updates.
        for update in &param_updates {
            match update {
                ParamUpdate::Int32 {
                    reason,
                    addr,
                    value,
                } => {
                    let pv = ParamSetValue::new(*reason, *addr, ParamValue::Int32(*value));
                    if *addr == 0 {
                        addr0.push(pv);
                    } else {
                        extra.entry(*addr).or_default().push(pv);
                    }
                }
                ParamUpdate::Float64 {
                    reason,
                    addr,
                    value,
                } => {
                    let pv = ParamSetValue::new(*reason, *addr, ParamValue::Float64(*value));
                    if *addr == 0 {
                        addr0.push(pv);
                    } else {
                        extra.entry(*addr).or_default().push(pv);
                    }
                }
                ParamUpdate::Octet {
                    reason,
                    addr,
                    value,
                } => {
                    let pv = ParamSetValue::new(
                        *reason,
                        *addr,
                        ParamValue::Octet(value.clone().into_bytes()),
                    );
                    if *addr == 0 {
                        addr0.push(pv);
                    } else {
                        extra.entry(*addr).or_default().push(pv);
                    }
                }
                ParamUpdate::Float64Array {
                    reason,
                    addr,
                    value,
                } => {
                    let pv = ParamSetValue::new(
                        *reason,
                        *addr,
                        ParamValue::Float64Array(value.clone().into()),
                    );
                    if *addr == 0 {
                        addr0.push(pv);
                    } else {
                        extra.entry(*addr).or_default().push(pv);
                    }
                }
            }
        }

        ProcessOutput {
            // NDArrayCallbacks==0 suppresses downstream delivery (C++
            // endProcessCallbacks early-return); the metadata params above are
            // still published.
            arrays: if deliver { output_arrays } else { Vec::new() },
            scatter,
            batch: ParamBatch { addr0, extra },
        }
    }
}

/// Output from processing: arrays to publish + param batch to flush.
struct ProcessOutput {
    arrays: Vec<Arc<NDArray>>,
    scatter: bool,
    batch: ParamBatch,
}

impl ProcessOutput {
    /// Publish arrays to downstream senders (async, concurrent fan-out).
    ///
    /// Broadcast frames are published to every sender concurrently (independent
    /// backpressure per sender). Scatter frames are routed to a single consumer
    /// via `scatter_publish`, which advances the persistent `scatter_cursor`.
    /// Arrays are published in order — the next array's fan-out starts only
    /// after the previous one completes.
    async fn publish_arrays(&self, senders: &[NDArraySender], scatter_cursor: &mut usize) {
        for arr in &self.arrays {
            if self.scatter {
                Self::scatter_publish(arr, senders, scatter_cursor).await;
            } else {
                let futs = senders.iter().map(|s| s.publish(arr.clone()));
                futures_util::future::join_all(futs).await;
            }
        }
    }

    /// Deliver one array to the next downstream consumer in round-robin order,
    /// rerouting past full queues — a port of C++
    /// `NDPluginScatter::doNDArrayCallbacks` (NDPluginScatter.cpp:59-90).
    ///
    /// `cursor` is the persistent `nextClient_`: it advances by one per
    /// *attempt*, so a frame that reroutes past a full consumer leaves the
    /// cursor pointing just past the consumer it actually delivered to (not
    /// merely one past the starting point). Walking begins at `cursor % n` and
    /// makes at most `n` attempts. A full (or disabled/closed) consumer is
    /// rerouted past unless this is the last attempt; only the last node is
    /// allowed to drop the array (C++ sets `auxStatus=asynSuccess` for the last
    /// node so its full queue drops rather than reroutes). Earlier full
    /// consumers are passed `is_last=false` so the rerouted-away drop is *not*
    /// counted (C++ `ignoreQueueFull`, NDPluginDriver.cpp:406,433-442).
    ///
    /// Routing is over the *enabled* senders only: a downstream with callbacks
    /// disabled is unregistered from the interrupt list in C
    /// (`setArrayInterrupt(0)`) and is therefore not a scatter target, so it
    /// must not consume a round-robin slot.
    async fn scatter_publish(arr: &Arc<NDArray>, senders: &[NDArraySender], cursor: &mut usize) {
        let active: Vec<&NDArraySender> = senders.iter().filter(|s| s.is_enabled()).collect();
        let n = active.len();
        if n == 0 {
            return;
        }
        for attempt in 0..n {
            let target = *cursor % n;
            *cursor = cursor.wrapping_add(1);
            let is_last = attempt == n - 1;
            match active[target].publish_scatter(arr.clone(), is_last).await {
                // Delivered: done. (In blocking mode publish always delivers,
                // so the loop breaks on the first attempt — matching C++ where
                // a blocking scatter calls processCallbacks inline and breaks.)
                PublishOutcome::Delivered => break,
                // Full / disabled / closed: reroute to the next consumer unless
                // this was the last attempt (then the array is dropped — already
                // counted by publish_scatter when is_last).
                PublishOutcome::DroppedQueueFull
                | PublishOutcome::Disabled
                | PublishOutcome::ChannelClosed => {
                    if is_last {
                        break;
                    }
                }
                // C reroutes on `auxStatus` alone (NDPluginScatter.cpp:86-90),
                // which `driverCallback` sets on a refused `trySend` and on
                // nothing else — a compressed or time-throttled array leaves
                // it clear, so the scatter treats the node as served.
                PublishOutcome::DroppedCompressed | PublishOutcome::Throttled => break,
            }
        }
    }
}

/// Collected param updates ready to be flushed to the actor.
/// Produced by `build_publish_batch()`, consumed by async `flush()`.
struct ParamBatch {
    addr0: Vec<asyn_rs::request::ParamSetValue>,
    extra: std::collections::HashMap<i32, Vec<asyn_rs::request::ParamSetValue>>,
}

impl ParamBatch {
    fn empty() -> Self {
        Self {
            addr0: Vec::new(),
            extra: std::collections::HashMap::new(),
        }
    }

    fn merge(&mut self, other: ParamBatch) {
        self.addr0.extend(other.addr0);
        for (addr, updates) in other.extra {
            self.extra.entry(addr).or_default().extend(updates);
        }
    }

    /// Flush via reliable async enqueue. Call from async context.
    async fn flush(self, port: &asyn_rs::port_handle::PortHandle) {
        if !self.addr0.is_empty() {
            if let Err(e) = port.set_params_and_notify(0, self.addr0).await {
                eprintln!("plugin param flush error (addr 0): {e}");
            }
        }
        for (addr, updates) in self.extra {
            if let Err(e) = port.set_params_and_notify(addr, updates).await {
                eprintln!("plugin param flush error (addr {addr}): {e}");
            }
        }
    }
}

/// The `NDArrayAddr`, `maxThreads` and initial `numThreads` that C takes as
/// `NDPluginDriver` constructor arguments (NDPluginDriver.cpp:153, :158, :159).
///
/// The port's `*Configure` entry points take none of them, so the defaults
/// stand in: `maxThreads` floored at 1 as C floors it (`:117`) and published
/// to `MaxThreads` (`:158`), `numThreads` starting at 1 (`:159`). Both are
/// writable at run time, and a `NumThreads` or `QueueSize` write respawns the
/// callback-thread pool exactly as C does (`:730-733`). They are named rather
/// than inlined so the constructor's param block and `SharedProcessorInner`'s
/// initial state cannot drift — publishing one value to the RBV and running on
/// another is what a reader of MaxThreads_RBV would have no way to detect.
const PLUGIN_NDARRAY_ADDR: i32 = 0;
const PLUGIN_MAX_THREADS: i32 = 1;
const PLUGIN_NUM_THREADS: i32 = 1;

/// PortDriver implementation for a plugin's control plane.
#[allow(dead_code)]
pub struct PluginPortDriver {
    base: PortDriverBase,
    ndarray_params: NDArrayDriverParams,
    plugin_params: PluginBaseParams,
    param_change_tx: tokio::sync::mpsc::UnboundedSender<PluginParamMsg>,
    /// Optional handle to the latest NDArray for array read methods (used by StdArrays).
    array_data: Option<Arc<parking_lot::Mutex<Option<Arc<NDArray>>>>>,
    /// Param index for STD_ARRAY_DATA (triggers I/O Intr on ArrayData waveform).
    std_array_data_param: Option<usize>,
}

impl PluginPortDriver {
    fn new<P: NDPluginProcess>(
        port_name: &str,
        plugin_type_name: &str,
        queue_size: usize,
        ndarray_port: &str,
        max_addr: usize,
        param_change_tx: tokio::sync::mpsc::UnboundedSender<PluginParamMsg>,
        processor: &mut P,
        array_data: Option<Arc<parking_lot::Mutex<Option<Arc<NDArray>>>>>,
        pool: &NDArrayPool,
    ) -> AsynResult<Self> {
        let mut base = PortDriverBase::new(
            port_name,
            max_addr,
            PortFlags {
                can_block: true,
                ..Default::default()
            },
        );

        let ndarray_params = NDArrayDriverParams::create(&mut base)?;
        let plugin_params = PluginBaseParams::create(&mut base)?;

        // C++ `NDPluginDriver::NDPluginDriver` (NDPluginDriver.cpp:152-160)
        // initialises exactly these nine read-only / read-back params here,
        // under a comment that states the mechanism: "If a value is not set
        // here then the read request will return an error (uninitialized)".
        // `PluginPortDriver` does not override `read_int32`, so an unset one
        // reaches `get_int32_strict` (asyn-rs/src/port.rs:1663-1665) and the
        // record it feeds sits UDF/INVALID until something else happens to
        // write it — which for MaxThreads_RBV (SCAN "I/O Intr", no PINI, no
        // output partner) is never.
        base.set_string_param(plugin_params.nd_array_port, 0, ndarray_port)?;
        base.set_int32_param(plugin_params.nd_array_addr, 0, PLUGIN_NDARRAY_ADDR)?;
        base.set_int32_param(plugin_params.dropped_arrays, 0, 0)?;
        base.set_int32_param(plugin_params.dropped_output_arrays, 0, 0)?;
        base.set_int32_param(plugin_params.queue_size, 0, queue_size as i32)?;
        // C `:157` — an empty queue has every slot free. The param is spelled
        // QUEUE_FREE (params.rs:40) whatever the struct field is called, so 0
        // here published a permanently full queue on an idle plugin.
        base.set_int32_param(plugin_params.queue_use, 0, queue_size as i32)?;
        base.set_int32_param(plugin_params.max_threads, 0, PLUGIN_MAX_THREADS)?;
        base.set_int32_param(plugin_params.num_threads, 0, PLUGIN_NUM_THREADS)?;
        // C `:160` passes its `blockingCallbacks` constructor argument; the
        // port has no such argument and starts `blocking_mode` false.
        base.set_int32_param(plugin_params.blocking_callbacks, 0, 0)?;

        // EnableCallbacks=0 (Disable) by default, matching EPICS ADCore.
        base.set_int32_param(plugin_params.enable_callbacks, 0, 0)?;
        base.set_string_param(plugin_params.plugin_type, 0, plugin_type_name)?;

        // C++ `NDPluginDriver` derives from `asynNDArrayDriver`, so the base
        // constructor's read-only block (asynNDArrayDriver.cpp:954-1005) runs
        // for every plugin as well.
        crate::driver::ndarray_driver::init_read_only_params(
            &mut base,
            &ndarray_params,
            port_name,
        )?;
        crate::driver::ndarray_driver::refresh_pool_stats(&mut base, &ndarray_params, pool)?;
        // Not in C's block: NDArrayCallbacks is a database parameter there,
        // but a plugin that never emits arrays must publish 0 rather than let
        // the DB turn its output on.
        base.set_int32_param(
            ndarray_params.array_callbacks,
            0,
            processor.does_array_callbacks() as i32,
        )?;
        // Not in C's block either — `NDFullFileName_RBV` is only written by a
        // file plugin after a successful write.
        base.set_string_param(ndarray_params.full_file_name, 0, "")?;
        // Create STD_ARRAY_DATA param for StdArrays plugins (triggers I/O Intr on ArrayData waveform)
        let std_array_data_param = if array_data.is_some() {
            Some(base.create_param("STD_ARRAY_DATA", asyn_rs::param::ParamType::GenericPointer)?)
        } else {
            None
        };

        // Let the processor register its plugin-specific params
        processor.register_params(&mut base)?;

        Ok(Self {
            base,
            ndarray_params,
            plugin_params,
            param_change_tx,
            array_data,
            std_array_data_param,
        })
    }
}

/// Copy source slice directly into destination buffer, returning elements copied.
fn copy_direct<T: Copy>(src: &[T], dst: &mut [T]) -> usize {
    let n = src.len().min(dst.len());
    dst[..n].copy_from_slice(&src[..n]);
    n
}

/// Convert and copy source slice into destination buffer element-by-element.
fn copy_convert<S, D>(src: &[S], dst: &mut [D]) -> usize
where
    S: CastToF64 + Copy,
    D: CastFromF64 + Copy,
{
    let n = src.len().min(dst.len());
    for i in 0..n {
        dst[i] = D::cast_from_f64(src[i].cast_to_f64());
    }
    n
}

/// Cast an integer source element to an integer destination element with C
/// cast semantics. C++ `NDArrayPool::convert` (`NDArrayPool.cpp:387`,
/// `convertType`: `*pDataOut++ = (dataTypeOut)(*pDataIn++)`; and `:465`,
/// `convertDim`) performs a plain C cast between integer types. A C cast:
///   - same-width sign change is a bitwise reinterpret
///     (`(epicsInt8)(epicsUInt8)255 == -1`);
///   - narrowing truncates to the low bits, wrapping
///     (`(epicsInt8)(epicsUInt16)300 == 44`);
///   - widening sign/zero-extends exactly.
///
/// Rust's `as` between integer types implements exactly these semantics. The
/// f64 round-trip in [`copy_convert`] does NOT: it saturates on narrowing
/// (`300.0 as i8 == 127`), diverging from C++. So every integer-source ->
/// integer-target NDArray array read must go through this C-cast path, not
/// `copy_convert`.
trait CCastTo<D> {
    fn ccast(self) -> D;
}
macro_rules! impl_ccast {
    ( $src:ty => $( $dst:ty ),+ ) => {
        $(
            impl CCastTo<$dst> for $src {
                #[inline]
                fn ccast(self) -> $dst {
                    self as $dst
                }
            }
        )+
    };
}
impl_ccast!(i8 => i16, i32, i64);
impl_ccast!(u8 => i8, i16, i32, i64);
impl_ccast!(i16 => i8, i32, i64);
impl_ccast!(u16 => i8, i16, i32, i64);
impl_ccast!(i32 => i8, i16, i64);
impl_ccast!(u32 => i8, i16, i32, i64);
impl_ccast!(i64 => i8, i16, i32);
impl_ccast!(u64 => i8, i16, i32, i64);

/// Copy an integer source slice into an integer destination buffer using C
/// cast semantics (see [`CCastTo`]) — truncating on narrowing, never
/// saturating.
fn copy_ccast<S, D>(src: &[S], dst: &mut [D]) -> usize
where
    S: CCastTo<D> + Copy,
    D: Copy,
{
    let n = src.len().min(dst.len());
    for i in 0..n {
        dst[i] = src[i].ccast();
    }
    n
}

/// Helper trait for `as f64` casts (handles lossy conversions like i64/u64).
trait CastToF64 {
    fn cast_to_f64(self) -> f64;
}

impl CastToF64 for i8 {
    fn cast_to_f64(self) -> f64 {
        self as f64
    }
}
impl CastToF64 for u8 {
    fn cast_to_f64(self) -> f64 {
        self as f64
    }
}
impl CastToF64 for i16 {
    fn cast_to_f64(self) -> f64 {
        self as f64
    }
}
impl CastToF64 for u16 {
    fn cast_to_f64(self) -> f64 {
        self as f64
    }
}
impl CastToF64 for i32 {
    fn cast_to_f64(self) -> f64 {
        self as f64
    }
}
impl CastToF64 for u32 {
    fn cast_to_f64(self) -> f64 {
        self as f64
    }
}
impl CastToF64 for i64 {
    fn cast_to_f64(self) -> f64 {
        self as f64
    }
}
impl CastToF64 for u64 {
    fn cast_to_f64(self) -> f64 {
        self as f64
    }
}
impl CastToF64 for f32 {
    fn cast_to_f64(self) -> f64 {
        self as f64
    }
}
impl CastToF64 for f64 {
    fn cast_to_f64(self) -> f64 {
        self
    }
}

/// Helper trait for `as` casts from f64.
trait CastFromF64 {
    fn cast_from_f64(v: f64) -> Self;
}

impl CastFromF64 for i8 {
    fn cast_from_f64(v: f64) -> Self {
        v as i8
    }
}
impl CastFromF64 for i16 {
    fn cast_from_f64(v: f64) -> Self {
        v as i16
    }
}
impl CastFromF64 for i32 {
    fn cast_from_f64(v: f64) -> Self {
        v as i32
    }
}
impl CastFromF64 for i64 {
    fn cast_from_f64(v: f64) -> Self {
        v as i64
    }
}
impl CastFromF64 for f32 {
    fn cast_from_f64(v: f64) -> Self {
        v as f32
    }
}
impl CastFromF64 for f64 {
    fn cast_from_f64(v: f64) -> Self {
        v
    }
}

/// Copy NDArray data into the output buffer with type conversion.
/// Returns the number of elements copied, or 0 if no data is available.
macro_rules! impl_read_array {
    (
        $self:expr, $buf:expr, $direct_variant:ident,
        ccast: [ $( $ccast_variant:ident ),* ],
        convert: [ $( $variant:ident ),* ]
    ) => {{
        use crate::ndarray::NDDataBuffer;
        let handle = match &$self.array_data {
            Some(h) => h,
            None => return Ok(0),
        };
        let guard = handle.lock();
        let array = match &*guard {
            Some(a) => a,
            None => return Ok(0),
        };
        let n = match &array.data {
            NDDataBuffer::$direct_variant(v) => copy_direct(v, $buf),
            $( NDDataBuffer::$ccast_variant(v) => copy_ccast(v, $buf), )*
            $( NDDataBuffer::$variant(v) => copy_convert(v, $buf), )*
        };
        Ok(n)
    }};
}

impl PortDriver for PluginPortDriver {
    fn base(&self) -> &PortDriverBase {
        &self.base
    }

    fn base_mut(&mut self) -> &mut PortDriverBase {
        &mut self.base
    }

    fn io_write_int32(&mut self, user: &mut AsynUser, value: i32) -> AsynResult<()> {
        let reason = user.reason;
        let addr = user.addr;
        self.base.set_int32_param(reason, addr, value)?;
        self.base.call_param_callbacks(addr)?;
        // B14: reliable send on an unbounded channel — never drop param changes.
        let _ = self.param_change_tx.send(PluginParamMsg::Change(
            reason,
            addr,
            ParamChangeValue::Int32(value),
        ));
        Ok(())
    }

    fn io_write_float64(&mut self, user: &mut AsynUser, value: f64) -> AsynResult<()> {
        let reason = user.reason;
        let addr = user.addr;
        self.base.set_float64_param(reason, addr, value)?;
        self.base.call_param_callbacks(addr)?;
        let _ = self.param_change_tx.send(PluginParamMsg::Change(
            reason,
            addr,
            ParamChangeValue::Float64(value),
        ));
        Ok(())
    }

    fn io_write_octet(&mut self, user: &mut AsynUser, data: &[u8]) -> AsynResult<usize> {
        let reason = user.reason;
        let addr = user.addr;
        let s = String::from_utf8_lossy(data).into_owned();
        self.base.set_string_param(reason, addr, s.clone())?;
        self.base.call_param_callbacks(addr)?;
        let _ = self.param_change_tx.send(PluginParamMsg::Change(
            reason,
            addr,
            ParamChangeValue::Octet(s),
        ));
        Ok(data.len())
    }

    fn read_int8_array(&mut self, _user: &AsynUser, buf: &mut [i8]) -> AsynResult<usize> {
        // Every integer source -> i8 is a C cast (truncating, per C++
        // NDArrayPool.cpp:387); float sources keep the numeric f64 conversion.
        impl_read_array!(
            self, buf, I8,
            ccast: [U8, I16, U16, I32, U32, I64, U64],
            convert: [F32, F64]
        )
    }

    fn read_int16_array(&mut self, _user: &AsynUser, buf: &mut [i16]) -> AsynResult<usize> {
        impl_read_array!(
            self, buf, I16,
            ccast: [I8, U8, U16, I32, U32, I64, U64],
            convert: [F32, F64]
        )
    }

    fn read_int32_array(&mut self, _user: &AsynUser, buf: &mut [i32]) -> AsynResult<usize> {
        impl_read_array!(
            self, buf, I32,
            ccast: [I8, U8, I16, U16, U32, I64, U64],
            convert: [F32, F64]
        )
    }

    fn read_int64_array(&mut self, _user: &AsynUser, buf: &mut [i64]) -> AsynResult<usize> {
        impl_read_array!(
            self, buf, I64,
            ccast: [I8, U8, I16, U16, I32, U32, U64],
            convert: [F32, F64]
        )
    }

    fn read_float32_array(&mut self, _user: &AsynUser, buf: &mut [f32]) -> AsynResult<usize> {
        impl_read_array!(
            self, buf, F32,
            ccast: [],
            convert: [I8, U8, I16, U16, I32, U32, I64, U64, F64]
        )
    }

    fn read_float64_array(&mut self, _user: &AsynUser, buf: &mut [f64]) -> AsynResult<usize> {
        impl_read_array!(
            self, buf, F64,
            ccast: [],
            convert: [I8, U8, I16, U16, I32, U32, I64, U64, F32]
        )
    }
}

/// Handle to a running plugin runtime. Provides access to sender and port handle.
#[derive(Clone)]
pub struct PluginRuntimeHandle {
    port_runtime: PortRuntimeHandle,
    array_sender: NDArraySender,
    array_output: Arc<parking_lot::Mutex<NDArrayOutput>>,
    port_name: String,
    param_tx: tokio::sync::mpsc::UnboundedSender<PluginParamMsg>,
    pub ndarray_params: NDArrayDriverParams,
    pub plugin_params: PluginBaseParams,
}

impl PluginRuntimeHandle {
    pub fn port_runtime(&self) -> &PortRuntimeHandle {
        &self.port_runtime
    }

    pub fn array_sender(&self) -> &NDArraySender {
        &self.array_sender
    }

    pub fn array_output(&self) -> &Arc<parking_lot::Mutex<NDArrayOutput>> {
        &self.array_output
    }

    /// Block until the plugin's data thread has applied every control-plane
    /// param change submitted before this call.
    ///
    /// `write_*_blocking` on the port handle returns once the port actor has
    /// recorded the write and queued it for the data plane; the data thread
    /// applies it (the EnableCallbacks flip, NDArrayPort/NDArrayAddr rewiring,
    /// processor param updates) asynchronously. This is the fence between the
    /// two planes: it enqueues a barrier behind every already-queued change
    /// and waits for the data thread to acknowledge it — the param channel is
    /// FIFO, so the ack implies every earlier change is fully applied.
    ///
    /// The ack additionally waits for the array queue to drain, so it also
    /// implies every array published before this call has been fully handled
    /// (processed or throttled). Under continuous array traffic the ack is
    /// therefore delayed until the queue momentarily empties.
    ///
    /// Returns `false` if the data thread has exited or `timeout` elapsed.
    pub fn wait_params_applied(&self, timeout: std::time::Duration) -> bool {
        let (ack_tx, ack_rx) = std::sync::mpsc::sync_channel(1);
        if self.param_tx.send(PluginParamMsg::Barrier(ack_tx)).is_err() {
            return false;
        }
        ack_rx.recv_timeout(timeout).is_ok()
    }

    /// Apply C's `maxThreads` `*Configure` argument (NDPluginDriver.cpp:117,
    /// `:158`) to an already-built plugin.
    ///
    /// In C this is a constructor argument and nothing else: `writeInt32` has
    /// no `MaxThreads` arm, and `MaxThreads_RBV` is a `longin` with no output
    /// partner (NDPluginBase.template:290-295), so the configure line is an
    /// operator's only route to the ceiling. `create_plugin_runtime*` takes no
    /// such argument, so the configure command applies it here instead.
    ///
    /// It goes through the one owner allowed to assign `max_threads` — the
    /// data loop's `MAX_THREADS` arm, which floors the value, clamps
    /// `NumThreads` to the new ceiling and writes both readbacks back. Seeding
    /// the shared state from a second site instead would let the ceiling the
    /// pool runs on drift from the one `MaxThreads_RBV` publishes, which is
    /// the one disagreement a reader of that record cannot detect.
    ///
    /// Blocks until the data thread has applied it, so the ceiling is in place
    /// before the configure command returns, as it is in C.
    pub fn set_max_threads(&self, max_threads: i32) {
        if self
            .param_tx
            .send(PluginParamMsg::Change(
                self.plugin_params.max_threads,
                0,
                ParamChangeValue::Int32(max_threads),
            ))
            .is_err()
        {
            return;
        }
        let _ = self.wait_params_applied(std::time::Duration::from_secs(5));
    }

    pub fn port_name(&self) -> &str {
        &self.port_name
    }
}

/// Create a plugin runtime with control plane (PortActor) and data plane (processing thread).
///
/// Returns:
/// - `PluginRuntimeHandle` for wiring and control
/// - `PortRuntimeHandle` for param I/O
/// - `JoinHandle` for the data processing thread
pub fn create_plugin_runtime<P: NDPluginProcess>(
    port_name: &str,
    processor: P,
    pool: Arc<NDArrayPool>,
    queue_size: usize,
    ndarray_port: &str,
    wiring: Arc<WiringRegistry>,
) -> (PluginRuntimeHandle, thread::JoinHandle<()>) {
    create_plugin_runtime_multi_addr(
        port_name,
        processor,
        pool,
        queue_size,
        ndarray_port,
        wiring,
        1,
    )
}

/// Create a plugin runtime with multi-addr support.
///
/// `max_addr` specifies the number of addresses (sub-devices) the port supports.
pub fn create_plugin_runtime_multi_addr<P: NDPluginProcess>(
    port_name: &str,
    mut processor: P,
    pool: Arc<NDArrayPool>,
    queue_size: usize,
    ndarray_port: &str,
    wiring: Arc<WiringRegistry>,
    max_addr: usize,
) -> (PluginRuntimeHandle, thread::JoinHandle<()>) {
    // Param change channel (control plane -> data plane)
    // B14: unbounded so control-plane param changes (e.g. autosave restoring
    // hundreds of PVs at IOC init) are never silently dropped before the
    // data plane sees them.
    let (param_tx, param_rx) = tokio::sync::mpsc::unbounded_channel::<PluginParamMsg>();
    let handle_param_tx = param_tx.clone();

    // Capture plugin type and array data handle before mutable borrow
    let plugin_type_name = processor.plugin_type().to_string();
    let compression_aware = processor.compression_aware();
    let does_array_callbacks = processor.does_array_callbacks();
    let array_data = processor.array_data_handle();

    // Create the port driver for control plane
    let driver = PluginPortDriver::new(
        port_name,
        &plugin_type_name,
        queue_size,
        ndarray_port,
        max_addr,
        param_tx,
        &mut processor,
        array_data,
        &pool,
    )
    .expect("failed to create plugin port driver");

    let ndarray_params = driver.ndarray_params;
    let plugin_params = driver.plugin_params;
    let std_array_data_param = driver.std_array_data_param;

    // Create port runtime (actor thread for param I/O).
    //
    // Constructor-shaped, so a failure here is fatal and cannot be anything
    // else: this function hands back the built plugin and has no error channel
    // to its `*Configure` caller. C's equivalent — `asynPortDriver`'s
    // constructor printing and `throw`ing on a failed `registerPort`
    // (asynPortDriver.cpp:4036-4040) — is caught by iocsh
    // (iocsh.cpp:1269-1279) and the script continues by default
    // (iocsh.cpp:995, :1123), leaving the C IOC serving without the port. We
    // deviate on purpose: see `port_runtime_unavailable`.
    let (port_runtime, _actor_jh) = create_port_runtime(driver, RuntimeConfig::default())
        .unwrap_or_else(|e| port_runtime_unavailable(port_name, &e));

    // Clone port handle for the data thread to write params back
    let port_handle = port_runtime.port_handle().clone();

    // Array channel (data plane)
    let (array_sender, array_rx) = ndarray_channel(port_name, queue_size);

    // Shared mode flags
    let enabled = Arc::new(AtomicBool::new(false));
    let blocking_mode = Arc::new(AtomicBool::new(false));

    // Shared processor (accessible from data thread)
    let array_output = Arc::new(parking_lot::Mutex::new(NDArrayOutput::new()));
    let array_output_for_handle = array_output.clone();
    // B13/G6: register this plugin's output so the WiringRegistry is the
    // single source of truth for runtime rewiring (PluginManager::add_plugin
    // would also register it, but direct callers must not bypass the
    // registry). Registered under every address in 0..max_addr so a
    // downstream plugin can select a non-zero NDArrayAddr.
    wiring.register_output_addrs(port_name, max_addr, array_output.clone());
    // G1/B1: the DroppedArrays counter is owned by this plugin and shared with
    // every upstream sender so full-queue drops on our input queue are counted.
    let dropped_arrays_counter = array_sender.dropped_arrays_counter().clone();
    // The pre-queue gate: C runs both halves of `driverCallback` on the
    // producer's thread before `trySend`, so they live with the sender.
    let admission = array_sender.admission().clone();
    admission.set_compression_aware(compression_aware);
    // The processor is shared with the callback threads rather than owned by
    // `SharedProcessorInner`: the whole point of `&self` on `process_array` is
    // that it runs off the data loop, outside this mutex.
    let processor = Arc::new(processor);
    let shared = Arc::new(parking_lot::Mutex::new(SharedProcessorInner {
        output: array_output,
        pool,
        ndarray_params,
        plugin_params,
        port_handle,
        array_counter: 0,
        std_array_data_param,
        // C++ default NDArrayCallbacks = 1 (deliver downstream); terminal
        // plugins (StdArrays/Attribute/File) override `does_array_callbacks` to 0.
        array_callbacks: does_array_callbacks,
        sort_mode: 0,
        sort_time: 0.0,
        sort_size: 10,
        sort_buffer: SortBuffer::new(),
        dropped_arrays: dropped_arrays_counter,
        max_byte_rate: 0.0,
        throttler: super::throttler::Throttler::new(0.0),
        prev_input_array: None,
        last_output_array: None,
        num_pre_alloc_buffers: 0,
        dims_prev: vec![0i32; crate::ndarray::ND_ARRAY_MAX_DIMS],
        nd_array_addr: PLUGIN_NDARRAY_ADDR,
        max_threads: PLUGIN_MAX_THREADS,
        num_threads: PLUGIN_NUM_THREADS,
    }));

    let data_enabled = enabled.clone();
    let data_blocking = blocking_mode.clone();

    let mut array_sender = array_sender;
    array_sender.set_mode_flags(enabled, blocking_mode);
    // The data loop owns queue replacement; the handle is weak so it does not
    // keep the channel open past the last real sender.
    let data_queue_handle = array_sender.self_queue_handle();

    // Capture wiring info for data loop
    let sender_port_name = port_name.to_string();
    let initial_upstream = ndarray_port.to_string();

    // Spawn data processing thread
    let data_jh = MandatoryThread::new(
        format!("plugin-data-{port_name}"),
        // `asynNDArrayDriver.cpp:878` — `if (priority <= 0) priority =
        // epicsThreadPriorityMedium`, and that is what `NDPluginDriver` hands
        // its callback threads (`NDPluginDriver.cpp:1000`).
        ThreadPriority::Medium,
        // `asynNDArrayDriver.cpp:876` — `if (stackSize <= 0) stackSize =
        // epicsThreadGetStackSize(epicsThreadStackMedium)`.
        StackSizeClass::Medium,
    )
    .spawn(move || {
        plugin_data_loop(
            shared,
            processor,
            array_rx,
            data_queue_handle,
            param_rx,
            plugin_params,
            data_enabled,
            data_blocking,
            sender_port_name,
            initial_upstream,
            wiring,
        );
    });

    let handle = PluginRuntimeHandle {
        port_runtime,
        array_sender,
        array_output: array_output_for_handle,
        port_name: port_name.to_string(),
        param_tx: handle_param_tx,
        ndarray_params,
        plugin_params,
    };

    (handle, data_jh)
}

/// Build a param batch reporting the input-queue depth.
///
/// `QUEUE_SIZE` = total capacity, `QUEUE_FREE` = free slots. G2: the param is
/// named `QUEUE_FREE` and the reconciled semantics are *free slots*, matching
/// C++ `NDPluginDriverQueueFree = queueSize - pending()`.
fn queue_status_batch(
    plugin_params: &PluginBaseParams,
    max_capacity: usize,
    free: i32,
) -> ParamBatch {
    use asyn_rs::request::ParamSetValue;
    ParamBatch {
        addr0: vec![
            ParamSetValue::new(
                plugin_params.queue_size,
                0,
                ParamValue::Int32(max_capacity as i32),
            ),
            ParamSetValue::new(plugin_params.queue_use, 0, ParamValue::Int32(free)),
        ],
        extra: std::collections::HashMap::new(),
    }
}

/// Write a validated/clamped int32 value back into the param library so the
/// RBV reflects the accepted value (G4 NumThreads/MaxThreads clamping).
async fn clamp_writeback(port: &PortHandle, reason: usize, value: i32) {
    use asyn_rs::request::ParamSetValue;
    let _ = port
        .set_params_and_notify(
            0,
            vec![ParamSetValue::new(
                reason,
                0,
                asyn_rs::param::ParamValue::Int32(value),
            )],
        )
        .await;
}

/// One array handed to a callback thread.
///
/// The `ArrayMessage` travels with the work because dropping it is what
/// signals completion to a blocking sender and decrements the queued-array
/// counter — that must happen once the frame has been processed and published,
/// not at dispatch.
struct WorkItem {
    msg: ArrayMessage,
}

/// A processed frame on its way back to the data loop.
struct DoneItem {
    msg: ArrayMessage,
    result: ProcessResult,
    elapsed_ms: f64,
}

/// The port of C's `pToThreadMsgQ_` (NDPluginDriver.cpp:985): one queue, N
/// blocking receivers. `tokio::sync::mpsc` is single-consumer and
/// `crossbeam-channel` is not a dependency of this crate, so this is a plain
/// mutex plus condvar. It is unbounded because the real bound is the plugin's
/// input channel: the data loop dispatches at most `numThreads` frames before
/// it stops receiving, which is what leaves `QueueFree` meaning what C means
/// by it.
struct WorkQueue {
    inner: parking_lot::Mutex<WorkQueueInner>,
    ready: parking_lot::Condvar,
}

struct WorkQueueInner {
    items: std::collections::VecDeque<WorkItem>,
    closed: bool,
}

impl WorkQueue {
    fn new() -> Self {
        Self {
            inner: parking_lot::Mutex::new(WorkQueueInner {
                items: std::collections::VecDeque::new(),
                closed: false,
            }),
            ready: parking_lot::Condvar::new(),
        }
    }

    fn push(&self, item: WorkItem) {
        self.inner.lock().items.push_back(item);
        self.ready.notify_one();
    }

    /// Block until an item is available; `None` once the queue is closed and
    /// drained.
    fn pop(&self) -> Option<WorkItem> {
        let mut guard = self.inner.lock();
        loop {
            if let Some(item) = guard.items.pop_front() {
                return Some(item);
            }
            if guard.closed {
                return None;
            }
            self.ready.wait(&mut guard);
        }
    }

    /// C sends one `ToThreadMessageExit` per thread (NDPluginDriver.cpp:1037);
    /// closing the queue wakes all of them at once.
    fn close(&self) {
        self.inner.lock().closed = true;
        self.ready.notify_all();
    }
}

/// The callback threads of C's `createCallbackThreads`
/// (NDPluginDriver.cpp:996-1001): `numThreads` OS threads all receiving from
/// one queue, so a plugin whose `process_array` is slow overlaps frames
/// instead of serialising them.
struct WorkerPool {
    queue: Arc<WorkQueue>,
    threads: Vec<thread::JoinHandle<()>>,
    size: usize,
}

impl WorkerPool {
    /// How many frames may be in flight at once — the dispatch bound the data
    /// loop applies so undispatched arrays stay counted in `QueueFree`.
    fn size(&self) -> usize {
        self.size
    }

    fn dispatch(&self, msg: ArrayMessage) {
        self.queue.push(WorkItem { msg });
    }

    /// Close the queue and join every worker, C's `deleteCallbackThreads`
    /// (NDPluginDriver.cpp:1015-1074).
    ///
    /// A worker never waits on the data loop — it pushes its `DoneItem` into
    /// an unbounded channel and returns — so joining from the loop cannot
    /// deadlock. Frames finished during the join stay in the done channel for
    /// the caller to drain before it swaps in the new pool.
    ///
    /// Idempotent, because `Drop` runs it again on a pool that was respawned
    /// out of the way.
    fn shutdown(&mut self) {
        self.queue.close();
        for t in self.threads.drain(..) {
            let _ = t.join();
        }
    }
}

impl Drop for WorkerPool {
    /// A dropped pool must not leave its threads parked on the queue forever:
    /// the data loop drops it when the plugin's input channel closes.
    fn drop(&mut self) {
        self.shutdown();
    }
}

/// Spawn `num_threads` callback threads over one shared queue.
fn spawn_worker_pool<P: NDPluginProcess>(
    port_name: &str,
    shared: Arc<parking_lot::Mutex<SharedProcessorInner>>,
    processor: Arc<P>,
    array_pool: Arc<NDArrayPool>,
    num_threads: usize,
) -> (WorkerPool, tokio::sync::mpsc::UnboundedReceiver<DoneItem>) {
    let queue = Arc::new(WorkQueue::new());
    let (done_tx, done_rx) = tokio::sync::mpsc::unbounded_channel();
    let size = num_threads.max(1);
    let mut threads = Vec::with_capacity(size);
    for i in 0..size {
        let queue = Arc::clone(&queue);
        let shared = Arc::clone(&shared);
        let processor = Arc::clone(&processor);
        let array_pool = Arc::clone(&array_pool);
        let done_tx = done_tx.clone();
        // C names them `%s_Plugin_%d` from 1 (NDPluginDriver.cpp:997-998) and
        // gives them the driver's stack size and priority (`:1000`).
        let jh = MandatoryThread::new(
            format!("{port_name}_Plugin_{}", i + 1),
            ThreadPriority::Medium,
            StackSizeClass::Medium,
        )
        .spawn(move || {
            while let Some(item) = queue.pop() {
                // C's `beginProcessCallbacks` runs here, on the callback
                // thread at the top of `processCallbacks` — so the cached
                // input names the frame being worked on, not the newest one
                // queued behind it.
                shared.lock().cache_input_array(&item.msg.array);
                let t0 = std::time::Instant::now();
                let result = processor.process_array(&item.msg.array, &array_pool);
                let elapsed_ms = t0.elapsed().as_secs_f64() * 1000.0;
                if done_tx
                    .send(DoneItem {
                        msg: item.msg,
                        result,
                        elapsed_ms,
                    })
                    .is_err()
                {
                    break;
                }
            }
        });
        threads.push(jh);
    }
    (
        WorkerPool {
            queue,
            threads,
            size,
        },
        done_rx,
    )
}

/// Finish a frame a callback thread has processed: fold the result into the
/// shared state and publish it.
///
/// Everything here runs on the data loop, so the counters, the sort buffer and
/// the output senders keep a single owner however many workers are running —
/// which is what C gets from taking the port lock around `processCallbacks`
/// while the plugin releases it for the time-consuming part
/// (NDPluginDriver.cpp:516-517).
async fn complete_frame(
    shared: &Arc<parking_lot::Mutex<SharedProcessorInner>>,
    item: DoneItem,
    scatter_cursor: &mut usize,
) {
    let DoneItem {
        msg,
        result,
        elapsed_ms,
    } = item;
    let (output, senders, port) = {
        let mut guard = shared.lock();
        let output = guard.post_process(&msg.array, result, elapsed_ms);
        let senders = guard.output.lock().senders_clone();
        let port = guard.port_handle.clone();
        (output, senders, port)
    };
    output.publish_arrays(&senders, scatter_cursor).await;
    output.batch.flush(&port).await;
    // msg dropped here → completion signaled (if tracked)
}

fn plugin_data_loop<P: NDPluginProcess>(
    shared: Arc<parking_lot::Mutex<SharedProcessorInner>>,
    processor: Arc<P>,
    mut array_rx: NDArrayReceiver,
    // Lets a QueueSize write replace the queue behind every upstream port's
    // sender. Weak, so holding it cannot stop this loop seeing its own
    // shutdown.
    queue_handle: crate::plugin::channel::SelfQueueHandle,
    mut param_rx: tokio::sync::mpsc::UnboundedReceiver<PluginParamMsg>,
    plugin_params: PluginBaseParams,
    enabled: Arc<AtomicBool>,
    blocking_mode: Arc<AtomicBool>,
    sender_port_name: String,
    initial_upstream: String,
    wiring: Arc<WiringRegistry>,
) {
    let enable_callbacks_reason = plugin_params.enable_callbacks;
    let blocking_callbacks_reason = plugin_params.blocking_callbacks;
    let min_callback_time_reason = plugin_params.min_callback_time;
    let sort_mode_reason = plugin_params.sort_mode;
    let sort_time_reason = plugin_params.sort_time;
    let sort_size_reason = plugin_params.sort_size;
    let nd_array_port_reason = plugin_params.nd_array_port;
    let nd_array_addr_reason = plugin_params.nd_array_addr;
    let process_plugin_reason = plugin_params.process_plugin;
    let max_byte_rate_reason = plugin_params.max_byte_rate;
    let num_threads_reason = plugin_params.num_threads;
    let max_threads_reason = plugin_params.max_threads;
    let queue_size_reason = plugin_params.queue_size;
    let array_callbacks_reason = shared.lock().ndarray_params.array_callbacks;
    // G6: the upstream connection is keyed by (port, addr). `current_upstream`
    // is the base port name; `current_addr` is the selected NDArrayAddr; the
    // effective WiringRegistry key is computed by `upstream_key`.
    let mut current_upstream = initial_upstream;
    let mut current_addr: i32 = 0;
    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();
    rt.block_on(async {
        // Sort flush timer — starts disabled (very long interval).
        // Re-created when sort_time changes.
        let mut sort_flush_interval = tokio::time::interval(std::time::Duration::from_secs(3600));
        let mut sort_flush_active = false;
        // Last published QueueFree value — only flush the queue params when it
        // changes, so a steady queue depth does not spam param callbacks.
        let mut last_queue_free: Option<i32> = None;
        // Persistent scatter cursor (C++ NDPluginScatter::nextClient_): advances
        // per delivery *attempt* across frames so the round-robin survives
        // reroutes past full consumers. One per plugin instance, for its
        // lifetime — matching `nextClient_(1)` set once at construction.
        let mut scatter_cursor: usize = 0;
        // Barriers held until the array queue drains. A barrier acks only at
        // full quiescence — params applied AND no queued arrays — because a
        // param applied while an older array still waits in the queue would
        // retroactively change that array's processing (e.g. a
        // MinCallbackTime reset un-throttling it). See `PluginParamMsg`.
        let mut held_barriers: Vec<std::sync::mpsc::SyncSender<()>> = Vec::new();
        // The callback threads (C++ `createCallbackThreads`,
        // NDPluginDriver.cpp:996-1001) and the frames currently out with them.
        // `in_flight` is the dispatch bound: an array is left in `array_rx`
        // until a worker is free, so `QueueFree` keeps counting the same thing
        // C counts.
        let array_pool = shared.lock().pool.clone();
        let initial_threads = shared.lock().num_threads.max(1) as usize;
        let (mut pool, mut done_rx) = spawn_worker_pool(
            &sender_port_name,
            Arc::clone(&shared),
            Arc::clone(&processor),
            Arc::clone(&array_pool),
            initial_threads,
        );
        let mut in_flight: usize = 0;
        // The plugin's admission gate. Held across queue replacements: it
        // belongs to the plugin, not to the queue in front of it, so a
        // QueueSize write must not reset the MinCallbackTime clock.
        let admission = Arc::clone(array_rx.admission());

        loop {
            // Release held barriers once the array queue is empty AND every
            // dispatched frame has come back — a barrier acks at quiescence,
            // and with N workers running an empty input queue no longer implies
            // the last array is finished.
            if !held_barriers.is_empty() && array_rx.pending() == 0 && in_flight == 0 {
                for ack in held_barriers.drain(..) {
                    let _ = ack.try_send(());
                }
            }
            tokio::select! {
                // Stop receiving once every callback thread is busy: the array
                // then stays queued, which is what C's blocking
                // `pToThreadMsgQ_->receive()` does on each of its N threads.
                msg = array_rx.recv_msg(), if in_flight < pool.size() => {
                    match msg {
                        Some(msg) => {
                            // Note there is no `enabled` check here. In C
                            // `EnableCallbacks=0` cancels the array interrupt
                            // (`:711-720` -> `setArrayInterrupt(0)`) and
                            // touches nothing else, so the callback threads go
                            // on emptying `pToThreadMsgQ_` — which is exactly
                            // what `deleteCallbackThreads` relies on when it
                            // disables the interrupt and then waits for
                            // `pending()` to reach zero (`:1026-1034`). The
                            // flag means "is the interrupt registered", and
                            // the producer is the only place that can answer.
                            //
                            // Every array that reaches here already passed both
                            // halves of C's `driverCallback`
                            // (NDPluginDriver.cpp:383-418): the producer ran
                            // them before `trySend`, which is where C runs
                            // them. Nothing is left to classify.
                            let port = shared.lock().port_handle.clone();
                            // G2: update QueueSize/QueueFree from the channel
                            // depth (C++ NDPluginDriver.cpp:512-513). QueueFree
                            // = max_capacity - pending. Only flush when the
                            // value changed to avoid no-op param callbacks.
                            let max_cap = array_rx.max_capacity();
                            let free = max_cap.saturating_sub(array_rx.pending()) as i32;
                            let queue_batch = if last_queue_free != Some(free) {
                                last_queue_free = Some(free);
                                Some(queue_status_batch(&plugin_params, max_cap, free))
                            } else {
                                None
                            };
                            pool.dispatch(msg);
                            in_flight += 1;
                            if let Some(qb) = queue_batch {
                                qb.flush(&port).await;
                            }
                        }
                        None => break,
                    }
                }
                done = done_rx.recv(), if in_flight > 0 => {
                    match done {
                        Some(item) => {
                            in_flight -= 1;
                            complete_frame(&shared, item, &mut scatter_cursor).await;
                        }
                        // Every worker is gone without a reply: nothing can
                        // come back, so stop waiting on frames that are lost.
                        None => in_flight = 0,
                    }
                }
                // A drop the producer counted. C publishes DroppedArrays from
                // `driverCallback` itself (`callParamCallbacks()` at `:449`),
                // on the producer's thread and whether or not the frame was
                // ever processed; here only the data loop may touch the port,
                // so the producer wakes it instead. Without this a run that
                // drops every array — every array compressed on a non-aware
                // plugin, say — would leave the readback reading 0 forever.
                _ = admission.counted_drop() => {
                    let (out, port) = {
                        let guard = shared.lock();
                        (guard.dropped_arrays_only_batch(), guard.port_handle.clone())
                    };
                    out.batch.flush(&port).await;
                }
                param = param_rx.recv() => {
                    match param {
                        // Barrier: every Change enqueued before it has been
                        // applied by the arms below (FIFO channel). Ack is
                        // deferred to the top-of-loop release, which also
                        // requires the array queue to be drained — a gone
                        // waiter is not an error.
                        Some(PluginParamMsg::Barrier(ack)) => {
                            held_barriers.push(ack);
                        }
                        Some(PluginParamMsg::Change(reason, addr, value)) => {
                            // The base-class pool branch, which a plugin port
                            // inherits (NDPluginDriver.cpp:690-692). Answered
                            // first because none of these is a plugin
                            // parameter and none of the arms below can match.
                            if let Some(out) = {
                                let mut guard = shared.lock();
                                guard.pool_control_write(reason, value.as_i32())
                            } {
                                let port = shared.lock().port_handle.clone();
                                out.batch.flush(&port).await;
                            }
                            if reason == enable_callbacks_reason {
                                let on = value.as_i32() != 0;
                                enabled.store(on, Ordering::Release);
                                // B6: disabling releases the cached input array
                                // (C++ writeInt32 NDPluginDriver.cpp:712-722).
                                if !on {
                                    shared.lock().prev_input_array = None;
                                }
                            }
                            if reason == blocking_callbacks_reason {
                                blocking_mode.store(value.as_i32() != 0, Ordering::Release);
                            }
                            // NDArrayCallbacks gates downstream array delivery
                            // (C++ endProcessCallbacks NDPluginDriver.cpp:
                            // 257-265). Processing still runs; only delivery is
                            // suppressed when 0.
                            if reason == array_callbacks_reason {
                                shared.lock().array_callbacks = value.as_i32() != 0;
                            }
                            // Handle MinCallbackTime param change
                            // The gate is on the producer side, where C reads
                            // MinCallbackTime (`:398`).
                            if reason == min_callback_time_reason {
                                admission.set_min_callback_time(value.as_f64());
                            }
                            // G7: MaxByteRate change resets the output throttler
                            // (C++ writeFloat64 NDPluginDriver.cpp:788-790).
                            if reason == max_byte_rate_reason {
                                let rate = value.as_f64();
                                let mut guard = shared.lock();
                                guard.max_byte_rate = rate;
                                guard.throttler.reset(rate);
                            }
                            // G4: NumThreads / MaxThreads are validated and
                            // clamped on write (C++ `createCallbackThreads`,
                            // NDPluginDriver.cpp:955-971), and the accepted
                            // value is written back so the RBV reports what is
                            // actually running.
                            if reason == max_threads_reason {
                                // Scope the guard so it is released before await.
                                let (port, clamped, mt) = {
                                    let mut guard = shared.lock();
                                    guard.max_threads = value.as_i32().max(1);
                                    let clamped =
                                        guard.num_threads.clamp(1, guard.max_threads);
                                    guard.num_threads = clamped;
                                    (guard.port_handle.clone(), clamped, guard.max_threads)
                                };
                                clamp_writeback(&port, num_threads_reason, clamped).await;
                                clamp_writeback(&port, max_threads_reason, mt).await;
                            }
                            if reason == num_threads_reason {
                                let (port, clamped) = {
                                    let mut guard = shared.lock();
                                    let clamped =
                                        value.as_i32().clamp(1, guard.max_threads.max(1));
                                    guard.num_threads = clamped;
                                    (guard.port_handle.clone(), clamped)
                                };
                                clamp_writeback(&port, num_threads_reason, clamped).await;
                            }
                            // C `writeInt32` (NDPluginDriver.cpp:730-733): a
                            // NumThreads or QueueSize write deletes the
                            // callback threads and creates them again. A
                            // MaxThreads write that clamps NumThreads downward
                            // lands here too, because the pool is respawned
                            // whenever the accepted NumThreads stops matching
                            // it.
                            //
                            // QueueSize respawns the threads too, because C
                            // creates queue and threads together; the queue
                            // itself was already replaced just above.
                            // C's QueueSize write ends in `new
                            // epicsMessageQueue(queueSize, ...)`
                            // (NDPluginDriver.cpp:730-733 -> :985), so the
                            // depth an operator asks for is the depth they
                            // get. Replace the queue first: publishes racing
                            // this land in one queue or the other and none is
                            // refused. Then drain whatever the old one still
                            // holds, so the resize costs no frames — C loses
                            // any that arrive while it waits for the old queue
                            // to empty with the interrupt switched off
                            // (`deleteCallbackThreads`).
                            if reason == queue_size_reason {
                                let want = value.as_i32().max(1) as usize;
                                let mut old_rx = queue_handle
                                    .replace_queue(want)
                                    .map(|new_rx| std::mem::replace(&mut array_rx, new_rx));
                                while let Some(old) = old_rx.as_mut() {
                                    if in_flight < pool.size() {
                                        if let Some(msg) = old.try_recv_msg() {
                                            // Already through the gate when it
                                            // was queued, and C's resize
                                            // processes the old queue to
                                            // empty with the interrupt already
                                            // off (`:1026-1034`) — so neither
                                            // the gate nor the enable flag is
                                            // consulted again here.
                                            pool.dispatch(msg);
                                            in_flight += 1;
                                            continue;
                                        }
                                    }
                                    if in_flight == 0 {
                                        break;
                                    }
                                    match done_rx.recv().await {
                                        Some(item) => {
                                            in_flight -= 1;
                                            complete_frame(
                                                &shared,
                                                item,
                                                &mut scatter_cursor,
                                            )
                                            .await;
                                        }
                                        None => {
                                            in_flight = 0;
                                            break;
                                        }
                                    }
                                }
                                // C republishes QueueFree at the new depth
                                // (`:1008`).
                                last_queue_free = None;
                            }
                            if reason == num_threads_reason
                                || reason == max_threads_reason
                                || reason == queue_size_reason
                            {
                                let want = shared.lock().num_threads.max(1) as usize;
                                if reason == queue_size_reason || want != pool.size() {
                                    let (new_pool, new_done) = spawn_worker_pool(
                                        &sender_port_name,
                                        Arc::clone(&shared),
                                        Arc::clone(&processor),
                                        Arc::clone(&array_pool),
                                        want,
                                    );
                                    let mut old_pool =
                                        std::mem::replace(&mut pool, new_pool);
                                    let mut old_done =
                                        std::mem::replace(&mut done_rx, new_done);
                                    old_pool.shutdown();
                                    // Every worker has exited, so `recv` runs
                                    // out exactly when the last frame the old
                                    // pool finished has been published.
                                    while let Some(item) = old_done.recv().await {
                                        in_flight = in_flight.saturating_sub(1);
                                        complete_frame(&shared, item, &mut scatter_cursor)
                                            .await;
                                    }
                                    // C republishes QueueFree after
                                    // createCallbackThreads (`:1006`).
                                    last_queue_free = None;
                                }
                            }
                            // G6: NDArrayAddr selects a source address of a
                            // multi-address driver — reconnect on change
                            // (C++ writeInt32 NDPluginDriver.cpp:724-728).
                            if reason == nd_array_addr_reason {
                                let new_addr = value.as_i32();
                                if new_addr != current_addr {
                                    let old_key = upstream_key(&current_upstream, current_addr);
                                    let new_key = upstream_key(&current_upstream, new_addr);
                                    shared.lock().nd_array_addr = new_addr;
                                    match wiring.rewire_by_name(
                                        &sender_port_name,
                                        &old_key,
                                        &new_key,
                                    ) {
                                        Ok(()) => current_addr = new_addr,
                                        Err(e) => {
                                            eprintln!("NDArrayAddr reconnect failed: {e}");
                                            shared.lock().nd_array_addr = current_addr;
                                        }
                                    }
                                }
                            }
                            // G5: ProcessPlugin re-injects the cached input
                            // array (C++ writeInt32 NDPluginDriver.cpp:739-746)
                            // by handing it to `driverCallback` (`:741`) — the
                            // same entry point a detector array arrives
                            // through. So it is not a shortcut past the queue:
                            // it is classified and enqueued like any arrival,
                            // waits behind whatever is already queued, is
                            // refused and counted when the queue is full, and
                            // runs on a callback thread. Only `driverCallback`'s
                            // blocking arm (`:419-422`) runs the work on the
                            // writer's own thread, which is the inline branch
                            // below.
                            if reason == process_plugin_reason && value.as_i32() != 0 {
                                let cached = shared.lock().prev_input_array.clone();
                                let blocking = blocking_mode.load(Ordering::Acquire);
                                match cached {
                                    Some(array) if !blocking => {
                                        // `try_enqueue` runs both gates before
                                        // it touches the queue, exactly where
                                        // `driverCallback` runs them, so a
                                        // re-injection inside the
                                        // MinCallbackTime window takes no queue
                                        // slot and a compressed one on a
                                        // non-aware plugin is counted without
                                        // taking one either.
                                        if let Some(
                                            PublishOutcome::DroppedQueueFull
                                            | PublishOutcome::DroppedCompressed,
                                        ) = queue_handle.try_enqueue(array)
                                        {
                                            // The counter the enqueue bumped is
                                            // the plugin's own; only the
                                            // readback is owed (C `:387-390`
                                            // and `:436-441`, then
                                            // `callParamCallbacks` at `:751`).
                                            let (out, port) = {
                                                let guard = shared.lock();
                                                (
                                                    guard.dropped_arrays_only_batch(),
                                                    guard.port_handle.clone(),
                                                )
                                            };
                                            out.batch.flush(&port).await;
                                        }
                                    }
                                    Some(array) => {
                                        // Blocking mode runs the work on the
                                        // writer's own thread, but C decides
                                        // admission BEFORE it reads
                                        // `blockingCallbacks` (`:385-407` vs
                                        // `:419`), so the same two gates apply
                                        // to the inline arm.
                                        match admission.classify(&array) {
                                            Admission::Throttled => {}
                                            Admission::DropCompressed => {
                                                let (out, port) = {
                                                    let guard = shared.lock();
                                                    guard
                                                        .dropped_arrays
                                                        .fetch_add(1, Ordering::AcqRel);
                                                    (
                                                        guard.dropped_arrays_only_batch(),
                                                        guard.port_handle.clone(),
                                                    )
                                                };
                                                out.batch.flush(&port).await;
                                            }
                                            Admission::Admit => {
                                                let (po, senders, port) = {
                                                    let mut guard = shared.lock();
                                                    let output = guard
                                                        .process_and_publish(&*processor, &array);
                                                    let senders =
                                                        guard.output.lock().senders_clone();
                                                    let port = guard.port_handle.clone();
                                                    (output, senders, port)
                                                };
                                                po.publish_arrays(&senders, &mut scatter_cursor)
                                                    .await;
                                                po.batch.flush(&port).await;
                                            }
                                        }
                                    }
                                    None => {
                                        // C parity: NDPluginDriver::writeInt32
                                        // (NDPluginDriver.cpp:743) logs this at
                                        // ASYN_TRACE_WARNING, which is OFF in the
                                        // default port trace mask. Gate through the
                                        // port's trace facility so iocInit stays
                                        // silent unless WARNING is enabled, instead
                                        // of an unconditional eprintln! that spams
                                        // stderr on every PINI ProcessPlugin trigger.
                                        // The asyn port registry that exposes the
                                        // per-port mask only exists with the `ioc`
                                        // integration; a bare plugin build has no
                                        // trace facility to consult, so it stays
                                        // silent there too.
                                        #[cfg(feature = "ioc")]
                                        if let Some(entry) =
                                            asyn_rs::asyn_record::get_port(&sender_port_name)
                                        {
                                            asyn_rs::asyn_trace!(
                                                entry.trace,
                                                sender_port_name.as_str(),
                                                asyn_rs::trace::TraceMask::WARNING,
                                                "plugin {sender_port_name}: ProcessPlugin \
                                                 requested but no input array cached"
                                            );
                                        }
                                    }
                                }
                            }
                            // B12/B15: a control-plane write of any mirrored
                            // counter resets it. `resync_counter` owns the whole
                            // family so none can be published without a reset.
                            shared.lock().resync_counter(reason, value.as_i32());
                            // Handle sort param changes
                            if reason == sort_mode_reason {
                                let mode = value.as_i32();
                                // Scope the guard so clippy can verify the lock
                                // is released before any await.
                                let flush_work = {
                                    let mut guard = shared.lock();
                                    guard.sort_mode = mode;
                                    if mode == 0 {
                                        let output = guard.flush_sort_buffer();
                                        let senders = guard.output.lock().senders_clone();
                                        let port = guard.port_handle.clone();
                                        sort_flush_active = false;
                                        Some((output, senders, port))
                                    } else {
                                        sort_flush_active = guard.sort_time > 0.0;
                                        if sort_flush_active {
                                            let dur = epics_libcom_rs::runtime::time::duration_from_secs(
                                                guard.sort_time,
                                            );
                                            sort_flush_interval = tokio::time::interval(dur);
                                        }
                                        None
                                    }
                                };
                                if let Some((output, senders, port)) = flush_work {
                                    output.publish_arrays(&senders, &mut scatter_cursor).await;
                                    output.batch.flush(&port).await;
                                }
                            }
                            if reason == sort_time_reason {
                                let t = value.as_f64();
                                let mut guard = shared.lock();
                                guard.sort_time = t;
                                if guard.sort_mode != 0 && t > 0.0 {
                                    sort_flush_active = true;
                                    let dur = epics_libcom_rs::runtime::time::duration_from_secs(t);
                                    sort_flush_interval = tokio::time::interval(dur);
                                } else {
                                    sort_flush_active = false;
                                }
                                drop(guard);
                            }
                            if reason == sort_size_reason {
                                shared.lock().sort_size = value.as_i32();
                            }
                            // Handle NDArrayPort rewiring — keyed by (port, addr).
                            if reason == nd_array_port_reason {
                                if let Some(new_port) = value.as_string() {
                                    if new_port != current_upstream {
                                        let old_key =
                                            upstream_key(&current_upstream, current_addr);
                                        let new_key = upstream_key(new_port, current_addr);
                                        match wiring.rewire_by_name(
                                            &sender_port_name,
                                            &old_key,
                                            &new_key,
                                        ) {
                                            Ok(()) => current_upstream = new_port.to_string(),
                                            Err(e) => {
                                                eprintln!("NDArrayPort rewire failed: {e}")
                                            }
                                        }
                                    }
                                }
                            }
                            let snapshot = PluginParamSnapshot {
                                enable_callbacks: enabled.load(Ordering::Acquire),
                                reason,
                                addr,
                                value,
                            };
                            let (process_output, senders, port) = {
                                let mut guard = shared.lock();
                                let t0 = std::time::Instant::now();
                                let result = processor.on_param_change(reason, &snapshot);
                                let elapsed_ms = t0.elapsed().as_secs_f64() * 1000.0;
                                let output = if !result.output_arrays.is_empty() || !result.param_updates.is_empty() {
                                    let deliver = guard.array_callbacks;
                                    Some(guard.build_publish_batch(result.output_arrays, result.param_updates, false, None, elapsed_ms, deliver, true))
                                } else {
                                    None
                                };
                                let senders = guard.output.lock().senders_clone();
                                (output, senders, guard.port_handle.clone())
                            };
                            if let Some(po) = process_output {
                                po.publish_arrays(&senders, &mut scatter_cursor).await;
                                po.batch.flush(&port).await;
                            }
                        }
                        None => break,
                    }
                }
                _ = sort_flush_interval.tick(), if sort_flush_active => {
                    // B3: drain head-first while contiguous or past the
                    // staleness deadline — NOT the whole buffer.
                    let (output, senders, port) = {
                        let mut guard = shared.lock();
                        let output = guard.tick_sort_buffer();
                        let senders = guard.output.lock().senders_clone();
                        let port = guard.port_handle.clone();
                        (output, senders, port)
                    };
                    output.publish_arrays(&senders, &mut scatter_cursor).await;
                    output.batch.flush(&port).await;
                }
            }
        }
    });
}

/// Connect a downstream plugin's sender to a plugin runtime's output.
///
/// B13: the upstream's `array_output` is the same `Arc` that every
/// `create_plugin_runtime*` entry point registers in the `WiringRegistry`, so
/// adding a sender here mutates the registry-tracked output — the registry
/// remains the single source of truth for `rewire_by_name`.
pub fn wire_downstream(upstream: &PluginRuntimeHandle, downstream_sender: NDArraySender) {
    upstream.array_output().lock().add(downstream_sender);
}

/// Create a plugin runtime with a pre-wired output (for testing and direct wiring).
pub fn create_plugin_runtime_with_output<P: NDPluginProcess>(
    port_name: &str,
    mut processor: P,
    pool: Arc<NDArrayPool>,
    queue_size: usize,
    output: NDArrayOutput,
    ndarray_port: &str,
    wiring: Arc<WiringRegistry>,
) -> (PluginRuntimeHandle, thread::JoinHandle<()>) {
    // B14: unbounded so control-plane param changes (e.g. autosave restoring
    // hundreds of PVs at IOC init) are never silently dropped before the
    // data plane sees them.
    let (param_tx, param_rx) = tokio::sync::mpsc::unbounded_channel::<PluginParamMsg>();
    let handle_param_tx = param_tx.clone();

    let plugin_type_name = processor.plugin_type().to_string();
    let compression_aware = processor.compression_aware();
    let does_array_callbacks = processor.does_array_callbacks();
    let array_data = processor.array_data_handle();
    let driver = PluginPortDriver::new(
        port_name,
        &plugin_type_name,
        queue_size,
        ndarray_port,
        1,
        param_tx,
        &mut processor,
        array_data,
        &pool,
    )
    .expect("failed to create plugin port driver");

    let ndarray_params = driver.ndarray_params;
    let plugin_params = driver.plugin_params;
    let std_array_data_param = driver.std_array_data_param;

    // Fatal for the same reason as `create_plugin_runtime_multi_addr` above:
    // a constructor-shaped creator has nowhere to report to, and the only
    // alternative is a handle to a port that does not exist.
    let (port_runtime, _actor_jh) = create_port_runtime(driver, RuntimeConfig::default())
        .unwrap_or_else(|e| port_runtime_unavailable(port_name, &e));

    let port_handle = port_runtime.port_handle().clone();

    let (array_sender, array_rx) = ndarray_channel(port_name, queue_size);

    let enabled = Arc::new(AtomicBool::new(false));
    let blocking_mode = Arc::new(AtomicBool::new(false));

    let array_output = Arc::new(parking_lot::Mutex::new(output));
    let array_output_for_handle = array_output.clone();
    // B13: register this plugin's output so the WiringRegistry is the single
    // source of truth — an output created via this entry point is otherwise
    // invisible to runtime rewiring.
    wiring.register_output(port_name, array_output.clone());
    // G1/B1: DroppedArrays counter shared with upstream senders.
    let dropped_arrays_counter = array_sender.dropped_arrays_counter().clone();
    // The pre-queue gate: C runs both halves of `driverCallback` on the
    // producer's thread before `trySend`, so they live with the sender.
    let admission = array_sender.admission().clone();
    admission.set_compression_aware(compression_aware);
    // The processor is shared with the callback threads rather than owned by
    // `SharedProcessorInner`: the whole point of `&self` on `process_array` is
    // that it runs off the data loop, outside this mutex.
    let processor = Arc::new(processor);
    let shared = Arc::new(parking_lot::Mutex::new(SharedProcessorInner {
        output: array_output,
        pool,
        ndarray_params,
        plugin_params,
        port_handle,
        array_counter: 0,
        std_array_data_param,
        // C++ default NDArrayCallbacks = 1 (deliver downstream); terminal
        // plugins (StdArrays/Attribute/File) override `does_array_callbacks` to 0.
        array_callbacks: does_array_callbacks,
        sort_mode: 0,
        sort_time: 0.0,
        sort_size: 10,
        sort_buffer: SortBuffer::new(),
        dropped_arrays: dropped_arrays_counter,
        max_byte_rate: 0.0,
        throttler: super::throttler::Throttler::new(0.0),
        prev_input_array: None,
        last_output_array: None,
        num_pre_alloc_buffers: 0,
        dims_prev: vec![0i32; crate::ndarray::ND_ARRAY_MAX_DIMS],
        nd_array_addr: PLUGIN_NDARRAY_ADDR,
        max_threads: PLUGIN_MAX_THREADS,
        num_threads: PLUGIN_NUM_THREADS,
    }));

    let data_enabled = enabled.clone();
    let data_blocking = blocking_mode.clone();

    let mut array_sender = array_sender;
    array_sender.set_mode_flags(enabled, blocking_mode);
    // The data loop owns queue replacement; the handle is weak so it does not
    // keep the channel open past the last real sender.
    let data_queue_handle = array_sender.self_queue_handle();

    // Capture wiring info for data loop
    let sender_port_name = port_name.to_string();
    let initial_upstream = ndarray_port.to_string();

    let data_jh = MandatoryThread::new(
        format!("plugin-data-{port_name}"),
        // `asynNDArrayDriver.cpp:878` — `if (priority <= 0) priority =
        // epicsThreadPriorityMedium`, and that is what `NDPluginDriver` hands
        // its callback threads (`NDPluginDriver.cpp:1000`).
        ThreadPriority::Medium,
        // `asynNDArrayDriver.cpp:876` — `if (stackSize <= 0) stackSize =
        // epicsThreadGetStackSize(epicsThreadStackMedium)`.
        StackSizeClass::Medium,
    )
    .spawn(move || {
        plugin_data_loop(
            shared,
            processor,
            array_rx,
            data_queue_handle,
            param_rx,
            plugin_params,
            data_enabled,
            data_blocking,
            sender_port_name,
            initial_upstream,
            wiring,
        );
    });

    let handle = PluginRuntimeHandle {
        port_runtime,
        array_sender,
        array_output: array_output_for_handle,
        port_name: port_name.to_string(),
        param_tx: handle_param_tx,
        ndarray_params,
        plugin_params,
    };

    (handle, data_jh)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::ndarray::{NDDataType, NDDimension};
    use crate::plugin::channel::ndarray_channel;

    /// # Invariant
    ///
    /// MUST: every `plugin-data-*` thread and every `<port>_Plugin_<n>`
    /// callback thread be created through [`MandatoryThread`], so that a
    /// thread the plugin cannot process without takes the process down rather
    /// than the caller's thread.
    ///
    /// The reason this is not the `errlog-and-continue` class:
    /// `NDPluginDriver::createCallbackThreads` builds its workers as
    /// `new epicsThread(...)` (`NDPluginDriver.cpp:1000`), whose constructor
    /// calls `epicsThreadCreateOpt` and `throw unableToCreateThread()` on
    /// failure (`epicsThread.cpp:214-220`) — a thrown failure, not a status
    /// code the plugin inspects and carries on from.
    ///
    /// Where C ends up is **not** where we do, and the difference is
    /// deliberate: iocsh catches whatever a command throws
    /// (`iocsh.cpp:1269-1279`, `"C++ error: ..."`) and a startup script's
    /// default `on error` is `Continue` (`iocsh.cpp:995`, `:1123`), so C runs
    /// the rest of st.cmd with the plugin's port registered and its worker
    /// threads absent — arrays queue to it and are never processed, silently,
    /// for the life of the IOC. `MandatoryThread::spawn` refuses that state.
    /// The `.expect` it replaced reached neither: on a `panic = "unwind"`
    /// target it unwound one thread and left the same zombie plugin behind.
    ///
    /// Contrast the auxiliary AD threads, which genuinely do errlog-and-continue
    /// and have no site here: the sorting thread (`NDPluginDriver.cpp:1089-1098`,
    /// `asynPrint` + `return asynError`), the queued-array counter
    /// (`asynNDArrayDriver.cpp:1013-1021`, `asynPrint` and no error at all) and
    /// the HDF5 flush task (`NDFileHDF5.cpp:2421-2429`, `printf` + `return`).
    ///
    /// Source inspection, because the defect is a call that is *absent*.
    #[test]
    fn plugin_data_threads_are_mandatory() {
        let prod =
            source_guard::production(include_str!("runtime.rs"), source_guard::Comments::Strip);
        assert_eq!(
            prod.matches("MandatoryThread::new(").count(),
            3,
            "`create_plugin_runtime_multi_addr`, `create_plugin_runtime_with_output` \
             and `spawn_worker_pool`"
        );
        let bare = concat!("thread", "::Builder::new()");
        let strays: Vec<&str> = prod
            .lines()
            .map(str::trim)
            .filter(|l| l.contains(bare) || l.contains(concat!("thread", "::spawn(")))
            .collect();
        assert!(
            strays.is_empty(),
            "a plugin data thread created outside `MandatoryThread` resolves its \
             own spawn failure locally: {strays:?}"
        );
    }

    /// Passthrough processor: returns the input array as-is.
    struct PassthroughProcessor;

    impl NDPluginProcess for PassthroughProcessor {
        fn process_array(&self, array: &NDArray, _pool: &NDArrayPool) -> ProcessResult {
            ProcessResult::arrays(vec![Arc::new(array.clone())])
        }
        fn plugin_type(&self) -> &str {
            "Passthrough"
        }
    }

    /// Sink processor: consumes arrays, returns nothing.
    struct SinkProcessor {
        count: std::sync::atomic::AtomicUsize,
    }

    impl NDPluginProcess for SinkProcessor {
        fn process_array(&self, _array: &NDArray, _pool: &NDArrayPool) -> ProcessResult {
            self.count
                .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
            ProcessResult::empty()
        }
        fn plugin_type(&self) -> &str {
            "Sink"
        }
    }

    fn make_test_array(id: i32) -> Arc<NDArray> {
        let mut arr = NDArray::new(vec![NDDimension::new(4)], NDDataType::UInt8);
        arr.unique_id = id;
        Arc::new(arr)
    }

    fn test_wiring() -> Arc<WiringRegistry> {
        Arc::new(WiringRegistry::new())
    }

    /// Fence: wait until the data thread has applied every param change
    /// submitted so far. `write_*_blocking` only guarantees the change is
    /// queued for the data plane; asserting on data-plane behaviour without
    /// this fence is a race.
    fn params_applied(handle: &PluginRuntimeHandle) {
        assert!(
            handle.wait_params_applied(std::time::Duration::from_secs(10)),
            "data thread did not apply queued param changes"
        );
    }

    /// Fence for tests that deliberately keep the input queue occupied.
    ///
    /// `params_applied` cannot serve there: its barrier also waits for the
    /// array queue to drain, which is the very state these tests hold. The
    /// param channel is FIFO, so a later write whose *accepted* value only the
    /// data thread can produce is enough — NumThreads is clamped to
    /// `[1, MaxThreads]` and written back (NDPluginDriver.cpp:955-971), so an
    /// out-of-range write reading back as the clamp proves every earlier param
    /// message has been applied.
    ///
    /// Timing out here is itself the finding these tests exist for: the data
    /// thread is the one that applies params, so a param that never lands
    /// means the loop is stuck inside work a callback thread should be doing.
    fn param_fifo_drained(handle: &PluginRuntimeHandle) {
        let port = handle.port_runtime().port_handle();
        port.write_int32_blocking(handle.plugin_params.num_threads, 0, 99)
            .unwrap();
        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(2);
        while port
            .read_int32_blocking(handle.plugin_params.num_threads, 0)
            .unwrap()
            != 1
        {
            assert!(
                std::time::Instant::now() < deadline,
                "the data thread stopped applying param writes — it is busy \
                 inside work a callback thread should be doing"
            );
            std::thread::sleep(std::time::Duration::from_millis(1));
        }
    }

    /// Poll `cond` until it holds; panic after 10 s. Waits on the observable
    /// state itself instead of sleeping a guessed duration, so a loaded
    /// machine cannot flake the test.
    fn wait_until(what: &str, mut cond: impl FnMut() -> bool) {
        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10);
        while !cond() {
            assert!(
                std::time::Instant::now() < deadline,
                "timed out waiting for {what}"
            );
            std::thread::sleep(std::time::Duration::from_millis(2));
        }
    }

    /// Enable callbacks on a plugin handle (plugins default to disabled) and
    /// fence until the data thread has actually flipped the enable flag.
    fn enable_callbacks(handle: &PluginRuntimeHandle) {
        handle
            .port_runtime()
            .port_handle()
            .write_int32_blocking(handle.plugin_params.enable_callbacks, 0, 1)
            .unwrap();
        params_applied(handle);
    }

    /// Send an array via the sender from a sync test context.
    /// Uses a dedicated thread with a current-thread runtime to avoid
    /// interfering with the plugin's own runtime.
    fn send_array(sender: &NDArraySender, array: Arc<NDArray>) {
        let sender = sender.clone();
        let jh = std::thread::spawn(move || {
            let rt = tokio::runtime::Builder::new_current_thread()
                .enable_all()
                .build()
                .unwrap();
            rt.block_on(sender.publish(array));
        });
        jh.join().unwrap();
    }

    #[test]
    fn test_passthrough_runtime() {
        let pool = Arc::new(NDArrayPool::new(1_000_000));

        // Create downstream receiver
        let (downstream_sender, mut downstream_rx) = ndarray_channel("DOWNSTREAM", 10);
        let mut output = NDArrayOutput::new();
        output.add(downstream_sender);

        let (handle, _data_jh) = create_plugin_runtime_with_output(
            "PASS1",
            PassthroughProcessor,
            pool,
            10,
            output,
            "",
            test_wiring(),
        );
        enable_callbacks(&handle);

        // Send an array
        send_array(handle.array_sender(), make_test_array(42));

        // Should come out the other side
        let received = downstream_rx.blocking_recv().unwrap();
        assert_eq!(received.unique_id, 42);
    }

    #[test]
    fn test_sink_runtime() {
        let pool = Arc::new(NDArrayPool::new(1_000_000));

        let (handle, _data_jh) = create_plugin_runtime(
            "SINK1",
            SinkProcessor {
                count: std::sync::atomic::AtomicUsize::new(0),
            },
            pool,
            10,
            "",
            test_wiring(),
        );
        enable_callbacks(&handle);

        // Send arrays - they should be consumed silently
        send_array(handle.array_sender(), make_test_array(1));
        send_array(handle.array_sender(), make_test_array(2));

        // ArrayCounter advances once per processed frame — both consumed.
        let port = handle.port_runtime().port_handle().clone();
        let counter = handle.ndarray_params.array_counter;
        wait_until("sink to process both arrays", || {
            port.read_int32_blocking(counter, 0).is_ok_and(|v| v == 2)
        });
        assert_eq!(handle.port_name(), "SINK1");
    }

    #[test]
    fn test_plugin_type_param() {
        let pool = Arc::new(NDArrayPool::new(1_000_000));

        let (handle, _data_jh) = create_plugin_runtime(
            "TYPE_TEST",
            PassthroughProcessor,
            pool,
            10,
            "",
            test_wiring(),
        );

        // Verify port name
        assert_eq!(handle.port_name(), "TYPE_TEST");
        assert_eq!(handle.port_runtime().port_name(), "TYPE_TEST");
    }

    #[test]
    fn test_ndtimestamp_param_is_the_standalone_double() {
        // R8-66 family: C `setDoubleParam(NDTimeStamp, pArray->timeStamp)`
        // (NDPluginDriver.cpp:217) publishes the array's standalone double —
        // which a driver with a hardware clock sets independently of epicsTS —
        // while NDEpicsTSSec/nSec carry epicsTS (`:218-219`). The plugin runtime
        // published `timestamp.as_f64()` for all three, so NDTimeStamp_RBV
        // reported the epicsTS-derived value.
        let pool = Arc::new(NDArrayPool::new(1_000_000));
        let (ds, _rx) = ndarray_channel("DS_TS", 10);
        let mut output = NDArrayOutput::new();
        output.add(ds);
        let (handle, _jh) = create_plugin_runtime_with_output(
            "TS_PARAM",
            PassthroughProcessor,
            pool,
            10,
            output,
            "",
            test_wiring(),
        );
        enable_callbacks(&handle);
        let port = handle.port_runtime().port_handle().clone();

        let mut arr = NDArray::new(vec![NDDimension::new(4)], NDDataType::UInt8);
        arr.timestamp = crate::timestamp::EpicsTimestamp {
            sec: 1234,
            nsec: 5678,
        };
        arr.time_stamp = 100.5; // hardware clock, unrelated to epicsTS
        send_array(handle.array_sender(), Arc::new(arr));
        std::thread::sleep(std::time::Duration::from_millis(50));

        assert_eq!(
            port.read_float64_blocking(handle.ndarray_params.timestamp_rbv, 0)
                .unwrap(),
            100.5,
            "NDTimeStamp publishes pArray->timeStamp"
        );
        assert_eq!(
            port.read_int32_blocking(handle.ndarray_params.epics_ts_sec, 0)
                .unwrap(),
            1234
        );
        assert_eq!(
            port.read_int32_blocking(handle.ndarray_params.epics_ts_nsec, 0)
                .unwrap(),
            5678
        );
    }

    #[test]
    fn test_shutdown_on_handle_drop() {
        let pool = Arc::new(NDArrayPool::new(1_000_000));

        let (handle, data_jh) = create_plugin_runtime(
            "SHUTDOWN_TEST",
            PassthroughProcessor,
            pool,
            10,
            "",
            test_wiring(),
        );

        // Drop the handle (closes sender channel, which should cause data thread to exit)
        let sender = handle.array_sender().clone();
        drop(handle);
        drop(sender);

        // Data thread should terminate
        let result = data_jh.join();
        assert!(result.is_ok());
    }

    #[test]
    fn test_wire_to_nonzero_ndarray_addr() {
        // G6: a multi-address upstream plugin registers its output under every
        // address in 0..max_addr. A downstream consumer must be able to select
        // NDArrayAddr=1 and actually receive arrays — previously the output was
        // registered under the bare port name only, so the "PORT:1" key was
        // missing and rewire failed with "not found".
        use crate::plugin::wiring::upstream_key;
        let pool = Arc::new(NDArrayPool::new(1_000_000));
        let wiring = test_wiring();

        // Upstream plugin advertises 2 addresses.
        let (up_handle, _up_jh) = create_plugin_runtime_multi_addr(
            "UP_MULTI",
            PassthroughProcessor,
            pool,
            10,
            "",
            wiring.clone(),
            2,
        );
        enable_callbacks(&up_handle);

        // The "PORT:1" key must resolve to the same output as the bare port.
        let addr0 = wiring.lookup_output("UP_MULTI");
        let addr1 = wiring.lookup_output(&upstream_key("UP_MULTI", 1));
        assert!(addr0.is_some(), "addr 0 output must be registered");
        assert!(
            addr1.is_some(),
            "addr 1 output must be registered for a max_addr=2 port"
        );

        // Wire a downstream consumer to UP_MULTI address 1.
        let (downstream_sender, mut downstream_rx) = ndarray_channel("DOWN_ADDR1", 10);
        wiring
            .rewire(&downstream_sender, "", &upstream_key("UP_MULTI", 1))
            .expect("wiring a consumer to NDArrayAddr=1 must succeed");

        // An array sent through the upstream must reach the addr-1 consumer.
        send_array(up_handle.array_sender(), make_test_array(99));
        let received = downstream_rx.blocking_recv().unwrap();
        assert_eq!(
            received.unique_id, 99,
            "consumer wired to NDArrayAddr=1 must receive upstream arrays"
        );
    }

    #[test]
    fn test_nonblocking_passthrough() {
        let pool = Arc::new(NDArrayPool::new(1_000_000));
        let (downstream_sender, mut downstream_rx) = ndarray_channel("DOWNSTREAM", 10);
        let mut output = NDArrayOutput::new();
        output.add(downstream_sender);

        let (handle, _data_jh) = create_plugin_runtime_with_output(
            "NB_TEST",
            PassthroughProcessor,
            pool,
            10,
            output,
            "",
            test_wiring(),
        );
        enable_callbacks(&handle);

        send_array(handle.array_sender(), make_test_array(42));

        let received = downstream_rx.blocking_recv().unwrap();
        assert_eq!(received.unique_id, 42);
    }

    #[test]
    fn test_blocking_to_nonblocking_switch() {
        let pool = Arc::new(NDArrayPool::new(1_000_000));
        let (downstream_sender, mut downstream_rx) = ndarray_channel("DOWNSTREAM", 10);
        let mut output = NDArrayOutput::new();
        output.add(downstream_sender);

        let (handle, _data_jh) = create_plugin_runtime_with_output(
            "SWITCH_TEST",
            PassthroughProcessor,
            pool,
            10,
            output,
            "",
            test_wiring(),
        );
        enable_callbacks(&handle);

        // Start in blocking mode
        handle
            .port_runtime()
            .port_handle()
            .write_int32_blocking(handle.plugin_params.blocking_callbacks, 0, 1)
            .unwrap();
        params_applied(&handle);

        send_array(handle.array_sender(), make_test_array(1));
        let received = downstream_rx.blocking_recv().unwrap();
        assert_eq!(received.unique_id, 1);

        // Switch back to non-blocking
        handle
            .port_runtime()
            .port_handle()
            .write_int32_blocking(handle.plugin_params.blocking_callbacks, 0, 0)
            .unwrap();
        params_applied(&handle);

        // Send in non-blocking mode — goes through channel to data thread
        send_array(handle.array_sender(), make_test_array(2));
        let received = downstream_rx.blocking_recv().unwrap();
        assert_eq!(received.unique_id, 2);
    }

    #[test]
    fn test_enable_callbacks_disables_processing() {
        let pool = Arc::new(NDArrayPool::new(1_000_000));
        let (downstream_sender, mut downstream_rx) = ndarray_channel("DOWNSTREAM", 10);
        let mut output = NDArrayOutput::new();
        output.add(downstream_sender);

        let (handle, _data_jh) = create_plugin_runtime_with_output(
            "ENABLE_TEST",
            PassthroughProcessor,
            pool,
            10,
            output,
            "",
            test_wiring(),
        );

        // Disable callbacks
        handle
            .port_runtime()
            .port_handle()
            .write_int32_blocking(handle.plugin_params.enable_callbacks, 0, 0)
            .unwrap();
        params_applied(&handle);

        // Send array — should be silently dropped by sender (callbacks disabled)
        send_array(handle.array_sender(), make_test_array(99));

        // Verify nothing received (with timeout)
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let result = rt.block_on(async {
            tokio::time::timeout(std::time::Duration::from_millis(100), downstream_rx.recv()).await
        });
        assert!(
            result.is_err(),
            "should not receive array when callbacks disabled"
        );
    }

    /// Disabling a busy plugin does not throw away what it already accepted.
    ///
    /// C's `EnableCallbacks=0` cancels the array interrupt and nothing else
    /// (NDPluginDriver.cpp:711-720), so the callback threads keep emptying
    /// `pToThreadMsgQ_` — `deleteCallbackThreads` depends on precisely that,
    /// disabling the interrupt and then waiting for `pending()` to fall to
    /// zero (`:1026-1034`), which it never would if disabling discarded the
    /// backlog instead.
    ///
    /// The observable is the work: on an NDPluginFile those frames are files
    /// written, on an NDPluginStats they are counts a client reads back.
    #[test]
    fn disabling_callbacks_does_not_discard_the_queued_backlog() {
        const LAST: i32 = 5;
        let pool = Arc::new(NDArrayPool::new(1_000_000));
        let (processor, entered, gated) = GatedProcessor::new();
        let (handle, _jh) =
            create_plugin_runtime("DISABLE_MID", processor, pool, 8, "", test_wiring());
        let port = handle.port_runtime().port_handle();
        enable_callbacks(&handle);

        // Park the one callback thread on frame 1, then build a backlog
        // behind it — frames the plugin has accepted but not yet worked on.
        send_array(handle.array_sender(), make_test_array(1));
        wait_until("frame 1 reached the callback thread", || {
            entered.load(Ordering::Acquire) == 1
        });
        for id in 2..=LAST {
            send_array(handle.array_sender(), make_test_array(id));
        }
        port.write_int32_blocking(handle.plugin_params.enable_callbacks, 0, 0)
            .unwrap();
        param_fifo_drained(&handle);

        // A frame offered after the disable is refused at the interrupt, the
        // one place C can refuse it.
        send_array(handle.array_sender(), make_test_array(LAST + 1));
        gated.release();
        wait_until("the backlog drained", || {
            gated.processed().len() == LAST as usize
        });
        params_applied(&handle);

        assert_eq!(
            gated.processed(),
            (1..=LAST).collect::<Vec<_>>(),
            "everything already queued is processed; only new arrivals stop"
        );
        assert_eq!(
            port.read_int32_blocking(handle.ndarray_params.array_counter, 0)
                .unwrap(),
            LAST,
            "and each of them counts against ArrayCounter"
        );
        assert_eq!(
            port.read_int32_blocking(handle.plugin_params.dropped_arrays, 0)
                .unwrap(),
            0,
            "a frame refused at the interrupt is not a dropped array (C never \
             reaches driverCallback for it)"
        );
    }

    /// A plugin port answers the pool-control PVs, because
    /// `NDPluginDriver::writeInt32` hands every parameter below
    /// `FIRST_NDPLUGIN_PARAM` to `asynNDArrayDriver::writeInt32`
    /// (NDPluginDriver.cpp:690-692), whose pool branch is at
    /// asynNDArrayDriver.cpp:684-694. Leaving them unhandled froze
    /// `PoolAllocBuffers_RBV` and friends at their construction values for the
    /// plugin's whole life, so a client polling a plugin's pool read 0 no
    /// matter what the pool was doing.
    ///
    /// One case per branch of that C `if`, not one per story.
    #[test]
    fn a_plugin_port_answers_the_pool_control_writes() {
        let pool = Arc::new(NDArrayPool::new(1_000_000));
        let (handle, _jh) = create_plugin_runtime(
            "POOLCTL",
            PassthroughProcessor,
            Arc::clone(&pool),
            4,
            "",
            test_wiring(),
        );
        let port = handle.port_runtime().port_handle();
        let p = handle.ndarray_params;
        let alloc = || port.read_int32_blocking(p.pool_alloc_buffers, 0).unwrap();
        let free = || port.read_int32_blocking(p.pool_free_buffers, 0).unwrap();

        // PoolPollStats: the readbacks follow the pool.
        let held: Vec<_> = (0..3)
            .map(|_| {
                pool.alloc(vec![NDDimension::new(64)], NDDataType::UInt8)
                    .unwrap()
            })
            .collect();
        port.write_int32_blocking(p.pool_poll_stats, 0, 1).unwrap();
        params_applied(&handle);
        assert_eq!((alloc(), free()), (3, 0), "three buffers out, none free");

        for arr in held {
            pool.release(arr);
        }
        port.write_int32_blocking(p.pool_poll_stats, 0, 1).unwrap();
        params_applied(&handle);
        assert_eq!((alloc(), free()), (3, 3), "returned buffers are free");

        // PoolEmptyFreeList: drops them and republishes without a poll.
        port.write_int32_blocking(p.pool_empty_free_list, 0, 1)
            .unwrap();
        params_applied(&handle);
        assert_eq!((alloc(), free()), (0, 0), "the free list is gone");

        // PoolPreAllocBuffers: needs an array to take dimensions from, so run
        // one frame through first (C refuses outright without `pArrays[0]`,
        // asynNDArrayDriver.cpp:715-721).
        enable_callbacks(&handle);
        send_array(handle.array_sender(), make_test_array(1));
        params_applied(&handle);
        port.write_int32_blocking(p.pool_num_pre_alloc_buffers, 0, 2)
            .unwrap();
        port.write_int32_blocking(p.pool_pre_alloc, 0, 1).unwrap();
        params_applied(&handle);
        assert_eq!(free(), 2, "two buffers pre-allocated onto the free list");
        assert_eq!(
            port.read_int32_blocking(p.pool_pre_alloc, 0).unwrap(),
            0,
            "the command PV clears itself once it has run"
        );
    }

    #[test]
    fn test_downstream_receives_multiple() {
        let pool = Arc::new(NDArrayPool::new(1_000_000));

        let (ds1, mut rx1) = ndarray_channel("DS1", 10);
        let (ds2, mut rx2) = ndarray_channel("DS2", 10);
        let mut output = NDArrayOutput::new();
        output.add(ds1);
        output.add(ds2);

        let (handle, _data_jh) = create_plugin_runtime_with_output(
            "DS_TEST",
            PassthroughProcessor,
            pool,
            10,
            output,
            "",
            test_wiring(),
        );
        enable_callbacks(&handle);

        send_array(handle.array_sender(), make_test_array(77));

        // Both downstream receivers should have the array
        let r1 = rx1.blocking_recv().unwrap();
        let r2 = rx2.blocking_recv().unwrap();
        assert_eq!(r1.unique_id, 77);
        assert_eq!(r2.unique_id, 77);
    }

    #[test]
    fn test_param_updates_after_send() {
        let pool = Arc::new(NDArrayPool::new(1_000_000));

        struct ParamTracker;
        impl NDPluginProcess for ParamTracker {
            fn process_array(&self, array: &NDArray, _pool: &NDArrayPool) -> ProcessResult {
                ProcessResult::arrays(vec![Arc::new(array.clone())])
            }
            fn plugin_type(&self) -> &str {
                "ParamTracker"
            }
        }

        let (downstream_sender, mut downstream_rx) = ndarray_channel("DOWNSTREAM", 10);
        let mut output = NDArrayOutput::new();
        output.add(downstream_sender);

        let (handle, _data_jh) = create_plugin_runtime_with_output(
            "PARAM_TEST",
            ParamTracker,
            pool,
            10,
            output,
            "",
            test_wiring(),
        );
        enable_callbacks(&handle);

        // Send array
        send_array(handle.array_sender(), make_test_array(1));
        let received = downstream_rx.blocking_recv().unwrap();
        assert_eq!(received.unique_id, 1);

        // Write enable_callbacks — should not crash
        handle
            .port_runtime()
            .port_handle()
            .write_int32_blocking(handle.plugin_params.enable_callbacks, 0, 1)
            .unwrap();
        params_applied(&handle);

        // Still works after param update
        send_array(handle.array_sender(), make_test_array(2));
        let received = downstream_rx.blocking_recv().unwrap();
        assert_eq!(received.unique_id, 2);
    }

    #[test]
    fn test_sort_buffer_reorders_by_unique_id() {
        let mut buf = SortBuffer::new();

        // Insert out of order: 3, 1, 2
        buf.insert(3, vec![make_test_array(3)], 10);
        buf.insert(1, vec![make_test_array(1)], 10);
        buf.insert(2, vec![make_test_array(2)], 10);

        assert_eq!(buf.len(), 3);

        let drained = buf.drain_all();
        let ids: Vec<i32> = drained.iter().map(|(id, _)| *id).collect();
        assert_eq!(ids, vec![1, 2, 3], "should drain in sorted uniqueId order");
        assert_eq!(buf.len(), 0);
        assert_eq!(buf.prev_unique_id, 3);
    }

    #[test]
    fn test_sort_buffer_drain_ready_contiguous() {
        // B3: drain_ready releases the head while the next-expected uniqueId
        // is contiguous, even when later ids are still missing.
        let mut buf = SortBuffer::new();
        // Mark a prior emission (prev=0) so the contiguity path is active;
        // C++ only uses the deadline for the very first output array.
        buf.note_emitted(0);
        buf.insert(1, vec![make_test_array(1)], 10);
        buf.insert(2, vec![make_test_array(2)], 10);
        buf.insert(5, vec![make_test_array(5)], 10); // gap: 3,4 missing

        // sort_time large → only contiguity drives release.
        let drained = buf.drain_ready(100.0);
        let ids: Vec<i32> = drained.iter().map(|(id, _)| *id).collect();
        assert_eq!(ids, vec![1, 2], "contiguous run released; id=5 held by gap");
        assert_eq!(buf.len(), 1);
    }

    #[test]
    fn test_sort_buffer_drain_ready_deadline() {
        // B3: a stale head is released past sort_time even with a gap.
        let mut buf = SortBuffer::new();
        buf.note_emitted(1); // prev=1
        buf.insert(5, vec![make_test_array(5)], 10); // out of order
        std::thread::sleep(std::time::Duration::from_millis(30));
        // sort_time=0.01s → head aged past deadline → released.
        let drained = buf.drain_ready(0.01);
        let ids: Vec<i32> = drained.iter().map(|(id, _)| *id).collect();
        assert_eq!(ids, vec![5], "stale head released via deadline");
    }

    #[test]
    fn test_sort_buffer_detects_disordered_on_emit() {
        // B4: disorder is counted at emission time.
        let mut buf = SortBuffer::new();
        buf.note_emitted(5); // prev=5, first_output now false
        buf.note_emitted(3); // 3 != 5 and != 6 → disordered
        assert_eq!(buf.disordered_arrays, 1);
        buf.note_emitted(4); // 4 != 3 and != 4? 4 == prev+1 → ordered
        assert_eq!(buf.disordered_arrays, 1);
    }

    #[test]
    fn test_sort_buffer_drops_when_full() {
        let mut buf = SortBuffer::new();

        // sort_size=2: third insert is refused.
        assert!(buf.insert(1, vec![make_test_array(1)], 2));
        assert!(buf.insert(2, vec![make_test_array(2)], 2));
        assert!(!buf.insert(3, vec![make_test_array(3)], 2));

        assert_eq!(buf.len(), 2);
        assert_eq!(buf.dropped_output_arrays, 1);
    }

    #[test]
    fn test_constructor_initialises_c_read_only_params() {
        // C++ NDPluginDriver.cpp:152-160 initialises these in the constructor
        // precisely so a read before any array flows returns a value instead
        // of "uninitialized". Nothing here enables callbacks or sends an
        // array: this is the idle plugin an operator sees right after
        // iocInit, which is when QueueFree_RBV read 0 (full) and
        // MaxThreads_RBV read UDF/INVALID.
        let pool = Arc::new(NDArrayPool::new(1_000_000));
        let (handle, _data_jh) = create_plugin_runtime_with_output(
            "CTOR_TEST",
            PassthroughProcessor,
            pool,
            20,
            NDArrayOutput::new(),
            "",
            test_wiring(),
        );
        let port = handle.port_runtime().port_handle();
        let read = |reason: usize| port.read_int32_blocking(reason, 0);

        // C `:156` / `:157` — an idle queue is entirely free.
        assert_eq!(read(handle.plugin_params.queue_size).unwrap(), 20);
        assert_eq!(read(handle.plugin_params.queue_use).unwrap(), 20);
        // C `:153` / `:158` / `:159`. These have no other writer: MaxThreads
        // is a longin with SCAN "I/O Intr", no PINI and no output partner, so
        // an unset param leaves it UDF forever.
        assert_eq!(
            read(handle.plugin_params.nd_array_addr).unwrap(),
            PLUGIN_NDARRAY_ADDR
        );
        assert_eq!(
            read(handle.plugin_params.max_threads).unwrap(),
            PLUGIN_MAX_THREADS
        );
        assert_eq!(
            read(handle.plugin_params.num_threads).unwrap(),
            PLUGIN_NUM_THREADS
        );
        // C `:154` / `:155`.
        assert_eq!(read(handle.plugin_params.dropped_arrays).unwrap(), 0);
        assert_eq!(read(handle.plugin_params.dropped_output_arrays).unwrap(), 0);
    }

    /// Observable half of `OverlapProcessor`, kept in the test so the
    /// processor itself can be moved into the runtime.
    #[derive(Clone, Default)]
    struct OverlapProbe {
        inside: Arc<parking_lot::Mutex<usize>>,
        gate: Arc<parking_lot::Condvar>,
        peak: Arc<std::sync::atomic::AtomicUsize>,
    }

    /// Records the peak number of concurrent `process_array` calls by holding
    /// every frame until `want` of them are inside at once. A plugin that
    /// processes frames one at a time can never drive the peak above 1,
    /// whatever the timing.
    struct OverlapProcessor {
        probe: OverlapProbe,
        want: usize,
        budget: std::time::Duration,
    }

    impl NDPluginProcess for OverlapProcessor {
        fn process_array(&self, _array: &NDArray, _pool: &NDArrayPool) -> ProcessResult {
            let mut inside = self.probe.inside.lock();
            *inside += 1;
            let peak = self
                .probe
                .peak
                .fetch_max(*inside, Ordering::AcqRel)
                .max(*inside);
            if peak >= self.want {
                self.probe.gate.notify_all();
            } else {
                // Wait on the *peak*, not on the live count: it only ever
                // rises, so a frame that returns before its neighbours wake
                // cannot re-park them.
                let deadline = std::time::Instant::now() + self.budget;
                while self.probe.peak.load(Ordering::Acquire) < self.want
                    && std::time::Instant::now() < deadline
                {
                    self.probe.gate.wait_until(&mut inside, deadline);
                }
            }
            *inside -= 1;
            ProcessResult::empty()
        }

        fn plugin_type(&self) -> &str {
            "Overlap"
        }
    }

    #[test]
    fn test_num_threads_write_spawns_overlapping_callback_threads() {
        // C runs `NumThreads` callback threads over one input queue
        // (NDPluginDriver.cpp:996-1001) and tears them down and respawns them
        // when NumThreads is written (`:730-733`). Four frames that each hold
        // until four are inside can only all return if four threads really
        // are running — which is what the NumThreads PV exists to buy.
        let pool = Arc::new(NDArrayPool::new(1_000_000));
        let probe = OverlapProbe::default();
        let (handle, _data_jh) = create_plugin_runtime_with_output(
            "NTHREADS_TEST",
            OverlapProcessor {
                probe: probe.clone(),
                want: 4,
                budget: std::time::Duration::from_secs(2),
            },
            pool,
            10,
            NDArrayOutput::new(),
            "",
            test_wiring(),
        );
        let port = handle.port_runtime().port_handle();
        // NumThreads is clamped to [1, MaxThreads] (C `createCallbackThreads`
        // NDPluginDriver.cpp:955-971), so MaxThreads has to move first.
        port.write_int32_blocking(handle.plugin_params.max_threads, 0, 4)
            .unwrap();
        params_applied(&handle);
        port.write_int32_blocking(handle.plugin_params.num_threads, 0, 4)
            .unwrap();
        params_applied(&handle);
        assert_eq!(
            port.read_int32_blocking(handle.plugin_params.num_threads, 0)
                .unwrap(),
            4,
            "NumThreads_RBV must report the accepted value"
        );

        enable_callbacks(&handle);
        for id in 1..=4 {
            send_array(handle.array_sender(), make_test_array(id));
        }
        // Quiescence: the queue is empty AND every dispatched frame is back.
        params_applied(&handle);

        assert_eq!(
            probe.peak.load(Ordering::Acquire),
            4,
            "four callback threads must process four frames concurrently"
        );
    }

    #[test]
    fn test_default_num_threads_processes_one_frame_at_a_time() {
        // C's constructor starts NumThreads at 1 (NDPluginDriver.cpp:159), so
        // an untouched plugin must still serialise its frames.
        let pool = Arc::new(NDArrayPool::new(1_000_000));
        let probe = OverlapProbe::default();
        let (handle, _data_jh) = create_plugin_runtime_with_output(
            "NTHREADS1_TEST",
            OverlapProcessor {
                probe: probe.clone(),
                want: 2,
                budget: std::time::Duration::from_millis(300),
            },
            pool,
            10,
            NDArrayOutput::new(),
            "",
            test_wiring(),
        );
        enable_callbacks(&handle);
        for id in 1..=2 {
            send_array(handle.array_sender(), make_test_array(id));
        }
        params_applied(&handle);

        assert_eq!(
            probe.peak.load(Ordering::Acquire),
            1,
            "one callback thread must not overlap frames"
        );
    }

    #[test]
    fn test_constructor_initialises_the_ndarray_read_only_block() {
        // C++ `NDPluginDriver` derives from `asynNDArrayDriver`, so the base
        // constructor's block (asynNDArrayDriver.cpp:954-1005) runs for a
        // plugin as well. Read through the port handle, which is the same
        // path a record takes: an unwritten param comes back as an error.
        let pool = Arc::new(NDArrayPool::new(2_097_152));
        let (handle, _data_jh) = create_plugin_runtime_with_output(
            "NDCTOR_TEST",
            PassthroughProcessor,
            pool,
            20,
            NDArrayOutput::new(),
            "",
            test_wiring(),
        );
        let port = handle.port_runtime().port_handle();
        let p = &handle.ndarray_params;
        for (name, reason, want) in [
            ("ARRAY_SIZE_X", p.array_size_x, 0),
            ("ARRAY_SIZE_Y", p.array_size_y, 0),
            ("ARRAY_SIZE_Z", p.array_size_z, 0),
            ("ARRAY_SIZE", p.array_size, 0),
            ("ND_DIMENSIONS", p.n_dimensions, 0),
            (
                "COLOR_MODE",
                p.color_mode,
                crate::color::NDColorMode::Mono as i32,
            ),
            ("UNIQUE_ID", p.unique_id, 0),
            ("EPICS_TS_SEC", p.epics_ts_sec, 0),
            ("EPICS_TS_NSEC", p.epics_ts_nsec, 0),
            ("BAYER_PATTERN", p.bayer_pattern, 0),
            ("ARRAY_COUNTER", p.array_counter, 0),
            ("NUM_CAPTURED", p.num_captured, 0),
            ("FREE_CAPTURE", p.free_capture, 0),
            (
                "ND_ATTRIBUTES_STATUS",
                p.attributes_status,
                crate::driver::ndarray_driver::ATTR_STATUS_FILE_NOT_FOUND,
            ),
            ("NUM_QUEUED_ARRAYS", p.num_queued_arrays, 0),
            ("POOL_ALLOC_BUFFERS", p.pool_alloc_buffers, 0),
            ("POOL_FREE_BUFFERS", p.pool_free_buffers, 0),
        ] {
            assert_eq!(
                port.read_int32_blocking(reason, 0)
                    .unwrap_or_else(|e| panic!("{name} unset after construction: {e:?}")),
                want,
                "{name}"
            );
        }
        assert_eq!(
            port.read_float64_blocking(p.pool_max_memory, 0)
                .expect("POOL_MAX_MEMORY unset after construction"),
            2.0
        );
        assert_eq!(
            port.read_float64_blocking(p.pool_used_memory, 0)
                .expect("POOL_USED_MEMORY unset after construction"),
            0.0
        );
        assert_eq!(
            port.read_float64_blocking(p.timestamp_rbv, 0)
                .expect("TIME_STAMP unset after construction"),
            0.0
        );
    }

    /// A full sort buffer refuses arrays from the very first frame, because
    /// C judges the first output array by the same `orderOK` rule as every
    /// later one: `prevUniqueId_` starts at -1000 (NDPluginDriver.cpp:99), so
    /// nothing a detector ever sends is adjacent to it.
    ///
    /// Waving the first array past the buffer instead makes every later one
    /// in-order behind it, and the buffer that C fills and overflows here
    /// stays empty for the plugin's whole life.
    #[test]
    fn a_full_sort_buffer_drops_output_arrays_from_the_first_frame() {
        const SORT_SIZE: i32 = 3;
        const SENT: i32 = 8;
        let pool = Arc::new(NDArrayPool::new(1_000_000));
        let (downstream_sender, mut downstream_rx) = ndarray_channel("DOWNSTREAM", 32);
        let mut output = NDArrayOutput::new();
        output.add(downstream_sender);
        let (handle, _jh) = create_plugin_runtime_with_output(
            "SORT_FULL",
            PassthroughProcessor,
            pool,
            10,
            output,
            "",
            test_wiring(),
        );
        let port = handle.port_runtime().port_handle();
        enable_callbacks(&handle);
        port.write_int32_blocking(handle.plugin_params.sort_size, 0, SORT_SIZE)
            .unwrap();
        // Long enough that the sort deadline cannot fire during the test, so
        // what is measured is the admission rule and not a race with it.
        port.write_float64_blocking(handle.plugin_params.sort_time, 0, 30.0)
            .unwrap();
        port.write_int32_blocking(handle.plugin_params.sort_mode, 0, 1)
            .unwrap();
        params_applied(&handle);

        for id in 1..=SENT {
            send_array(handle.array_sender(), make_test_array(id));
        }
        params_applied(&handle);

        let mut emitted = Vec::new();
        while let Some(m) = downstream_rx.try_recv_msg() {
            emitted.push(m.array.unique_id);
        }
        assert!(
            emitted.is_empty(),
            "nothing may leave before the sort deadline, got {emitted:?}"
        );
        assert_eq!(
            port.read_int32_blocking(handle.plugin_params.sort_free, 0)
                .unwrap(),
            0,
            "the buffer holds SORT_SIZE arrays"
        );
        assert_eq!(
            port.read_int32_blocking(handle.plugin_params.dropped_output_arrays, 0)
                .unwrap(),
            SENT - SORT_SIZE,
            "every array past the buffer's capacity is a dropped output array"
        );
    }

    #[test]
    fn test_sort_mode_runtime_integration() {
        let pool = Arc::new(NDArrayPool::new(1_000_000));
        let (downstream_sender, mut downstream_rx) = ndarray_channel("DOWNSTREAM", 10);
        let mut output = NDArrayOutput::new();
        output.add(downstream_sender);

        let (handle, _data_jh) = create_plugin_runtime_with_output(
            "SORT_TEST",
            PassthroughProcessor,
            pool,
            10,
            output,
            "",
            test_wiring(),
        );
        enable_callbacks(&handle);

        // Enable sort mode with sort_size=10 and a sort_time deadline.
        handle
            .port_runtime()
            .port_handle()
            .write_int32_blocking(handle.plugin_params.sort_size, 0, 10)
            .unwrap();
        handle
            .port_runtime()
            .port_handle()
            .write_float64_blocking(handle.plugin_params.sort_time, 0, 0.1)
            .unwrap();
        handle
            .port_runtime()
            .port_handle()
            .write_int32_blocking(handle.plugin_params.sort_mode, 0, 1)
            .unwrap();
        params_applied(&handle);

        // Arrays 1,2,3 are all out of order against the -1000 sentinel, so
        // sorted mode buffers every one of them: nothing reaches the client
        // until the sort deadline releases the head (C sortingTask `:646`,
        // `deltaTime > sortTime`), and the rest then cascade out behind it
        // because each is adjacent to the one just emitted.
        send_array(handle.array_sender(), make_test_array(1));
        send_array(handle.array_sender(), make_test_array(2));
        send_array(handle.array_sender(), make_test_array(3));

        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let early = rt.block_on(async {
            tokio::time::timeout(std::time::Duration::from_millis(50), downstream_rx.recv()).await
        });
        assert!(
            early.is_err(),
            "the first array is held for the sort deadline, not emitted at once"
        );
        assert_eq!(downstream_rx.blocking_recv().unwrap().unique_id, 1);
        assert_eq!(downstream_rx.blocking_recv().unwrap().unique_id, 2);
        assert_eq!(downstream_rx.blocking_recv().unwrap().unique_id, 3);

        // B3: now send out of order (5 before 4). prev=3, so 4 is in-order
        // and emitted immediately; 5 arrives first, is buffered, then 4
        // unblocks it.
        send_array(handle.array_sender(), make_test_array(5));
        send_array(handle.array_sender(), make_test_array(4));
        // 4 emitted immediately (in order), then 5 released by contiguity;
        // blocking_recv below is the wait.
        assert_eq!(downstream_rx.blocking_recv().unwrap().unique_id, 4);
        assert_eq!(downstream_rx.blocking_recv().unwrap().unique_id, 5);
    }

    #[test]
    fn test_control_plane_write_resets_every_mirrored_counter() {
        // B12/B15 boundary test. C++ holds ArrayCounter, DroppedArrays,
        // DroppedOutputArrays and DisorderedArrays in the param library and
        // increments each with a read-modify-write, so a PV write resets the
        // count. Rust mirrors each in a working copy; if a write does not land
        // back in that copy, the next status publish silently reverts it.
        //
        // One case per counter, not one per scenario: the boundary is
        // "written value survives the next publish", and it is the same
        // boundary for all four regardless of how they came to be non-zero.
        let pool = Arc::new(NDArrayPool::new(1_000_000));
        let (handle, _data_jh) = create_plugin_runtime(
            "COUNTER_RESET_TEST",
            PassthroughProcessor,
            pool,
            10,
            "",
            test_wiring(),
        );
        enable_callbacks(&handle);

        let port = handle.port_runtime().port_handle().clone();
        let counters: [(&str, usize); 4] = [
            ("ARRAY_COUNTER", handle.ndarray_params.array_counter),
            ("DROPPED_ARRAYS", handle.plugin_params.dropped_arrays),
            (
                "DROPPED_OUTPUT_ARRAYS",
                handle.plugin_params.dropped_output_arrays,
            ),
            ("DISORDERED_ARRAYS", handle.plugin_params.disordered_arrays),
        ];

        for (name, reason) in counters {
            // Write a distinct non-zero value through the control plane.
            port.write_int32_blocking(reason, 0, 7).unwrap();
            params_applied(&handle);

            // Drive one array so the data path publishes its status batch.
            // Before the fix the batch carried the working copy's stale 0 and
            // stomped the 7; ArrayCounter additionally increments, so it must
            // read 8 rather than restarting from 1.
            send_array(handle.array_sender(), make_test_array(1));
            params_applied(&handle);

            let want = if reason == handle.ndarray_params.array_counter {
                8
            } else {
                7
            };
            wait_until(&format!("{name} to keep its control-plane value"), || {
                port.read_int32_blocking(reason, 0).is_ok_and(|v| v == want)
            });
            assert_eq!(
                port.read_int32_blocking(reason, 0).unwrap(),
                want,
                "{name}: a control-plane write was reverted by the next publish"
            );
        }
    }

    #[test]
    fn test_throttle_drops_output_arrays() {
        // G7: with a tiny MaxByteRate, output arrays exceeding the byte budget
        // are dropped and counted into DroppedOutputArrays.
        let pool = Arc::new(NDArrayPool::new(1_000_000));
        let (downstream_sender, mut downstream_rx) = ndarray_channel("DOWNSTREAM", 10);
        let mut output = NDArrayOutput::new();
        output.add(downstream_sender);

        let (handle, _data_jh) = create_plugin_runtime_with_output(
            "THROTTLE_TEST",
            PassthroughProcessor,
            pool,
            10,
            output,
            "",
            test_wiring(),
        );
        enable_callbacks(&handle);

        // MaxByteRate = 8 bytes/sec. Each test array is 4 bytes; the bucket
        // starts full at 8, so the first two pass and the rest are dropped.
        handle
            .port_runtime()
            .port_handle()
            .write_float64_blocking(handle.plugin_params.max_byte_rate, 0, 8.0)
            .unwrap();
        params_applied(&handle);

        for id in 1..=5 {
            send_array(handle.array_sender(), make_test_array(id));
        }
        // ArrayCounter counts every processed frame (throttle drops happen on
        // the output side, after counting), and its flush follows the array
        // publish — so counter == 5 means everything that will ever reach the
        // downstream queue is already there.
        let port = handle.port_runtime().port_handle().clone();
        let counter = handle.ndarray_params.array_counter;
        wait_until("all 5 frames to be processed", || {
            port.read_int32_blocking(counter, 0).is_ok_and(|v| v == 5)
        });

        // Drain whatever made it through — strictly fewer than 5.
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let mut received = 0;
        while rt
            .block_on(async {
                tokio::time::timeout(std::time::Duration::from_millis(20), downstream_rx.recv())
                    .await
            })
            .map(|o| o.is_some())
            .unwrap_or(false)
        {
            received += 1;
        }
        assert!(
            received < 5,
            "throttle must drop some arrays (got {received})"
        );
        assert!(received >= 1, "first array within budget must pass");
    }

    #[test]
    fn test_process_plugin_reprocesses_last_input() {
        // G5: writing ProcessPlugin re-injects the cached last input array.
        let pool = Arc::new(NDArrayPool::new(1_000_000));
        let (downstream_sender, mut downstream_rx) = ndarray_channel("DOWNSTREAM", 10);
        let mut output = NDArrayOutput::new();
        output.add(downstream_sender);

        let (handle, _data_jh) = create_plugin_runtime_with_output(
            "PROCESS_PLUGIN_TEST",
            PassthroughProcessor,
            pool,
            10,
            output,
            "",
            test_wiring(),
        );
        enable_callbacks(&handle);

        send_array(handle.array_sender(), make_test_array(7));
        assert_eq!(downstream_rx.blocking_recv().unwrap().unique_id, 7);

        // Trigger ProcessPlugin — the cached input (id=7) is reprocessed.
        handle
            .port_runtime()
            .port_handle()
            .write_int32_blocking(handle.plugin_params.process_plugin, 0, 1)
            .unwrap();
        let reprocessed = downstream_rx.blocking_recv().unwrap();
        assert_eq!(
            reprocessed.unique_id, 7,
            "ProcessPlugin re-emits last input"
        );
    }

    #[test]
    fn test_min_callback_time_throttle_not_counted() {
        // A MinCallbackTime-throttled array is silently skipped, NOT counted.
        // C++ driverCallback (NDPluginDriver.cpp:405-450) falls through the
        // `deltaTime <= minCallbackTime` gate straight to callParamCallbacks()
        // without touching DroppedArrays — that counter is incremented ONLY on
        // a compression-unaware array (:388) or a full message queue (:440).
        // Verify (a) the throttled array is not emitted and (b) DroppedArrays
        // stays zero.
        let pool = Arc::new(NDArrayPool::new(1_000_000));
        let (downstream_sender, mut downstream_rx) = ndarray_channel("DOWNSTREAM", 10);
        let mut output = NDArrayOutput::new();
        output.add(downstream_sender);

        let (handle, _data_jh) = create_plugin_runtime_with_output(
            "MIN_CB_TEST",
            PassthroughProcessor,
            pool,
            10,
            output,
            "",
            test_wiring(),
        );
        enable_callbacks(&handle);
        let dropped = handle.array_sender().dropped_arrays_counter().clone();

        // 10s minimum between callbacks — only the first array gets through.
        handle
            .port_runtime()
            .port_handle()
            .write_float64_blocking(handle.plugin_params.min_callback_time, 0, 10.0)
            .unwrap();
        params_applied(&handle);

        send_array(handle.array_sender(), make_test_array(1));
        send_array(handle.array_sender(), make_test_array(2));

        assert_eq!(downstream_rx.blocking_recv().unwrap().unique_id, 1);
        // Fence the param channel: the barrier acks only once the array queue
        // has drained, so array 1's processing is finished before the negative
        // checks below. Array 2 was refused at the gate and never queued.
        params_applied(&handle);
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let second = rt.block_on(async {
            tokio::time::timeout(std::time::Duration::from_millis(50), downstream_rx.recv()).await
        });
        assert!(
            second.is_err(),
            "second array throttled out by MinCallbackTime"
        );
        assert_eq!(
            dropped.load(Ordering::Acquire),
            0,
            "a MinCallbackTime-throttled frame must NOT increment DroppedArrays"
        );
    }

    #[test]
    fn test_array_callbacks_zero_withholds_downstream_delivery() {
        // ADC-2: NDArrayCallbacks==0 stops downstream NDArray delivery while
        // the plugin still processes and updates its metadata params — C++
        // endProcessCallbacks (NDPluginDriver.cpp:257-265) caches the array and
        // returns before doCallbacksGenericPointer. Distinct from
        // EnableCallbacks, which gates whether the plugin processes at all.
        let pool = Arc::new(NDArrayPool::new(1_000_000));
        let (downstream_sender, mut downstream_rx) = ndarray_channel("DOWNSTREAM", 10);
        let mut output = NDArrayOutput::new();
        output.add(downstream_sender);

        let (handle, _data_jh) = create_plugin_runtime_with_output(
            "ARRAY_CB_TEST",
            PassthroughProcessor,
            pool,
            10,
            output,
            "",
            test_wiring(),
        );
        enable_callbacks(&handle);
        let port = handle.port_runtime().port_handle().clone();

        // Disable downstream array callbacks.
        port.write_int32_blocking(handle.ndarray_params.array_callbacks, 0, 0)
            .unwrap();
        params_applied(&handle);

        send_array(handle.array_sender(), make_test_array(1));
        // Processing fence: the counter flush follows any downstream publish,
        // so once it reads 1, a delivery that was going to happen already has.
        wait_until("frame 1 to be processed", || {
            port.read_int32_blocking(handle.ndarray_params.array_counter, 0)
                .is_ok_and(|v| v == 1)
        });

        // No downstream delivery.
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let got = rt.block_on(async {
            tokio::time::timeout(std::time::Duration::from_millis(50), downstream_rx.recv()).await
        });
        assert!(
            got.is_err(),
            "NDArrayCallbacks=0 must withhold downstream delivery"
        );
        // But the plugin still processed: ArrayCounter advanced.
        assert_eq!(
            port.read_int32_blocking(handle.ndarray_params.array_counter, 0)
                .unwrap(),
            1,
            "processing (and metadata params) must continue while delivery is off"
        );

        // Re-enable: the next array IS delivered downstream.
        port.write_int32_blocking(handle.ndarray_params.array_callbacks, 0, 1)
            .unwrap();
        params_applied(&handle);
        send_array(handle.array_sender(), make_test_array(2));
        assert_eq!(downstream_rx.blocking_recv().unwrap().unique_id, 2);
    }

    #[test]
    fn test_plugin_output_publishes_compressed_size() {
        // ADC-3: every processed array publishes NDCodec / NDCompressedSize
        // (C++ beginProcessCallbacks NDPluginDriver.cpp:213-214). An
        // uncompressed output carries an empty codec name and compressedSize ==
        // raw bytes; a compressed output carries the codec name and its
        // compressed size. CompressedSize_RBV (Int32) exercises both arms.
        struct CompressProcessor;
        impl NDPluginProcess for CompressProcessor {
            fn process_array(&self, array: &NDArray, _pool: &NDArrayPool) -> ProcessResult {
                let mut out = array.clone();
                out.codec = Some(crate::codec::Codec {
                    name: crate::codec::CodecName::JPEG,
                    compressed_size: 7,
                    level: 0,
                    shuffle: 0,
                    compressor: 0,
                    original_data_type: NDDataType::UInt8,
                });
                ProcessResult::arrays(vec![Arc::new(out)])
            }
            fn plugin_type(&self) -> &str {
                "Compress"
            }
        }

        // Uncompressed passthrough: a 4-byte UInt8 array → compressedSize == 4.
        {
            let pool = Arc::new(NDArrayPool::new(1_000_000));
            let (ds, _rx) = ndarray_channel("DS_RAW", 10);
            let mut output = NDArrayOutput::new();
            output.add(ds);
            let (handle, _jh) = create_plugin_runtime_with_output(
                "CODEC_RAW",
                PassthroughProcessor,
                pool,
                10,
                output,
                "",
                test_wiring(),
            );
            enable_callbacks(&handle);
            let port = handle.port_runtime().port_handle().clone();
            send_array(handle.array_sender(), make_test_array(1));
            // The read errors with ParamUndefined until the first flush — treat
            // that as "not yet".
            wait_until(
                "uncompressed output to publish CompressedSize == raw byte count",
                || {
                    port.read_int32_blocking(handle.ndarray_params.compressed_size, 0)
                        .is_ok_and(|v| v == 4)
                },
            );
        }

        // Compressed output: compressedSize == codec.compressed_size (7).
        {
            let pool = Arc::new(NDArrayPool::new(1_000_000));
            let (ds, _rx) = ndarray_channel("DS_CMP", 10);
            let mut output = NDArrayOutput::new();
            output.add(ds);
            let (handle, _jh) = create_plugin_runtime_with_output(
                "CODEC_CMP",
                CompressProcessor,
                pool,
                10,
                output,
                "",
                test_wiring(),
            );
            enable_callbacks(&handle);
            let port = handle.port_runtime().port_handle().clone();
            send_array(handle.array_sender(), make_test_array(1));
            wait_until(
                "compressed output to publish CompressedSize == codec.compressed_size",
                || {
                    port.read_int32_blocking(handle.ndarray_params.compressed_size, 0)
                        .is_ok_and(|v| v == 7)
                },
            );
        }
    }

    #[test]
    fn test_process_plugin_skips_throttled_input() {
        // a MinCallbackTime-throttled frame must NOT be cached as the
        // ProcessPlugin input. After array 1 is processed and array 2 is
        // dropped by the throttle, ProcessPlugin must re-inject array 1
        // (the last *processed* array), not the dropped array 2.
        let pool = Arc::new(NDArrayPool::new(1_000_000));
        let (downstream_sender, mut downstream_rx) = ndarray_channel("DOWNSTREAM", 10);
        let mut output = NDArrayOutput::new();
        output.add(downstream_sender);

        let (handle, _data_jh) = create_plugin_runtime_with_output(
            "PROCESS_THROTTLE_TEST",
            PassthroughProcessor,
            pool,
            10,
            output,
            "",
            test_wiring(),
        );
        enable_callbacks(&handle);

        // 10s minimum between callbacks — only the first array is processed.
        handle
            .port_runtime()
            .port_handle()
            .write_float64_blocking(handle.plugin_params.min_callback_time, 0, 10.0)
            .unwrap();
        params_applied(&handle);

        send_array(handle.array_sender(), make_test_array(1));
        send_array(handle.array_sender(), make_test_array(2));

        // Array 1 was processed and emitted; array 2 was throttled out.
        assert_eq!(downstream_rx.blocking_recv().unwrap().unique_id, 1);
        // Fence before the reset below un-gates the throttle: array 2 was
        // refused at the gate on the producer's thread, so nothing of it is
        // still in flight, but array 1 may still be processing.
        params_applied(&handle);

        // ProcessPlugin re-injects the cached input. The cache must still hold
        // array 1, because array 2 never passed the throttle gate. The
        // re-injected array itself is also subject to the throttle, so reset
        // MinCallbackTime to 0 first so the re-injected frame is processed.
        handle
            .port_runtime()
            .port_handle()
            .write_float64_blocking(handle.plugin_params.min_callback_time, 0, 0.0)
            .unwrap();
        // No fence needed: the MinCallbackTime reset and the ProcessPlugin
        // trigger below travel the same FIFO param channel, so the reset is
        // applied before the trigger by construction.
        handle
            .port_runtime()
            .port_handle()
            .write_int32_blocking(handle.plugin_params.process_plugin, 0, 1)
            .unwrap();
        let reprocessed = downstream_rx.blocking_recv().unwrap();
        assert_eq!(
            reprocessed.unique_id, 1,
            "ProcessPlugin must re-inject the last processed array (1), not the throttled array (2)"
        );
    }

    #[test]
    fn test_g3_compressed_array_dropped_on_non_aware_plugin() {
        // G3: a non-compression-aware plugin drops a compressed array.
        let pool = Arc::new(NDArrayPool::new(1_000_000));
        let (downstream_sender, mut downstream_rx) = ndarray_channel("DOWNSTREAM", 10);
        let mut output = NDArrayOutput::new();
        output.add(downstream_sender);

        let (handle, _data_jh) = create_plugin_runtime_with_output(
            "G3_TEST",
            PassthroughProcessor, // compression_aware() defaults to false
            pool,
            10,
            output,
            "",
            test_wiring(),
        );
        enable_callbacks(&handle);

        // A compressed array must be dropped, not forwarded.
        let mut compressed = NDArray::new(vec![NDDimension::new(4)], NDDataType::UInt8);
        compressed.unique_id = 1;
        compressed.codec = Some(crate::codec::Codec {
            name: crate::codec::CodecName::JPEG,
            compressed_size: 16,
            level: 0,
            shuffle: 0,
            compressor: 0,
            original_data_type: NDDataType::UInt8,
        });
        send_array(handle.array_sender(), Arc::new(compressed));

        // An uncompressed array passes through normally.
        send_array(handle.array_sender(), make_test_array(2));

        let r = downstream_rx.blocking_recv().unwrap();
        assert_eq!(
            r.unique_id, 2,
            "compressed array dropped; only the raw array reaches downstream"
        );
    }

    #[test]
    fn test_drop_on_full_increments_dropped_counter() {
        // B1/G1: a slow downstream plugin with a tiny input queue drops arrays
        // when the queue is full; the drop is counted in the plugin's shared
        // DroppedArrays counter rather than back-pressuring the producer.
        struct SlowProcessor;
        impl NDPluginProcess for SlowProcessor {
            fn process_array(&self, _a: &NDArray, _p: &NDArrayPool) -> ProcessResult {
                std::thread::sleep(std::time::Duration::from_millis(200));
                ProcessResult::empty()
            }
            fn plugin_type(&self) -> &str {
                "Slow"
            }
        }
        let pool = Arc::new(NDArrayPool::new(1_000_000));

        // Downstream plugin with queue size 1 and a slow processor.
        let (downstream_handle, _ds_jh) =
            create_plugin_runtime("B1_DOWNSTREAM", SlowProcessor, pool, 1, "", test_wiring());
        enable_callbacks(&downstream_handle);
        let ds_sender = downstream_handle.array_sender().clone();
        let dropped = ds_sender.dropped_arrays_counter().clone();

        // First array is taken by the data loop (now sleeping 200ms); second
        // fills the 1-slot queue; the rest find a full queue → dropped.
        send_array(&ds_sender, make_test_array(1));
        send_array(&ds_sender, make_test_array(2));
        send_array(&ds_sender, make_test_array(3));
        send_array(&ds_sender, make_test_array(4));

        assert!(
            dropped.load(Ordering::Acquire) >= 1,
            "arrays dropped on a full queue must be counted (got {})",
            dropped.load(Ordering::Acquire)
        );
    }

    #[test]
    fn test_cross_width_narrowing_array_read_truncates() {
        // Cross-width integer narrowing array reads must TRUNCATE (wrapping),
        // matching the C cast in C++ NDArrayPool.cpp:387 `convertType`
        //   *pDataOut++ = (dataTypeOut)(*pDataIn++);
        // A C cast `(epicsInt8)(epicsUInt16)300` keeps the low 8 bits == 44.
        // The f64 round-trip in copy_convert would SATURATE (`300.0 as i8`
        // == 127) and diverge from C++ — copy_ccast must be used instead.

        // U16 -> i8: 300 = 0x012C; low byte 0x2C = 44.
        let mut out = [0i8; 1];
        let n = copy_ccast(&[300u16], &mut out);
        assert_eq!(n, 1);
        assert_eq!(out[0], 44, "(epicsInt8)(epicsUInt16)300 == 44 (low 8 bits)");
        // copy_convert would have saturated:
        let mut sat = [0i8; 1];
        copy_convert(&[300u16], &mut sat);
        assert_eq!(sat[0], 127, "f64 round-trip saturates — the wrong behavior");

        // I32 -> i8: 0x1234_5678 -> low byte 0x78 = 120.
        let mut out2 = [0i8; 1];
        copy_ccast(&[0x1234_5678i32], &mut out2);
        assert_eq!(out2[0], 0x78);

        // I32 -> i8: -1 stays -1 (all-ones low byte).
        let mut out3 = [0i8; 1];
        copy_ccast(&[-1i32], &mut out3);
        assert_eq!(out3[0], -1);

        // U16 -> i8: 0x00FF = 255 -> low byte 0xFF reinterpreted as i8 == -1.
        let mut out4 = [0i8; 1];
        copy_ccast(&[255u16], &mut out4);
        assert_eq!(out4[0], -1);

        // I64 -> i32: 0x0000_0001_0000_002A -> low 32 bits == 42.
        let mut out5 = [0i32; 1];
        copy_ccast(&[0x0000_0001_0000_002Ai64], &mut out5);
        assert_eq!(out5[0], 42);

        // U32 -> i16: 70000 = 0x0001_1170 -> low 16 bits 0x1170 == 4464.
        let mut out6 = [0i16; 1];
        copy_ccast(&[70000u32], &mut out6);
        assert_eq!(out6[0], 4464);

        // Same-width sign change still works as a bitwise reinterpret:
        // U8 255 -> i8 -1.
        let mut out7 = [0i8; 1];
        copy_ccast(&[255u8], &mut out7);
        assert_eq!(out7[0], -1);

        // F64 out-of-range -> i32 still routes through copy_convert (the
        // `convert:` arm for float sources). C++ converts float->int with a
        // C cast too, but the runtime keeps the f64 numeric path for float
        // sources; this asserts the integer-narrowing fix did not change the
        // float-source path.
        let mut fout = [0i32; 1];
        copy_convert(&[42.9f64], &mut fout);
        assert_eq!(fout[0], 42, "f64 -> i32 truncates toward zero");
    }

    // ---- ADP-45: scatter overflow-reroute (C++ NDPluginScatter) ----

    /// Run an async body on a throwaway current-thread runtime.
    fn block<F: std::future::Future>(f: F) -> F::Output {
        tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap()
            .block_on(f)
    }

    #[test]
    fn test_scatter_reroutes_past_full_consumer() {
        // 3 consumers, queue size 1. Pre-fill A so its queue is full; a scatter
        // that would target A must reroute to B (C++ auxStatus=asynOverflow),
        // and the rerouted-away full queue must NOT count a dropped array
        // (driverCallback ignoreQueueFull, NDPluginDriver.cpp:406,433-442).
        let (sa, mut ra) = ndarray_channel("A", 1);
        let (sb, mut rb) = ndarray_channel("B", 1);
        let (sc, _rc) = ndarray_channel("C", 1);
        block(async {
            assert_eq!(
                sa.publish(make_test_array(99)).await,
                PublishOutcome::Delivered
            );
            let senders = vec![sa.clone(), sb.clone(), sc.clone()];
            let mut cursor = 0usize;
            ProcessOutput::scatter_publish(&make_test_array(1), &senders, &mut cursor).await;
            // A rerouted (attempt 0), B delivered (attempt 1): cursor +2.
            assert_eq!(cursor, 2);
            assert_eq!(rb.recv().await.unwrap().unique_id, 1);
            // A still holds only its filler; the rerouted-away drop was not counted.
            assert_eq!(ra.recv().await.unwrap().unique_id, 99);
            assert_eq!(sa.dropped_arrays_counter().load(Ordering::Acquire), 0);
        });
    }

    #[test]
    fn test_scatter_drops_on_last_when_all_full_counts_once() {
        // Both consumers full. The array is dropped on the last node only, and
        // the drop is counted exactly once (C++ sets auxStatus=asynSuccess for
        // the last node so its full queue drops and counts).
        let (sa, mut ra) = ndarray_channel("A", 1);
        let (sb, mut rb) = ndarray_channel("B", 1);
        block(async {
            sa.publish(make_test_array(91)).await;
            sb.publish(make_test_array(92)).await;
            let senders = vec![sa.clone(), sb.clone()];
            let mut cursor = 0usize;
            ProcessOutput::scatter_publish(&make_test_array(7), &senders, &mut cursor).await;
            assert_eq!(cursor, 2); // both attempted
            // A rerouted-away (not counted); B last (dropped, counted once).
            assert_eq!(sa.dropped_arrays_counter().load(Ordering::Acquire), 0);
            assert_eq!(sb.dropped_arrays_counter().load(Ordering::Acquire), 1);
            // Neither queue received frame 7 — both still hold their fillers.
            assert_eq!(ra.recv().await.unwrap().unique_id, 91);
            assert_eq!(rb.recv().await.unwrap().unique_id, 92);
        });
    }

    #[test]
    fn test_scatter_cursor_advances_per_attempt_across_frames() {
        // A is permanently full; B and C are free. Frame 0 reroutes A->B, so
        // the persistent cursor (C++ nextClient_) ends past B. Frame 1 must
        // therefore start at C, NOT back at B: a per-frame cursor would send
        // frame 1 to B; the per-attempt cursor sends it to C.
        let (sa, _ra) = ndarray_channel("A", 1);
        let (sb, mut rb) = ndarray_channel("B", 10);
        let (sc, mut rc) = ndarray_channel("C", 10);
        block(async {
            sa.publish(make_test_array(90)).await; // fill A permanently
            let senders = vec![sa.clone(), sb.clone(), sc.clone()];
            let mut cursor = 0usize;
            ProcessOutput::scatter_publish(&make_test_array(0), &senders, &mut cursor).await;
            assert_eq!(cursor, 2); // A(reroute) + B(deliver)
            assert_eq!(rb.recv().await.unwrap().unique_id, 0);
            ProcessOutput::scatter_publish(&make_test_array(1), &senders, &mut cursor).await;
            assert_eq!(cursor, 3); // C(deliver) on the first attempt
            assert_eq!(rc.recv().await.unwrap().unique_id, 1);
        });
    }

    #[test]
    fn test_scatter_skips_disabled_consumer() {
        // A disabled downstream is unregistered from the interrupt list in C++
        // (setArrayInterrupt(0)) and must not consume a round-robin slot.
        let (sa, mut ra) = ndarray_channel("A", 10);
        let (mut sb, _rb) = ndarray_channel("B", 10);
        let (sc, mut rc) = ndarray_channel("C", 10);
        sb.set_mode_flags(
            Arc::new(AtomicBool::new(false)),
            Arc::new(AtomicBool::new(false)),
        );
        block(async {
            let senders = vec![sa.clone(), sb.clone(), sc.clone()];
            let mut cursor = 0usize;
            // Active set = [A, C] (n=2): frame 0 -> A, frame 1 -> C.
            ProcessOutput::scatter_publish(&make_test_array(0), &senders, &mut cursor).await;
            ProcessOutput::scatter_publish(&make_test_array(1), &senders, &mut cursor).await;
            assert_eq!(ra.recv().await.unwrap().unique_id, 0);
            assert_eq!(rc.recv().await.unwrap().unique_id, 1);
        });
    }

    /// Holds every frame in `process_array` until released, so the input queue
    /// is the only thing absorbing the backlog.
    struct GatedProcessor {
        open: Arc<parking_lot::Mutex<bool>>,
        gate: Arc<parking_lot::Condvar>,
        /// Bumped on entry, ahead of the gate: a frame that has reached a
        /// callback thread at all shows up here even while the gate is shut.
        entered: Arc<std::sync::atomic::AtomicUsize>,
        /// uniqueIds in completion order.
        log: Arc<parking_lot::Mutex<Vec<i32>>>,
    }

    impl GatedProcessor {
        fn new() -> (Self, Arc<std::sync::atomic::AtomicUsize>, GatedControls) {
            let controls = GatedControls {
                open: Arc::new(parking_lot::Mutex::new(false)),
                gate: Arc::new(parking_lot::Condvar::new()),
                log: Arc::new(parking_lot::Mutex::new(Vec::new())),
            };
            let entered = Arc::new(std::sync::atomic::AtomicUsize::new(0));
            (
                Self {
                    open: controls.open.clone(),
                    gate: controls.gate.clone(),
                    entered: entered.clone(),
                    log: controls.log.clone(),
                },
                entered,
                controls,
            )
        }
    }

    struct GatedControls {
        open: Arc<parking_lot::Mutex<bool>>,
        gate: Arc<parking_lot::Condvar>,
        log: Arc<parking_lot::Mutex<Vec<i32>>>,
    }

    impl GatedControls {
        fn release(&self) {
            *self.open.lock() = true;
            self.gate.notify_all();
        }
        fn processed(&self) -> Vec<i32> {
            self.log.lock().clone()
        }
    }

    impl NDPluginProcess for GatedProcessor {
        fn process_array(&self, array: &NDArray, _pool: &NDArrayPool) -> ProcessResult {
            self.entered.fetch_add(1, Ordering::AcqRel);
            let mut open = self.open.lock();
            while !*open {
                self.gate.wait(&mut open);
            }
            drop(open);
            self.log.lock().push(array.unique_id);
            ProcessResult::empty()
        }
        fn plugin_type(&self) -> &str {
            "Gated"
        }
        fn does_array_callbacks(&self) -> bool {
            false
        }
    }

    /// A QueueSize write must resize the input queue, not just the readback.
    ///
    /// C deletes `pToThreadMsgQ_` and news it at the new depth
    /// (NDPluginDriver.cpp:730-733 -> :985), so an operator who raises
    /// QueueSize gets a deeper queue. This port fixed the depth at
    /// construction, so the PV moved and the queue did not — a backlog past
    /// the ORIGINAL depth was still dropped and counted in DroppedArrays while
    /// QueueFree claimed room. That is the observable: same PV writes, same
    /// frames offered, arrays lost.
    #[test]
    fn test_queue_size_write_resizes_the_input_queue() {
        const OFFERED: usize = 20;
        let pool = Arc::new(NDArrayPool::new(1_000_000));
        let (processor, _entered, gated) = GatedProcessor::new();
        let (handle, _jh) =
            create_plugin_runtime("QSIZE_TEST", processor, pool, 2, "", test_wiring());
        let port = handle.port_runtime().port_handle();
        enable_callbacks(&handle);

        // Raise the depth while nothing is in flight, exactly as an operator
        // would before starting acquisition.
        port.write_int32_blocking(handle.plugin_params.queue_size, 0, 32)
            .unwrap();
        params_applied(&handle);
        assert_eq!(
            port.read_int32_blocking(handle.plugin_params.queue_size, 0)
                .unwrap(),
            32,
            "QueueSize readback did not take"
        );

        // One frame occupies the single worker; the remaining 19 must fit in
        // the queue the operator just asked for. The gate is shut, so nothing
        // drains while they are offered.
        for id in 0..OFFERED as i32 {
            send_array(handle.array_sender(), make_test_array(id));
        }

        gated.release();

        // Let the backlog drain, then count. A frame the queue refused never
        // reaches `process_array` at all.
        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
        while gated.processed().len() < OFFERED && std::time::Instant::now() < deadline {
            std::thread::sleep(std::time::Duration::from_millis(2));
        }
        params_applied(&handle);

        let processed = gated.processed().len();
        let dropped = port
            .read_int32_blocking(handle.plugin_params.dropped_arrays, 0)
            .unwrap();
        assert_eq!(
            processed, OFFERED,
            "{OFFERED} frames offered to a queue the operator set to 32, but only {processed} \
             reached the plugin ({dropped} counted dropped): the write moved the PV, not the queue"
        );
        assert_eq!(dropped, 0, "DroppedArrays should not have moved");
    }

    /// MinCallbackTime exists to relieve queue pressure, so a frame it gates
    /// must occupy no queue slot — and therefore can never push a later frame
    /// out of one. C decides this inside `driverCallback` before `trySend`
    /// (NDPluginDriver.cpp:404-433): a throttled array returns at `:449`
    /// having touched neither the queue nor `DroppedArrays`.
    ///
    /// Deciding it after `recv` instead inverts the observable: the throttled
    /// frames fill the queue, the queue refuses a later one, and the plugin
    /// reports a drop C never reports.
    #[test]
    fn min_callback_time_gate_takes_no_queue_slot() {
        const OFFERED: usize = 20;
        const DEPTH: usize = 2;
        let pool = Arc::new(NDArrayPool::new(1_000_000));
        let (processor, entered, gated) = GatedProcessor::new();
        let (handle, _jh) =
            create_plugin_runtime("MINCB_PRESSURE", processor, pool, DEPTH, "", test_wiring());
        let port = handle.port_runtime().port_handle();
        enable_callbacks(&handle);
        // Long enough that nothing after the first frame can pass the gate,
        // however slowly the test runs.
        port.write_float64_blocking(handle.plugin_params.min_callback_time, 0, 30.0)
            .unwrap();
        params_applied(&handle);

        // Park the single callback thread on frame 0, so from here on the
        // queue is the only thing that could absorb an arrival.
        send_array(handle.array_sender(), make_test_array(0));
        wait_until("frame 0 reached the callback thread", || {
            entered.load(Ordering::Acquire) == 1
        });
        for id in 1..OFFERED as i32 {
            send_array(handle.array_sender(), make_test_array(id));
        }
        gated.release();
        params_applied(&handle);

        assert_eq!(
            gated.processed(),
            vec![0],
            "only the frame that passed the gate is processed"
        );
        assert_eq!(
            port.read_int32_blocking(handle.plugin_params.dropped_arrays, 0)
                .unwrap(),
            0,
            "the throttle gate counts nothing: C reaches DroppedArrays only \
             from the compression check (:390) and a refused trySend (:440)"
        );
        assert_eq!(
            port.read_int32_blocking(handle.plugin_params.queue_use, 0)
                .unwrap(),
            DEPTH as i32,
            "and it consumed no queue slot on the way"
        );
    }

    /// Every compressed array offered to a non-compression-aware plugin is
    /// counted, one per array (NDPluginDriver.cpp:383-394). The overflow
    /// episode cannot silence them: `auxStatus` is only read inside the
    /// `minCallbackTime` branch at `:405`, which the compression gate returns
    /// before ever reaching.
    ///
    /// Letting them into the queue instead collapses the count to the number
    /// of overflow episodes — here 3 rather than 12 — because they then reach
    /// `trySend` and arm the episode cell.
    #[test]
    fn every_compressed_array_counts_against_dropped_arrays() {
        const COMPRESSED: usize = 12;
        fn compressed_array(id: i32) -> Arc<NDArray> {
            let mut a = NDArray::new(vec![NDDimension::new(4)], NDDataType::UInt8);
            a.unique_id = id;
            a.codec = Some(crate::codec::Codec {
                name: crate::codec::CodecName::JPEG,
                compressed_size: 16,
                level: 0,
                shuffle: 0,
                compressor: 0,
                original_data_type: NDDataType::UInt8,
            });
            Arc::new(a)
        }
        let pool = Arc::new(NDArrayPool::new(1_000_000));
        let (processor, entered, gated) = GatedProcessor::new();
        let (handle, _jh) =
            create_plugin_runtime("COMPR_PRESSURE", processor, pool, 2, "", test_wiring());
        let port = handle.port_runtime().port_handle();
        enable_callbacks(&handle);

        // One raw frame parks the single callback thread, so the queue is the
        // only thing that could absorb what follows — the pressure that makes
        // the episode collapse visible.
        send_array(handle.array_sender(), make_test_array(0));
        wait_until("frame 0 reached the callback thread", || {
            entered.load(Ordering::Acquire) == 1
        });
        for id in 1..=COMPRESSED as i32 {
            send_array(handle.array_sender(), compressed_array(id));
        }
        gated.release();
        params_applied(&handle);

        assert_eq!(
            port.read_int32_blocking(handle.plugin_params.dropped_arrays, 0)
                .unwrap(),
            COMPRESSED as i32,
            "one count per compressed array, not one per overflow episode"
        );
        assert_eq!(gated.processed(), vec![0], "and none of them was processed");
    }

    /// ProcessPlugin re-injects through the queue, not around it.
    ///
    /// C hands the cached array to `driverCallback`
    /// (NDPluginDriver.cpp:741) — the detector's own entry point — so the
    /// re-processed frame takes its turn behind whatever is already queued and
    /// is run by a callback thread. Re-injecting on the data loop instead put
    /// it ahead of the backlog and ran it on a thread that is not a callback
    /// thread at all, so with every callback thread busy a client saw the
    /// duplicate arrive *before* frames the detector had already delivered.
    #[test]
    fn test_process_plugin_queues_behind_the_backlog() {
        let pool = Arc::new(NDArrayPool::new(1_000_000));
        let (processor, entered, gated) = GatedProcessor::new();
        let (handle, _jh) =
            create_plugin_runtime("PPORDER_TEST", processor, pool, 8, "", test_wiring());
        let port = handle.port_runtime().port_handle();
        enable_callbacks(&handle);

        // Frame 1 occupies the single callback thread and is the array
        // ProcessPlugin will re-inject; 2 and 3 pile up behind it.
        send_array(handle.array_sender(), make_test_array(1));
        wait_until("frame 1 reached the callback thread", || {
            entered.load(Ordering::Acquire) == 1
        });
        send_array(handle.array_sender(), make_test_array(2));
        send_array(handle.array_sender(), make_test_array(3));

        port.write_int32_blocking(handle.plugin_params.process_plugin, 0, 1)
            .unwrap();
        param_fifo_drained(&handle);
        assert_eq!(
            entered.load(Ordering::Acquire),
            1,
            "the re-injected frame entered process_array with the only callback \
             thread still busy, so it did not go through the queue"
        );

        gated.release();
        wait_until("every frame drained", || gated.processed().len() == 4);
        params_applied(&handle);
        assert_eq!(
            gated.processed(),
            vec![1, 2, 3, 1],
            "the re-injected frame must arrive after the queued backlog"
        );
    }

    /// A full queue refuses the re-injection and counts it.
    ///
    /// `driverCallback` `trySend`s (NDPluginDriver.cpp:430) and on failure
    /// bumps DroppedArrays (`:433-441`), and ProcessPlugin goes through it like
    /// any array. Running the work inline instead meant the one request C
    /// refuses always succeeded: the operator got an output frame where C
    /// gives them a dropped-array count.
    #[test]
    fn test_process_plugin_is_refused_when_the_queue_is_full() {
        let pool = Arc::new(NDArrayPool::new(1_000_000));
        let (processor, entered, gated) = GatedProcessor::new();
        let (handle, _jh) =
            create_plugin_runtime("PPFULL_TEST", processor, pool, 1, "", test_wiring());
        let port = handle.port_runtime().port_handle();
        enable_callbacks(&handle);

        send_array(handle.array_sender(), make_test_array(1));
        wait_until("frame 1 reached the callback thread", || {
            entered.load(Ordering::Acquire) == 1
        });
        // Frame 2 fills the one-deep queue, so there is no room for a
        // re-injection.
        send_array(handle.array_sender(), make_test_array(2));

        port.write_int32_blocking(handle.plugin_params.process_plugin, 0, 1)
            .unwrap();
        param_fifo_drained(&handle);
        assert_eq!(
            entered.load(Ordering::Acquire),
            1,
            "the re-injection was processed even though the queue was full"
        );
        assert_eq!(
            port.read_int32_blocking(handle.plugin_params.dropped_arrays, 0)
                .unwrap(),
            1,
            "a refused re-injection must be counted in DroppedArrays"
        );

        gated.release();
        wait_until("every frame drained", || gated.processed().len() == 2);
        params_applied(&handle);
        assert_eq!(gated.processed(), vec![1, 2]);
    }
}