cloacina 0.11.1

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

//! Accumulator trait, runtime, and supporting types.
//!
//! An accumulator is a long-lived process that consumes events from a source,
//! optionally aggregates them, and pushes typed boundaries to a reactor.
//! See CLOACI-S-0004 for the full specification.

use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;

use serde::de::DeserializeOwned;
use serde::Serialize;
use tokio::sync::{mpsc, watch};

use super::types::{self, SourceName};

// =============================================================================
// Accumulator Health
// =============================================================================

/// Health state of an accumulator, reported via watch channel.
///
/// The reactor watches these to gate its own startup (Warming → Live)
/// and detect degradation (Live → Degraded).
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
#[serde(rename_all = "snake_case")]
pub enum AccumulatorHealth {
    /// Loading checkpoint from DAL.
    Starting,
    /// Checkpoint loaded, connecting to source. Socket already active.
    Connecting,
    /// Connected, processing events, pushing boundaries.
    Live,
    /// Was live, lost source connection. Socket still active. Retrying.
    Disconnected,
    /// Passthrough — no source to connect to. Healthy by definition.
    SocketOnly,
}

impl std::fmt::Display for AccumulatorHealth {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Starting => write!(f, "starting"),
            Self::Connecting => write!(f, "connecting"),
            Self::Live => write!(f, "live"),
            Self::Disconnected => write!(f, "disconnected"),
            Self::SocketOnly => write!(f, "socket_only"),
        }
    }
}

impl AccumulatorHealth {
    /// Project the AccumulatorHealth state machine onto the bounded
    /// `cloacina_component_health` gauge label values
    /// (`starting | healthy | degraded`). See I-0099 / T-0585 for the
    /// rationale: operators want a small, cross-component vocabulary;
    /// per-component nuance lives in /v1/health endpoints.
    pub fn as_state_label(&self) -> &'static str {
        match self {
            Self::Starting | Self::Connecting => "starting",
            Self::Live | Self::SocketOnly => "healthy",
            Self::Disconnected => "degraded",
        }
    }
}

/// Create a health reporting channel for an accumulator.
pub fn health_channel() -> (
    watch::Sender<AccumulatorHealth>,
    watch::Receiver<AccumulatorHealth>,
) {
    watch::channel(AccumulatorHealth::Starting)
}

/// Errors from accumulator operations.
#[derive(Debug, thiserror::Error)]
pub enum AccumulatorError {
    #[error("accumulator init failed: {0}")]
    Init(String),
    #[error("accumulator run failed: {0}")]
    Run(String),
    #[error("send failed: {0}")]
    Send(String),
    #[error("checkpoint error: {0}")]
    Checkpoint(String),
}

/// An accumulator consumes events from a source and pushes boundaries to a reactor.
///
/// Two input paths:
/// - Event source (optional): an [`EventSource`] actively pulls from a source
/// - Socket receiver: events pushed in from outside (always active)
///
/// Both paths feed through `process()` which is called sequentially.
///
/// To add an active event loop, implement [`EventSource`] and pass it to
/// [`accumulator_runtime_with_source`]. The processor owns `&mut self` for
/// `process()` while the event source runs independently on its own task.
#[async_trait::async_trait]
pub trait Accumulator: Send + 'static {
    /// The typed boundary produced for the reactor.
    type Output: Serialize + Send + 'static;

    /// Process raw event bytes and optionally produce a boundary.
    /// The implementor owns deserialization — the runtime is format-agnostic.
    /// Called sequentially by the processor task — no concurrent `&mut self`.
    ///
    /// CLOACI-T-0739: a trait-level default body was investigated and rejected.
    /// A `where Self::Output: DeserializeOwned` default cannot be called from
    /// the generic runtime (`accumulator_runtime<A: Accumulator>`), which calls
    /// `process` for *any* `A` whose `Output` is only bound by `Serialize`;
    /// tightening that bound would break the format-agnostic contract. The
    /// boilerplate-free passthrough path is instead the
    /// `#[passthrough_accumulator]` macro, which generates this method.
    fn process(&mut self, event: Vec<u8>) -> Option<Self::Output>;

    /// Called on startup before first receive.
    /// Use to restore state from last checkpoint.
    async fn init(&mut self, _ctx: &AccumulatorContext) -> Result<(), AccumulatorError> {
        Ok(())
    }
}

/// An event source actively pulls events from an external source and pushes
/// them into the accumulator's merge channel. Runs on its own tokio task,
/// independently of the processor that calls [`Accumulator::process()`].
///
/// This is the correct way to add an active event loop to an accumulator.
/// The event source takes ownership (`self`, not `&mut self`) so it can
/// run concurrently with the processor without borrowing conflicts.
///
/// For stream-backed sources, see also [`StreamBackend`](super::stream_backend::StreamBackend).
#[async_trait::async_trait]
pub trait EventSource: Send + 'static {
    /// Run the event loop. Push raw event bytes into `events` until shutdown
    /// fires or the source is exhausted. The runtime shuts down if this returns.
    async fn run(
        self,
        events: mpsc::Sender<Vec<u8>>,
        shutdown: watch::Receiver<bool>,
    ) -> Result<(), AccumulatorError>;
}

/// Handle for persisting accumulator state via the DAL.
///
/// Wraps simple key-value checkpoint storage keyed by (graph_name, accumulator_name).
/// Serialization uses bincode wire format.
#[derive(Clone)]
pub struct CheckpointHandle {
    dal: crate::dal::unified::DAL,
    graph_name: String,
    accumulator_name: String,
}

impl CheckpointHandle {
    /// Create a new checkpoint handle for the given graph and accumulator.
    pub fn new(
        dal: crate::dal::unified::DAL,
        graph_name: String,
        accumulator_name: String,
    ) -> Self {
        Self {
            dal,
            graph_name,
            accumulator_name,
        }
    }

    /// Persist accumulator state.
    pub async fn save<T: Serialize>(&self, state: &T) -> Result<(), AccumulatorError> {
        let bytes = types::serialize(state)
            .map_err(|e| AccumulatorError::Checkpoint(format!("serialization failed: {}", e)))?;
        let result = self
            .dal
            .checkpoint()
            .save_checkpoint(&self.graph_name, &self.accumulator_name, bytes)
            .await
            .map_err(|e| AccumulatorError::Checkpoint(e.to_string()));
        // Only count successful writes — failed checkpoints surface via
        // `cloacina_accumulator_events_total` already (the event still
        // emitted) plus the warning log emitted by the caller.
        if result.is_ok() {
            metrics::counter!(
                "cloacina_accumulator_checkpoint_writes_total",
                "graph" => self.graph_name.clone(),
                "accumulator" => self.accumulator_name.clone(),
            )
            .increment(1);
        }
        result
    }

    /// Load previously persisted accumulator state.
    pub async fn load<T: DeserializeOwned>(&self) -> Result<Option<T>, AccumulatorError> {
        let bytes = self
            .dal
            .checkpoint()
            .load_checkpoint(&self.graph_name, &self.accumulator_name)
            .await
            .map_err(|e| AccumulatorError::Checkpoint(e.to_string()))?;
        match bytes {
            Some(data) => {
                let state = types::deserialize(&data).map_err(|e| {
                    AccumulatorError::Checkpoint(format!("deserialization failed: {}", e))
                })?;
                Ok(Some(state))
            }
            None => Ok(None),
        }
    }

    /// Access the underlying DAL for direct checkpoint operations.
    pub fn dal(&self) -> &crate::dal::unified::DAL {
        &self.dal
    }

    /// Get the graph name this handle is scoped to.
    pub fn graph_name(&self) -> &str {
        &self.graph_name
    }

    /// Get the accumulator name this handle is scoped to.
    pub fn accumulator_name(&self) -> &str {
        &self.accumulator_name
    }
}

/// Context provided to the accumulator by the runtime.
pub struct AccumulatorContext {
    /// Send a boundary to the reactor.
    pub output: BoundarySender,
    /// Accumulator's name (used for registration and logging).
    pub name: String,
    /// Shutdown signal — accumulator should exit run() when this fires.
    pub shutdown: watch::Receiver<bool>,
    /// Handle to persist accumulator state. None when DAL is not available
    /// (e.g., embedded mode without database).
    pub checkpoint: Option<CheckpointHandle>,
    /// Health state reporter. None when health tracking is not needed
    /// (e.g., tests, embedded mode).
    pub health: Option<watch::Sender<AccumulatorHealth>>,
}

/// Sends serialized boundaries to the reactor.
///
/// Wire format: bincode.
/// Tracks a monotonically increasing sequence number per accumulator
/// for deduplication and ordering guarantees.
#[derive(Clone)]
pub struct BoundarySender {
    inner: mpsc::Sender<(SourceName, Vec<u8>)>,
    source_name: SourceName,
    /// Monotonically increasing sequence counter (shared across clones). This is
    /// also the accumulator's `events_total` (one boundary emitted per send).
    sequence: Arc<AtomicU64>,
    /// Wall-clock (unix millis) of the last successful emit; `0` = never
    /// (CLOACI-T-0765 freshness). Shared across clones.
    last_event_ms: Arc<std::sync::atomic::AtomicI64>,
    /// Current buffered-event count for buffering kinds (batch/state); `-1` =
    /// this accumulator kind doesn't buffer / untracked (CLOACI-T-0744).
    /// Shared across clones and into the `FreshnessHandle` the health API samples.
    buffer_depth: Arc<std::sync::atomic::AtomicI64>,
    /// Declared buffer capacity for bounded kinds (state `capacity = N`, batch
    /// `max_buffer_size`); `<= 0` = unbounded / not applicable (CLOACI-T-0744).
    buffer_capacity: Arc<std::sync::atomic::AtomicI64>,
}

