ktstr 0.4.14

//! Stimulus/phase correlation for scenario execution.
//!
//! Correlates [`StimulusEvent`]s (cgroup operations, cpuset changes)
//! with [`MonitorSample`] windows to
//! measure per-phase scheduler behavior degradation. Produces
//! [`Timeline`] entries consumed by the stats and reporting pipeline.

use std::fmt;

use crate::monitor::{MonitorSample, sample_looks_valid};

// ---------------------------------------------------------------------------
// TimelineContext — system context rendered as a header
// ---------------------------------------------------------------------------

/// System context for a timeline, rendered as a header block.
#[derive(Debug, Clone, Default)]
pub struct TimelineContext {
    /// Kernel version string (e.g. "6.14.0-rc3+").
    pub kernel: Option<String>,
    /// Topology description (e.g. "2n4l4c2t (16 cpus)").
    pub topology: Option<String>,
    /// Scheduler name (e.g. "scx_mitosis").
    pub scheduler: Option<String>,
    /// Scenario name.
    pub scenario: Option<String>,
    /// Total run duration in seconds.
    pub duration_s: Option<f64>,
}

// ---------------------------------------------------------------------------
// StimulusEvent — what happened and when
// ---------------------------------------------------------------------------

/// A discrete event during scenario execution that may cause observable
/// changes in scheduler behavior. Generated by step executors on the guest
/// side and carried in the VM output alongside monitor samples.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct StimulusEvent {
    /// Milliseconds since scenario start (guest monotonic clock).
    pub elapsed_ms: u64,
    /// Human-readable label: "ScenarioStart", "StepStart\[0\]", "StepEnd\[0\]".
    pub label: String,
    /// What kind of operation triggered this event.
    pub op_kind: Option<String>,
    /// Additional context (e.g. "4 cpus", "cgroup=cg_0").
    pub detail: Option<String>,
    /// Cumulative worker iterations at this event. Used to compute
    /// per-phase throughput (iterations/s).
    pub total_iterations: Option<u64>,
}

// ---------------------------------------------------------------------------
// Phase — a time window between consecutive stimulus events
// ---------------------------------------------------------------------------

/// Metrics aggregated from monitor samples within a phase.
#[derive(Debug, Clone, Default)]
pub struct PhaseMetrics {
    pub sample_count: usize,
    pub avg_imbalance: f64,
    pub max_imbalance: f64,
    pub avg_dsq_depth: f64,
    pub max_dsq_depth: u32,
    pub stall_count: usize,
    /// select_cpu_fallback events per second. None when event counters unavailable.
    pub fallback_rate: Option<f64>,
    /// dispatch_keep_last events per second. None when event counters unavailable.
    pub keep_last_rate: Option<f64>,
    /// Worker iterations per second during this phase. Computed from
    /// cumulative iteration counts in consecutive stimulus events.
    pub iteration_rate: Option<f64>,
}

/// Direction of change at a phase boundary.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ChangeDirection {
    Improved,
    Degraded,
}

impl fmt::Display for ChangeDirection {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            ChangeDirection::Improved => write!(f, "IMPROVEMENT"),
            ChangeDirection::Degraded => write!(f, "DEGRADATION"),
        }
    }
}

/// Detected change at a stimulus boundary.
#[derive(Debug, Clone)]
pub struct PhaseChange {
    pub direction: ChangeDirection,
    pub metric: String,
    pub before: f64,
    pub after: f64,
}

/// A time window between two consecutive stimulus events.
#[derive(Debug, Clone)]
pub struct Phase {
    pub index: usize,
    pub start_ms: u64,
    pub end_ms: u64,
    /// The stimulus event that starts this phase (None for the initial phase).
    pub stimulus: Option<StimulusEvent>,
    pub metrics: PhaseMetrics,
    /// Changes detected at this phase's stimulus boundary.
    pub changes: Vec<PhaseChange>,
}

// ---------------------------------------------------------------------------
// Timeline
// ---------------------------------------------------------------------------

/// Correlated timeline of stimulus events and monitor observations.
#[derive(Debug, Clone)]
pub struct Timeline {
    pub phases: Vec<Phase>,
}

/// Minimum delta in imbalance ratio to flag a change (avoids noise).
const IMBALANCE_THRESHOLD: f64 = 0.5;
/// Minimum delta in DSQ depth to flag a change.
const DSQ_THRESHOLD: f64 = 3.0;
/// Minimum delta in fallback rate (events/s) to flag a change.
const FALLBACK_RATE_THRESHOLD: f64 = 10.0;
/// Minimum delta in keep_last rate (events/s) to flag a change.
const KEEP_LAST_RATE_THRESHOLD: f64 = 10.0;
/// Minimum relative change in iteration rate to flag a throughput change.
/// 0.3 = 30% drop or increase.
const ITERATION_RATE_REL_THRESHOLD: f64 = 0.3;

/// Create a PhaseChange if the delta between `before` and `after` exceeds
/// `threshold`. `higher_is_worse` determines degradation direction: when
/// true, a positive delta means Degraded; when false, a negative delta
/// means Degraded.
fn detect_change(
    before: f64,
    after: f64,
    threshold: f64,
    metric: &str,
    higher_is_worse: bool,
) -> Option<PhaseChange> {
    let delta = after - before;
    if delta.abs() <= threshold {
        return None;
    }
    let degraded = if higher_is_worse {
        delta > 0.0
    } else {
        delta < 0.0
    };
    Some(PhaseChange {
        direction: if degraded {
            ChangeDirection::Degraded
        } else {
            ChangeDirection::Improved
        },
        metric: metric.to_string(),
        before,
        after,
    })
}

