freenet 0.2.82

use crate::ring::{Distance, Location, PeerKeyLocation};
use pav_regression::IsotonicRegression;
use pav_regression::Point;
use serde::Serialize;
use std::collections::{HashMap, VecDeque};

const MIN_POINTS_FOR_REGRESSION: usize = 5;

/// Maximum number of raw data points retained by the global regression.
/// Once reached, each new point evicts the oldest via `remove_points`.
const MAX_REGRESSION_POINTS: usize = 500;

/// EWMA smoothing factor for per-peer adjustments.
/// Alpha = 0.1 gives a half-life of ~6.6 events, meaning the influence of an
/// observation drops below 50% after about 7 newer observations.
const EWMA_ALPHA: f64 = 0.1;

/// Floor for the global base in [`AdjustmentMode::Multiplicative`]. The global
/// regression can extrapolate slightly negative near the edges of its data range;
/// clamping that base to exactly `0.0` before a multiplicative adjustment would
/// annihilate the peer's factor (`0 * exp(adj) == 0`), predicting a slow peer as
/// instant. A tiny positive floor keeps `base * exp(adj)` ordered by the per-peer
/// factor. It is far below any real response time (1 ns) / transfer rate, so it
/// only affects the degenerate `global <= 0` region.
const MULTIPLICATIVE_MIN_BASE: f64 = 1e-9;

/// `IsotonicEstimator` provides outcome estimation for a given action, such as
/// retrieving the state of a contract, based on the distance between the peer
/// and the contract. It uses an isotonic regression model from the `pav.rs`
/// library to estimate the outcome based on the distance between the peer and
/// the contract, but then also tracks an adjustment for each peer based on the
/// outcome of the peer's previous requests.
///
/// The global regression uses a rolling window: once `MAX_REGRESSION_POINTS`
/// raw points have been accumulated, each new point evicts the oldest via
/// `remove_points`. Per-peer adjustments use an exponentially-weighted moving
/// average (EWMA) so recent events have more influence than old ones.

#[derive(Debug, Clone, Serialize)]
pub(crate) struct IsotonicEstimator {
    pub global_regression: IsotonicRegression<f64>,
    pub peer_adjustments: HashMap<PeerKeyLocation, Adjustment>,
    /// How per-peer adjustments combine with the global estimate. See
    /// [`AdjustmentMode`].
    #[serde(skip)]
    adjustment_mode: AdjustmentMode,
    /// Raw input points in insertion order. When len exceeds
    /// `MAX_REGRESSION_POINTS`, the oldest is evicted via `remove_points`.
    #[serde(skip)]
    raw_points: VecDeque<Point<f64>>,
}

/// How a per-peer adjustment combines with the global isotonic estimate.
///
/// The per-peer adjustment is an EWMA that corrects the global distance→outcome
/// fit for a specific peer. Whether that correction is best expressed as an
/// absolute offset or a scaling factor depends on the target:
///
/// - **Additive** (`global + adjustment`): the EWMA averages absolute residuals
///   `observed - global`. Correct for a bounded target such as failure
///   probability, where "this peer fails 0.05 more often" is the natural unit.
/// - **Multiplicative** (`global * exp(adjustment)`): the EWMA averages log
///   ratios `ln(observed) - ln(global)` (a geometric mean of `observed/global`).
///   Correct for an unbounded, heavy-tailed, multiplicative-scale target such as
///   response time. Telemetry over 5.3 days / 631 peers showed a peer's deviation
///   from the global response-time curve is a near-constant *ratio*, not a
///   constant offset (log-residuals are level-independent for ~88% of peers,
///   96% observation-weighted, whereas additive residuals grow with the level).
///   It also makes a negative estimate impossible by construction: `global >= 0`
///   and `exp(_) > 0`.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub(crate) enum AdjustmentMode {
    #[default]
    Additive,
    Multiplicative,
}

impl AdjustmentMode {
    /// The per-event training residual fed to the peer's EWMA, or `None` when the
    /// event cannot be expressed in this mode's space. Multiplicative requires a
    /// strictly-positive observation and global estimate (`ln` is undefined at or
    /// below zero); such events are skipped rather than corrupting the EWMA with
    /// `NaN`/`-inf`.
    fn residual(self, observed: f64, global: f64) -> Option<f64> {
        match self {
            AdjustmentMode::Additive => Some(observed - global),
            AdjustmentMode::Multiplicative => {
                if observed > 0.0 && global > 0.0 {
                    Some(observed.ln() - global.ln())
                } else {
                    None
                }
            }
        }
    }

    /// Combine a global estimate with a peer's smoothed adjustment value. The
    /// neutral adjustment is `0.0` in both modes (`global + 0` and
    /// `global * e^0 = global`), so callers can pass `0.0` for a peer that has no
    /// usable adjustment yet. Shared by the router and the dashboard so both
    /// render the identical peer-adjusted value.
    pub(crate) fn apply(self, global: f64, adjustment: f64) -> f64 {
        match self {
            AdjustmentMode::Additive => global + adjustment,
            AdjustmentMode::Multiplicative => global * adjustment.exp(),
        }
    }

