lobe-core 0.1.4

//! Two-peak (bimodal) latency detection, ported from the web dashboard's
//! `detectBimodal` (apps/lobe-web/lib/analytics.ts) so the CLI can flag a
//! fast/slow cohort split natively — without shipping samples to the LLM.
//! Keep the constants in sync with the TypeScript implementation.

/// Log-spaced latency edges (ms). A latency lands in bucket i if it satisfies
/// edges[i] <= x < edges[i+1].
const LATENCY_EDGES_MS: [f64; 14] = [
    0.5, 1.0, 2.0, 5.0, 10.0, 20.0, 50.0, 100.0, 200.0, 500.0, 1000.0, 2000.0, 5000.0, 10_000.0,
];

/// Bimodality detection needs enough samples that the two-peak shape is
/// stable — below this floor, false positives dominate. 30 is the standard
/// threshold in production observability tools for histogram-based
/// distribution analysis.
pub const BIMODAL_MIN_SAMPLES: usize = 30;

const MIN_PEAK_SHARE: f64 = 0.15;
const MIN_PEAK_GAP: usize = 3;
const MAX_VALLEY_RATIO: f64 = 0.4;
/// Minimum slow-median / fast-median ratio before the split is worth
/// reporting; below this the two "peaks" are adjacent noise.
const MIN_COHORT_RATIO: f64 = 3.0;

#[derive(Debug, Clone, PartialEq)]
pub struct BimodalCohort {
    pub median_ms: u64,
    pub count: usize,
    /// Fraction of the total sample size in this cohort (0..1).
    pub share: f64,
}

#[derive(Debug, Clone, PartialEq)]
pub struct BimodalDetection {
    pub sample_size: usize,
    pub fast: BimodalCohort,
    pub slow: BimodalCohort,
    /// slow median / fast median.
    pub ratio: f64,
}

/// Detect a two-peak (bimodal) latency distribution. Returns `None` for too
/// few samples or unimodal shapes.
///
/// Algorithm:
///  1. Histogram latencies into log-scaled bins.
///  2. Smooth with a 3-bin center-weighted average to suppress single-bucket noise.
///  3. Find local maxima whose height exceeds MIN_PEAK_SHARE of the sample size.
///  4. Require ≥2 significant peaks separated by ≥ MIN_PEAK_GAP buckets with a
///     valley between them deeper than MAX_VALLEY_RATIO of the smaller peak.
///  5. Partition samples at the valley boundary and report each side's median.
pub fn detect_bimodal(latencies_ms: &[u64]) -> Option<BimodalDetection> {
    if latencies_ms.len() < BIMODAL_MIN_SAMPLES {
        return None;
    }

    let mut histogram = [0_usize; LATENCY_EDGES_MS.len()];
    for &ms in latencies_ms {
        histogram[latency_bucket_index(ms as f64)] += 1;
    }

    let smoothed = smooth_histogram(&histogram);
    let total = latencies_ms.len() as f64;

    let mut peaks = Vec::new();
    for (i, &value) in smoothed.iter().enumerate() {
        if value / total < MIN_PEAK_SHARE {
            continue;
        }
        let left_higher = i > 0 && smoothed[i - 1] > value;
        let right_higher = i < smoothed.len() - 1 && smoothed[i + 1] > value;
        if !left_higher && !right_higher {
            peaks.push(i);
        }
    }

    if peaks.len() < 2 {
        return None;
    }

    let mut by_height = peaks.clone();
    by_height.sort_by(|a, b| smoothed[*b].partial_cmp(&smoothed[*a]).unwrap());
    let mut fast_idx = None;
    let mut slow_idx = None;
    for candidate in by_height {
        match fast_idx {
            None => fast_idx = Some(candidate),
            Some(first) => {
                if candidate.abs_diff(first) >= MIN_PEAK_GAP {
                    slow_idx = Some(candidate);
                    break;
                }
            }
        }
    }
    let (mut fast_idx, mut slow_idx) = (fast_idx?, slow_idx?);
    if fast_idx > slow_idx {
        std::mem::swap(&mut fast_idx, &mut slow_idx);
    }

    let mut valley_idx = fast_idx;
    let mut valley_value = smoothed[fast_idx];
    for i in fast_idx + 1..slow_idx {
        if smoothed[i] < valley_value {
            valley_value = smoothed[i];
            valley_idx = i;
        }
    }
    let smaller_peak = smoothed[fast_idx].min(smoothed[slow_idx]);
    if smaller_peak == 0.0 || valley_value / smaller_peak > MAX_VALLEY_RATIO {
        return None;
    }

    let mut fast_values = Vec::new();
    let mut slow_values = Vec::new();
    for &ms in latencies_ms {
        if latency_bucket_index(ms as f64) <= valley_idx {
            fast_values.push(ms);
        } else {
            slow_values.push(ms);
        }
    }

    if fast_values.len() < 2 || slow_values.len() < 2 {
        return None;
    }

    let fast_median = median(&fast_values);
    let slow_median = median(&slow_values);
    if fast_median == 0 {
        return None;
    }
    let ratio = slow_median as f64 / fast_median as f64;
    if ratio < MIN_COHORT_RATIO {
        return None;
    }

    let sample_size = latencies_ms.len();
    Some(BimodalDetection {
        sample_size,
        fast: BimodalCohort {
            median_ms: fast_median,
            count: fast_values.len(),
            share: fast_values.len() as f64 / sample_size as f64,
        },
        slow: BimodalCohort {
            median_ms: slow_median,
            count: slow_values.len(),
            share: slow_values.len() as f64 / sample_size as f64,
        },
        ratio,
    })
}