impl Timeline {
    /// Build a timeline from stimulus events and monitor samples.
    ///
    /// Clock alignment: stimulus events use guest monotonic time (ms since
    /// scenario start). Monitor samples use host monotonic time (ms since
    /// VM boot). The first stimulus event's timestamp and the first
    /// non-trivial monitor sample (after 500ms warmup) approximately
    /// coincide. We compute an offset to align them.
    ///
    /// Returns an empty timeline if either input is empty.
    pub fn build(stimulus_events: &[StimulusEvent], monitor_samples: &[MonitorSample]) -> Self {
        if stimulus_events.is_empty() || monitor_samples.is_empty() {
            return Self { phases: Vec::new() };
        }

        let mut events = stimulus_events.to_vec();
        events.sort_by_key(|e| e.elapsed_ms);

        // Clock alignment: find the offset between guest stimulus time
        // and host monitor time. The first stimulus event (ScenarioStart)
        // and the first monitor sample with plausible data roughly coincide.
        let first_stimulus_ms = events[0].elapsed_ms;
        let first_monitor_ms = monitor_samples
            .iter()
            .find(|s| s.elapsed_ms > 500 && !s.cpus.is_empty())
            .map(|s| s.elapsed_ms)
            .unwrap_or_else(|| monitor_samples.first().map(|s| s.elapsed_ms).unwrap_or(0));

        // offset: add this to a stimulus timestamp to get monitor time
        let offset = first_monitor_ms as i64 - first_stimulus_ms as i64;

        // Define phase boundaries from consecutive stimulus events.
        // Each pair (events[i], events[i+1]) bounds a phase.
        // The last event to end-of-data is also a phase.
        let last_monitor_ms = monitor_samples.last().map(|s| s.elapsed_ms).unwrap_or(0);

        let mut boundaries: Vec<(u64, u64, Option<StimulusEvent>)> = Vec::new();
        for i in 0..events.len() {
            let start = (events[i].elapsed_ms as i64 + offset).max(0) as u64;
            let end = if i + 1 < events.len() {
                (events[i + 1].elapsed_ms as i64 + offset).max(0) as u64
            } else {
                last_monitor_ms.saturating_add(1)
            };
            let stimulus = if i == 0 {
                None
            } else {
                Some(events[i].clone())
            };
            boundaries.push((start, end, stimulus));
        }

        // Assign monitor samples to phases and compute metrics.
        let mut phases: Vec<Phase> = Vec::with_capacity(boundaries.len());
        for (idx, (start, end, stimulus)) in boundaries.into_iter().enumerate() {
            let phase_samples: Vec<&MonitorSample> = monitor_samples
                .iter()
                .filter(|s| s.elapsed_ms >= start && s.elapsed_ms < end && sample_looks_valid(s))
                .collect();

            let metrics = compute_metrics(&phase_samples);

            phases.push(Phase {
                index: idx,
                start_ms: start,
                end_ms: end,
                stimulus,
                metrics,
                changes: Vec::new(),
            });
        }

        // Compute per-phase iteration rate from consecutive stimulus
        // event iteration totals. Index-based: accesses both phases[i] and events[i+1].
        #[allow(clippy::needless_range_loop)]
        for i in 0..phases.len() {
            let iter_start = if i == 0 {
                events.first().and_then(|e| e.total_iterations)
            } else {
                events.get(i).and_then(|e| e.total_iterations)
            };
            let iter_end = events.get(i + 1).and_then(|e| e.total_iterations);
            if let (Some(s), Some(e)) = (iter_start, iter_end) {
                let duration_s =
                    phases[i].end_ms.saturating_sub(phases[i].start_ms) as f64 / 1000.0;
                if duration_s > 0.0 && e > s {
                    // `e > s` above, but `saturating_sub` is
                    // defense-in-depth — if a future change reorders
                    // the guard, the rate stays at 0 rather than
                    // panicking on underflow.
                    phases[i].metrics.iteration_rate =
                        Some(e.saturating_sub(s) as f64 / duration_s);
                }
            }
        }

        // Detect changes at each phase boundary.
        for i in 1..phases.len() {
            let before = &phases[i - 1].metrics;
            let after_metrics = &phases[i].metrics;
            let mut changes = Vec::new();

            if before.sample_count > 0 && after_metrics.sample_count > 0 {
                changes.extend(detect_change(
                    before.avg_imbalance,
                    after_metrics.avg_imbalance,
                    IMBALANCE_THRESHOLD,
                    "imbalance",
                    true,
                ));
                changes.extend(detect_change(
                    before.avg_dsq_depth,
                    after_metrics.avg_dsq_depth,
                    DSQ_THRESHOLD,
                    "dsq_depth",
                    true,
                ));
                if let (Some(bf), Some(af)) = (before.fallback_rate, after_metrics.fallback_rate) {
                    changes.extend(detect_change(
                        bf,
                        af,
                        FALLBACK_RATE_THRESHOLD,
                        "fallback",
                        true,
                    ));
                }
                if let (Some(bk), Some(ak)) = (before.keep_last_rate, after_metrics.keep_last_rate)
                {
                    changes.extend(detect_change(
                        bk,
                        ak,
                        KEEP_LAST_RATE_THRESHOLD,
                        "keep_last",
                        true,
                    ));
                }
                if let (Some(bi), Some(ai)) = (before.iteration_rate, after_metrics.iteration_rate)
                    && bi > 0.0
                {
                    let rel_delta = (ai - bi) / bi;
                    if rel_delta.abs() > ITERATION_RATE_REL_THRESHOLD {
                        changes.push(PhaseChange {
                            direction: if rel_delta < 0.0 {
                                ChangeDirection::Degraded
                            } else {
                                ChangeDirection::Improved
                            },
                            metric: "throughput".to_string(),
                            before: bi,
                            after: ai,
                        });
                    }
                }
            }

            phases[i].changes = changes;
        }

        Self { phases }
    }

    /// Format the timeline with a system context header.
    ///
    /// Tests without a real context pass `&TimelineContext::default()`;
    /// the header lines (`kernel:`, `topology:`, etc.) are omitted but
    /// the `--- timeline ---` prefix is preserved.
    // No parameterless format() sibling: output with default context
    // is byte-identical, but the only non-test caller (eval.rs) always
    // has real context, so format() would be dead code.
    pub fn format_with_context(&self, ctx: &TimelineContext) -> String {
        if self.phases.is_empty() {
            return String::new();
        }

        let mut out = String::from("--- timeline ---\n");

        // Render context header.
        let mut header_parts = Vec::new();
        if let Some(ref k) = ctx.kernel {
            header_parts.push(format!("kernel: {k}"));
        }
        if let Some(ref t) = ctx.topology {
            header_parts.push(format!("topology: {t}"));
        }
        if let Some(ref s) = ctx.scheduler {
            header_parts.push(format!("scheduler: {s}"));
        }
        if let Some(ref s) = ctx.scenario {
            header_parts.push(format!("scenario: {s}"));
        }
        if let Some(d) = ctx.duration_s {
            header_parts.push(format!("duration: {d:.1}s"));
        }
        if !header_parts.is_empty() {
            for part in &header_parts {
                out.push_str(part);
                out.push_str("  ");
            }
            // Trim trailing "  " appended by the last iteration.
            // Explicit length guard so a future edit that stops
            // appending the separator here can't underflow.
            if out.len() >= 2 {
                out.truncate(out.len() - 2);
            }
            out.push('\n');
        }

        self.format_phases(&mut out);
        out
    }