/// Read-only freshness probe for an accumulator (CLOACI-T-0765): the monotonic
/// emit count + the wall-clock of the last emit, shared (Arc) with the live
/// `BoundarySender` so the registry/server can sample it without locking.
#[derive(Clone)]
pub struct FreshnessHandle {
    events_total: Arc<AtomicU64>,
    last_event_ms: Arc<std::sync::atomic::AtomicI64>,
    buffer_depth: Arc<std::sync::atomic::AtomicI64>,
    buffer_capacity: Arc<std::sync::atomic::AtomicI64>,
}

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

impl FreshnessHandle {
    /// A fresh probe (zeroed counters). Share it into a `BoundarySender` via
    /// `BoundarySender::with_freshness` and register it with the graph registry.
    pub fn new() -> Self {
        Self {
            events_total: Arc::new(AtomicU64::new(0)),
            last_event_ms: Arc::new(std::sync::atomic::AtomicI64::new(0)),
            buffer_depth: Arc::new(std::sync::atomic::AtomicI64::new(-1)),
            buffer_capacity: Arc::new(std::sync::atomic::AtomicI64::new(-1)),
        }
    }

    /// Total boundaries emitted since load (monotonic).
    pub fn events_total(&self) -> u64 {
        self.events_total.load(Ordering::Relaxed)
    }
    /// Unix-millis of the last emit, or `None` if nothing has been emitted yet.
    pub fn last_event_ms(&self) -> Option<i64> {
        let v = self.last_event_ms.load(Ordering::Relaxed);
        if v > 0 {
            Some(v)
        } else {
            None
        }
    }
    /// Buffered-event count for buffering kinds (batch/state), or `None` for
    /// kinds that don't buffer / runtimes predating the gauge (CLOACI-T-0744).
    pub fn buffer_depth(&self) -> Option<u64> {
        let v = self.buffer_depth.load(Ordering::Relaxed);
        if v >= 0 {
            Some(v as u64)
        } else {
            None
        }
    }
    /// Declared buffer capacity for bounded kinds, or `None` when unbounded /
    /// not applicable (CLOACI-T-0744). The UI renders `depth/capacity`.
    pub fn buffer_capacity(&self) -> Option<u64> {
        let v = self.buffer_capacity.load(Ordering::Relaxed);
        if v > 0 {
            Some(v as u64)
        } else {
            None
        }
    }
}

fn now_unix_ms() -> i64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_millis() as i64)
        .unwrap_or(0)
}

impl BoundarySender {
    pub fn new(sender: mpsc::Sender<(SourceName, Vec<u8>)>, source_name: SourceName) -> Self {
        Self {
            inner: sender,
            source_name,
            sequence: Arc::new(AtomicU64::new(0)),
            last_event_ms: Arc::new(std::sync::atomic::AtomicI64::new(0)),
            buffer_depth: Arc::new(std::sync::atomic::AtomicI64::new(-1)),
            buffer_capacity: Arc::new(std::sync::atomic::AtomicI64::new(-1)),
        }
    }

    /// Create a sender that shares its events_total + last-event with a
    /// pre-made `FreshnessHandle` (CLOACI-T-0765), so the registry can sample
    /// freshness for the live accumulator.
    pub fn with_freshness(
        sender: mpsc::Sender<(SourceName, Vec<u8>)>,
        source_name: SourceName,
        freshness: FreshnessHandle,
    ) -> Self {
        Self {
            inner: sender,
            source_name,
            sequence: freshness.events_total.clone(),
            last_event_ms: freshness.last_event_ms.clone(),
            buffer_depth: freshness.buffer_depth.clone(),
            buffer_capacity: freshness.buffer_capacity.clone(),
        }
    }

    /// Create a sender with a specific starting sequence number (for restart recovery).
    pub fn with_sequence(
        sender: mpsc::Sender<(SourceName, Vec<u8>)>,
        source_name: SourceName,
        start_sequence: u64,
    ) -> Self {
        Self {
            inner: sender,
            source_name,
            sequence: Arc::new(AtomicU64::new(start_sequence)),
            last_event_ms: Arc::new(std::sync::atomic::AtomicI64::new(0)),
            buffer_depth: Arc::new(std::sync::atomic::AtomicI64::new(-1)),
            buffer_capacity: Arc::new(std::sync::atomic::AtomicI64::new(-1)),
        }
    }

    /// Serialize and send a boundary to the reactor.
    /// Increments the sequence counter atomically after successful send.
    pub async fn send<T: Serialize>(&self, boundary: &T) -> Result<(), AccumulatorError> {
        let bytes = types::serialize(boundary)
            .map_err(|e| AccumulatorError::Send(format!("serialization failed: {}", e)))?;
        self.inner
            .send((self.source_name.clone(), bytes))
            .await
            .map_err(|e| AccumulatorError::Send(format!("channel send failed: {}", e)))?;
        self.sequence.fetch_add(1, Ordering::SeqCst);
        // CLOACI-T-0765: stamp last-emit for the freshness probe.
        self.last_event_ms.store(now_unix_ms(), Ordering::Relaxed);
        Ok(())
    }

    /// Get the source name this sender is associated with.
    pub fn source_name(&self) -> &SourceName {
        &self.source_name
    }

    /// Get the current sequence number (last emitted).
    pub fn sequence_number(&self) -> u64 {
        self.sequence.load(Ordering::SeqCst)
    }

    /// Record the current buffered-event count so the health API can report it
    /// (CLOACI-T-0744). Called by the buffering runtimes (batch/state) alongside
    /// the Prometheus gauge.
    pub fn set_buffer_depth(&self, depth: u64) {
        self.buffer_depth
            .store(depth as i64, std::sync::atomic::Ordering::Relaxed);
    }

    /// Record the declared buffer capacity for bounded kinds (CLOACI-T-0744).
    /// Pass a positive value; unbounded/none kinds just never call this.
    pub fn set_buffer_capacity(&self, capacity: u64) {
        self.buffer_capacity
            .store(capacity as i64, std::sync::atomic::Ordering::Relaxed);
    }

    /// A shared freshness probe (events_total + last_event), for registration
    /// with the graph registry so the health endpoint can report freshness.
    pub fn freshness(&self) -> FreshnessHandle {
        FreshnessHandle {
            events_total: self.sequence.clone(),
            last_event_ms: self.last_event_ms.clone(),
            buffer_depth: self.buffer_depth.clone(),
            buffer_capacity: self.buffer_capacity.clone(),
        }
    }
}

/// Configuration for the accumulator runtime.
pub struct AccumulatorRuntimeConfig {
    /// Merge channel capacity (backpressure).
    pub merge_channel_capacity: usize,
}

impl Default for AccumulatorRuntimeConfig {
    fn default() -> Self {
        Self {
            merge_channel_capacity: 1024,
        }
    }
}

/// Run an accumulator as 2-3 tokio tasks connected by a merge channel.
///
/// Socket-only mode (no event source):
/// ```text
/// ┌─────────────────┐     ┌─────────────────┐
/// │  Socket task     │──→  │  Processor task  │──→ BoundarySender ──→ Reactor
/// │  (always active) │     │  (calls process) │
/// └─────────────────┘     └─────────────────┘
/// ```
///
/// With event source (use [`accumulator_runtime_with_source`]):
/// ```text
/// ┌─────────────────┐
/// │  Event source    │──→ mpsc<Event> ──┐
/// │  (pulls events)  │                  │     ┌─────────────────┐
/// └─────────────────┘                  ├────→│  Processor task  │──→ BoundarySender ──→ Reactor
/// ┌─────────────────┐                  │     │  (calls process) │
/// │  Socket task     │──→ mpsc<Event> ──┘     └─────────────────┘
/// │  (always active) │
/// └─────────────────┘
/// ```
pub async fn accumulator_runtime<A: Accumulator>(
    acc: A,
    ctx: AccumulatorContext,
    socket_rx: mpsc::Receiver<Vec<u8>>,
    config: AccumulatorRuntimeConfig,
) {
    accumulator_runtime_inner::<A, NoEventSource>(acc, ctx, socket_rx, config, None).await
}

/// Run an accumulator with an active event source that pulls events from
/// an external system. The event source runs on its own task and pushes
/// raw bytes into the merge channel concurrently with the socket receiver.
pub async fn accumulator_runtime_with_source<A, S>(
    acc: A,
    ctx: AccumulatorContext,
    socket_rx: mpsc::Receiver<Vec<u8>>,
    config: AccumulatorRuntimeConfig,
    source: S,
) where
    A: Accumulator,
    S: EventSource,
{
    accumulator_runtime_inner(acc, ctx, socket_rx, config, Some(source)).await
}

/// Placeholder type for when no event source is provided.
struct NoEventSource;

#[async_trait::async_trait]
impl EventSource for NoEventSource {
    async fn run(
        self,
        _events: mpsc::Sender<Vec<u8>>,
        _shutdown: watch::Receiver<bool>,
    ) -> Result<(), AccumulatorError> {
        std::future::pending().await
    }
}