    /// Clamp the global estimate to a non-negative base before applying the
    /// adjustment. The regression can extrapolate slightly negative near its
    /// data-range edges; both modes must start from a non-negative base.
    ///
    /// Additive floors at exactly `0.0` (`0 + adj` still carries the peer's
    /// absolute correction). Multiplicative floors at a tiny POSITIVE value
    /// instead: flooring to `0.0` would make `0 * exp(adj) == 0` and erase the
    /// peer's learned factor entirely — a consistently-slow peer would be
    /// predicted as instant. The tiny floor keeps `base * exp(adj)` ordered by the
    /// per-peer factor in that degenerate region while staying far below any real
    /// value, so normal (positive-base) predictions are unchanged.
    fn floor_base(self, global: f64) -> f64 {
        match self {
            AdjustmentMode::Additive => global.max(0.0),
            AdjustmentMode::Multiplicative => global.max(MULTIPLICATIVE_MIN_BASE),
        }
    }
}

impl IsotonicEstimator {
    // Minimum sample size before we apply per-peer adjustments; keeps peer curves from being
    // dominated by sparse/noisy data.
    const ADJUSTMENT_PRIOR_SIZE: u64 = 10;

    /// Creates a new `IsotonicEstimator` from a list of historical events, using
    /// [`AdjustmentMode::Additive`] per-peer adjustments.
    pub fn new<I>(history: I, estimator_type: EstimatorType) -> Self
    where
        I: IntoIterator<Item = IsotonicEvent>,
    {
        Self::new_with_mode(history, estimator_type, AdjustmentMode::Additive)
    }

    /// Like [`new`](Self::new) but selects how per-peer adjustments combine with
    /// the global estimate. The mode must be fixed at construction because the
    /// per-peer EWMA is trained on residuals computed in that mode's space (see
    /// [`AdjustmentMode::residual`]); it cannot be changed afterwards without
    /// recomputing every peer's history.
    pub fn new_with_mode<I>(
        history: I,
        estimator_type: EstimatorType,
        adjustment_mode: AdjustmentMode,
    ) -> Self
    where
        I: IntoIterator<Item = IsotonicEvent>,
    {
        let mut all_events: Vec<IsotonicEvent> = history.into_iter().collect();

        // If history exceeds the window, keep only the most recent events.
        // Both the regression points and the peer adjustment deltas are computed
        // from the same windowed subset to avoid stale-data bias.
        if all_events.len() > MAX_REGRESSION_POINTS {
            all_events.drain(..all_events.len() - MAX_REGRESSION_POINTS);
        }

        let mut all_points = VecDeque::with_capacity(all_events.len());
        let mut peer_events: HashMap<PeerKeyLocation, Vec<IsotonicEvent>> = HashMap::new();

        for event in all_events {
            let point = Point::new(event.route_distance().as_f64(), event.result);
            all_points.push_back(point);
            peer_events
                .entry(event.peer.clone())
                .or_default()
                .push(event);
        }

        let points: Vec<Point<f64>> = all_points.iter().cloned().collect();
        let global_regression = match estimator_type {
            EstimatorType::Positive => IsotonicRegression::new_ascending(&points),
            EstimatorType::Negative => IsotonicRegression::new_descending(&points),
        }
        .expect("Failed to create isotonic regression");

        let global_regression_big_enough =
            global_regression.len() >= Self::ADJUSTMENT_PRIOR_SIZE as usize;

        let mut peer_adjustments: HashMap<PeerKeyLocation, Adjustment> = HashMap::new();

        if global_regression_big_enough {
            for (peer_location, events) in peer_events.iter() {
                let mut adjustment = Adjustment::new();
                // Seed with ADJUSTMENT_PRIOR_SIZE phantom neutral observations
                // so peers with few real observations are shrunk toward zero.
                adjustment.effective_count = Self::ADJUSTMENT_PRIOR_SIZE as f64;

                for event in events {
                    let global_estimate = global_regression
                        .interpolate(event.route_distance().as_f64())
                        .expect("Regression should always produce an estimate");
                    if let Some(delta) = adjustment_mode.residual(event.result, global_estimate) {
                        adjustment.add(delta);
                    }
                }
                peer_adjustments.insert(peer_location.clone(), adjustment);
            }
        }

        IsotonicEstimator {
            global_regression,
            peer_adjustments,
            adjustment_mode,
            raw_points: all_points,
        }
    }

