//! `CumSum` — running total of one `f64` input with O(1) scalar state.
//!
//! The generic run-length accumulator (#281). The domain-embedded precedents
//! (`SimBroker.cum`, `PositionManagement.cum_realized_r`, `CostRunner.cum`)
//! stay as they are; this cell is the vocabulary form, so counters compose:
//! `count_true = CumSum(Select(cond, 1.0, 0.0))`.
//!
//! The running sum carries a compensation term for the same reason
//! [`Sma`](crate::Sma) documents: an accumulative running sum has nowhere for
//! rounding error to go, so over millions of adds an uncompensated total
//! drifts. The compensated total is folded in at the read (`sum + comp`); a
//! `NaN` input poisons the total (same posture as `Sma`, not guarded).
//!
//! **Neumaier, not Sma's plain Kahan.** `Sma`'s private `kahan()` helper uses
//! the branchless Kahan formula, sound because a windowed mean's addends stay
//! the same order of magnitude as the running window sum. A free-standing
//! accumulator has no such guarantee — a large running total can absorb a
//! comparatively tiny sample (`1e16 + 1.0`), or the reverse — so this cell uses
//! the branching Neumaier variant, which stays exact in *both* magnitude orders
//! (`|sum| >= |x|` and `|x| > |sum|`). Plain Kahan drops the small addend in the
//! latter case (`[1e16, 1.0, -1e16]` -> `0.0`, not `1.0`); Neumaier keeps it.
//! That is why the helper is not reused here — the algorithms differ, and this
//! one is deliberately the more robust of the two.
//!
//! One `f64` input (`Firing::Any`), one `f64` output, no params. Emits from
//! the first sample (`cum = x0`); state advances once per fired eval (the
//! engine's freshness gating guarantees one eval per fresh sample).

use aura_core::{Cell, Ctx, FieldSpec, Firing, Node, NodeSchema, PortSpec, PrimitiveBuilder, ScalarKind};

/// Neumaier-compensated running total: emits `x0 + x1 + ... + xn` at each
/// fired cycle. Emits `None` until the input is warm (C8).
pub struct CumSum {
    // Running sum and its Neumaier compensation term (the low-order bits each
    // add dropped, folded into the emitted total at the read).
    sum: f64,
    comp: f64,
    out: [Cell; 1],
}

impl CumSum {
    /// Build a `CumSum` node, initially zero.
    pub fn new() -> Self {
        Self { sum: 0.0, comp: 0.0, out: [Cell::from_f64(0.0)] }
    }

    /// The param-generic recipe for a blueprint primitive: paramless, builds
    /// through `CumSum::new`.
    pub fn builder() -> PrimitiveBuilder {
        PrimitiveBuilder::new(
            "CumSum",
            NodeSchema {
                inputs: vec![PortSpec { kind: ScalarKind::F64, firing: Firing::Any, name: "series".into() }],
                output: vec![FieldSpec { name: "value".into(), kind: ScalarKind::F64 }],
                params: vec![],
            },
            |_| Box::new(CumSum::new()),
        )
    }
}

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

impl Node for CumSum {
    fn lookbacks(&self) -> Vec<usize> {
        vec![1]
    }

    fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> {
        let v = ctx.f64_in(0);
        if v.is_empty() {
            return None; // warm-up gate (C8)
        }
        let x = v[0];
        // Neumaier add: capture the low-order bits the naive add drops.
        let t = self.sum + x;
        if self.sum.abs() >= x.abs() {
            self.comp += (self.sum - t) + x;
        } else {
            self.comp += (x - t) + self.sum;
        }
        self.sum = t;
        self.out[0] = Cell::from_f64(self.sum + self.comp);
        Some(&self.out)
    }

    fn label(&self) -> String {
        "CumSum".to_string()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use aura_core::{AnyColumn, Scalar, Timestamp};

    #[test]
    fn cumsum_accumulates_from_the_first_sample() {
        // PROPERTY: emits the running total from the very first sample on,
        // advancing once per fired eval. Exact-representable values keep the
        // compensation term at zero, so the expectations are exact.
        let mut node = CumSum::new();
        let mut inputs = vec![AnyColumn::with_capacity(ScalarKind::F64, 1)];
        let feed = [2.0, -0.5, 3.5, 0.0];
        let want = [2.0, 1.5, 5.0, 5.0];
        for (x, w) in feed.iter().zip(want) {
            inputs[0].push(Scalar::f64(*x)).unwrap();
            let got = node.eval(Ctx::new(&inputs, Timestamp(0)));
            assert_eq!(got, Some([Cell::from_f64(w)].as_slice()));
        }
    }

    #[test]
    fn cumsum_is_none_until_warm() {
        let mut node = CumSum::new();
        let inputs = vec![AnyColumn::with_capacity(ScalarKind::F64, 1)];
        assert_eq!(node.eval(Ctx::new(&inputs, Timestamp(0))), None);
    }

    #[test]
    fn cumsum_compensation_survives_magnitude_spread() {
        // The Neumaier pin: 1e16 + 1.0 == 1e16 in naive f64 (the 1.0 is below
        // the ULP), so a naive running sum over [1e16, 1.0, -1e16] ends at 0.0.
        // The compensated total keeps the small addend and ends at exactly 1.0.
        let mut node = CumSum::new();
        let mut inputs = vec![AnyColumn::with_capacity(ScalarKind::F64, 1)];
        let mut last = 0.0;
        for x in [1e16, 1.0, -1e16] {
            inputs[0].push(Scalar::f64(x)).unwrap();
            last = node.eval(Ctx::new(&inputs, Timestamp(0))).unwrap()[0].f64();
        }
        assert_eq!(last, 1.0, "the compensated running sum keeps the small addend");
    }

    #[test]
    fn input_slot_is_named_series() {
        let names: Vec<String> =
            CumSum::builder().schema().inputs.iter().map(|p| p.name.clone()).collect();
        assert_eq!(names, ["series"]);
    }
}