/// Inner runtime shared by both `accumulator_runtime` and `accumulator_runtime_with_source`.
async fn accumulator_runtime_inner<A: Accumulator, S: EventSource>(
    mut acc: A,
    ctx: AccumulatorContext,
    socket_rx: mpsc::Receiver<Vec<u8>>,
    config: AccumulatorRuntimeConfig,
    event_source: Option<S>,
) {
    // Report starting health
    set_health(&ctx, AccumulatorHealth::Starting);

    // Passthrough/stream accumulators don't buffer — seed the gauge at 0 so
    // dashboards have a stable series per (graph, accumulator).
    set_accumulator_buffer_depth(&ctx, 0.0);

    // Kind label is fixed at runtime startup: with an event source we are
    // stream-backed, otherwise passthrough. Hardcoded into the bounded enum
    // documented on `cloacina_accumulator_events_total`.
    let kind: &'static str = if event_source.is_some() {
        "stream"
    } else {
        "passthrough"
    };

    // Initialize — may restore state from checkpoint
    if let Err(e) = acc.init(&ctx).await {
        tracing::error!(name = %ctx.name, "accumulator init failed: {}", e);
        return;
    }

    // Create merge channel — carries raw bytes from all sources
    let (event_tx, mut event_rx) = mpsc::channel::<Vec<u8>>(config.merge_channel_capacity);

    // Spawn event source task (or no-op wait if none provided)
    let name_loop = ctx.name.clone();
    let loop_handle = if let Some(source) = event_source {
        set_health(&ctx, AccumulatorHealth::Connecting);
        let shutdown_source = ctx.shutdown.clone();
        let event_tx_source = event_tx.clone();
        let name_source = name_loop.clone();
        let handle = tokio::spawn(async move {
            match source.run(event_tx_source, shutdown_source).await {
                Ok(()) => tracing::debug!(name = %name_source, "event source completed"),
                Err(e) => tracing::error!(name = %name_source, "event source failed: {}", e),
            }
        });
        // Advance past Connecting to Live now that the source task is running.
        // The reactor's startup gate only proceeds once every accumulator
        // reports Live/SocketOnly; a stream accumulator that stayed Connecting
        // gated the reactor forever, so it never consumed the boundaries the
        // source delivered (CLOACI-T-0715). Optimistic + symmetric with the
        // socket path below, which reports SocketOnly before any data arrives;
        // the degraded-mode monitor downgrades health if the source later fails.
        set_health(&ctx, AccumulatorHealth::Live);
        handle
    } else {
        set_health(&ctx, AccumulatorHealth::SocketOnly);
        let mut shutdown_loop = ctx.shutdown.clone();
        tokio::spawn(async move {
            let _ = shutdown_loop.changed().await;
            tracing::debug!(name = %name_loop, "event loop task shutting down");
        })
    };

    // Spawn socket receiver task — forwards raw bytes without deserialization
    let event_tx_socket = event_tx.clone();
    let mut shutdown_socket = ctx.shutdown.clone();
    let name_socket = ctx.name.clone();
    let socket_handle = tokio::spawn(async move {
        let mut socket_rx = socket_rx;
        loop {
            tokio::select! {
                Some(bytes) = socket_rx.recv() => {
                    if event_tx_socket.send(bytes).await.is_err() {
                        break; // merge channel closed
                    }
                }
                _ = shutdown_socket.changed() => {
                    tracing::debug!(name = %name_socket, "socket task shutting down");
                    break;
                }
            }
        }
    });

    // Processor task (runs on current task — owns &mut acc)
    let mut shutdown_proc = ctx.shutdown.clone();
    loop {
        tokio::select! {
            Some(event) = event_rx.recv() => {
                let emit_started = std::time::Instant::now();
                if let Some(boundary) = acc.process(event) {
                    if let Err(e) = ctx.output.send(&boundary).await {
                        tracing::error!(name = %ctx.name, "boundary send failed: {}", e);
                    } else {
                        persist_boundary(&ctx, &boundary).await;
                    }
                }
                record_accumulator_event(&ctx, kind, emit_started);
            }
            _ = shutdown_proc.changed() => {
                tracing::debug!(name = %ctx.name, "processor task shutting down");
                break;
            }
        }
    }

    // Wait for spawned tasks to finish
    let _ = loop_handle.await;
    let _ = socket_handle.await;
}

/// Create a shutdown signal pair.
pub fn shutdown_signal() -> (watch::Sender<bool>, watch::Receiver<bool>) {
    watch::channel(false)
}

// =============================================================================
// Polling Accumulator
// =============================================================================

/// A polling accumulator periodically calls an async poll function to query
/// pull-based data sources (databases, APIs, config files).
///
/// Returns `Option<Output>` — `Some` emits a boundary, `None` means "no change".
#[async_trait::async_trait]
pub trait PollingAccumulator: Send + 'static {
    /// The typed boundary produced for the reactor.
    type Output: Serialize + DeserializeOwned + Send + 'static;

    /// Poll the data source. Called on each timer tick.
    /// Return `Some(output)` to emit a boundary, `None` to skip.
    async fn poll(&mut self) -> Option<Self::Output>;

    /// Polling interval.
    fn interval(&self) -> std::time::Duration;
}

/// Run a polling accumulator as a timer-based loop.
///
/// On each tick: calls `poll()`, if Some → serializes and sends boundary.
/// Also accepts socket events (same as passthrough — external pushes still work).
pub async fn polling_accumulator_runtime<P: PollingAccumulator>(
    mut poller: P,
    ctx: AccumulatorContext,
    socket_rx: mpsc::Receiver<Vec<u8>>,
) {
    set_health(&ctx, AccumulatorHealth::Starting);
    // Polling accumulators don't buffer — emit a stable 0 series for dashboards.
    set_accumulator_buffer_depth(&ctx, 0.0);

    // Restore last poll output from checkpoint and emit to reactor
    if let Some(ref handle) = ctx.checkpoint {
        match handle.load::<P::Output>().await {
            Ok(Some(output)) => {
                tracing::info!(name = %ctx.name, "polling accumulator restored last output from checkpoint");
                if let Err(e) = ctx.output.send(&output).await {
                    tracing::warn!(name = %ctx.name, "failed to emit restored poll output: {}", e);
                }
            }
            Ok(None) => {}
            Err(e) => {
                tracing::warn!(name = %ctx.name, "failed to load polling checkpoint: {}", e);
            }
        }
    }

    let interval = poller.interval();
    let mut timer = tokio::time::interval(interval);
    // Skip the first immediate tick — we want to wait one interval before first poll
    timer.tick().await;

    // Polling accumulators are Live once the timer starts
    set_health(&ctx, AccumulatorHealth::Live);

    let mut shutdown = ctx.shutdown.clone();
    let mut socket_rx = socket_rx;

    loop {
        tokio::select! {
            _ = timer.tick() => {
                let emit_started = std::time::Instant::now();
                if let Some(output) = poller.poll().await {
                    if let Err(e) = ctx.output.send(&output).await {
                        tracing::error!(name = %ctx.name, "polling boundary send failed: {}", e);
                    } else {
                        persist_boundary(&ctx, &output).await;
                        // Checkpoint the last successful poll output
                        if let Some(ref handle) = ctx.checkpoint {
                            if let Err(e) = handle.save(&output).await {
                                tracing::warn!(name = %ctx.name, "polling checkpoint save failed: {}", e);
                                record_accumulator_persist_failure(&ctx, "checkpoint");
                            }
                        }
                    }
                    record_accumulator_event(&ctx, "polling", emit_started);
                }
            }
            Some(bytes) = socket_rx.recv() => {
                let emit_started = std::time::Instant::now();
                // Socket receives JSON from external sources
                match serde_json::from_slice::<P::Output>(&bytes) {
                    Ok(output) => {
                        if let Err(e) = ctx.output.send(&output).await {
                            tracing::error!(name = %ctx.name, "socket boundary send failed: {}", e);
                        } else {
                            persist_boundary(&ctx, &output).await;
                        }
                    }
                    Err(e) => {
                        tracing::warn!(name = %ctx.name, "socket deserialize error: {}", e);
                    }
                }
                record_accumulator_event(&ctx, "polling", emit_started);
            }
            _ = shutdown.changed() => {
                tracing::debug!(name = %ctx.name, "polling accumulator shutting down");
                break;
            }
        }
    }
}

// =============================================================================
// Batch Accumulator
// =============================================================================

/// A batch accumulator buffers incoming events and processes them all at once
/// on a flush signal. Emits a single boundary containing the batch result.
///
/// Flush triggers:
/// - Timer-based flush interval
/// - Buffer size threshold (optional)
/// - Shutdown (drains remaining buffer)
#[async_trait::async_trait]
pub trait BatchAccumulator: Send + 'static {
    /// The typed boundary produced from the batch.
    type Output: Serialize + Send + 'static;

    /// Process a batch of raw event bytes and optionally produce a boundary.
    /// The implementor owns deserialization — the runtime is format-agnostic.
    /// Called when the buffer is flushed. Empty batches are never passed.
    fn process_batch(&mut self, events: Vec<Vec<u8>>) -> Option<Self::Output>;
}

/// Configuration for the batch accumulator runtime.
#[derive(Default)]
pub struct BatchAccumulatorConfig {
    /// Optional timer-based flush interval. If None, only flushes on signal or size threshold.
    pub flush_interval: Option<std::time::Duration>,
    /// Optional: flush when buffer reaches this size.
    pub max_buffer_size: Option<usize>,
}