    /// Adds a new event to the estimator.
    pub fn add_event(&mut self, event: IsotonicEvent) {
        let route_distance = event.route_distance();
        let point = Point::new(route_distance.as_f64(), event.result);

        // Add the new point to the regression and raw-point FIFO.
        self.global_regression.add_points(&[point]);
        self.raw_points.push_back(point);

        // Evict the oldest point if the window is full.
        if self.raw_points.len() > MAX_REGRESSION_POINTS {
            if let Some(oldest) = self.raw_points.pop_front() {
                self.global_regression.remove_points(&[oldest]);
            }
        }

        if self.global_regression.len() >= Self::ADJUSTMENT_PRIOR_SIZE as usize {
            let global_estimate = self
                .global_regression
                .interpolate(route_distance.as_f64())
                .unwrap();

            if let Some(delta) = self.adjustment_mode.residual(event.result, global_estimate) {
                self.peer_adjustments
                    .entry(event.peer)
                    .or_default()
                    .add(delta);
            }
        }
    }

    pub fn estimate_retrieval_time(
        &self,
        peer: &PeerKeyLocation,
        contract_location: Location,
    ) -> Result<f64, EstimationError> {
        if self.global_regression.len() < MIN_POINTS_FOR_REGRESSION {
            return Err(EstimationError::InsufficientData);
        }

        let peer_location = peer.location().ok_or(EstimationError::InsufficientData)?;
        let distance: f64 = contract_location.distance(peer_location).as_f64();

        let global_estimate = self
            .global_regression
            .interpolate(distance)
            .ok_or(EstimationError::InsufficientData)?;

        // Regression can sometimes produce negative estimates. Floor the base
        // non-negative before applying the per-peer adjustment; in multiplicative
        // mode the floor is a tiny positive value so the peer's factor is not
        // annihilated (see `AdjustmentMode::floor_base`).
        let global_estimate = self.adjustment_mode.floor_base(global_estimate);

        let adjusted_estimate =
            self.peer_adjustments
                .get(peer)
                .map_or(global_estimate, |peer_adjustment| {
                    let should_use_peer_adjustment =
                        peer_adjustment.effective_count >= MIN_POINTS_FOR_REGRESSION as f64;
                    if should_use_peer_adjustment {
                        self.adjustment_mode
                            .apply(global_estimate, peer_adjustment.value())
                    } else {
                        global_estimate
                    }
                });

        // The per-peer adjustment is applied *after* the global clamp above. In
        // additive mode it can be negative (a peer faster / more reliable than the
        // global fit); in multiplicative mode `global * exp(_)` is already >= 0.
        // Re-clamp either way so the per-peer estimate can never go below zero —
        // these targets (response time, transfer rate, failure probability) are
        // all physically non-negative, and a negative prediction would both
        // distort routing cost formulas and render below the x-axis on the
        // dashboard's "Peer-adjusted" curve. The failure path applies the same
        // clamp downstream (see `predict_routing_outcome`).
        Ok(adjusted_estimate.max(0.0))
    }

    pub(crate) fn len(&self) -> usize {
        self.global_regression.len()
    }

    /// The per-peer adjustment mode this estimator was constructed with.
    /// Test-only: used to pin each estimator's mode (see the router test
    /// `estimators_use_intended_adjustment_modes`). When the dashboard is wired to
    /// read the mode (the #4547 follow-up), drop the `cfg(test)` to make it real API.
    #[cfg(test)]
    pub(crate) fn adjustment_mode(&self) -> AdjustmentMode {
        self.adjustment_mode
    }

    /// Return the x-range of actual regression data points, or (0, 0) if empty.
    pub(crate) fn data_x_range(&self) -> (f64, f64) {
        let sorted = self.global_regression.get_points_sorted();
        if sorted.is_empty() {
            return (0.0, 0.0);
        }
        (*sorted.first().unwrap().x(), *sorted.last().unwrap().x())
    }

    /// Sample the regression's `interpolate()` across the full distance range [0, 0.5],
    /// clamping outputs to `[y_clamp_min, y_clamp_max]`. This produces the actual
    /// predictions the estimator would make, including centroid-based extrapolation
    /// beyond the data range.
    ///
    /// Returns `(sampled_points, data_x_min, data_x_max)` where `data_x_min/max`
    /// are the bounds of the actual regression data (for distinguishing interpolation
    /// from extrapolation in charts).
    /// Requires `num_samples >= 2` to produce a meaningful curve.
    pub(crate) fn sampled_curve(
        &self,
        y_clamp_min: f64,
        y_clamp_max: f64,
        num_samples: usize,
    ) -> Vec<(f64, f64)> {
        if num_samples < 2 || self.global_regression.get_points_sorted().is_empty() {
            return Vec::new();
        }

        let mut points = Vec::with_capacity(num_samples);
        for i in 0..num_samples {
            let x = (i as f64 / (num_samples - 1) as f64) * 0.5;
            if let Some(y) = self.global_regression.interpolate(x) {
                points.push((x, y.clamp(y_clamp_min, y_clamp_max)));
            }
        }

        points
    }