    /// Render phase details into the output buffer.
    fn format_phases(&self, out: &mut String) {
        for phase in &self.phases {
            let duration_ms = phase.end_ms.saturating_sub(phase.start_ms);

            if phase.index == 0 {
                // Phase 0 is the settle window before any stimulus.
                out.push_str(&format!(
                    "\nBASELINE (settle, {}ms, {} samples):\n",
                    duration_ms, phase.metrics.sample_count,
                ));
            } else {
                let label_start = phase
                    .stimulus
                    .as_ref()
                    .map(|s| {
                        let mut l = s.label.clone();
                        if let Some(op) = &s.op_kind {
                            l.push(' ');
                            l.push_str(op);
                        }
                        l
                    })
                    .unwrap_or_else(|| "?".to_string());

                out.push_str(&format!(
                    "\nPhase {}: {} ({}ms, {} samples):\n",
                    phase.index, label_start, duration_ms, phase.metrics.sample_count,
                ));
            }

            let m = &phase.metrics;
            if m.sample_count > 0 {
                out.push_str(&format!(
                    "  imbalance: avg={:.1} max={:.1} | dsq: avg={:.0} max={}",
                    m.avg_imbalance, m.max_imbalance, m.avg_dsq_depth, m.max_dsq_depth,
                ));
                if let Some(fb) = m.fallback_rate {
                    out.push_str(&format!(" | fallback: {:.0}/s", fb));
                }
                if let Some(kl) = m.keep_last_rate {
                    out.push_str(&format!(" | keep_last: {:.0}/s", kl));
                }
                if let Some(ir) = m.iteration_rate {
                    out.push_str(&format!(" | throughput: {:.0} iter/s", ir));
                }
                out.push('\n');
                if m.stall_count > 0 {
                    out.push_str(&format!("  stalls: {}\n", m.stall_count));
                }
            } else {
                out.push_str("  [no samples]\n");
            }

            if let Some(ref stim) = phase.stimulus {
                let detail = stim.detail.as_deref().unwrap_or("");
                let op = stim.op_kind.as_deref().unwrap_or("?");
                out.push_str(&format!("  >>> {}: {op}", stim.label));
                if !detail.is_empty() {
                    out.push_str(&format!(" ({detail})"));
                }
                out.push('\n');
            }

            for change in &phase.changes {
                let delta = change.after - change.before;
                let sign = if delta > 0.0 { "+" } else { "" };
                out.push_str(&format!(
                    "  >>> {}: {} {sign}{:.1}\n",
                    change.direction, change.metric, delta,
                ));
            }
        }
    }

    /// Test helper — collect all degradation changes across phases.
    /// Retained after the gauntlet analyzer was removed; the scenarios
    /// pipeline consumes `Timeline` via `format_with_context` and does
    /// not read degradations directly.
    #[cfg(test)]
    pub fn degradations(&self) -> Vec<(&Phase, &PhaseChange)> {
        let mut out = Vec::new();
        for phase in &self.phases {
            for change in &phase.changes {
                if change.direction == ChangeDirection::Degraded {
                    out.push((phase, change));
                }
            }
        }
        out
    }
}

// ---------------------------------------------------------------------------
// Metric computation
// ---------------------------------------------------------------------------