/// Create a flush signal pair for batch accumulators.
///
/// The sender is held by the reactor (or external code) and used to trigger
/// a flush. The receiver is passed to `batch_accumulator_runtime`.
pub fn flush_signal() -> (mpsc::Sender<()>, mpsc::Receiver<()>) {
    mpsc::channel(16)
}

/// Run a batch accumulator that buffers events and flushes on signal, timer, or size threshold.
///
/// Primary flush trigger is the `flush_rx` channel — typically sent by the reactor
/// after each graph execution ("give me everything since last run").
/// Timer and size threshold are secondary/fallback triggers.
pub async fn batch_accumulator_runtime<B: BatchAccumulator>(
    mut acc: B,
    ctx: AccumulatorContext,
    socket_rx: mpsc::Receiver<Vec<u8>>,
    mut flush_rx: mpsc::Receiver<()>,
    config: BatchAccumulatorConfig,
) {
    set_health(&ctx, AccumulatorHealth::Starting);
    set_accumulator_buffer_depth(&ctx, 0.0);
    // CLOACI-T-0744: surface the bounded flush threshold as the buffer
    // capacity so the health API can render fill (`depth/capacity`).
    if let Some(cap) = config.max_buffer_size {
        ctx.output.set_buffer_capacity(cap as u64);
    }

    // Restore buffered events from checkpoint if available
    let mut buffer: Vec<Vec<u8>> = Vec::new();
    if let Some(ref handle) = ctx.checkpoint {
        match handle.load::<Vec<Vec<u8>>>().await {
            Ok(Some(raw_events)) => {
                buffer = raw_events;
                if !buffer.is_empty() {
                    tracing::info!(name = %ctx.name, events = buffer.len(), "batch buffer restored from checkpoint");
                    set_accumulator_buffer_depth(&ctx, buffer.len() as f64);
                }
            }
            Ok(None) => {}
            Err(e) => {
                tracing::warn!(name = %ctx.name, "failed to load batch checkpoint: {}", e);
            }
        }
    }

    // Create timer only if interval is configured
    let mut timer = config.flush_interval.map(tokio::time::interval);
    if let Some(ref mut t) = timer {
        // Skip the first immediate tick
        t.tick().await;
    }

    // Batch accumulators are Live once ready to receive events
    set_health(&ctx, AccumulatorHealth::Live);

    let mut shutdown = ctx.shutdown.clone();
    let mut socket_rx = socket_rx;

    loop {
        tokio::select! {
            Some(bytes) = socket_rx.recv() => {
                buffer.push(bytes);
                set_accumulator_buffer_depth(&ctx, buffer.len() as f64);
                // Persist buffer snapshot for crash resilience
                persist_batch_buffer(&ctx, &buffer).await;
                // Check size threshold
                if let Some(max) = config.max_buffer_size {
                    if buffer.len() >= max {
                        flush_batch(&mut acc, &mut buffer, &ctx).await;
                        set_accumulator_buffer_depth(&ctx, 0.0);
                    }
                }
            }
            Some(()) = flush_rx.recv() => {
                flush_batch(&mut acc, &mut buffer, &ctx).await;
                set_accumulator_buffer_depth(&ctx, 0.0);
                // Clear checkpoint after flush (buffer is empty)
                persist_batch_buffer(&ctx, &[]).await;
            }
            _ = async {
                match timer.as_mut() {
                    Some(t) => t.tick().await,
                    None => std::future::pending().await,
                }
            } => {
                flush_batch(&mut acc, &mut buffer, &ctx).await;
                set_accumulator_buffer_depth(&ctx, 0.0);
            }
            _ = shutdown.changed() => {
                tracing::debug!(name = %ctx.name, "batch accumulator shutting down, draining buffer");
                // Drain remaining buffer on shutdown
                flush_batch(&mut acc, &mut buffer, &ctx).await;
                set_accumulator_buffer_depth(&ctx, 0.0);
                break;
            }
        }
    }
}

/// Persist batch buffer snapshot to DAL for crash resilience (best-effort).
async fn persist_batch_buffer(ctx: &AccumulatorContext, buffer: &[Vec<u8>]) {
    if let Some(ref handle) = ctx.checkpoint {
        if let Err(e) = handle.save(&buffer.to_vec()).await {
            tracing::warn!(name = %ctx.name, "batch buffer checkpoint failed: {}", e);
            record_accumulator_persist_failure(ctx, "batch_buffer");
        }
    }
}

/// Flush the buffer through the batch accumulator and send boundary if produced.
async fn flush_batch<B: BatchAccumulator>(
    acc: &mut B,
    buffer: &mut Vec<Vec<u8>>,
    ctx: &AccumulatorContext,
) {
    if buffer.is_empty() {
        return;
    }
    let emit_started = std::time::Instant::now();
    let batch = std::mem::take(buffer);
    let count = batch.len();
    if let Some(output) = acc.process_batch(batch) {
        if let Err(e) = ctx.output.send(&output).await {
            tracing::error!(name = %ctx.name, "batch boundary send failed: {}", e);
        } else {
            tracing::debug!(name = %ctx.name, events = count, "batch flushed");
            persist_boundary(ctx, &output).await;
        }
    }
    // One emit_total + emit_duration per flush — operators see flush rate
    // and per-flush latency, while `cloacina_accumulator_buffer_depth` shows
    // the size that drove the flush.
    record_accumulator_event(ctx, "batch", emit_started);
}

// =============================================================================
// Internal helpers
// =============================================================================

/// Set health state (best-effort, no-op if health channel not configured).
fn set_health(ctx: &AccumulatorContext, health: AccumulatorHealth) {
    if let Some(ref sender) = ctx.health {
        let _ = sender.send(health);
    }
}

/// Increment `cloacina_accumulator_persist_failures_total{graph,accumulator,kind}`
/// with a bounded `kind ∈ {checkpoint, boundary, batch_buffer}` label.
/// Replaces the silent `let _ = persist_*` failure paths flagged as OPS-15
/// (CLOACI-I-0108 / T-0590).
fn record_accumulator_persist_failure(ctx: &AccumulatorContext, kind: &'static str) {
    metrics::counter!(
        "cloacina_accumulator_persist_failures_total",
        "graph" => graph_label(ctx),
        "accumulator" => ctx.name.clone(),
        "kind" => kind,
    )
    .increment(1);
}

/// Derive the bounded `graph` metric label for an accumulator. When the
/// checkpoint handle is present (production / DAL-backed deployments) we use
/// the deployed graph name; embedded / test runtimes without a DAL fall
/// back to the `embedded` sentinel so the label space stays closed.
fn graph_label(ctx: &AccumulatorContext) -> String {
    ctx.checkpoint
        .as_ref()
        .map(|c| c.graph_name().to_string())
        .unwrap_or_else(|| "embedded".to_string())
}

/// Record one accumulator event and its emit duration. Called by each
/// runtime once per event processed; `kind` is set by the runtime
/// (`passthrough` / `stream` / `polling` / `batch`).
fn record_accumulator_event(
    ctx: &AccumulatorContext,
    kind: &'static str,
    emit_started: std::time::Instant,
) {
    let graph = graph_label(ctx);
    metrics::counter!(
        "cloacina_accumulator_events_total",
        "graph" => graph.clone(),
        "accumulator" => ctx.name.clone(),
        "kind" => kind,
    )
    .increment(1);
    metrics::histogram!(
        "cloacina_accumulator_emit_duration_seconds",
        "graph" => graph,
        "accumulator" => ctx.name.clone(),
    )
    .record(emit_started.elapsed().as_secs_f64());
}

/// Update the `cloacina_accumulator_buffer_depth` gauge. Only batch and
/// stateful stream accumulators have meaningful buffers; the other kinds
/// emit `0` from runtime startup so dashboards see a consistent series
/// per `(graph, accumulator)` tuple.
fn set_accumulator_buffer_depth(ctx: &AccumulatorContext, depth: f64) {
    metrics::gauge!(
        "cloacina_accumulator_buffer_depth",
        "graph" => graph_label(ctx),
        "accumulator" => ctx.name.clone(),
    )
    .set(depth);
    // CLOACI-T-0744: mirror into the freshness probe so the polled health API
    // reports the same gauge the Prometheus series carries.
    ctx.output.set_buffer_depth(depth as u64);
}

/// Persist last-emitted boundary with sequence number to DAL (best-effort, logs on failure).
async fn persist_boundary<T: Serialize>(ctx: &AccumulatorContext, boundary: &T) {
    if let Some(ref handle) = ctx.checkpoint {
        let bytes = match types::serialize(boundary) {
            Ok(b) => b,
            Err(e) => {
                tracing::warn!(name = %ctx.name, "boundary persistence serialization failed: {}", e);
                record_accumulator_persist_failure(ctx, "boundary");
                return;
            }
        };
        let seq = ctx.output.sequence_number() as i64;
        match handle
            .dal()
            .checkpoint()
            .save_boundary(handle.graph_name(), handle.accumulator_name(), bytes, seq)
            .await
        {
            Ok(_) => {
                metrics::counter!(
                    "cloacina_accumulator_checkpoint_writes_total",
                    "graph" => handle.graph_name().to_string(),
                    "accumulator" => handle.accumulator_name().to_string(),
                )
                .increment(1);
            }
            Err(e) => {
                tracing::warn!(name = %ctx.name, "boundary persistence failed: {}", e);
                record_accumulator_persist_failure(ctx, "boundary");
            }
        }
    }
}