    /// Downsampled raw `(distance, outcome)` observations for visualization, in
    /// insertion order, at most `max` points (evenly strided across the retained
    /// window). These are the actual events the isotonic fit is computed from, so
    /// the dashboard can show the scatter behind the curve. Empty when no data.
    pub(crate) fn sampled_raw_points(&self, max: usize) -> Vec<(f64, f64)> {
        let n = self.raw_points.len();
        if n == 0 || max == 0 {
            return Vec::new();
        }
        if n <= max {
            return self.raw_points.iter().map(|p| (*p.x(), *p.y())).collect();
        }
        let stride = n as f64 / max as f64;
        (0..max)
            .map(|i| {
                let idx = ((i as f64 * stride) as usize).min(n - 1);
                let p = &self.raw_points[idx];
                (*p.x(), *p.y())
            })
            .collect()
    }
}

#[derive(Debug, Clone)]
pub(crate) enum EstimatorType {
    /// Where the estimated value is expected to increase as distance increases
    Positive,
    /// Where the estimated value is expected to decrease as distance increases
    Negative,
}

#[derive(Debug, PartialEq, Eq, thiserror::Error)]
pub(crate) enum EstimationError {
    #[error("Insufficient data for estimation")]
    InsufficientData,
}

/// A routing event is a single request to a peer for a contract, and some value indicating
/// the result of the request, such as the time it took to retrieve the contract.
#[derive(Debug, Clone)]
pub(crate) struct IsotonicEvent {
    pub peer: PeerKeyLocation,
    pub contract_location: Location,
    /// The result of the routing event, which is used to train the estimator, typically the time
    /// but could also represent request success as 0.0 and failure as 1.0, and then be used
    /// to predict the probability of success.
    pub result: f64,
}

impl IsotonicEvent {
    fn route_distance(&self) -> Distance {
        let peer_location = self
            .peer
            .location()
            .ok_or(EstimationError::InsufficientData)
            .expect("IsotonicEvent should always carry a peer location");
        self.contract_location.distance(peer_location)
    }
}

/// Per-peer adjustment using an exponentially-weighted moving average (EWMA).
///
/// Each new observation is blended with the running average:
///   smoothed = alpha * new_value + (1 - alpha) * smoothed
///
/// `effective_count` tracks the decayed sample size so callers can decide
/// whether the peer has enough data to be trusted.
#[derive(Debug, Clone, Serialize)]
pub(crate) struct Adjustment {
    smoothed: f64,
    effective_count: f64,
    #[serde(skip)]
    alpha: f64,
}

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

impl Adjustment {
    fn new() -> Self {
        Self {
            smoothed: 0.0,
            effective_count: 0.0,
            alpha: EWMA_ALPHA,
        }
    }

    fn add(&mut self, value: f64) {
        if self.effective_count < 1.0 {
            // First real observation — set directly rather than blending with zero.
            self.smoothed = value;
        } else {
            self.smoothed = self.alpha * value + (1.0 - self.alpha) * self.smoothed;
        }
        // Decay the effective count and add 1 for this new observation.
        self.effective_count = 1.0 + (1.0 - self.alpha) * self.effective_count;
    }

    /// EWMA smoothed adjustment value.
    pub(crate) fn value(&self) -> f64 {
        self.smoothed
    }

    /// Effective number of events contributing to this adjustment (decayed).
    pub(crate) fn event_count(&self) -> u64 {
        self.effective_count.round() as u64
    }
}

// Tests

#[cfg(test)]
mod tests {

    use super::*;
    use tracing::debug;

    #[test]
    fn test_positive_peer_time_estimator() {
        let mut events = Vec::new();
        for _ in 0..100 {
            let peer = PeerKeyLocation::random();
            if peer.location().is_none() {
                debug!("Peer location is none for {peer:?}");
            }
            let contract_location = Location::random();
            events.push(simulate_positive_request(peer, contract_location));
        }

        let (training_events, testing_events) = events.split_at(events.len() / 2);

        let estimator =
            IsotonicEstimator::new(training_events.iter().cloned(), EstimatorType::Positive);

        let mut errors = Vec::new();
        for event in testing_events {
            let estimated_time = estimator
                .estimate_retrieval_time(&event.peer, event.contract_location)
                .unwrap();
            let actual_time = event.result;
            let error = (estimated_time - actual_time).abs();
            errors.push(error);
        }

        let average_error = errors.iter().sum::<f64>() / errors.len() as f64;
        debug!("Average error: {average_error}");
        // Threshold 0.02 to avoid flaky failures from random seed variation
        assert!(average_error < 0.02);
    }

    #[test]
    fn test_negative_peer_time_estimator() {
        let mut events = Vec::new();
        for _ in 0..100 {
            let peer = PeerKeyLocation::random();
            if peer.location().is_none() {
                debug!("Peer location is none for {peer:?}");
            }
            let contract_location = Location::random();
            events.push(simulate_negative_request(peer, contract_location));
        }

        let (training_events, testing_events) = events.split_at(events.len() / 2);

        let estimator =
            IsotonicEstimator::new(training_events.iter().cloned(), EstimatorType::Negative);

        let mut errors = Vec::new();
        for event in testing_events {
            let estimated_time = estimator
                .estimate_retrieval_time(&event.peer, event.contract_location)
                .unwrap();
            let actual_time = event.result;
            let error = (estimated_time - actual_time).abs();
            errors.push(error);
        }

        let average_error = errors.iter().sum::<f64>() / errors.len() as f64;
        debug!("Average error: {average_error}");
        // Threshold 0.02 to avoid flaky failures from random seed variation
        assert!(average_error < 0.02);
    }