fn latency_bucket_index(latency_ms: f64) -> usize {
    for i in (0..LATENCY_EDGES_MS.len()).rev() {
        if latency_ms >= LATENCY_EDGES_MS[i] {
            return i;
        }
    }
    0
}

fn smooth_histogram(counts: &[usize; LATENCY_EDGES_MS.len()]) -> [f64; LATENCY_EDGES_MS.len()] {
    let mut out = [0.0; LATENCY_EDGES_MS.len()];
    for i in 0..counts.len() {
        let left = if i > 0 { counts[i - 1] } else { 0 } as f64;
        let right = if i + 1 < counts.len() { counts[i + 1] } else { 0 } as f64;
        out[i] = (left + 2.0 * counts[i] as f64 + right) / 4.0;
    }
    out
}

/// Median using the "P × N ceil" convention, matching the web dashboard's
/// `percentile(values, 50)`.
fn median(values: &[u64]) -> u64 {
    let mut sorted = values.to_vec();
    sorted.sort_unstable();
    let rank = ((0.5 * sorted.len() as f64).ceil() as usize).max(1);
    sorted[rank.min(sorted.len()) - 1]
}

#[cfg(test)]
mod tests {
    use super::{detect_bimodal, BIMODAL_MIN_SAMPLES};

    #[test]
    fn too_few_samples_returns_none() {
        let values = vec![40; BIMODAL_MIN_SAMPLES - 1];
        assert!(detect_bimodal(&values).is_none());
    }

    #[test]
    fn unimodal_distribution_returns_none() {
        // Everything clustered around 40ms.
        let values: Vec<u64> = (0..60).map(|i| 38 + (i % 5)).collect();
        assert!(detect_bimodal(&values).is_none());
    }

    #[test]
    fn pool_exhaustion_shape_is_detected() {
        // Fast cohort ~40ms (pool hit), slow cohort ~600ms (queued) — the
        // classic connection-pool saturation shape.
        let mut values: Vec<u64> = (0..20).map(|i| 38 + (i % 6)).collect();
        values.extend((0..40).map(|i| 590 + (i % 30)));

        let detection = detect_bimodal(&values).expect("should detect bimodal split");
        assert!(detection.fast.median_ms < 50);
        assert!(detection.slow.median_ms > 500);
        assert!(detection.ratio >= 3.0);
        assert_eq!(detection.fast.count + detection.slow.count, values.len());
    }

    #[test]
    fn close_cohorts_below_ratio_threshold_return_none() {
        // Two clusters, but only ~1.5x apart — not a meaningful split.
        let mut values: Vec<u64> = (0..30).map(|i| 40 + (i % 4)).collect();
        values.extend((0..30).map(|i| 60 + (i % 4)));
        assert!(detect_bimodal(&values).is_none());
    }
}