// =============================================================================
// State Accumulator
// =============================================================================

/// A state accumulator holds a bounded VecDeque<T> that receives values from
/// the computation graph (collector or mid-graph writes), persists to DAL on
/// every write, and loads from DAL on startup. Enables cyclic state patterns
/// where the graph's output feeds back as input on the next execution.
///
/// Capacity modes:
/// - `capacity > 0`: bounded — evicts oldest when at capacity
/// - `capacity < 0` (e.g., -1): unbounded — grows without limit
/// - `capacity == 0`: write-only sink — no history emitted back
pub struct StateAccumulator<T: Serialize + DeserializeOwned + Send + Clone + 'static> {
    buffer: std::collections::VecDeque<T>,
    capacity: i32,
}

impl<T: Serialize + DeserializeOwned + Send + Clone + 'static> StateAccumulator<T> {
    pub fn new(capacity: i32) -> Self {
        Self {
            buffer: std::collections::VecDeque::new(),
            capacity,
        }
    }
}

/// Run a state accumulator. Receives values via socket, appends to VecDeque,
/// evicts if over capacity, persists to DAL, and emits the full list as boundary.
///
/// On startup: loads from DAL and emits current list to reactor.
/// Encode a state window for the boundary wire (CLOACI-T-0842).
///
/// The window ships in the SAME shape passthrough events use —
/// `bincode(Vec<u8>)` of JSON bytes (here, a JSON array) — because the
/// previous `bincode(Vec<T>)` encoding was WRITE-ONLY for the
/// `serde_json::Value` instantiation the factory uses: `Value` can't
/// deserialize from bincode (non-self-describing, `deserialize_any`), so the
/// fires log rendered `null`, the FFI cache conversion errored, and no
/// consumer could read the window. One wire shape for every boundary means
/// every existing decoder just works.
fn state_window_frame<T: Serialize>(list: &[T]) -> Result<Vec<u8>, String> {
    serde_json::to_vec(list).map_err(|e| e.to_string())
}