    #[test]
    fn test_peer_adjustment_cannot_produce_negative_estimate() {
        // A strongly-negative per-peer EWMA adjustment (a peer far faster / more
        // reliable than the global fit) must not drive the estimate below zero:
        // these targets are physically non-negative, and a negative estimate both
        // distorts routing and renders below the x-axis on the dashboard chart.
        let mut events = Vec::new();
        for _ in 0..100 {
            let peer = PeerKeyLocation::random();
            let contract_location = Location::random();
            events.push(simulate_positive_request(peer, contract_location));
        }
        let mut estimator = IsotonicEstimator::new(events, EstimatorType::Positive);

        // Overwrite one peer's adjustment with a large negative value and enough
        // effective observations that it is actually applied.
        let fast_peer = PeerKeyLocation::random();
        let mut adjustment = Adjustment::new();
        for _ in 0..10 {
            adjustment.add(-1000.0);
        }
        assert!(
            adjustment.effective_count >= MIN_POINTS_FOR_REGRESSION as f64,
            "adjustment must have enough effective observations to be applied"
        );
        assert!(
            adjustment.value() < -1.0,
            "adjustment must be strongly negative"
        );
        estimator
            .peer_adjustments
            .insert(fast_peer.clone(), adjustment);

        // Estimate at the peer's own location (distance 0 → smallest global value).
        let contract_location = fast_peer.location().unwrap();
        let estimate = estimator
            .estimate_retrieval_time(&fast_peer, contract_location)
            .expect("estimator has enough data to produce an estimate");

        assert!(
            estimate >= 0.0,
            "per-peer adjusted estimate must be clamped to a non-negative value, got {estimate}"
        );
    }

    #[test]
    fn adjustment_mode_residual_and_apply() {
        // Additive: residual is the absolute difference; apply adds it back.
        assert_eq!(
            AdjustmentMode::Additive.residual(7.0, 10.0),
            Some(-3.0),
            "additive residual is observed - global"
        );
        assert_eq!(AdjustmentMode::Additive.apply(10.0, -3.0), 7.0);

        // Multiplicative: residual is the log-ratio; apply scales by exp().
        let r = AdjustmentMode::Multiplicative
            .residual(20.0, 10.0)
            .expect("positive inputs");
        assert!(
            (r - std::f64::consts::LN_2).abs() < 1e-12,
            "multiplicative residual of 20/10 must be ln(2), got {r}"
        );
        let applied = AdjustmentMode::Multiplicative.apply(10.0, std::f64::consts::LN_2);
        assert!(
            (applied - 20.0).abs() < 1e-9,
            "applying ln(2) to 10 must give 20, got {applied}"
        );

        // Multiplicative is undefined for non-positive inputs → skipped (None).
        assert_eq!(AdjustmentMode::Multiplicative.residual(0.0, 10.0), None);
        assert_eq!(AdjustmentMode::Multiplicative.residual(5.0, 0.0), None);
        assert_eq!(AdjustmentMode::Multiplicative.residual(-1.0, 10.0), None);

        // Neutral adjustment (0.0) is a no-op in both modes.
        assert_eq!(AdjustmentMode::Additive.apply(42.0, 0.0), 42.0);
        assert_eq!(AdjustmentMode::Multiplicative.apply(42.0, 0.0), 42.0);

        // Multiplicative can never produce a negative estimate, even for a huge
        // negative log-adjustment (the additive failure mode this design avoids).
        assert!(AdjustmentMode::Multiplicative.apply(10.0, -1000.0) >= 0.0);
    }

    #[test]
    fn multiplicative_estimate_scales_global_by_geometric_factor() {
        // Build a multiplicative-mode estimator with a real global fit.
        let mut events = Vec::new();
        for _ in 0..100 {
            let peer = PeerKeyLocation::random();
            let contract_location = Location::random();
            events.push(simulate_positive_request(peer, contract_location));
        }
        let mut estimator = IsotonicEstimator::new_with_mode(
            events,
            EstimatorType::Positive,
            AdjustmentMode::Multiplicative,
        );

        // A peer whose log-adjustment is ln(2): it is consistently ~2x the global.
        let peer = PeerKeyLocation::random();
        let mut adjustment = Adjustment::new();
        for _ in 0..10 {
            adjustment.add(std::f64::consts::LN_2);
        }
        assert!(adjustment.effective_count >= MIN_POINTS_FOR_REGRESSION as f64);
        estimator.peer_adjustments.insert(peer.clone(), adjustment);

        // At the peer's own location distance is 0; compare against global * 2.
        let contract_location = peer.location().unwrap();
        let global = estimator
            .global_regression
            .interpolate(0.0)
            .unwrap()
            .max(0.0);
        let estimate = estimator
            .estimate_retrieval_time(&peer, contract_location)
            .expect("enough data");
        let expected = global * 2.0;
        assert!(
            (estimate - expected).abs() <= 1e-6 + expected * 1e-9,
            "multiplicative estimate must be global*exp(ln2)=2*global ({expected}), got {estimate}"
        );

        // A strongly-negative log-adjustment scales toward (but not below) zero.
        let mut neg = Adjustment::new();
        for _ in 0..10 {
            neg.add(-1000.0);
        }
        estimator.peer_adjustments.insert(peer.clone(), neg);
        let est_neg = estimator
            .estimate_retrieval_time(&peer, contract_location)
            .expect("enough data");
        assert!(
            (0.0..1e-3).contains(&est_neg),
            "global*exp(-1000) must round to ~0 and stay non-negative, got {est_neg}"
        );
    }