fn compute_metrics(samples: &[&MonitorSample]) -> PhaseMetrics {
    if samples.is_empty() {
        return PhaseMetrics::default();
    }

    // Filter out samples with implausible data (e.g. garbage DSQ depths
    // from uninitialized guest memory) before computing metrics.
    let valid: Vec<&MonitorSample> = samples
        .iter()
        .copied()
        .filter(|s| !s.cpus.is_empty() && sample_looks_valid(s))
        .collect();

    if valid.is_empty() {
        return PhaseMetrics {
            sample_count: 0,
            ..PhaseMetrics::default()
        };
    }

    let mut total_imbalance = 0.0f64;
    let mut max_imbalance = 0.0f64;
    let mut total_dsq = 0.0f64;
    let mut max_dsq = 0u32;
    let mut stall_count = 0usize;

    for sample in &valid {
        for cpu in &sample.cpus {
            max_dsq = max_dsq.max(cpu.local_dsq_depth);
        }
        let ratio = sample.imbalance_ratio();
        total_imbalance += ratio;
        if ratio > max_imbalance {
            max_imbalance = ratio;
        }

        let avg_dsq_this: f64 = sample
            .cpus
            .iter()
            .map(|c| c.local_dsq_depth as f64)
            .sum::<f64>()
            / sample.cpus.len() as f64;
        total_dsq += avg_dsq_this;
    }

    // Stall detection between consecutive valid samples in this phase.
    for w in valid.windows(2) {
        let prev = w[0];
        let curr = w[1];
        let cpu_count = prev.cpus.len().min(curr.cpus.len());
        for cpu in 0..cpu_count {
            let idle = curr.cpus[cpu].nr_running == 0 && prev.cpus[cpu].nr_running == 0;
            if curr.cpus[cpu].rq_clock != 0
                && curr.cpus[cpu].rq_clock == prev.cpus[cpu].rq_clock
                && !idle
            {
                stall_count += 1;
            }
        }
    }

    // Event counter rates: sum counters across CPUs for first/last valid
    // samples that have event_counters, compute delta / duration.
    let has_events = |s: &&MonitorSample| s.cpus.iter().any(|c| c.event_counters.is_some());
    let first_ev = valid.iter().copied().find(|s| has_events(s));
    let last_ev = valid.iter().copied().rev().find(|s| has_events(s));

    let (fallback_rate, keep_last_rate) = match (first_ev, last_ev) {
        (Some(first), Some(last)) if first.elapsed_ms < last.elapsed_ms => {
            // `<` guard above is expected to rule out underflow, but
            // `saturating_sub` is defense-in-depth: if a future change
            // loosens the guard, the worst outcome becomes
            // `duration_s == 0.0` (which disables the rate below) rather
            // than a panic.
            let duration_s = last.elapsed_ms.saturating_sub(first.elapsed_ms) as f64 / 1000.0;
            // Event counters can reset mid-run (scheduler restart) and
            // produce a negative raw delta. Shared helper clamps to
            // >= 0 so the computed rate never goes negative; same
            // semantics as MonitorSummary::compute_event_deltas.
            let fb_delta = crate::monitor::counter_delta(
                last.sum_event_field(|e| e.select_cpu_fallback).unwrap_or(0),
                first
                    .sum_event_field(|e| e.select_cpu_fallback)
                    .unwrap_or(0),
            );
            let kl_delta = crate::monitor::counter_delta(
                last.sum_event_field(|e| e.dispatch_keep_last).unwrap_or(0),
                first.sum_event_field(|e| e.dispatch_keep_last).unwrap_or(0),
            );
            (
                Some(fb_delta as f64 / duration_s),
                Some(kl_delta as f64 / duration_s),
            )
        }
        _ => (None, None),
    };

    let n = valid.len() as f64;
    PhaseMetrics {
        sample_count: valid.len(),
        avg_imbalance: total_imbalance / n,
        max_imbalance,
        avg_dsq_depth: total_dsq / n,
        max_dsq_depth: max_dsq,
        stall_count,
        fallback_rate,
        keep_last_rate,
        iteration_rate: None,
    }
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

#[cfg(test)]
mod tests {
    use super::*;
    use crate::monitor::{CpuSnapshot, MonitorSample};

    fn sample(elapsed_ms: u64, cpus: Vec<(u32, u32, u64)>) -> MonitorSample {
        MonitorSample {
            prog_stats: None,
            elapsed_ms,
            cpus: cpus
                .into_iter()
                .map(|(nr_running, dsq, rq_clock)| CpuSnapshot {
                    nr_running,
                    scx_nr_running: 0,
                    local_dsq_depth: dsq,
                    rq_clock,
                    scx_flags: 0,
                    event_counters: None,
                    schedstat: None,
                    vcpu_cpu_time_ns: None,
                    vcpu_perf: None,
                    sched_domains: None,
                })
                .collect(),
        }
    }

    fn stimulus(elapsed_ms: u64, label: &str) -> StimulusEvent {
        StimulusEvent {
            elapsed_ms,
            label: label.to_string(),
            op_kind: None,
            detail: None,
            total_iterations: None,
        }
    }

    #[test]
    fn empty_inputs_empty_timeline() {
        let t = Timeline::build(&[], &[]);
        assert!(t.phases.is_empty());
    }

    #[test]
    fn no_stimulus_empty_timeline() {
        let samples = vec![sample(1000, vec![(2, 1, 100)])];
        let t = Timeline::build(&[], &samples);
        assert!(t.phases.is_empty());
    }

    #[test]
    fn no_monitor_empty_timeline() {
        let events = vec![stimulus(0, "ScenarioStart")];
        let t = Timeline::build(&events, &[]);
        assert!(t.phases.is_empty());
    }

    #[test]
    fn single_event_single_phase() {
        let events = vec![stimulus(0, "ScenarioStart")];
        let samples = vec![
            sample(600, vec![(2, 1, 100), (2, 1, 200)]),
            sample(700, vec![(2, 1, 300), (2, 1, 400)]),
        ];
        let t = Timeline::build(&events, &samples);
        assert_eq!(t.phases.len(), 1);
        assert!(t.phases[0].metrics.sample_count > 0);
    }

    #[test]
    fn two_events_two_phases() {
        let events = vec![stimulus(0, "ScenarioStart"), stimulus(3000, "StepStart[0]")];
        let samples: Vec<MonitorSample> = (5..65)
            .map(|i| sample(i * 100, vec![(2, 1, i * 1000), (2, 1, i * 1000 + 100)]))
            .collect();
        let t = Timeline::build(&events, &samples);
        assert_eq!(t.phases.len(), 2);
        assert!(t.phases[0].metrics.sample_count > 0);
        assert!(t.phases[1].metrics.sample_count > 0);
    }

    #[test]
    fn improvement_detected() {
        // Phase 0: imbalanced
        // Phase 1: balanced
        let events = vec![stimulus(0, "ScenarioStart"), stimulus(1000, "StepStart[0]")];
        let mut samples = Vec::new();
        for i in 5..15 {
            samples.push(sample(
                i * 100,
                vec![(1, 1, i * 1000), (5, 1, i * 1000 + 100)],
            ));
        }
        for i in 15..25 {
            samples.push(sample(
                i * 100,
                vec![(2, 1, i * 1000), (2, 1, i * 1000 + 100)],
            ));
        }
        let t = Timeline::build(&events, &samples);
        let improvements: Vec<_> = t
            .phases
            .iter()
            .flat_map(|p| p.changes.iter())
            .filter(|c| c.direction == ChangeDirection::Improved)
            .collect();
        assert!(!improvements.is_empty());
    }

    #[test]
    fn format_non_empty() {
        let events = vec![stimulus(0, "ScenarioStart"), stimulus(1000, "StepStart[0]")];
        let samples: Vec<MonitorSample> = (5..25)
            .map(|i| sample(i * 100, vec![(2, 1, i * 1000), (2, 1, i * 1000 + 100)]))
            .collect();
        let t = Timeline::build(&events, &samples);
        let formatted = t.format_with_context(&TimelineContext::default());
        assert!(formatted.contains("BASELINE"));
        assert!(formatted.contains("Phase 1"));
        assert!(formatted.contains("imbalance"));
    }

    #[test]
    fn unsorted_events_sorted() {
        let events = vec![stimulus(3000, "StepStart[0]"), stimulus(0, "ScenarioStart")];
        let samples: Vec<MonitorSample> = (5..35)
            .map(|i| sample(i * 100, vec![(2, 1, i * 1000), (2, 1, i * 1000 + 100)]))
            .collect();
        let t = Timeline::build(&events, &samples);
        assert_eq!(t.phases.len(), 2);
        // First phase should be from ScenarioStart (earliest).
        assert!(t.phases[0].stimulus.is_none());
    }

    #[test]
    fn stall_detected_in_phase() {
        let events = vec![stimulus(0, "ScenarioStart")];
        let samples = vec![
            sample(600, vec![(1, 0, 5000), (1, 0, 6000)]),
            sample(700, vec![(1, 0, 5000), (1, 0, 7000)]), // cpu0 stalled
        ];
        let t = Timeline::build(&events, &samples);
        assert_eq!(t.phases[0].metrics.stall_count, 1);
    }

    #[test]
    fn compute_metrics_empty() {
        let m = compute_metrics(&[]);
        assert_eq!(m.sample_count, 0);
        assert_eq!(m.avg_imbalance, 0.0);
        assert_eq!(m.max_dsq_depth, 0);
    }

    #[test]
    fn stimulus_event_with_detail() {
        let e = StimulusEvent {
            elapsed_ms: 100,
            label: "StepStart[0]".to_string(),
            op_kind: Some("SetCpuset".to_string()),
            detail: Some("4 cpus".to_string()),
            total_iterations: None,
        };
        let events = vec![stimulus(0, "ScenarioStart"), e];
        let samples: Vec<MonitorSample> = (5..25)
            .map(|i| sample(i * 100, vec![(2, 1, i * 1000), (2, 1, i * 1000 + 100)]))
            .collect();
        let t = Timeline::build(&events, &samples);
        let formatted = t.format_with_context(&TimelineContext::default());
        assert!(formatted.contains("SetCpuset"));
        assert!(formatted.contains("4 cpus"));
    }

    #[test]
    fn many_phases() {
        let events: Vec<StimulusEvent> = (0..10)
            .map(|i| stimulus(i * 500, &format!("Step[{i}]")))
            .collect();
        let samples: Vec<MonitorSample> = (5..55)
            .map(|i| sample(i * 100, vec![(2, 1, i * 1000)]))
            .collect();
        let t = Timeline::build(&events, &samples);
        assert_eq!(t.phases.len(), 10);
    }

    #[test]
    fn phase_metrics_accuracy() {
        let s1 = sample(600, vec![(1, 3, 100), (4, 5, 200)]); // ratio=4, avg_dsq=4
        let s2 = sample(700, vec![(2, 1, 300), (2, 7, 400)]); // ratio=1, avg_dsq=4
        let refs: Vec<&MonitorSample> = vec![&s1, &s2];
        let m = compute_metrics(&refs);
        assert_eq!(m.sample_count, 2);
        assert!((m.avg_imbalance - 2.5).abs() < 0.01); // (4+1)/2
        assert!((m.max_imbalance - 4.0).abs() < 0.01);
        assert_eq!(m.max_dsq_depth, 7);
    }

    // -- ChangeDirection Display tests --

    #[test]
    fn change_direction_display() {
        assert_eq!(format!("{}", ChangeDirection::Improved), "IMPROVEMENT");
        assert_eq!(format!("{}", ChangeDirection::Degraded), "DEGRADATION");
    }

    // -- compute_metrics with event counters --

    #[test]
    fn compute_metrics_with_event_counters() {
        use crate::monitor::ScxEventCounters;

        let s1 = MonitorSample {
            prog_stats: None,
            elapsed_ms: 600,
            cpus: vec![CpuSnapshot {
                nr_running: 2,
                local_dsq_depth: 1,
                rq_clock: 100,
                scx_nr_running: 0,
                scx_flags: 0,
                event_counters: Some(ScxEventCounters {
                    select_cpu_fallback: 10,
                    dispatch_keep_last: 5,
                    ..Default::default()
                }),
                schedstat: None,
                vcpu_cpu_time_ns: None,
                vcpu_perf: None,
                sched_domains: None,
            }],
        };
        let s2 = MonitorSample {
            prog_stats: None,
            elapsed_ms: 1600,
            cpus: vec![CpuSnapshot {
                nr_running: 2,
                local_dsq_depth: 1,
                rq_clock: 200,
                scx_nr_running: 0,
                scx_flags: 0,
                event_counters: Some(ScxEventCounters {
                    select_cpu_fallback: 110,
                    dispatch_keep_last: 55,
                    ..Default::default()
                }),
                schedstat: None,
                vcpu_cpu_time_ns: None,
                vcpu_perf: None,
                sched_domains: None,
            }],
        };
        let refs: Vec<&MonitorSample> = vec![&s1, &s2];
        let m = compute_metrics(&refs);
        // fallback delta: 110 - 10 = 100 over 1.0s = 100.0/s
        assert!((m.fallback_rate.unwrap() - 100.0).abs() < 0.01);
        // keep_last delta: 55 - 5 = 50 over 1.0s = 50.0/s
        assert!((m.keep_last_rate.unwrap() - 50.0).abs() < 0.01);
    }

    #[test]
    fn compute_metrics_no_event_counters() {
        let s1 = sample(600, vec![(2, 1, 100)]);
        let s2 = sample(700, vec![(2, 1, 200)]);
        let refs: Vec<&MonitorSample> = vec![&s1, &s2];
        let m = compute_metrics(&refs);
        assert!(m.fallback_rate.is_none());
        assert!(m.keep_last_rate.is_none());
    }

    #[test]
    fn compute_metrics_counter_reset_clamps_rates_to_non_negative() {
        // A scheduler restart between samples resets event counters
        // to smaller (or zero) values. Raw `last - first` then
        // produces a negative delta, which would flow into
        // `fallback_rate = delta / duration` and report a negative
        // rate. The shared counter_delta helper clamps to 0.
        use crate::monitor::ScxEventCounters;

        let s1 = MonitorSample {
            prog_stats: None,
            elapsed_ms: 0,
            cpus: vec![CpuSnapshot {
                nr_running: 2,
                local_dsq_depth: 1,
                rq_clock: 100,
                scx_nr_running: 0,
                scx_flags: 0,
                event_counters: Some(ScxEventCounters {
                    select_cpu_fallback: 1000,
                    dispatch_keep_last: 500,
                    ..Default::default()
                }),
                schedstat: None,
                vcpu_cpu_time_ns: None,
                vcpu_perf: None,
                sched_domains: None,
            }],
        };
        let s2 = MonitorSample {
            prog_stats: None,
            elapsed_ms: 1000,
            cpus: vec![CpuSnapshot {
                nr_running: 2,
                local_dsq_depth: 1,
                rq_clock: 200,
                scx_nr_running: 0,
                scx_flags: 0,
                event_counters: Some(ScxEventCounters {
                    select_cpu_fallback: 5,
                    dispatch_keep_last: 2,
                    ..Default::default()
                }),
                schedstat: None,
                vcpu_cpu_time_ns: None,
                vcpu_perf: None,
                sched_domains: None,
            }],
        };
        let refs: Vec<&MonitorSample> = vec![&s1, &s2];
        let m = compute_metrics(&refs);
        let fb = m.fallback_rate.expect("reset still produces Some rate");
        let kl = m.keep_last_rate.expect("reset still produces Some rate");
        assert!(
            fb >= 0.0,
            "reset must not produce negative fallback_rate, got {fb}"
        );
        assert!(
            kl >= 0.0,
            "reset must not produce negative keep_last_rate, got {kl}"
        );
    }

    // -- format with stalls --

    #[test]
    fn format_with_stalls_shown() {
        let events = vec![stimulus(0, "ScenarioStart")];
        let samples = vec![
            sample(600, vec![(1, 0, 5000), (1, 0, 6000)]),
            sample(700, vec![(1, 0, 5000), (1, 0, 7000)]), // cpu0 stalled
        ];
        let t = Timeline::build(&events, &samples);
        let formatted = t.format_with_context(&TimelineContext::default());
        assert!(formatted.contains("stalls: 1"));
    }

    // -- format with no samples in a phase --

    #[test]
    fn format_phase_no_samples() {
        // Create a phase with no samples by making a phase boundary far
        // beyond the last monitor sample's time.
        let events = vec![
            stimulus(0, "ScenarioStart"),
            stimulus(100, "StepStart[0]"),
            stimulus(50000, "StepStart[1]"),
        ];
        // All samples are in the middle phase window.
        let samples: Vec<MonitorSample> = (5..15)
            .map(|i| sample(i * 100, vec![(2, 1, i * 1000)]))
            .collect();
        let t = Timeline::build(&events, &samples);
        let formatted = t.format_with_context(&TimelineContext::default());
        // The last phase (50000+offset to end) should have no samples.
        assert!(formatted.contains("[no samples]"));
    }

    // -- timeline with fallback rate change detection --

    #[test]
    fn fallback_rate_degradation_detected() {
        use crate::monitor::ScxEventCounters;

        let events = vec![stimulus(0, "ScenarioStart"), stimulus(1000, "StepStart[0]")];
        let mut samples = Vec::new();
        // Phase 0: zero fallback rate (counter stays constant).
        for i in 5..15 {
            samples.push(MonitorSample {
                prog_stats: None,
                elapsed_ms: i * 100,
                cpus: vec![CpuSnapshot {
                    nr_running: 2,
                    local_dsq_depth: 1,
                    rq_clock: i * 1000,
                    scx_nr_running: 0,
                    scx_flags: 0,
                    event_counters: Some(ScxEventCounters {
                        select_cpu_fallback: 0,
                        dispatch_keep_last: 0,
                        ..Default::default()
                    }),
                    schedstat: None,
                    vcpu_cpu_time_ns: None,
                    vcpu_perf: None,
                    sched_domains: None,
                }],
            });
        }
        // Phase 1: very high fallback rate.
        // 10 samples over 1s. Counter goes from 0 to 500.
        // Rate = 500/1.0 = 500/s, well above threshold 10.0.
        for i in 15..25 {
            samples.push(MonitorSample {
                prog_stats: None,
                elapsed_ms: i * 100,
                cpus: vec![CpuSnapshot {
                    nr_running: 2,
                    local_dsq_depth: 1,
                    rq_clock: i * 1000,
                    scx_nr_running: 0,
                    scx_flags: 0,
                    event_counters: Some(ScxEventCounters {
                        select_cpu_fallback: (i as i64 - 15) * 50,
                        dispatch_keep_last: 0,
                        ..Default::default()
                    }),
                    schedstat: None,
                    vcpu_cpu_time_ns: None,
                    vcpu_perf: None,
                    sched_domains: None,
                }],
            });
        }
        let t = Timeline::build(&events, &samples);
        let degs: Vec<_> = t
            .degradations()
            .into_iter()
            .filter(|(_, c)| c.metric == "fallback")
            .collect();
        assert!(!degs.is_empty());
    }

    // -- format_with_context tests --

    #[test]
    fn format_with_context_includes_header() {
        let events = vec![stimulus(0, "ScenarioStart")];
        let samples = vec![
            sample(600, vec![(2, 1, 100), (2, 1, 200)]),
            sample(700, vec![(2, 1, 300), (2, 1, 400)]),
        ];
        let t = Timeline::build(&events, &samples);
        let ctx = TimelineContext {
            kernel: Some("6.14.0-rc3+".to_string()),
            topology: Some("2n4l4c2t (16 cpus)".to_string()),
            scheduler: Some("scx_mitosis".to_string()),
            scenario: Some("proportional".to_string()),
            duration_s: Some(20.5),
        };
        let formatted = t.format_with_context(&ctx);
        assert!(formatted.contains("--- timeline ---"));
        assert!(formatted.contains("kernel: 6.14.0-rc3+"));
        assert!(formatted.contains("topology: 2n4l4c2t (16 cpus)"));
        assert!(formatted.contains("scheduler: scx_mitosis"));
        assert!(formatted.contains("scenario: proportional"));
        assert!(formatted.contains("duration: 20.5s"));
        assert!(formatted.contains("BASELINE"));
    }

    #[test]
    fn format_with_context_partial_fields() {
        let events = vec![stimulus(0, "ScenarioStart")];
        let samples = vec![sample(600, vec![(2, 1, 100)])];
        let t = Timeline::build(&events, &samples);
        let ctx = TimelineContext {
            kernel: None,
            topology: Some("1n1l1c1t (1 cpus)".to_string()),
            scheduler: None,
            scenario: Some("basic".to_string()),
            duration_s: None,
        };
        let formatted = t.format_with_context(&ctx);
        assert!(formatted.contains("topology: 1n1l1c1t"));
        assert!(formatted.contains("scenario: basic"));
        assert!(!formatted.contains("kernel:"));
        assert!(!formatted.contains("scheduler:"));
        assert!(!formatted.contains("duration:"));
    }

    #[test]
    fn format_with_context_empty_timeline() {
        let t = Timeline { phases: vec![] };
        let ctx = TimelineContext {
            kernel: Some("6.14.0".to_string()),
            ..Default::default()
        };
        assert!(t.format_with_context(&ctx).is_empty());
    }

    #[test]
    fn format_with_context_empty_context() {
        let events = vec![stimulus(0, "ScenarioStart")];
        let samples = vec![sample(600, vec![(2, 1, 100)])];
        let t = Timeline::build(&events, &samples);
        let ctx = TimelineContext::default();
        let formatted = t.format_with_context(&ctx);
        // Should have the timeline header and phases but no context line.
        assert!(formatted.contains("--- timeline ---"));
        assert!(formatted.contains("BASELINE"));
        // The line after "--- timeline ---\n" should be "\nBASELINE" (no context line).
        let after_header = &formatted["--- timeline ---\n".len()..];
        assert!(after_header.starts_with('\n'));
    }

    #[test]
    fn garbage_dsq_samples_filtered_from_metrics() {
        // Samples with DSQ depth above DSQ_PLAUSIBILITY_CEILING should be
        // excluded from phase metrics (the bug: garbage values like 1.5B
        // were flowing into timeline output).
        let events = vec![stimulus(0, "ScenarioStart")];
        let garbage_dsq = 1_550_435_906u32;
        let samples = vec![
            // Garbage sample (DSQ above ceiling).
            MonitorSample {
                prog_stats: None,
                elapsed_ms: 600,
                cpus: vec![CpuSnapshot {
                    nr_running: 1,
                    local_dsq_depth: garbage_dsq,
                    rq_clock: 1000,
                    ..Default::default()
                }],
            },
            // Valid sample.
            sample(700, vec![(2, 3, 2000)]),
        ];
        let t = Timeline::build(&events, &samples);
        assert_eq!(t.phases.len(), 1);
        // Only the valid sample should be counted.
        assert_eq!(t.phases[0].metrics.sample_count, 1);
        assert_eq!(t.phases[0].metrics.max_dsq_depth, 3);
    }

    #[test]
    fn all_garbage_samples_yield_no_metrics() {
        let events = vec![stimulus(0, "ScenarioStart")];
        let samples = vec![MonitorSample {
            prog_stats: None,
            elapsed_ms: 600,
            cpus: vec![CpuSnapshot {
                nr_running: 1,
                local_dsq_depth: 50_000,
                rq_clock: 1000,
                ..Default::default()
            }],
        }];
        let t = Timeline::build(&events, &samples);
        assert_eq!(t.phases[0].metrics.sample_count, 0);
    }

    // ---------------------------------------------------------------
    // Negative test: timeline detects degradation at phase transition
    // ---------------------------------------------------------------

    #[test]
    fn neg_timeline_detects_imbalance_degradation() {
        let events = vec![stimulus(0, "ScenarioStart"), stimulus(2000, "StepStart[0]")];
        let mut samples = Vec::new();
        for i in 6..25 {
            samples.push(sample(
                i * 100,
                vec![(2, 1, i * 1000), (2, 1, i * 1000 + 100)],
            ));
        }
        for i in 26..45 {
            samples.push(sample(
                i * 100,
                vec![(1, 1, i * 1000), (10, 1, i * 1000 + 100)],
            ));
        }
        let t = Timeline::build(&events, &samples);
        assert_eq!(t.phases.len(), 2, "must have 2 phases");
        assert!(!t.degradations().is_empty());

        // Phase 0 (baseline) must have samples and reasonable metrics.
        assert!(
            t.phases[0].metrics.sample_count > 0,
            "baseline must have samples"
        );
        assert!(
            (t.phases[0].metrics.avg_imbalance - 1.0).abs() < 0.5,
            "baseline imbalance should be ~1.0, got {:.1}",
            t.phases[0].metrics.avg_imbalance,
        );

        // Phase 1 must have the stimulus label and degradation.
        assert!(
            t.phases[1].metrics.sample_count > 0,
            "phase 1 must have samples"
        );
        assert!(
            t.phases[1]
                .stimulus
                .as_ref()
                .is_some_and(|s| s.label == "StepStart[0]"),
            "phase 1 stimulus must be StepStart[0]",
        );

        let degs = t.degradations();
        assert!(!degs.is_empty());
        let (phase, change) = &degs[0];
        assert_eq!(phase.index, 1);
        assert_eq!(change.metric, "imbalance");
        assert_eq!(change.direction, ChangeDirection::Degraded);
        let delta = change.after - change.before;
        assert!(delta > 0.0, "delta must be positive for degradation");
        assert!(
            delta > IMBALANCE_THRESHOLD,
            "delta {:.1} must exceed threshold {:.1}",
            delta,
            IMBALANCE_THRESHOLD
        );
        assert!(
            change.before < 2.0,
            "before should be low: {:.1}",
            change.before
        );
        assert!(
            change.after > 5.0,
            "after should be high: {:.1}",
            change.after
        );

        // Format output must be parseable.
        let formatted = t.format_with_context(&TimelineContext::default());
        assert!(
            formatted.contains("BASELINE"),
            "format must include BASELINE phase"
        );
        assert!(formatted.contains("Phase 1"), "format must include Phase 1");
        assert!(
            formatted.contains("DEGRADATION"),
            "format must include DEGRADATION label"
        );
        assert!(
            formatted.contains("imbalance"),
            "format must name the metric"
        );
    }

    #[test]
    fn neg_timeline_detects_dsq_depth_degradation() {
        let events = vec![stimulus(0, "ScenarioStart"), stimulus(2000, "StepStart[0]")];
        let mut samples = Vec::new();
        for i in 6..25 {
            samples.push(sample(
                i * 100,
                vec![(2, 1, i * 1000), (2, 1, i * 1000 + 100)],
            ));
        }
        for i in 26..45 {
            samples.push(sample(
                i * 100,
                vec![(2, 20, i * 1000), (2, 20, i * 1000 + 100)],
            ));
        }
        let t = Timeline::build(&events, &samples);
        assert!(
            !t.degradations().is_empty(),
            "DSQ depth jump must be detected"
        );
        let degs = t.degradations();
        let dsq_deg = degs.iter().find(|(_, c)| c.metric == "dsq_depth");
        assert!(dsq_deg.is_some(), "must detect dsq_depth degradation");
        let (phase, change) = dsq_deg.unwrap();
        assert_eq!(phase.index, 1);
        assert_eq!(change.direction, ChangeDirection::Degraded);
        let delta = change.after - change.before;
        assert!(
            delta > DSQ_THRESHOLD,
            "dsq delta {:.1} must exceed threshold {:.1}",
            delta,
            DSQ_THRESHOLD
        );
        assert!(
            change.before < 5.0,
            "before dsq should be low: {:.1}",
            change.before
        );
        assert!(
            change.after > 15.0,
            "after dsq should be high: {:.1}",
            change.after
        );

        let formatted = t.format_with_context(&TimelineContext::default());
        assert!(
            formatted.contains("dsq_depth"),
            "format must name dsq_depth"
        );
        assert!(
            formatted.contains("DEGRADATION"),
            "format must label degradation"
        );
    }

    #[test]
    fn neg_timeline_no_degradation_when_stable() {
        let events = vec![stimulus(0, "ScenarioStart"), stimulus(2000, "StepStart[0]")];
        let mut samples = Vec::new();
        for i in 6..45 {
            samples.push(sample(
                i * 100,
                vec![(2, 1, i * 1000), (2, 1, i * 1000 + 100)],
            ));
        }
        let t = Timeline::build(&events, &samples);
        assert_eq!(t.phases.len(), 2, "must have 2 phases");
        assert!(t.phases[0].metrics.sample_count > 0);
        assert!(t.phases[1].metrics.sample_count > 0);
        assert!(
            t.degradations().is_empty(),
            "stable phases must not show degradation"
        );
        assert!(t.degradations().is_empty());
        // All phase changes should be empty.
        for phase in &t.phases {
            assert!(
                phase.changes.is_empty(),
                "phase {} should have no changes",
                phase.index
            );
        }
    }

    // -- detect_change direct tests --

    #[test]
    fn detect_change_higher_is_worse_positive_delta_degraded() {
        let c = detect_change(1.0, 5.0, 0.5, "imbalance", true).unwrap();
        assert_eq!(c.direction, ChangeDirection::Degraded);
        assert_eq!(c.metric, "imbalance");
        assert!((c.before - 1.0).abs() < f64::EPSILON);
        assert!((c.after - 5.0).abs() < f64::EPSILON);
    }

    #[test]
    fn detect_change_higher_is_worse_negative_delta_improved() {
        let c = detect_change(5.0, 1.0, 0.5, "imbalance", true).unwrap();
        assert_eq!(c.direction, ChangeDirection::Improved);
    }

    #[test]
    fn detect_change_lower_is_worse_negative_delta_degraded() {
        let c = detect_change(100.0, 50.0, 10.0, "throughput", false).unwrap();
        assert_eq!(c.direction, ChangeDirection::Degraded);
    }

    #[test]
    fn detect_change_lower_is_worse_positive_delta_improved() {
        let c = detect_change(50.0, 100.0, 10.0, "throughput", false).unwrap();
        assert_eq!(c.direction, ChangeDirection::Improved);
    }

    #[test]
    fn detect_change_below_threshold_returns_none() {
        assert!(detect_change(1.0, 1.3, 0.5, "imbalance", true).is_none());
    }

    #[test]
    fn detect_change_exactly_at_threshold_returns_none() {
        assert!(detect_change(1.0, 1.5, 0.5, "imbalance", true).is_none());
    }

    // -- iteration_rate computation tests --

    fn stimulus_with_iters(elapsed_ms: u64, label: &str, total_iterations: u64) -> StimulusEvent {
        StimulusEvent {
            elapsed_ms,
            label: label.to_string(),
            op_kind: None,
            detail: None,
            total_iterations: Some(total_iterations),
        }
    }

    #[test]
    fn iteration_rate_computed_from_consecutive_events() {
        // Two events with total_iterations: phase 0 spans 0..3000ms
        // (aligned). iterations: 0 -> 3000 over ~3s = 1000 iter/s.
        let events = vec![
            stimulus_with_iters(0, "ScenarioStart", 0),
            stimulus_with_iters(3000, "StepStart[0]", 3000),
        ];
        let samples: Vec<MonitorSample> = (5..35)
            .map(|i| sample(i * 100, vec![(2, 1, i * 1000), (2, 1, i * 1000 + 100)]))
            .collect();
        let t = Timeline::build(&events, &samples);
        assert_eq!(t.phases.len(), 2);
        let rate = t.phases[0].metrics.iteration_rate;
        assert!(rate.is_some(), "phase 0 should have iteration_rate");
        let r = rate.unwrap();
        // Duration is phase boundary difference, not exactly 3s due to
        // clock alignment offset. Check that the rate is reasonable.
        assert!(r > 500.0 && r < 2000.0, "rate {r} outside expected range");
    }

    #[test]
    fn iteration_rate_none_without_total_iterations() {
        // Events without total_iterations: iteration_rate should be None.
        let events = vec![stimulus(0, "ScenarioStart"), stimulus(3000, "StepStart[0]")];
        let samples: Vec<MonitorSample> = (5..35)
            .map(|i| sample(i * 100, vec![(2, 1, i * 1000), (2, 1, i * 1000 + 100)]))
            .collect();
        let t = Timeline::build(&events, &samples);
        assert!(t.phases[0].metrics.iteration_rate.is_none());
        assert!(t.phases[1].metrics.iteration_rate.is_none());
    }

    #[test]
    fn throughput_degradation_detected() {
        // Phase 0: high throughput (0 -> 10000 iters over ~2s = ~5000/s)
        // Phase 1: low throughput (10000 -> 11000 iters over ~2s = ~500/s)
        // 90% drop exceeds ITERATION_RATE_REL_THRESHOLD (0.3).
        let events = vec![
            stimulus_with_iters(0, "ScenarioStart", 0),
            stimulus_with_iters(2000, "StepStart[0]", 10000),
            stimulus_with_iters(4000, "StepEnd[0]", 11000),
        ];
        let samples: Vec<MonitorSample> = (5..45)
            .map(|i| sample(i * 100, vec![(2, 1, i * 1000), (2, 1, i * 1000 + 100)]))
            .collect();
        let t = Timeline::build(&events, &samples);
        assert_eq!(t.phases.len(), 3);
        // Phase 0 should have high iteration_rate.
        assert!(t.phases[0].metrics.iteration_rate.is_some());
        // Phase 1 should have low iteration_rate.
        assert!(t.phases[1].metrics.iteration_rate.is_some());
        let r0 = t.phases[0].metrics.iteration_rate.unwrap();
        let r1 = t.phases[1].metrics.iteration_rate.unwrap();
        assert!(
            r0 > r1,
            "phase 0 rate ({r0}) should exceed phase 1 rate ({r1})"
        );

        // Throughput degradation should be detected at phase 1 boundary.
        let degs: Vec<_> = t
            .degradations()
            .into_iter()
            .filter(|(_, c)| c.metric == "throughput")
            .collect();
        assert!(!degs.is_empty(), "throughput degradation must be detected");
        let (phase, change) = &degs[0];
        assert_eq!(phase.index, 1);
        assert_eq!(change.direction, ChangeDirection::Degraded);
        assert!(change.before > change.after);
    }

    #[test]
    fn throughput_improvement_detected() {
        // Phase 0: low throughput (0 -> 500 iters over ~2s = ~250/s)
        // Phase 1: high throughput (500 -> 10500 iters over ~2s = ~5000/s)
        // >30% increase should be flagged as improvement.
        let events = vec![
            stimulus_with_iters(0, "ScenarioStart", 0),
            stimulus_with_iters(2000, "StepStart[0]", 500),
            stimulus_with_iters(4000, "StepEnd[0]", 10500),
        ];
        let samples: Vec<MonitorSample> = (5..45)
            .map(|i| sample(i * 100, vec![(2, 1, i * 1000), (2, 1, i * 1000 + 100)]))
            .collect();
        let t = Timeline::build(&events, &samples);
        let improvements: Vec<_> = t
            .phases
            .iter()
            .flat_map(|p| p.changes.iter())
            .filter(|c| c.metric == "throughput" && c.direction == ChangeDirection::Improved)
            .collect();
        assert!(
            !improvements.is_empty(),
            "throughput improvement must be detected"
        );
    }

    #[test]
    fn throughput_stable_below_threshold() {
        // Phase 0: 1000 iter/s
        // Phase 1: ~900 iter/s (10% drop, below 30% threshold)
        // No throughput change should be detected.
        let events = vec![
            stimulus_with_iters(0, "ScenarioStart", 0),
            stimulus_with_iters(2000, "StepStart[0]", 2000),
            stimulus_with_iters(4000, "StepEnd[0]", 3800),
        ];
        let samples: Vec<MonitorSample> = (5..45)
            .map(|i| sample(i * 100, vec![(2, 1, i * 1000), (2, 1, i * 1000 + 100)]))
            .collect();
        let t = Timeline::build(&events, &samples);
        let throughput_changes: Vec<_> = t
            .phases
            .iter()
            .flat_map(|p| p.changes.iter())
            .filter(|c| c.metric == "throughput")
            .collect();
        assert!(
            throughput_changes.is_empty(),
            "10% change should not trigger throughput change detection"
        );
    }

    #[test]
    fn iteration_rate_in_formatted_output() {
        let events = vec![
            stimulus_with_iters(0, "ScenarioStart", 0),
            stimulus_with_iters(2000, "StepStart[0]", 5000),
        ];
        let samples: Vec<MonitorSample> = (5..25)
            .map(|i| sample(i * 100, vec![(2, 1, i * 1000), (2, 1, i * 1000 + 100)]))
            .collect();
        let t = Timeline::build(&events, &samples);
        let formatted = t.format_with_context(&TimelineContext::default());
        assert!(
            formatted.contains("throughput:"),
            "format output must contain throughput when iteration_rate is set"
        );
        assert!(formatted.contains("iter/s"));
    }
}