pub async fn state_accumulator_runtime<T: Serialize + DeserializeOwned + Send + Clone + 'static>(
    mut acc: StateAccumulator<T>,
    ctx: AccumulatorContext,
    socket_rx: mpsc::Receiver<Vec<u8>>,
) {
    set_health(&ctx, AccumulatorHealth::Starting);

    // Load from DAL on startup
    if let Some(ref handle) = ctx.checkpoint {
        match handle
            .dal()
            .checkpoint()
            .load_state_buffer(handle.graph_name(), handle.accumulator_name())
            .await
        {
            Ok(Some((data, _cap))) => {
                // JSON first (the current write format — see the save site for
                // why bincode cannot work here), then bincode for rows written
                // by older builds with a concrete T. A row that parses as
                // NEITHER is reported loudly: the silent `if let Ok` this
                // replaces is exactly how packaged windows vanished on
                // takeover without a single log line.
                match serde_json::from_slice::<std::collections::VecDeque<T>>(&data)
                    .ok()
                    .or_else(|| types::deserialize::<std::collections::VecDeque<T>>(&data).ok())
                {
                    Some(buffer) => {
                        acc.buffer = buffer;
                        tracing::info!(name = %ctx.name, entries = acc.buffer.len(), "state accumulator restored from DAL");
                    }
                    None => {
                        tracing::warn!(
                            name = %ctx.name,
                            bytes = data.len(),
                            "persisted state buffer could not be decoded as JSON or legacy \
                             bincode — the window is LOST and will restart empty"
                        );
                    }
                }
            }
            Ok(None) => {
                tracing::debug!(name = %ctx.name, "no persisted state accumulator buffer found");
            }
            Err(e) => {
                tracing::warn!(name = %ctx.name, "failed to load state buffer: {}", e);
            }
        }

        // Emit current list to reactor immediately (so reactor has state on startup)
        if !acc.buffer.is_empty() && acc.capacity != 0 {
            let list: Vec<T> = acc.buffer.iter().cloned().collect();
            match state_window_frame(&list) {
                Ok(frame) => {
                    if let Err(e) = ctx.output.send(&frame).await {
                        tracing::error!(name = %ctx.name, "state accumulator initial emit failed: {}", e);
                    }
                }
                Err(e) => {
                    tracing::error!(name = %ctx.name, "state window encode failed: {}", e)
                }
            }
        }
    }

    // CLOACI-T-0744: state buffers were previously invisible to BOTH the
    // Prometheus gauge and the health API — report depth (incl. the restored
    // buffer) and the bounded capacity so the UI can render `N/capacity`.
    set_accumulator_buffer_depth(&ctx, acc.buffer.len() as f64);
    if acc.capacity > 0 {
        ctx.output.set_buffer_capacity(acc.capacity as u64);
    }

    set_health(&ctx, AccumulatorHealth::SocketOnly);

    let mut shutdown = ctx.shutdown.clone();
    let mut socket_rx = socket_rx;

    loop {
        tokio::select! {
            Some(bytes) = socket_rx.recv() => {
                // Socket receives JSON from external sources
                match serde_json::from_slice::<T>(&bytes) {
                    Ok(value) => {
                        // Append to buffer
                        acc.buffer.push_back(value);

                        // Evict if over capacity (bounded mode)
                        if acc.capacity > 0 {
                            while acc.buffer.len() > acc.capacity as usize {
                                acc.buffer.pop_front();
                            }
                        }
                        set_accumulator_buffer_depth(&ctx, acc.buffer.len() as f64);

                        // Persist to DAL
                        if let Some(ref handle) = ctx.checkpoint {
                            // serde_json, NOT the bincode used elsewhere
                            // (CLOACI-T-0851): the packaged path instantiates
                            // this runtime with T = serde_json::Value, and
                            // bincode can SERIALIZE a Value but can never
                            // DESERIALIZE one (Value requires deserialize_any,
                            // which bincode does not support). Bincode here
                            // meant packaged state windows were persisted in a
                            // format that could not be read back — write-only
                            // durability, found when a takeover restored an
                            // empty window on a live cluster. JSON is
                            // self-describing, so every T this runtime accepts
                            // round-trips.
                            let data = match serde_json::to_vec(&acc.buffer) {
                                Ok(d) => d,
                                Err(e) => {
                                    tracing::warn!(name = %ctx.name, "state buffer serialization failed: {}", e);
                                    continue;
                                }
                            };
                            if let Err(e) = handle
                                .dal()
                                .checkpoint()
                                .save_state_buffer(
                                    handle.graph_name(),
                                    handle.accumulator_name(),
                                    data,
                                    acc.capacity,
                                )
                                .await
                            {
                                tracing::warn!(name = %ctx.name, "state buffer persistence failed: {}", e);
                            }
                        }

                        // Emit full list as boundary (unless write-only mode)
                        if acc.capacity != 0 {
                            let list: Vec<T> = acc.buffer.iter().cloned().collect();
                            match state_window_frame(&list) {
                                Ok(frame) => {
                                    if let Err(e) = ctx.output.send(&frame).await {
                                        tracing::error!(name = %ctx.name, "state accumulator emit failed: {}", e);
                                    } else {
                                        persist_boundary(&ctx, &list).await;
                                    }
                                }
                                Err(e) => {
                                    tracing::error!(name = %ctx.name, "state window encode failed: {}", e)
                                }
                            }
                        }
                    }
                    Err(e) => {
                        tracing::warn!(name = %ctx.name, "state accumulator deserialize error: {}", e);
                    }
                }
            }
            _ = shutdown.changed() => {
                tracing::debug!(name = %ctx.name, "state accumulator shutting down");
                break;
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use serde::{Deserialize, Serialize};

    #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
    struct TestEvent {
        value: f64,
    }

    #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
    struct TestBoundary {
        result: f64,
    }

    struct DoubleAccumulator;

    #[async_trait::async_trait]
    impl Accumulator for DoubleAccumulator {
        type Output = TestBoundary;

        fn process(&mut self, event: Vec<u8>) -> Option<TestBoundary> {
            let parsed: TestEvent = serde_json::from_slice(&event).ok()?;
            Some(TestBoundary {
                result: parsed.value * 2.0,
            })
        }
    }

    #[tokio::test]
    async fn test_boundary_sender_round_trip() {
        let (tx, mut rx) = mpsc::channel(10);
        let sender = BoundarySender::new(tx, SourceName::new("test"));

        let boundary = TestBoundary { result: 42.0 };
        sender.send(&boundary).await.unwrap();

        let (name, bytes) = rx.recv().await.unwrap();
        assert_eq!(name, SourceName::new("test"));

        let decoded: TestBoundary = types::deserialize(&bytes).unwrap();
        assert_eq!(decoded, boundary);
    }

    // CLOACI-T-0715: a stream/event-source accumulator must advance to Live so
    // the reactor's startup health-gate clears. It previously stuck at
    // Connecting, gating Kafka-fed reactors forever (they received messages but
    // never fired).
    #[tokio::test]
    async fn test_stream_accumulator_reaches_live() {
        struct IdleSource;
        #[async_trait::async_trait]
        impl EventSource for IdleSource {
            async fn run(
                self,
                _events: mpsc::Sender<Vec<u8>>,
                mut shutdown: watch::Receiver<bool>,
            ) -> Result<(), AccumulatorError> {
                let _ = shutdown.changed().await;
                Ok(())
            }
        }

        let (boundary_tx, _boundary_rx) = mpsc::channel(10);
        let (_socket_tx, socket_rx) = mpsc::channel(10);
        let (shutdown_tx, shutdown_rx) = shutdown_signal();
        let (health_tx, health_rx) = health_channel();

        let ctx = AccumulatorContext {
            output: BoundarySender::new(boundary_tx, SourceName::new("stream_acc")),
            name: "stream_acc".to_string(),
            shutdown: shutdown_rx,
            checkpoint: None,
            health: Some(health_tx),
        };

        let handle = tokio::spawn(accumulator_runtime_with_source(
            DoubleAccumulator,
            ctx,
            socket_rx,
            AccumulatorRuntimeConfig::default(),
            IdleSource,
        ));

        // Health must reach Live (not stay Connecting) shortly after start.
        let mut ok = false;
        for _ in 0..50 {
            if matches!(*health_rx.borrow(), AccumulatorHealth::Live) {
                ok = true;
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(20)).await;
        }
        assert!(
            ok,
            "stream accumulator should reach Live; was {:?}",
            *health_rx.borrow()
        );

        shutdown_tx.send(true).unwrap();
        let _ = tokio::time::timeout(std::time::Duration::from_secs(2), handle).await;
    }

    #[tokio::test]
    async fn test_accumulator_runtime_processes_socket_events() {
        let (boundary_tx, mut boundary_rx) = mpsc::channel(10);
        let (socket_tx, socket_rx) = mpsc::channel(10);
        let (shutdown_tx, shutdown_rx) = shutdown_signal();

        let sender = BoundarySender::new(boundary_tx, SourceName::new("test_acc"));
        let ctx = AccumulatorContext {
            output: sender,
            name: "test_acc".to_string(),
            shutdown: shutdown_rx,
            checkpoint: None,
            health: None,
        };

        let acc = DoubleAccumulator;

        // Spawn the runtime
        let handle = tokio::spawn(accumulator_runtime(
            acc,
            ctx,
            socket_rx,
            AccumulatorRuntimeConfig::default(),
        ));

        // Push an event via socket
        let event = TestEvent { value: 5.0 };
        let event_bytes = serde_json::to_vec(&event).unwrap();
        socket_tx.send(event_bytes).await.unwrap();

        // Read the boundary
        let (name, bytes) = boundary_rx.recv().await.unwrap();
        assert_eq!(name, SourceName::new("test_acc"));
        let boundary: TestBoundary = types::deserialize(&bytes).unwrap();
        assert_eq!(boundary.result, 10.0);

        // Shutdown
        shutdown_tx.send(true).unwrap();
        handle.await.unwrap();
    }

    #[tokio::test]
    async fn test_accumulator_runtime_multiple_events() {
        let (boundary_tx, mut boundary_rx) = mpsc::channel(10);
        let (socket_tx, socket_rx) = mpsc::channel(10);
        let (shutdown_tx, shutdown_rx) = shutdown_signal();

        let sender = BoundarySender::new(boundary_tx, SourceName::new("multi"));
        let ctx = AccumulatorContext {
            output: sender,
            name: "multi".to_string(),
            shutdown: shutdown_rx,
            checkpoint: None,
            health: None,
        };

        let handle = tokio::spawn(accumulator_runtime(
            DoubleAccumulator,
            ctx,
            socket_rx,
            AccumulatorRuntimeConfig::default(),
        ));

        // Push 3 events
        for v in [1.0, 2.0, 3.0] {
            let bytes = serde_json::to_vec(&TestEvent { value: v }).unwrap();
            socket_tx.send(bytes).await.unwrap();
        }

        // Read 3 boundaries in order
        for expected in [2.0, 4.0, 6.0] {
            let (_, bytes) = boundary_rx.recv().await.unwrap();
            let boundary: TestBoundary = types::deserialize(&bytes).unwrap();
            assert_eq!(boundary.result, expected);
        }

        shutdown_tx.send(true).unwrap();
        handle.await.unwrap();
    }

    #[tokio::test]
    async fn test_accumulator_shutdown() {
        let (boundary_tx, _boundary_rx) = mpsc::channel(10);
        let (_socket_tx, socket_rx) = mpsc::channel(10);
        let (shutdown_tx, shutdown_rx) = shutdown_signal();

        let sender = BoundarySender::new(boundary_tx, SourceName::new("shutdown_test"));
        let ctx = AccumulatorContext {
            output: sender,
            name: "shutdown_test".to_string(),
            shutdown: shutdown_rx,
            checkpoint: None,
            health: None,
        };

        let handle = tokio::spawn(accumulator_runtime(
            DoubleAccumulator,
            ctx,
            socket_rx,
            AccumulatorRuntimeConfig::default(),
        ));

        // Shutdown immediately
        shutdown_tx.send(true).unwrap();

        // Should complete without hanging
        tokio::time::timeout(std::time::Duration::from_secs(2), handle)
            .await
            .expect("runtime should shut down within 2 seconds")
            .unwrap();
    }

    // --- Polling accumulator tests ---

    struct CountingPoller {
        count: u32,
        max: u32,
    }

    #[async_trait::async_trait]
    impl PollingAccumulator for CountingPoller {
        type Output = TestBoundary;

        async fn poll(&mut self) -> Option<TestBoundary> {
            self.count += 1;
            if self.count <= self.max {
                Some(TestBoundary {
                    result: self.count as f64,
                })
            } else {
                None // "no change" after max polls
            }
        }

        fn interval(&self) -> std::time::Duration {
            std::time::Duration::from_millis(50)
        }
    }

    #[tokio::test]
    async fn test_polling_accumulator_emits_on_some() {
        let (boundary_tx, mut boundary_rx) = mpsc::channel(10);
        let (_socket_tx, socket_rx) = mpsc::channel(10);
        let (shutdown_tx, shutdown_rx) = shutdown_signal();

        let sender = BoundarySender::new(boundary_tx, SourceName::new("poller"));
        let ctx = AccumulatorContext {
            output: sender,
            name: "poller".to_string(),
            shutdown: shutdown_rx,
            checkpoint: None,
            health: None,
        };

        let poller = CountingPoller { count: 0, max: 3 };
        let handle = tokio::spawn(polling_accumulator_runtime(poller, ctx, socket_rx));

        // Wait for 3 polls (50ms each + first tick skipped = ~200ms)
        tokio::time::sleep(std::time::Duration::from_millis(250)).await;

        // Should have received 3 boundaries
        let mut received = vec![];
        while let Ok((_name, bytes)) = boundary_rx.try_recv() {
            let b: TestBoundary = types::deserialize(&bytes).unwrap();
            received.push(b.result);
        }
        assert!(
            received.len() >= 3,
            "expected at least 3 polls, got {}",
            received.len()
        );
        assert_eq!(received[0], 1.0);
        assert_eq!(received[1], 2.0);
        assert_eq!(received[2], 3.0);

        shutdown_tx.send(true).unwrap();
        let _ = handle.await;
    }

    #[tokio::test]
    async fn test_polling_accumulator_skips_on_none() {
        let (boundary_tx, mut boundary_rx) = mpsc::channel(10);
        let (_socket_tx, socket_rx) = mpsc::channel(10);
        let (shutdown_tx, shutdown_rx) = shutdown_signal();

        let sender = BoundarySender::new(boundary_tx, SourceName::new("skip_poller"));
        let ctx = AccumulatorContext {
            output: sender,
            name: "skip_poller".to_string(),
            shutdown: shutdown_rx,
            checkpoint: None,
            health: None,
        };

        // max=0 means poll always returns None
        let poller = CountingPoller { count: 0, max: 0 };
        let handle = tokio::spawn(polling_accumulator_runtime(poller, ctx, socket_rx));

        // Wait for a few poll cycles
        tokio::time::sleep(std::time::Duration::from_millis(200)).await;

        // Should have received zero boundaries
        assert!(
            boundary_rx.try_recv().is_err(),
            "should not have received any boundary"
        );

        shutdown_tx.send(true).unwrap();
        let _ = handle.await;
    }

    #[tokio::test]
    async fn test_polling_accumulator_shutdown() {
        let (boundary_tx, _boundary_rx) = mpsc::channel(10);
        let (_socket_tx, socket_rx) = mpsc::channel(10);
        let (shutdown_tx, shutdown_rx) = shutdown_signal();

        let sender = BoundarySender::new(boundary_tx, SourceName::new("shutdown_poller"));
        let ctx = AccumulatorContext {
            output: sender,
            name: "shutdown_poller".to_string(),
            shutdown: shutdown_rx,
            checkpoint: None,
            health: None,
        };

        let poller = CountingPoller { count: 0, max: 100 };
        let handle = tokio::spawn(polling_accumulator_runtime(poller, ctx, socket_rx));

        // Shutdown immediately
        shutdown_tx.send(true).unwrap();

        tokio::time::timeout(std::time::Duration::from_secs(2), handle)
            .await
            .expect("polling runtime should shut down within 2 seconds")
            .unwrap();
    }

    // --- Batch accumulator tests ---

    struct SumBatchAccumulator;

    #[async_trait::async_trait]
    impl BatchAccumulator for SumBatchAccumulator {
        type Output = TestBoundary;

        fn process_batch(&mut self, events: Vec<Vec<u8>>) -> Option<TestBoundary> {
            let sum: f64 = events
                .iter()
                .filter_map(|raw| serde_json::from_slice::<TestEvent>(raw).ok())
                .map(|e| e.value)
                .sum();
            Some(TestBoundary { result: sum })
        }
    }

    #[tokio::test]
    async fn test_batch_accumulator_flush_on_signal() {
        let (boundary_tx, mut boundary_rx) = mpsc::channel(10);
        let (socket_tx, socket_rx) = mpsc::channel(10);
        let (flush_tx, flush_rx) = flush_signal();
        let (shutdown_tx, shutdown_rx) = shutdown_signal();

        let sender = BoundarySender::new(boundary_tx, SourceName::new("batch_signal"));
        let ctx = AccumulatorContext {
            output: sender,
            name: "batch_signal".to_string(),
            shutdown: shutdown_rx,
            checkpoint: None,
            health: None,
        };

        let config = BatchAccumulatorConfig::default(); // no timer, no size threshold

        let handle = tokio::spawn(batch_accumulator_runtime(
            SumBatchAccumulator,
            ctx,
            socket_rx,
            flush_rx,
            config,
        ));

        // Push 3 events
        for v in [10.0, 20.0, 30.0] {
            socket_tx
                .send(serde_json::to_vec(&TestEvent { value: v }).unwrap())
                .await
                .unwrap();
        }

        tokio::time::sleep(std::time::Duration::from_millis(50)).await;

        // No boundary yet — no flush signal sent
        assert!(boundary_rx.try_recv().is_err());

        // Send flush signal
        flush_tx.send(()).await.unwrap();
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;

        // Should get one boundary with sum = 60.0
        let (_name, bytes) = boundary_rx.recv().await.unwrap();
        let b: TestBoundary = types::deserialize(&bytes).unwrap();
        assert_eq!(b.result, 60.0);

        shutdown_tx.send(true).unwrap();
        let _ = handle.await;
    }

    #[tokio::test]
    async fn test_batch_accumulator_flush_on_timer() {
        let (boundary_tx, mut boundary_rx) = mpsc::channel(10);
        let (socket_tx, socket_rx) = mpsc::channel(10);
        let (_flush_tx, flush_rx) = flush_signal();
        let (shutdown_tx, shutdown_rx) = shutdown_signal();

        let sender = BoundarySender::new(boundary_tx, SourceName::new("batch"));
        let ctx = AccumulatorContext {
            output: sender,
            name: "batch".to_string(),
            shutdown: shutdown_rx,
            checkpoint: None,
            health: None,
        };

        let config = BatchAccumulatorConfig {
            flush_interval: Some(std::time::Duration::from_millis(100)),
            max_buffer_size: None,
        };

        let handle = tokio::spawn(batch_accumulator_runtime(
            SumBatchAccumulator,
            ctx,
            socket_rx,
            flush_rx,
            config,
        ));

        // Push 5 events quickly (before timer fires)
        for v in [1.0, 2.0, 3.0, 4.0, 5.0] {
            socket_tx
                .send(serde_json::to_vec(&TestEvent { value: v }).unwrap())
                .await
                .unwrap();
        }

        // Wait for timer flush
        tokio::time::sleep(std::time::Duration::from_millis(200)).await;

        // Should get ONE boundary with sum = 15.0
        let (_name, bytes) = boundary_rx.recv().await.unwrap();
        let b: TestBoundary = types::deserialize(&bytes).unwrap();
        assert_eq!(b.result, 15.0);

        shutdown_tx.send(true).unwrap();
        let _ = handle.await;
    }

    #[tokio::test]
    async fn test_batch_accumulator_empty_flush_skips() {
        let (boundary_tx, mut boundary_rx) = mpsc::channel(10);
        let (_socket_tx, socket_rx) = mpsc::channel(10);
        let (flush_tx, flush_rx) = flush_signal();
        let (shutdown_tx, shutdown_rx) = shutdown_signal();

        let sender = BoundarySender::new(boundary_tx, SourceName::new("empty_batch"));
        let ctx = AccumulatorContext {
            output: sender,
            name: "empty_batch".to_string(),
            shutdown: shutdown_rx,
            checkpoint: None,
            health: None,
        };

        let config = BatchAccumulatorConfig::default();

        let handle = tokio::spawn(batch_accumulator_runtime(
            SumBatchAccumulator,
            ctx,
            socket_rx,
            flush_rx,
            config,
        ));

        // Send flush with empty buffer
        flush_tx.send(()).await.unwrap();

        // Wait for a few flush cycles with no events
        tokio::time::sleep(std::time::Duration::from_millis(200)).await;

        // Should have no boundaries
        assert!(boundary_rx.try_recv().is_err());

        shutdown_tx.send(true).unwrap();
        let _ = handle.await;
    }

    #[tokio::test]
    async fn test_batch_accumulator_max_buffer_size() {
        let (boundary_tx, mut boundary_rx) = mpsc::channel(10);
        let (socket_tx, socket_rx) = mpsc::channel(10);
        let (_flush_tx, flush_rx) = flush_signal();
        let (shutdown_tx, shutdown_rx) = shutdown_signal();

        let sender = BoundarySender::new(boundary_tx, SourceName::new("size_batch"));
        let ctx = AccumulatorContext {
            output: sender,
            name: "size_batch".to_string(),
            shutdown: shutdown_rx,
            checkpoint: None,
            health: None,
        };

        let config = BatchAccumulatorConfig {
            flush_interval: None,     // no timer
            max_buffer_size: Some(3), // flush at 3 events
        };

        let handle = tokio::spawn(batch_accumulator_runtime(
            SumBatchAccumulator,
            ctx,
            socket_rx,
            flush_rx,
            config,
        ));

        // Push exactly 3 events — should trigger size-based flush
        for v in [10.0, 20.0, 30.0] {
            socket_tx
                .send(serde_json::to_vec(&TestEvent { value: v }).unwrap())
                .await
                .unwrap();
        }

        tokio::time::sleep(std::time::Duration::from_millis(100)).await;

        // Should get boundary with sum = 60.0
        let (_name, bytes) = boundary_rx.recv().await.unwrap();
        let b: TestBoundary = types::deserialize(&bytes).unwrap();
        assert_eq!(b.result, 60.0);

        shutdown_tx.send(true).unwrap();
        let _ = handle.await;
    }

    #[tokio::test]
    async fn test_batch_accumulator_shutdown_drains() {
        let (boundary_tx, mut boundary_rx) = mpsc::channel(10);
        let (socket_tx, socket_rx) = mpsc::channel(10);
        let (_flush_tx, flush_rx) = flush_signal();
        let (shutdown_tx, shutdown_rx) = shutdown_signal();

        let sender = BoundarySender::new(boundary_tx, SourceName::new("drain_batch"));
        let ctx = AccumulatorContext {
            output: sender,
            name: "drain_batch".to_string(),
            shutdown: shutdown_rx,
            checkpoint: None,
            health: None,
        };

        let config = BatchAccumulatorConfig::default(); // no timer, no size

        let handle = tokio::spawn(batch_accumulator_runtime(
            SumBatchAccumulator,
            ctx,
            socket_rx,
            flush_rx,
            config,
        ));

        // Push events without triggering flush
        for v in [1.0, 2.0] {
            socket_tx
                .send(serde_json::to_vec(&TestEvent { value: v }).unwrap())
                .await
                .unwrap();
        }

        tokio::time::sleep(std::time::Duration::from_millis(50)).await;

        // Shutdown — should drain remaining buffer
        shutdown_tx.send(true).unwrap();
        let _ = tokio::time::timeout(std::time::Duration::from_secs(2), handle).await;

        // Should get one boundary from the drain
        let (_name, bytes) = boundary_rx.recv().await.unwrap();
        let b: TestBoundary = types::deserialize(&bytes).unwrap();
        assert_eq!(b.result, 3.0); // 1.0 + 2.0
    }

    struct FilterAccumulator;

    #[async_trait::async_trait]
    impl Accumulator for FilterAccumulator {
        type Output = TestBoundary;

        fn process(&mut self, event: Vec<u8>) -> Option<TestBoundary> {
            let parsed: TestEvent = serde_json::from_slice(&event).ok()?;
            // Only produce boundary for values > 5
            if parsed.value > 5.0 {
                Some(TestBoundary {
                    result: parsed.value,
                })
            } else {
                None
            }
        }
    }

    #[tokio::test]
    async fn test_accumulator_process_returns_none() {
        let (boundary_tx, mut boundary_rx) = mpsc::channel(10);
        let (socket_tx, socket_rx) = mpsc::channel(10);
        let (shutdown_tx, shutdown_rx) = shutdown_signal();

        let sender = BoundarySender::new(boundary_tx, SourceName::new("filter"));
        let ctx = AccumulatorContext {
            output: sender,
            name: "filter".to_string(),
            shutdown: shutdown_rx,
            checkpoint: None,
            health: None,
        };

        let handle = tokio::spawn(accumulator_runtime(
            FilterAccumulator,
            ctx,
            socket_rx,
            AccumulatorRuntimeConfig::default(),
        ));

        // Push event below threshold (should produce no boundary)
        socket_tx
            .send(serde_json::to_vec(&TestEvent { value: 3.0 }).unwrap())
            .await
            .unwrap();

        // Push event above threshold (should produce boundary)
        socket_tx
            .send(serde_json::to_vec(&TestEvent { value: 10.0 }).unwrap())
            .await
            .unwrap();

        // Only one boundary should come through
        let (_, bytes) = boundary_rx.recv().await.unwrap();
        let boundary: TestBoundary = types::deserialize(&bytes).unwrap();
        assert_eq!(boundary.result, 10.0);

        shutdown_tx.send(true).unwrap();
        handle.await.unwrap();
    }

    // --- State accumulator runtime tests (CLOACI-T-0688) ---
    //
    // These exercise the RUNTIME wiring that `@cloaca.state_accumulator(capacity=N)`
    // (and `StateAccumulatorFactory`) drive: values pushed over the socket are
    // appended to a bounded `VecDeque`, oldest entries are evicted past capacity,
    // and the full bounded history is emitted back as the boundary on every write.
    // The boundary wire-type is `Vec<T>` (see `state_accumulator_runtime`), so we
    // decode boundaries as `Vec<serde_json::Value>` to mirror the
    // `serde_json::Value` instantiation used by `StateAccumulatorFactory`.

    /// Build a minimal socket-driven AccumulatorContext (no DAL/checkpoint, no
    /// health) mirroring the embedded/test wiring used by the other runtime tests.
    fn make_state_ctx(
        name: &str,
    ) -> (
        AccumulatorContext,
        mpsc::Receiver<(SourceName, Vec<u8>)>,
        watch::Sender<bool>,
    ) {
        let (boundary_tx, boundary_rx) = mpsc::channel(32);
        let (shutdown_tx, shutdown_rx) = shutdown_signal();
        let ctx = AccumulatorContext {
            output: BoundarySender::new(boundary_tx, SourceName::new(name)),
            name: name.to_string(),
            shutdown: shutdown_rx,
            checkpoint: None,
            health: None,
        };
        (ctx, boundary_rx, shutdown_tx)
    }

    /// Drive `state_accumulator_runtime::<serde_json::Value>` over the socket and
    /// assert that a bounded capacity evicts the oldest entries and emits the
    /// retained history (newest `capacity` values) as each boundary.
    #[tokio::test]
    async fn test_state_accumulator_runtime_bounded_evicts_and_emits_history() {
        let capacity = 2;
        let (socket_tx, socket_rx) = mpsc::channel::<Vec<u8>>(32);
        let (ctx, mut boundary_rx, shutdown_tx) = make_state_ctx("state_bounded");
        let acc = StateAccumulator::<serde_json::Value>::new(capacity);
        let handle = tokio::spawn(state_accumulator_runtime(acc, ctx, socket_rx));

        // Feed 3 values (> capacity 2): 1, 2, 3.
        for v in [1_i64, 2, 3] {
            socket_tx
                .send(serde_json::to_vec(&serde_json::json!(v)).unwrap())
                .await
                .unwrap();
        }

        // Expect 3 boundaries, each the bounded history after that write:
        //   after 1 -> [1]
        //   after 2 -> [1, 2]
        //   after 3 -> [2, 3]   (oldest "1" evicted)
        let expected: Vec<Vec<i64>> = vec![vec![1], vec![1, 2], vec![2, 3]];
        for exp in expected {
            let (name, bytes) =
                tokio::time::timeout(std::time::Duration::from_secs(2), boundary_rx.recv())
                    .await
                    .expect("boundary should arrive within 2s")
                    .expect("boundary channel open");
            assert_eq!(name, SourceName::new("state_bounded"));
            // CLOACI-T-0842: the window ships as JSON-array bytes in the
            // passthrough bincode(Vec<u8>) wrapper (one wire shape for every
            // boundary) — decode accordingly.
            let json_bytes: Vec<u8> = types::deserialize(&bytes).unwrap();
            let list: Vec<i64> = serde_json::from_slice(&json_bytes).unwrap();
            assert_eq!(
                list, exp,
                "bounded history mismatch (capacity={})",
                capacity
            );
        }

        shutdown_tx.send(true).unwrap();
        let _ = tokio::time::timeout(std::time::Duration::from_secs(2), handle).await;
    }

    /// CLOACI-T-0744: the freshness probe reports the state buffer's live
    /// depth and bounded capacity so the health API can render `N/capacity`.
    #[tokio::test]
    async fn test_state_accumulator_probe_reports_buffer_fill() {
        let capacity = 5;
        let (socket_tx, socket_rx) = mpsc::channel::<Vec<u8>>(32);
        let (boundary_tx, mut boundary_rx) = mpsc::channel(32);
        let (shutdown_tx, shutdown_rx) = shutdown_signal();
        let probe = FreshnessHandle::new();
        let ctx = AccumulatorContext {
            output: BoundarySender::with_freshness(
                boundary_tx,
                SourceName::new("state_probe"),
                probe.clone(),
            ),
            name: "state_probe".to_string(),
            shutdown: shutdown_rx,
            checkpoint: None,
            health: None,
        };
        let acc = StateAccumulator::<serde_json::Value>::new(capacity);
        let handle = tokio::spawn(state_accumulator_runtime(acc, ctx, socket_rx));

        for v in [1_i64, 2, 3] {
            socket_tx
                .send(serde_json::to_vec(&serde_json::json!(v)).unwrap())
                .await
                .unwrap();
            // Consume the emitted boundary so the ingest completed.
            tokio::time::timeout(std::time::Duration::from_secs(2), boundary_rx.recv())
                .await
                .expect("boundary should arrive within 2s")
                .expect("boundary channel open");
        }

        assert_eq!(probe.buffer_depth(), Some(3), "3 ingests -> depth 3");
        assert_eq!(
            probe.buffer_capacity(),
            Some(5),
            "bounded capacity reported"
        );
        assert_eq!(probe.events_total(), 3);

        shutdown_tx.send(true).unwrap();
        let _ = tokio::time::timeout(std::time::Duration::from_secs(2), handle).await;
    }

    /// `capacity == 0` is a write-only sink: values are buffered but NO boundary
    /// is emitted back to the reactor.
    #[tokio::test]
    async fn test_state_accumulator_runtime_write_only_emits_nothing() {
        let (socket_tx, socket_rx) = mpsc::channel::<Vec<u8>>(32);
        let (ctx, mut boundary_rx, shutdown_tx) = make_state_ctx("state_write_only");
        let acc = StateAccumulator::<serde_json::Value>::new(0);
        let handle = tokio::spawn(state_accumulator_runtime(acc, ctx, socket_rx));

        for v in [10_i64, 20, 30] {
            socket_tx
                .send(serde_json::to_vec(&serde_json::json!(v)).unwrap())
                .await
                .unwrap();
        }

        // Give the runtime time to process; assert no boundary was emitted.
        tokio::time::sleep(std::time::Duration::from_millis(150)).await;
        assert!(
            boundary_rx.try_recv().is_err(),
            "write-only (capacity==0) state accumulator must not emit a boundary"
        );

        shutdown_tx.send(true).unwrap();
        let _ = tokio::time::timeout(std::time::Duration::from_secs(2), handle).await;
    }

    /// CLOACI-T-0851: the packaged state path runs with T = serde_json::Value,
    /// and bincode can serialize a Value but can NEVER deserialize one
    /// (deserialize_any is unsupported). This test pins the failure that made
    /// packaged windows write-only — if it ever starts passing under bincode,
    /// the fallback ordering in the restore path deserves a rethink.
    #[test]
    fn bincode_cannot_round_trip_a_value_window() {
        use std::collections::VecDeque;
        let window: VecDeque<serde_json::Value> =
            vec![serde_json::json!(11), serde_json::json!(22)].into();
        let bytes = types::serialize(&window).expect("serializing succeeds — that is the trap");
        assert!(
            types::deserialize::<VecDeque<serde_json::Value>>(&bytes).is_err(),
            "bincode deserialized a serde_json::Value — the JSON buffer format \
             and its legacy fallback should be revisited"
        );
    }

    /// The fix: JSON round-trips every T the runtime accepts, including Value.
    #[test]
    fn json_round_trips_both_value_and_typed_windows() {
        use std::collections::VecDeque;
        let vals: VecDeque<serde_json::Value> =
            vec![serde_json::json!({"p": 1.5}), serde_json::json!(7)].into();
        let bytes = serde_json::to_vec(&vals).unwrap();
        let back: VecDeque<serde_json::Value> = serde_json::from_slice(&bytes).unwrap();
        assert_eq!(vals, back);

        let nums: VecDeque<u64> = vec![11, 22, 33].into();
        let bytes = serde_json::to_vec(&nums).unwrap();
        let back: VecDeque<u64> = serde_json::from_slice(&bytes).unwrap();
        assert_eq!(nums, back);
    }

    /// Legacy migration: a bincode row written by an older build with a
    /// CONCRETE T must still restore through the fallback arm.
    #[test]
    fn legacy_bincode_typed_rows_still_decode_via_the_fallback() {
        use std::collections::VecDeque;
        let nums: VecDeque<u64> = vec![1, 2, 3].into();
        let legacy = types::serialize(&nums).unwrap();
        // Mirrors the restore path's ordering: JSON first, bincode second.
        let restored = serde_json::from_slice::<VecDeque<u64>>(&legacy)
            .ok()
            .or_else(|| types::deserialize::<VecDeque<u64>>(&legacy).ok())
            .expect("legacy typed row must decode via the bincode fallback");
        assert_eq!(restored, nums);
    }
}