    #[test]
    fn multiplicative_adjustment_via_add_event_orders_peers_by_ratio() {
        // End-to-end: build the multiplicative adjustment through the production
        // path — `add_event` -> `residual` (log-ratio) -> EWMA -> `apply` — rather
        // than hand-inserting an `Adjustment`. A slow peer (results ~2.0) must end
        // up predicted slower than a fast peer (results ~0.5) at the same distance.
        // This is the path a sign error or `ln`/`exp` inversion would corrupt.
        let mut estimator = IsotonicEstimator::new_with_mode(
            std::iter::empty(),
            EstimatorType::Positive,
            AdjustmentMode::Multiplicative,
        );
        for _ in 0..40 {
            estimator.add_event(IsotonicEvent {
                peer: PeerKeyLocation::random(),
                contract_location: Location::random(),
                result: 1.0,
            });
        }
        let fast_peer = PeerKeyLocation::random();
        let slow_peer = PeerKeyLocation::random();
        for _ in 0..20 {
            estimator.add_event(IsotonicEvent {
                peer: fast_peer.clone(),
                contract_location: Location::random(),
                result: 0.5,
            });
            estimator.add_event(IsotonicEvent {
                peer: slow_peer.clone(),
                contract_location: Location::random(),
                result: 2.0,
            });
        }
        // Both query at distance 0 (contract at own location) → identical global
        // base, so the ordering is decided purely by each peer's learned factor.
        let est_fast = estimator
            .estimate_retrieval_time(&fast_peer, fast_peer.location().unwrap())
            .expect("enough data");
        let est_slow = estimator
            .estimate_retrieval_time(&slow_peer, slow_peer.location().unwrap())
            .expect("enough data");
        assert!(
            est_slow > est_fast && est_fast > 0.0,
            "slow peer (2.0) must estimate higher than fast peer (0.5): slow={est_slow} fast={est_fast}"
        );
    }

    #[test]
    fn multiplicative_skips_non_positive_observations() {
        // A zero/negative response time can't be expressed in log space; `residual`
        // returns None and `add_event` must skip the peer entirely rather than
        // poisoning its EWMA with NaN/-inf.
        let mut estimator = IsotonicEstimator::new_with_mode(
            std::iter::empty(),
            EstimatorType::Positive,
            AdjustmentMode::Multiplicative,
        );
        for _ in 0..15 {
            estimator.add_event(IsotonicEvent {
                peer: PeerKeyLocation::random(),
                contract_location: Location::random(),
                result: 1.0,
            });
        }
        let peer = PeerKeyLocation::random();
        estimator.add_event(IsotonicEvent {
            peer: peer.clone(),
            contract_location: Location::random(),
            result: 0.0,
        });
        assert!(
            estimator.peer_adjustments.get(&peer).is_none(),
            "a non-positive observation must be skipped in multiplicative mode (no adjustment entry created)"
        );
    }

    #[test]
    fn floor_base_preserves_multiplicative_factor_at_degenerate_base() {
        // Additive floors at exactly 0; multiplicative floors at a tiny positive
        // value so a degenerate (clamped) base does not annihilate the peer factor.
        assert_eq!(AdjustmentMode::Additive.floor_base(-1.0), 0.0);
        assert_eq!(AdjustmentMode::Additive.floor_base(5.0), 5.0);
        assert_eq!(AdjustmentMode::Multiplicative.floor_base(5.0), 5.0);
        let base = AdjustmentMode::Multiplicative.floor_base(-1.0);
        assert!(
            base > 0.0,
            "multiplicative base must stay positive, got {base}"
        );

        // The fix: at a clamped base, a slow peer (adj>0) is still predicted slower
        // than a fast peer (adj<0). Flooring to 0 (the bug) would tie both at 0.
        let slow = AdjustmentMode::Multiplicative.apply(base, 0.7);
        let fast = AdjustmentMode::Multiplicative.apply(base, -0.7);
        assert!(
            slow > fast && fast > 0.0,
            "multiplicative ordering must survive a degenerate base: slow={slow} fast={fast}"
        );
        assert_eq!(
            AdjustmentMode::Multiplicative.apply(0.0, 0.7),
            0.0,
            "sanity: a literal-0 base would annihilate the factor — the bug floor_base avoids"
        );
    }

    #[test]
    fn test_adjustment_ewma_recency() {
        let mut adj = Adjustment::new();

        // Feed 100 events with value 10.0 (simulating a "bad" period)
        for _ in 0..100 {
            adj.add(10.0);
        }
        let after_bad = adj.value();
        assert!(
            (after_bad - 10.0).abs() < 0.01,
            "EWMA should converge to 10.0, got {after_bad}"
        );

        // Now feed 20 events with value 0.0 (simulating recovery)
        for _ in 0..20 {
            adj.add(0.0);
        }
        let after_recovery = adj.value();
        // (1-0.1)^20 ≈ 0.12, so ~88% of the old 10.0 has decayed.
        assert!(
            after_recovery < 2.0,
            "EWMA should reflect recent 0.0 events after 20 observations, got {after_recovery}"
        );
    }

    #[test]
    fn test_adjustment_ewma_first_observation() {
        let mut adj = Adjustment {
            alpha: 0.5,
            ..Adjustment::new()
        };
        adj.add(5.0);
        assert_eq!(adj.value(), 5.0, "First observation should be set directly");
        assert_eq!(adj.event_count(), 1);

        adj.add(3.0);
        // alpha * 3.0 + (1-alpha) * 5.0 = 0.5 * 3.0 + 0.5 * 5.0 = 4.0
        assert!(
            (adj.value() - 4.0).abs() < 1e-10,
            "Second observation should blend via EWMA"
        );
    }

    #[test]
    fn test_rolling_window_eviction() {
        let peer = PeerKeyLocation::random();
        let contract = Location::random();

        let mut estimator = IsotonicEstimator::new(std::iter::empty(), EstimatorType::Positive);

        // Add more events than MAX_REGRESSION_POINTS
        for i in 0..(MAX_REGRESSION_POINTS + 100) {
            estimator.add_event(IsotonicEvent {
                peer: peer.clone(),
                contract_location: contract,
                result: i as f64,
            });
        }

        assert!(
            estimator.raw_points.len() <= MAX_REGRESSION_POINTS,
            "Raw points should be bounded, got {}",
            estimator.raw_points.len()
        );

        let result = estimator.estimate_retrieval_time(&peer, contract);
        assert!(
            result.is_ok(),
            "Estimator should produce estimates after eviction"
        );
    }

    #[test]
    fn sampled_raw_points_downsamples_and_bounds() {
        let peer = PeerKeyLocation::random();
        let contract = Location::random();
        let mut estimator = IsotonicEstimator::new(std::iter::empty(), EstimatorType::Positive);

        // No data yet.
        assert!(estimator.sampled_raw_points(50).is_empty());

        for i in 0..40 {
            estimator.add_event(IsotonicEvent {
                peer: peer.clone(),
                contract_location: contract,
                result: i as f64,
            });
        }

        // Fewer than the cap -> every retained point is returned.
        assert_eq!(estimator.sampled_raw_points(100).len(), 40);
        // More than the cap -> downsampled to exactly the cap.
        let strided = estimator.sampled_raw_points(10);
        assert_eq!(strided.len(), 10);
        // Striding preserves insertion order (outcomes were added monotonically
        // 0..40), so a returned-first-before-returned-newest check catches a
        // regression that returned the first N points or reversed the window.
        assert!(
            strided.first().unwrap().1 < strided.last().unwrap().1,
            "downsample should span oldest..newest in order, got {strided:?}",
        );
        // Degenerate cap.
        assert!(estimator.sampled_raw_points(0).is_empty());
    }

    #[test]
    fn test_estimator_adapts_to_regime_change() {
        let peer = PeerKeyLocation::random();
        let contract = Location::new(0.0);

        let mut estimator = IsotonicEstimator::new(std::iter::empty(), EstimatorType::Positive);

        // Phase 1: build up global regression with several peers at value 100.0
        let peers: Vec<PeerKeyLocation> = (0..5).map(|_| PeerKeyLocation::random()).collect();
        for _ in 0..30 {
            for p in &peers {
                estimator.add_event(IsotonicEvent {
                    peer: p.clone(),
                    contract_location: contract,
                    result: 100.0,
                });
            }
        }
        // Target peer is "slow" — higher than average
        for _ in 0..20 {
            estimator.add_event(IsotonicEvent {
                peer: peer.clone(),
                contract_location: contract,
                result: 200.0,
            });
        }

        let estimate_before = estimator
            .estimate_retrieval_time(&peer, contract)
            .unwrap_or(0.0);

        // Phase 2: peer becomes "fast"
        for _ in 0..20 {
            estimator.add_event(IsotonicEvent {
                peer: peer.clone(),
                contract_location: contract,
                result: 50.0,
            });
        }

        let estimate_after = estimator
            .estimate_retrieval_time(&peer, contract)
            .unwrap_or(0.0);

        assert!(
            estimate_after < estimate_before,
            "Estimate should decrease after peer improves: before={estimate_before}, after={estimate_after}"
        );
    }

    /// Deterministic per-peer noise derived from the public key hash.
    fn peer_noise(peer: &PeerKeyLocation) -> f64 {
        use std::hash::{Hash, Hasher};
        let mut hasher = std::collections::hash_map::DefaultHasher::new();
        format!("{}", peer.pub_key()).hash(&mut hasher);
        (hasher.finish() as u8) as f64
    }

    fn simulate_request(
        peer: PeerKeyLocation,
        contract_location: Location,
        result_fn: impl FnOnce(f64) -> f64,
    ) -> IsotonicEvent {
        let distance = peer
            .location()
            .unwrap()
            .distance(contract_location)
            .as_f64();
        let result = result_fn(distance) + peer_noise(&peer);
        IsotonicEvent {
            peer,
            contract_location,
            result,
        }
    }

    fn simulate_positive_request(
        peer: PeerKeyLocation,
        contract_location: Location,
    ) -> IsotonicEvent {
        simulate_request(peer, contract_location, |d| d.powf(0.5))
    }

    fn simulate_negative_request(
        peer: PeerKeyLocation,
        contract_location: Location,
    ) -> IsotonicEvent {
        simulate_request(peer, contract_location, |d| (100.0 - d).powf(0.5))
    }

    #[test]
    fn test_sampled_curve_empty_estimator() {
        let estimator = IsotonicEstimator::new(std::iter::empty(), EstimatorType::Positive);
        let curve = estimator.sampled_curve(0.0, 1.0, 50);
        assert!(
            curve.is_empty(),
            "Empty estimator should produce empty curve"
        );
        assert_eq!(estimator.data_x_range(), (0.0, 0.0));
    }

    #[test]
    fn test_sampled_curve_clamping() {
        // Create an ascending estimator where extrapolation could exceed [0, 1]
        let peer = PeerKeyLocation::random();
        let events: Vec<IsotonicEvent> = (0..50)
            .map(|i| {
                let x = i as f64 / 100.0; // distances 0.0 to 0.49
                IsotonicEvent {
                    peer: peer.clone(),
                    contract_location: Location::new(x),
                    result: x * 3.0, // values 0.0 to 1.47 -- will exceed 1.0 clamp
                }
            })
            .collect();
        let estimator = IsotonicEstimator::new(events, EstimatorType::Positive);
        let curve = estimator.sampled_curve(0.0, 1.0, 50);

        assert!(!curve.is_empty());
        for &(_, y) in &curve {
            assert!(y >= 0.0, "y should be >= 0, got {y}");
            assert!(y <= 1.0, "y should be <= 1.0 (clamped), got {y}");
        }
    }

    #[test]
    fn test_sampled_curve_covers_full_range() {
        let peer = PeerKeyLocation::random();
        let events: Vec<IsotonicEvent> = (0..20)
            .map(|i| IsotonicEvent {
                peer: peer.clone(),
                contract_location: Location::new(0.1 + i as f64 * 0.01),
                result: i as f64,
            })
            .collect();
        let estimator = IsotonicEstimator::new(events, EstimatorType::Positive);
        let curve = estimator.sampled_curve(0.0, f64::INFINITY, 50);

        assert_eq!(curve.len(), 50);
        // First point should be at x=0.0, last at x=0.5
        assert!((curve[0].0 - 0.0).abs() < 1e-10);
        assert!((curve[49].0 - 0.5).abs() < 1e-10);
    }

    #[test]
    fn test_data_x_range_reflects_actual_data() {
        let peer = PeerKeyLocation::random();
        let events: Vec<IsotonicEvent> = vec![
            IsotonicEvent {
                peer: peer.clone(),
                contract_location: Location::new(0.1),
                result: 1.0,
            },
            IsotonicEvent {
                peer: peer.clone(),
                contract_location: Location::new(0.3),
                result: 2.0,
            },
        ];
        let estimator = IsotonicEstimator::new(events, EstimatorType::Positive);
        let (lo, hi) = estimator.data_x_range();

        // Data range should approximately match the distances we fed in
        // (exact values depend on PeerKeyLocation's random location)
        assert!((0.0..=0.5).contains(&lo));
        assert!(hi >= lo);
        assert!(hi <= 0.5);
    }

    #[test]
    fn test_sampled_curve_guard_against_low_samples() {
        let estimator = IsotonicEstimator::new(std::iter::empty(), EstimatorType::Positive);
        // num_samples < 2 should return empty without panicking
        let curve = estimator.sampled_curve(0.0, 1.0, 0);
        assert!(curve.is_empty());
        let curve = estimator.sampled_curve(0.0, 1.0, 1);
        assert!(curve.is_empty());
    }
}