ktstr 0.17.0

//! 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. Produced as `"StepStart[k]"` by
    /// [`Self::from_wire`] (the 0-indexed scenario Step ordinal),
    /// `"ScenarioEnd"` by [`Self::terminal`], and the
    /// `"BASELINE"`/`"Step[k]"` bucket label by the
    /// `phase_from_bucket` placeholder. Test fixtures may carry any
    /// label.
    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. `Some(_)` for every
    /// event built from the wire (the wire counter is always present —
    /// see [`Self::from_wire`]); a cumulative counter for which
    /// `Some(0)` is a legitimate "no iterations accumulated yet"
    /// baseline, NOT a missing sample. `None` only for synthetic /
    /// placeholder events that carry no counter (the
    /// `phase_from_bucket` fallback and test fixtures). Used to
    /// compute per-phase throughput (iterations/s) as the delta
    /// between consecutive events.
    ///
    /// SEMANTICS: this is the sum of the iteration counters of the
    /// worker handles ALIVE at the event instant (step-local +
    /// Backdrop). Each step emits BOTH a StepStart event (counter at the
    /// step's start) and a StepEnd event ([`Self::is_step_end`], counter
    /// at the step's end-of-hold), so the per-phase iteration_rate is the
    /// STEP-LOCAL delta `StepEnd[k] - StepStart[k]` — each step's OWN
    /// workers measured start-to-end. That works for workers respawned
    /// per step (the cross-step `StepStart[k+1] - StepStart[k]` delta
    /// reads fresh~0 - fresh~0 and is dropped) AND is more accurate for
    /// persistent (Backdrop) workers (it excludes the inter-step
    /// teardown/respawn wall-time the cross-step window spanned). Bucket
    /// `k` is sourced ONLY by its `StepStart[k] -> StepEnd[k]` pair: the
    /// `iteration_rate` attribution loop in
    /// [`crate::assert::build_phase_buckets_with_stimulus`] skips any
    /// `is_step_end` `prev`, so a stalled step whose step-local delta is
    /// zero (`StepEnd[k] == StepStart[k]`) reports its MEASURED-ZERO rate
    /// `Some(0.0)` (see `Self::rate_to`) rather than leaking the
    /// inter-step gap rate from the `StepEnd[k] -> StepStart[k+1]` pair.
    /// The monitor-only
    /// [`Timeline::build`] fallback (no snapshot captures) computes the
    /// SAME step-local `StepStart[k] -> StepEnd[k]` rate — the StepEnd
    /// events reach it too (they are emitted independent of captures) — and
    /// falls back to cross-step (or the terminal for the last step) only
    /// when a step has no StepEnd (sched-died / legacy data); StepEnd is
    /// filtered only from that path's phase LAYOUT, not its rate.
    pub total_iterations: Option<u64>,
    /// 1-indexed scenario step this event belongs to (the same
    /// encoding the bridge stamps: `1..=N` for Step ordinals), or
    /// `None` for non-step events (including the terminal scenario-end
    /// boundary; see `is_terminal`). Carried explicitly from the wire
    /// `StimulusPayload.step_index` so the periodic-capture phase
    /// attribution can map a capture's workload-relative boundary
    /// offset onto the guest's own step timeline without parsing the
    /// human-readable `label`.
    pub step_index: Option<u16>,
    /// True only for the synthetic scenario-end boundary the eval
    /// walker appends from the `ScenarioEnd` wire frame's final
    /// `total_iterations`. On a CLEAN run the last step emits its own
    /// `StepEnd[N]`, which supplies that step's `iteration_rate` right
    /// boundary in BOTH rate consumers — the snapshot path
    /// ([`crate::assert::build_phase_buckets_with_stimulus`], the
    /// `StepStart[N]` -> `StepEnd[N]` pair) and the monitor-only
    /// [`Timeline::build`] fallback (which looks up each step's `StepEnd`
    /// by `step_index`) — and the terminal is then NOT consumed for a
    /// rate: the snapshot path's attribution loop skips the
    /// `(StepEnd[N], terminal)` pair via its `is_step_end` guard (before
    /// `rate_components` is reached), and `Timeline::build` reaches for the
    /// terminal only when a step's `StepEnd` lookup misses. The terminal
    /// is consumed as a step's rate boundary ONLY for legacy/synthetic
    /// data that carries a `ScenarioEnd` frame but no `StepEnd` frames
    /// (fresh guest output always pairs them). A sched-died step is NOT
    /// such a case: its early return skips BOTH the `StepEnd` emission AND
    /// `send_scenario_end`, so neither frame exists and the dead step
    /// reports no rate via the no-successor path. It is NOT a step start:
    /// `step_index` is `None` so it seeds no [`crate::assert::PhaseBucket`]
    /// (excluded from the step-start timeline), and [`Timeline::build`]
    /// skips it when laying out phase boundaries so it never renders a
    /// phantom trailing phase.
    pub is_terminal: bool,
    /// True for a per-step END event (decoded from a
    /// `crate::vmm::wire::MsgType::StepEnd` frame via
    /// [`Self::from_step_end`]). It carries the SAME 1-indexed
    /// `step_index` as its StepStart and its step's end-of-hold
    /// `total_iterations`, so [`crate::assert::build_phase_buckets_with_stimulus`]'s
    /// elapsed-sorted `windows(2)` pairs `StepStart[k]` -> `StepEnd[k]`
    /// first and `or_insert` keeps that step-local rate. NOT a step
    /// start, so [`Timeline::build`] (the monitor-only fallback's
    /// index-based cross-step pairing) filters it out of its step-start
    /// list to avoid a phantom phase.
    pub is_step_end: bool,
}

impl StimulusEvent {
    /// Build a timeline event from a deserialized wire stimulus event.
    /// Centralizes the wire→timeline mapping so the production eval path
    /// (`evaluate_vm_result`) and out-of-tree consumers — post_vm
    /// callbacks folding `VmResult::stimulus_timeline()` (which calls
    /// this internally) through
    /// [`crate::assert::build_phase_buckets_with_stimulus`] — produce
    /// identical events. The wire `step_index` is the bridge 1-indexed
    /// convention (`Step[k]` -> `k + 1`, BASELINE owns 0); the human
    /// `label` renders the 0-indexed Scenario-Step ordinal
    /// (`step_index - 1`) to match the `PhaseBucket` `Step[k]` labels,
    /// while the `step_index` field keeps the 1-indexed wire value for
    /// phase-bucket remap. `total_iterations` is carried verbatim as
    /// `Some(_)`: the wire field is a cumulative counter that is always
    /// populated (the guest sums live worker iterations at every step
    /// boundary), so `0` is a legitimate baseline reading — the FIRST
    /// step's frame fires right after its workers spawn and genuinely
    /// reads ~0. Collapsing that `0` to `None` (the old behavior) made
    /// the (first, second) delta pair fail the `Some`/`Some` guard in
    /// both rate consumers, silently dropping the first step's
    /// `iteration_rate`; carrying `Some(0)` lets the delta compute the
    /// first step's throughput for the PERSISTENT (Backdrop) population
    /// (see the `total_iterations` field doc for the persistent-vs-
    /// step-local semantics this delta measures).
    pub fn from_wire(ev: &crate::vmm::wire::StimulusEvent) -> Self {
        Self {
            elapsed_ms: ev.elapsed_ms as u64,
            label: format!("StepStart[{}]", ev.step_index.saturating_sub(1)),
            op_kind: Some(format!("ops={}", ev.op_count)),
            detail: Some(format!(
                "{} cgroups, {} workers",
                ev.cgroup_count, ev.worker_count,
            )),
            total_iterations: Some(ev.total_iterations),
            step_index: Some(ev.step_index),
            is_terminal: false,
            is_step_end: false,
        }
    }

    /// Build a per-step END event from a `crate::vmm::wire::MsgType::StepEnd`
    /// frame (reuses the `crate::vmm::wire::StimulusEvent` wire body).
    /// Carries the SAME 1-indexed `step_index` as the step's StepStart
    /// and the step's end-of-hold `total_iterations`, with `is_step_end`
    /// set. Elapsed-sorted, a step's events order `StepStart[k]` (start) <
    /// `StepEnd[k]` (end-of-hold) < `StepStart[k+1]`, so
    /// [`crate::assert::build_phase_buckets_with_stimulus`]'s `windows(2)`
    /// pairs `StepStart[k]` -> `StepEnd[k]` first and `or_insert` keeps that
    /// step-local rate. `is_terminal` is false (it is a real per-step
    /// boundary, not the scenario-end terminal).
    pub fn from_step_end(ev: &crate::vmm::wire::StimulusEvent) -> Self {
        Self {
            elapsed_ms: ev.elapsed_ms as u64,
            label: format!("StepEnd[{}]", ev.step_index.saturating_sub(1)),
            op_kind: Some(format!("ops={}", ev.op_count)),
            detail: Some(format!(
                "{} cgroups, {} workers",
                ev.cgroup_count, ev.worker_count,
            )),
            total_iterations: Some(ev.total_iterations),
            step_index: Some(ev.step_index),
            is_terminal: false,
            is_step_end: true,
        }
    }

    /// Build the synthetic terminal boundary event from the
    /// `ScenarioEnd` wire frame's final cumulative `total_iterations`
    /// and scenario-relative `elapsed_ms`. Appended once, after every
    /// per-step [`Self::from_wire`] event. On a clean run `StepEnd[N]`
    /// supplies the last step's `iteration_rate` right boundary in both
    /// rate consumers and the terminal is not consumed for a rate; it is
    /// consumed as a step's boundary ONLY for legacy/synthetic data with a
    /// `ScenarioEnd` frame but no `StepEnd` frames (a sched-died step has
    /// neither, since the early return skips both emissions) — see the
    /// [`Self::is_terminal`] field doc.
    /// `step_index` is `None` (it is not a step start — it seeds no
    /// [`crate::assert::PhaseBucket`]) and `is_terminal` is set so
    /// [`Timeline::build`] treats it as a right boundary only, never a
    /// phase. `elapsed_ms` is in the same guest-monotonic frame as the
    /// step events (both come from `scenario_start.elapsed()`), so the
    /// last-step duration is well-formed.
    pub fn terminal(elapsed_ms: u64, total_iterations: u64) -> Self {
        Self {
            elapsed_ms,
            label: "ScenarioEnd".to_string(),
            op_kind: None,
            detail: None,
            total_iterations: Some(total_iterations),
            step_index: None,
            is_terminal: true,
            is_step_end: false,
        }
    }

    /// Iterations-per-second from this event to `next`:
    /// `(next.total_iterations - self.total_iterations)` over the
    /// guest-clock elapsed-ms delta between them. Returns `None` ONLY when
    /// the measurement is genuinely undefined: either event lacks a
    /// `total_iterations` sample, the window is zero-length, or the count
    /// went BACKWARD (`next < self` — a counter reset; the delta is
    /// unmeasurable, not zero). The backward case is reachable only for the
    /// guard-skipped cross-step pairing or legacy/synthetic data, NOT for
    /// the live step-local `StepStart[k]` -> `StepEnd[k]` pair: teardown
    /// runs after `StepEnd` is emitted, so the handle set is stable within
    /// a step and the per-worker counters are monotone across the pair.
    ///
    /// MEASURED ZERO is distinct from not-measured: a step whose workers
    /// made exactly zero forward progress over a positive hold
    /// (`next == self`) returns `Some(0.0)`, not `None`. Zero throughput
    /// is a real, measured value — the strongest degradation signal — so
    /// it must surface, not vanish. With `Some(0.0)` a phase that
    /// collapsed to zero IS visible to the throughput-degradation detector
    /// ([`Timeline::build`] / [`Timeline::from_phase_buckets`]): when the
    /// prior phase had a positive rate (`before > 0.0`), the relative
    /// delta is `-1.0` and the drop is flagged. (A phase that was already
    /// zero before is still not relatively comparable — the detector's
    /// `before > 0.0` gate avoids a div-by-zero — but an *unchanged* zero
    /// is not a degradation.)
    ///
    /// This is the SINGLE iteration-rate formula, shared via its
    /// decomposition [`Self::rate_components`] by
    /// [`crate::assert::build_phase_buckets_with_stimulus`] (per-step
    /// windows attributed by `step_index`) and [`Timeline::build`]
    /// (per-phase windows attributed by index) — the two callers pair
    /// events differently but must compute the rate identically. The
    /// per-step metric producer inserts the `rate_components` pair (the
    /// `iteration_rate` Rate's `total_phase_iterations` /
    /// `total_phase_duration_sec` components); `rate_to` (the quotient) is
    /// the display/comparison form used by `Timeline::build` and the
    /// result-helper ratios.
    pub fn rate_to(&self, next: &StimulusEvent) -> Option<f64> {
        self.rate_components(next).map(|(iters, secs)| iters / secs)
    }

    /// The `(iteration_delta, window_seconds)` components of [`Self::rate_to`]
    /// — same `None` conditions (missing `total_iterations`, backward count,
    /// or zero-length window). The per-phase metric pipeline inserts these as
    /// the `total_phase_iterations` / `total_phase_duration_sec` Counter
    /// components rather than the ready ratio, so the `iteration_rate` Rate
    /// re-pools across phases/runs as `Σdelta / Σseconds`, not a mean of
    /// per-phase ratios. The ms→s `/1000` lives HERE (the seconds component)
    /// because `derive_rate_metrics` does a bare num/den with no scaling.
    pub fn rate_components(&self, next: &StimulusEvent) -> Option<(f64, f64)> {
        let s = self.total_iterations?;
        let e = next.total_iterations?;
        if e < s {
            return None;
        }
        let duration_ms = next.elapsed_ms.saturating_sub(self.elapsed_ms);
        if duration_ms == 0 {
            return None;
        }
        Some(((e - s) as f64, duration_ms as f64 / 1000.0))
    }

    /// The scenario [`Phase`](crate::assert::Phase) this event belongs to,
    /// or `None` for the terminal scenario-end boundary (which seeds no
    /// phase). Use THIS — not the raw [`Self::step_index`] field — to key
    /// per-phase lookups. `step_index` carries the bridge 1-indexed wire
    /// convention (`Step k` -> `Some(k + 1)`) while `label` renders the
    /// 0-indexed `k`, so reading the field directly invites the 0-vs-1
    /// off-by-one this method removes: it maps the wire value onto the same
    /// [`Phase`](crate::assert::Phase) newtype the
    /// [`ScenarioStats`](crate::assert::ScenarioStats) /
    /// [`PhaseBucket`](crate::assert::PhaseBucket) accessors are keyed by
    /// (`Phase::step(k)`). Step events carry `step_index >= 1`, so the
    /// `saturating_sub(1)` is exact.
    pub fn phase(&self) -> Option<crate::assert::Phase> {
        self.step_index
            .map(|si| crate::assert::Phase::step(si.saturating_sub(1)))
    }
}

// ---------------------------------------------------------------------------
// 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,
    /// Mean CPU-imbalance ratio over the phase's valid samples. `None`
    /// when the phase had no valid samples (monitor-only `Timeline::build`)
    /// or its source bucket carried no `avg_imbalance_ratio` metric
    /// (snapshot `from_phase_buckets`) — distinct from a real `Some(0.0)`
    /// (perfectly balanced). The change detector compares it only when
    /// both sides are `Some`, so an absent phase never reads as a false
    /// zero-imbalance.
    pub avg_imbalance: Option<f64>,
    /// Peak CPU-imbalance ratio over the phase's valid samples. `None` on
    /// the same no-data conditions as [`Self::avg_imbalance`].
    pub max_imbalance: Option<f64>,
    /// Mean local-DSQ depth over the phase's valid samples. `None` on the
    /// same no-data conditions as [`Self::avg_imbalance`].
    pub avg_dsq_depth: Option<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>,
    /// Per-cgroup raw telemetry for this phase, keyed by cgroup name. Carried
    /// from [`crate::assert::PhaseBucket::per_cgroup`] on the
    /// [`Self`]-via-`from_phase_buckets` path; empty on the monitor-only
    /// [`Timeline::build`] path (which has no carriers). Rendered as a
    /// per-cgroup sub-block by display-time reduction
    /// ([`crate::assert::PhaseCgroupStats::off_cpu_summary`] etc.); never a
    /// change-detection input (those are [`PhaseMetrics`] scalars only).
    pub per_cgroup: std::collections::BTreeMap<String, crate::assert::PhaseCgroupStats>,
    /// True when `(start_ms, end_ms)` is the normalized `(0, 0)` of an ORPHAN
    /// carrier (no measured host window) rather than a real window. Set in
    /// `phase_from_bucket` by the orphan shape signature `(0,0)` + empty
    /// `metrics` + non-empty `per_cgroup` (unique to
    /// `crate::assert::fold_guest_per_cgroup_into_host_buckets`'s orphan arm).
    /// The render shows "window not measured" instead of a misleading `0ms`.
    /// Always `false` on the [`Timeline::build`] path.
    pub not_measured_window: bool,
}

// ---------------------------------------------------------------------------
// 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,
    })
}

/// Per-boundary change set between two adjacent phases, shared by both
/// [`Timeline::build`] and [`Timeline::from_phase_buckets`].
///
/// Throughput (`iteration_rate`) is compared whenever BOTH phases carry
/// a rate and the earlier one is positive — INCLUDING a synthesized
/// zero-capture step, whose rate is stimulus-derived (total_iterations
/// deltas) rather than sampled. Throughput is the one metric that
/// survives a capture gap, so a throughput collapse entering or leaving
/// a synthesized step must still be flagged; gating it on `sample_count`
/// (as both call sites did before) silently dropped exactly the
/// degradation this timeline exists to surface.
///
/// Asymmetry from the `bi > 0.0` div-by-zero guard: a COLLAPSE into a
/// zero-rate (incl. a synthesized measured-zero) step is flagged
/// (before > 0), but a RECOVERY out of one (before == 0 -> positive) is
/// not — there is no relative baseline to divide by. This is a
/// deliberate tradeoff, not an oversight: an unchanged zero is genuinely
/// not a degradation, and the collapse direction (the one this task
/// targets) is caught.
///
/// The monitor-derived metrics (imbalance, dsq depth, fallback rate,
/// keep_last rate) ARE gated on both phases having real samples. The
/// values themselves can be real even on a synthesized phase (monitor
/// windows fold into the bucket), but they come from a DIFFERENT
/// sampling basis (folded monitor window vs captured periodic samples)
/// than a captured neighbor's, so cross-comparing them is
/// apples-to-oranges — suppressed. Throughput is exempt because it is
/// the same stimulus-derived quantity on both sides.
fn detect_boundary_changes(before: &PhaseMetrics, after: &PhaseMetrics) -> Vec<PhaseChange> {
    let mut changes = Vec::new();

    // Monitor-derived metrics (absolute-unit gauges/rates with
    // fixed-magnitude thresholds) — gated on both phases having real
    // samples. Pushed FIRST to preserve the historical render order
    // (these before throughput in a multi-change boundary; format_phases
    // renders changes in vec order).
    if before.sample_count > 0 && after.sample_count > 0 {
        if let (Some(bi), Some(ai)) = (before.avg_imbalance, after.avg_imbalance) {
            changes.extend(detect_change(
                bi,
                ai,
                IMBALANCE_THRESHOLD,
                "imbalance",
                true,
            ));
        }
        if let (Some(bd), Some(ad)) = (before.avg_dsq_depth, after.avg_dsq_depth) {
            changes.extend(detect_change(bd, ad, DSQ_THRESHOLD, "dsq_depth", true));
        }
        if let (Some(bf), Some(af)) = (before.fallback_rate, after.fallback_rate) {
            changes.extend(detect_change(
                bf,
                af,
                FALLBACK_RATE_THRESHOLD,
                "fallback",
                true,
            ));
        }
        if let (Some(bk), Some(ak)) = (before.keep_last_rate, after.keep_last_rate) {
            changes.extend(detect_change(
                bk,
                ak,
                KEEP_LAST_RATE_THRESHOLD,
                "keep_last",
                true,
            ));
        }
    }

    // Throughput is the SOLE rate-class field (counter-delta / elapsed), so
    // it uses a RELATIVE threshold (rel = (ai-bi)/bi vs
    // ITERATION_RATE_REL_THRESHOLD) and cannot route through detect_change's
    // absolute-delta gate above — fixed-unit metrics get absolute
    // thresholds, this gets a relative one (a semantic-class decision, not
    // arbitrary). Strict `>`: an exactly-at-threshold relative change is not
    // flagged. Pushed last so it renders after the monitor metrics.
    if let (Some(bi), Some(ai)) = (before.iteration_rate, after.iteration_rate)
        && bi > 0.0
    {
        let rel = (ai - bi) / bi;
        if rel.abs() > ITERATION_RATE_REL_THRESHOLD {
            changes.push(PhaseChange {
                direction: if rel < 0.0 {
                    ChangeDirection::Degraded
                } else {
                    ChangeDirection::Improved
                },
                metric: "throughput".to_string(),
                before: bi,
                after: ai,
            });
        }
    }

    changes
}

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.
    /// Build a Timeline from stimulus events + raw monitor
    /// samples via the per-window `compute_metrics` reduction.
    /// The production success path uses [`Self::from_phase_buckets`]
    /// (which folds pre-bucketed PhaseBuckets); `build` is the fallback
    /// evaluate_vm_result takes only for a run with an EMPTY PhaseBuckets
    /// vec but monitor samples present — i.e. no periodic captures AND no
    /// stimulus Steps. A monitor-only run that DID run Steps now
    /// synthesizes a capture-free bucket per StepStart (see
    /// [`crate::assert::build_phase_buckets_with_stimulus`]), so its vec is
    /// non-empty and it takes the from_phase_buckets path (whose
    /// fold_monitor_into_bucket recovers the same monitor-derived metric
    /// set this path computes). Both entry points produce the same
    /// Timeline field shape; from_phase_buckets is preferred when buckets
    /// are available because it avoids the per-MonitorSample reduction.
    ///
    /// `preemption_threshold_ns` threads the vCPU-preemption exemption
    /// window into the per-phase stall predicate (see `compute_metrics`);
    /// the production caller passes the run's
    /// `MonitorReport::preemption_threshold_ns`. `0` derives it from the
    /// guest kernel `CONFIG_HZ`.
    pub fn build(
        stimulus_events: &[StimulusEvent],
        monitor_samples: &[MonitorSample],
        preemption_threshold_ns: u64,
    ) -> Self {
        if stimulus_events.is_empty() || monitor_samples.is_empty() {
            return Self { phases: Vec::new() };
        }

        let mut events = stimulus_events.to_vec();
        // Total-order on an elapsed_ms tie: StepEnd before StepStart
        // (`!is_step_end` is false=0 for StepEnd) so a zero-length
        // inter-step gap at the guest's coarse-ms clock attributes the
        // step-local StepStart[k]->StepEnd[k] rate to bucket k, never the
        // cross-step StepStart[k]->StepStart[k+1] delta. Mirrors the same
        // sort in build_phase_buckets_with_stimulus so the two rate
        // consumers stay identical.
        events.sort_by_key(|e| (e.elapsed_ms, !e.is_step_end));

        // 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);

        // The terminal scenario-end event is a rate right
        // boundary ONLY — it seeds no phase. Extract it explicitly
        // rather than relying on it sorting last for positional
        // alignment: a corrupt / out-of-order step `elapsed_ms` (a u32
        // read off the wire) could otherwise shift it into the middle
        // of `events` and misalign the dense phase index against the
        // step events. `step_events` is the phase-bearing set.
        let terminal: Option<&StimulusEvent> = events.iter().find(|e| e.is_terminal);
        // StepStart events only — the PHASE-LAYOUT set. Per-step StepEnd
        // events are excluded here because a StepEnd seeds no new phase
        // (it is an end-of-hold marker, not a step boundary); including
        // them would produce a phantom extra phase and misalign the dense
        // phase index. StepEnd events are NOT discarded, though: the
        // step-local iteration_rate loop below pairs each StepStart[k]
        // with its own StepEnd[k] (looked up by step_index in the full
        // `events` vec), matching build_phase_buckets_with_stimulus. The
        // dense-index cross-step pairing is kept only as a fallback for
        // steps that have no StepEnd (a sched-died step, or legacy data
        // predating the StepEnd frame).
        let step_events: Vec<&StimulusEvent> = events
            .iter()
            .filter(|e| !e.is_terminal && !e.is_step_end)
            .collect();

        let mut boundaries: Vec<(u64, u64, Option<StimulusEvent>)> = Vec::new();
        for i in 0..step_events.len() {
            let start = (step_events[i].elapsed_ms as i64 + offset).max(0) as u64;
            // The LAST step phase extends to end-of-monitor-data, NOT to
            // the terminal event: the terminal is a rate boundary only,
            // and clamping the last phase's metric window to it would
            // drop trailing monitor samples (the host keeps sampling
            // through teardown). Preserves the pre-terminal window.
            let end = if i + 1 < step_events.len() {
                (step_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(step_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, preemption_threshold_ns);

            phases.push(Phase {
                // Enumerate position over the phase-bearing step_events,
                // NOT the bucket step_index `phase_from_bucket` uses. This
                // path's input is a monitor-only / legacy / test stream with
                // no step_index-bearing Stimulus frames (the settle is
                // step_events[0]; later events follow), so enumerate IS the
                // step identity and `index == 0 => BASELINE` holds for this
                // model. It diverges from the step_index model ONLY for a
                // step_index-bearing stream with no leading settle (the
                // production stimulus shape), which never reaches `build`:
                // build is the monitor-only fallback, taken only when there
                // are no PhaseBuckets, i.e. no StepStarts
                // (build_phase_buckets_with_stimulus synthesizes a bucket
                // per StepStart) — so a step_index-bearing production stream
                // always takes from_phase_buckets, never this path.
                index: idx,
                start_ms: start,
                end_ms: end,
                stimulus,
                metrics,
                changes: Vec::new(),
                // Monitor-only path: no per-cgroup carriers, real window.
                per_cgroup: std::collections::BTreeMap::new(),
                not_measured_window: false,
            });
        }

        // Per-phase iteration rate, STEP-LOCAL: each step's rate is its
        // own `StepStart[k] -> StepEnd[k]` delta — the step's OWN workers
        // measured start-to-end-of-hold, matching the snapshot path
        // (`build_phase_buckets_with_stimulus`). StepEnd events are
        // present in `events` (emitted independent of snapshot captures)
        // even on this monitor-only path, so the same step-local model
        // applies; without it, workers respawned fresh each step read
        // ~0 -> ~0 cross-step and every fresh-per-step phase but the last
        // silently reported no throughput. A step with NO StepEnd falls
        // back to the cross-step successor, or the terminal scenario-end
        // event for the last step — but that fallback yields a rate only
        // for legacy/synthetic data (a ScenarioEnd frame present without
        // per-step StepEnd frames). A sched-died step has neither a
        // StepEnd nor a terminal (the early return skips both emissions),
        // so its lookup and fallback both miss and it correctly reports no
        // rate. Duration is the guest-clock elapsed-ms delta between the
        // paired events — independent of the metric-sample window above
        // (whose last phase reaches end-of-monitor-data).
        #[allow(clippy::needless_range_loop)]
        for i in 0..phases.len() {
            let this = step_events[i];
            // Step-local boundary: this step's own StepEnd (same
            // step_index). Cross-step successor / terminal only when the
            // step has no StepEnd.
            let step_end: Option<&StimulusEvent> = this.step_index.and_then(|k| {
                events
                    .iter()
                    .find(|e| e.is_step_end && e.step_index == Some(k))
            });
            let next: Option<&StimulusEvent> = step_end.or_else(|| {
                if i + 1 < step_events.len() {
                    Some(step_events[i + 1])
                } else {
                    terminal
                }
            });
            // Timeline::build's display fallback: compute this phase's rate
            // directly via rate_to. The metric-pipeline producer
            // build_phase_buckets_with_stimulus shares the same rate
            // semantics via rate_components (it emits the two Counter
            // components that derive_rate_metrics re-pools into
            // iteration_rate); this display field reads the quotient.
            if let Some(next_ev) = next
                && let Some(rate) = this.rate_to(next_ev)
            {
                phases[i].metrics.iteration_rate = Some(rate);
            }
        }

        // Detect changes at each phase boundary. Throughput is compared
        // even across synthesized zero-capture steps; monitor-derived
        // metrics stay gated on both phases having real samples (see
        // [`detect_boundary_changes`]).
        for i in 1..phases.len() {
            let changes = detect_boundary_changes(&phases[i - 1].metrics, &phases[i].metrics);
            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
    // (crate::test_support::eval) 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);
            // An orphan carrier carries a normalized (0,0) window that is NOT a
            // measured zero-duration step; surface it as not-measured rather
            // than a misleading 0ms (the None-vs-Some discipline the per-metric
            // renders use, applied to the window).
            let window = if phase.not_measured_window {
                "window not measured".to_string()
            } else {
                format!("{duration_ms}ms")
            };

            if phase.index == 0 {
                // Phase 0 is the settle window before any stimulus.
                out.push_str(&format!(
                    "\nBASELINE (settle, {}, {} samples):\n",
                    window, 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 {}: {} ({}, {} samples):\n",
                    phase.index, label_start, window, phase.metrics.sample_count,
                ));
            }

            let m = &phase.metrics;
            // Render the metric block whenever the phase carries
            // monitor-derived metrics, not only when it captured periodic
            // samples: a SYNTHESIZED zero-capture bucket
            // (build_phase_buckets_with_stimulus) has sample_count 0 but
            // fold_monitor_into_bucket fills its imbalance / dsq / fallback
            // / stall from in-window monitor samples — render them for
            // parity with the legacy Timeline::build path (which a
            // zero-capture-with-monitor run took before the synthesize
            // seam flipped it onto from_phase_buckets).
            let has_monitor_metrics = m.avg_imbalance.is_some()
                || m.max_imbalance.is_some()
                || m.avg_dsq_depth.is_some()
                || m.max_dsq_depth > 0
                || m.fallback_rate.is_some()
                || m.keep_last_rate.is_some()
                || m.stall_count > 0;
            if m.sample_count > 0 || has_monitor_metrics {
                out.push_str(&format!(
                    "  imbalance: avg={} max={} | dsq: avg={} max={}",
                    m.avg_imbalance
                        .map_or_else(|| "n/a".to_string(), |v| format!("{v:.1}")),
                    m.max_imbalance
                        .map_or_else(|| "n/a".to_string(), |v| format!("{v:.1}")),
                    m.avg_dsq_depth
                        .map_or_else(|| "n/a".to_string(), |v| format!("{v:.0}")),
                    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 {
                    // A synthesized (sample_count==0) step's rate is
                    // stimulus-derived; label it consistently with the
                    // no-monitor-metrics branch below.
                    let suffix = if m.sample_count == 0 {
                        " (stimulus-derived)"
                    } else {
                        ""
                    };
                    out.push_str(&format!(" | throughput: {ir:.0} iter/s{suffix}"));
                }
                out.push('\n');
                if m.stall_count > 0 {
                    out.push_str(&format!("  stalls: {}\n", m.stall_count));
                }
            } else if let Some(ir) = m.iteration_rate {
                // Synthesized zero-capture step (the
                // build_phase_buckets_with_stimulus seam): no periodic
                // captures landed, but the stimulus StepStart/StepEnd
                // deltas still yield a throughput. Surface it so a short
                // interior step's recovered rate is visible in the
                // rendered timeline, not only via the structured
                // phase_metric/step_metric API.
                out.push_str(&format!(
                    "  [no samples] | throughput: {ir:.0} iter/s (stimulus-derived)\n"
                ));
            } else {
                out.push_str("  [no samples]\n");
            }

            format_phase_cgroups(out, &phase.per_cgroup);

            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,
                ));
            }
        }
    }

    /// Build a [`Timeline`] from pre-bucketed
    /// [`crate::assert::PhaseBucket`]s emitted by the metric pipeline.
    /// Preferred over [`Self::build`] when the caller already has
    /// `PhaseBucket`s in hand — avoids re-deriving phase boundaries
    /// from stimulus events + monitor samples by walking the buckets
    /// directly.
    ///
    /// One [`Phase`] is emitted per bucket, in `step_index` order.
    /// `PhaseMetrics` fields are populated from the bucket's
    /// `metrics` map via a name-keyed mapping:
    ///
    /// | PhaseBucket metric key  | PhaseMetrics field      |
    /// |-------------------------|-------------------------|
    /// | `max_imbalance_ratio`   | `max_imbalance`         |
    /// | `avg_imbalance_ratio`   | `avg_imbalance`         |
    /// | `max_dsq_depth`         | `max_dsq_depth`         |
    /// | `avg_dsq_depth`         | `avg_dsq_depth`         |
    /// | `stuck_count`           | `stall_count`           |
    /// | `total_fallback`        | `fallback_rate` (rate)  |
    /// | `total_keep_last`       | `keep_last_rate` (rate) |
    /// | `iteration_rate`        | `iteration_rate`        |
    ///
    /// Rate fields (`fallback_rate`, `keep_last_rate`) are computed
    /// by dividing the bucket's reduced counter delta by the
    /// bucket's window duration in seconds
    /// (`(end_ms - start_ms) / 1000.0`). When the window has zero
    /// duration (degenerate bucket) the rate stays `None`.
    ///
    /// Every PhaseMetrics field has a PhaseBucket source — but
    /// `iteration_rate` only when build_phase_buckets_with_stimulus
    /// (not the plain build_phase_buckets) produced the bucket.
    /// `iteration_rate` requires stimulus events that the per-test
    /// scenario produces; the plain bucket-builder used by some
    /// tests doesn't have access to them. Defaults to `None` when
    /// PhaseBucket.metrics has no `iteration_rate` key.
    ///
    /// `changes` (boundary degradation detection) IS computed
    /// here by diffing adjacent `PhaseMetrics` fields — same
    /// detection logic [`Self::build`] uses, applied after the
    /// per-bucket conversion. avg_imbalance + avg_dsq_depth are
    /// supplied by PhaseBucket so the detection runs on the same
    /// fields as the legacy path.
    pub fn from_phase_buckets(
        phase_buckets: &[crate::assert::PhaseBucket],
        stimulus_events: &[StimulusEvent],
        _ctx: &TimelineContext,
    ) -> Self {
        let mut sorted: Vec<&crate::assert::PhaseBucket> = phase_buckets.iter().collect();
        sorted.sort_by_key(|b| b.step_index);
        // Sort stimulus events by elapsed_ms so correlation finds
        // the closest event for each bucket window deterministically.
        // The terminal scenario-end event is excluded: it carries no
        // step ops/detail to render and its elapsed_ms lands past
        // every bucket window, so it would never correlate — filtering
        // it keeps the correlation set to real step starts only.
        // Per-step StepEnd events are likewise excluded so each bucket's
        // rendered op/detail label correlates to the step's defining
        // StepStart, not its end-of-hold marker (the bucket's iteration_rate
        // is already the step-local value computed upstream).
        let mut sorted_events: Vec<&StimulusEvent> = stimulus_events
            .iter()
            .filter(|e| !e.is_terminal && !e.is_step_end)
            .collect();
        sorted_events.sort_by_key(|e| e.elapsed_ms);
        let mut phases: Vec<Phase> = sorted
            .into_iter()
            .map(|b| phase_from_bucket(b, &sorted_events))
            .collect();
        // Boundary-change detection — shares [`detect_boundary_changes`]
        // with [`Self::build`]. Walks each adjacent (prev, curr) pair and
        // records significant deltas on the LATER phase's `changes` vec so
        // the operator sees "what changed when entering this phase".
        // Throughput is compared even when a side is a SYNTHESIZED
        // zero-capture bucket (build_phase_buckets_with_stimulus): its
        // `iteration_rate` is stimulus-derived and real, so a throughput
        // collapse entering or leaving it must surface. The monitor-derived
        // metrics (imbalance / dsq depth / fallback / keep_last) stay gated
        // inside the helper on both phases having real samples, so a
        // partial-metric phase never paints a phantom non-throughput change.
        // An orphan phase (a not-measured (0,0) window with all-None
        // PhaseMetrics) sits between its neighbors here, so detect_boundary_changes
        // compares each real neighbor against the orphan's None metrics — which
        // the helper gates away (both sides must be Some) — rather than across
        // the unmeasured window. INTENTIONAL: there is no data for the orphan's
        // step, so flagging a phase-k-1 -> phase-k+1 change as if they were
        // adjacent would assert a transition over an unmeasured interval. This is
        // render-only (Phase.changes has no verdict/sidecar/A-B consumer).
        for i in 1..phases.len() {
            let changes = detect_boundary_changes(&phases[i - 1].metrics, &phases[i].metrics);
            phases[i].changes = changes;
        }
        Self { phases }
    }

    /// 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
    }
}

// ---------------------------------------------------------------------------
// PhaseBucket → Phase conversion
// ---------------------------------------------------------------------------

/// Build a [`Phase`] from a [`crate::assert::PhaseBucket`]. The phase
/// index is the bucket's `step_index` (BASELINE = 0, scenario Step k =
/// k + 1), NOT the enumerate position in the vec — so `format_phases`
/// keys its BASELINE-vs-Step label on the true phase identity, and a run
/// whose first bucket is a Step (no BASELINE bucket, e.g. under
/// `--cell-parent-cgroup` where BASELINE captured nothing) renders that
/// Step correctly rather than mislabeling it as BASELINE. The metric map
/// is projected onto the named `PhaseMetrics` fields per the table in
/// [`Timeline::from_phase_buckets`]. BASELINE (`step_index` 0) emits
/// `stimulus = None`; later phases synthesize a [`StimulusEvent`] whose
/// label / op_kind come from the bucket label so the failure-message
/// renderer prints a recognizable phase header.
fn phase_from_bucket(b: &crate::assert::PhaseBucket, sorted_events: &[&StimulusEvent]) -> Phase {
    let duration_s = if b.end_ms > b.start_ms {
        (b.end_ms - b.start_ms) as f64 / 1000.0
    } else {
        0.0
    };
    // Rate computation: counter-delta / duration_s. duration_s == 0
    // disables the rate (None) — degenerate buckets shouldn't
    // produce spurious infinities.
    let rate = |key: &str| -> Option<f64> {
        if duration_s <= 0.0 {
            return None;
        }
        b.metrics.get(key).map(|v| v / duration_s)
    };
    let metrics = PhaseMetrics {
        sample_count: b.sample_count,
        avg_imbalance: b.metrics.get("avg_imbalance_ratio").copied(),
        max_imbalance: b.metrics.get("max_imbalance_ratio").copied(),
        avg_dsq_depth: b.metrics.get("avg_dsq_depth").copied(),
        max_dsq_depth: b
            .metrics
            .get("max_dsq_depth")
            .map(|v| v.round() as u32)
            .unwrap_or(0),
        stall_count: b
            .metrics
            .get("stuck_count")
            .map(|v| v.round() as usize)
            .unwrap_or(0),
        fallback_rate: rate("total_fallback"),
        keep_last_rate: rate("total_keep_last"),
        // iteration_rate is a derived Rate: derive_rate_metrics already
        // placed the Σiterations/Σseconds quotient into the bucket map, so
        // read it verbatim — do NOT divide by duration (unlike
        // fallback_rate / keep_last_rate above, which divide their Counter
        // by the window).
        iteration_rate: b.metrics.get("iteration_rate").copied(),
    };
    let stimulus = if b.step_index == 0 {
        None
    } else {
        // Correlate with the closest StimulusEvent whose
        // elapsed_ms falls in [start_ms, end_ms]. Carrying the
        // real event preserves op_kind + detail in the failure-
        // message timeline render — `phase_from_bucket`'s prior
        // synthesis of a placeholder StimulusEvent with op_kind
        // = None / detail = None produced "Step[N]: ?" headers
        // that lost the operator-facing per-phase context the
        // legacy Timeline::build path carried.
        let correlated = sorted_events.iter().find(|e| {
            if b.start_ms == b.end_ms {
                e.elapsed_ms == b.start_ms
            } else {
                e.elapsed_ms >= b.start_ms && e.elapsed_ms < b.end_ms
            }
        });
        match correlated {
            Some(ev) => Some((*ev).clone()),
            None => Some(StimulusEvent {
                elapsed_ms: b.start_ms,
                label: b.label.clone(),
                op_kind: None,
                detail: None,
                total_iterations: None,
                // Synthetic placeholder for a bucket with no
                // correlated stimulus event; no authoritative step
                // ordinal to carry.
                step_index: None,
                is_terminal: false,
                is_step_end: false,
            }),
        }
    };
    // Orphan signature: fold_guest_per_cgroup_into_host_buckets normalizes a
    // guest carrier with no paired host bucket to a (0,0) window carrying ONLY
    // per_cgroup (empty metrics). A captured bucket has metrics, so the
    // (0,0)+empty-metrics+non-empty-per_cgroup shape is the orphan arm's on
    // every NON-zero-duration window. The one other producer is a ZERO-duration
    // step at scenario start (StepStart[k] == StepEnd[k] == 0 -> a synthesized
    // host (0,0) window with empty metrics, since duration 0 yields no rate and
    // no in-window monitor sample, then MATCHED with a same-step guest carrier),
    // but that collision is HARMLESS: a zero-duration step has no window to
    // measure, so "window not measured" reads the same as the "0ms" it would
    // otherwise show. Display-only, with no verdict/sidecar consumer,
    // so the marker needs no serialized PhaseBucket flag.
    let not_measured_window =
        b.start_ms == 0 && b.end_ms == 0 && b.metrics.is_empty() && !b.per_cgroup.is_empty();
    Phase {
        index: b.step_index as usize,
        start_ms: b.start_ms,
        end_ms: b.end_ms,
        stimulus,
        metrics,
        changes: Vec::new(),
        per_cgroup: b.per_cgroup.clone(),
        not_measured_window,
    }
}

/// Cap on per-cgroup lines rendered per phase, bounding the failure-message
/// size for a many-cgroup scenario (the sched_log render caps similarly).
/// Truncation is by BTreeMap NAME order (deterministic, no ranking math); a
/// "+J more" note records the drop so the cap never reads as "all cgroups".
const MAX_RENDERED_CGROUPS: usize = 16;

/// Render the per-cgroup sub-block for one phase: one line per cgroup in
/// BTreeMap (name) order, reduced at display time via the
/// [`crate::assert::PhaseCgroupStats`] summaries. Empty `per_cgroup` (the
/// monitor-only path, or a phase that carried no per-cgroup components)
/// renders nothing. The None-vs-Some(0.0)
/// discipline carries through: an absent off-CPU reduction renders `n/a` (NOT
/// `0.0%`), and an empty wake / run-delay pool OMITS that segment rather than
/// painting a misleading `0µs`.
fn format_phase_cgroups(
    out: &mut String,
    per_cgroup: &std::collections::BTreeMap<String, crate::assert::PhaseCgroupStats>,
) {
    if per_cgroup.is_empty() {
        return;
    }
    out.push_str("  per-cgroup:\n");
    for (name, pcg) in per_cgroup.iter().take(MAX_RENDERED_CGROUPS) {
        out.push_str(&format!("    {name}: "));
        // A stripped carrier had its raw sample vectors dropped to fit the bulk
        // frame, so off-cpu / wake / run-delay summaries are all absent — but
        // that is NOT "not measured". Surface the size-limit drop explicitly so
        // the operator does not read it as a quiet cgroup.
        if pcg.stripped {
            out.push_str("samples stripped (size limit)");
        } else {
            match pcg.off_cpu_summary() {
                Some((avg, min, max, spread)) => out.push_str(&format!(
                    "off-cpu avg={avg:.1}% min={min:.1}% max={max:.1}% spread={spread:.1}%"
                )),
                None => out.push_str("off-cpu n/a"),
            }
            if let Some((p99, median)) = pcg.wake_summary() {
                out.push_str(&format!(
                    " | wake p99={p99:.0}\u{00b5}s median={median:.0}\u{00b5}s"
                ));
            }
            if let Some((mean, worst)) = pcg.run_delay_summary() {
                out.push_str(&format!(
                    " | run-delay mean={mean:.0}\u{00b5}s worst={worst:.0}\u{00b5}s"
                ));
            }
        }
        out.push_str(&format!(
            " | iters={} migrations={}",
            pcg.total_iterations, pcg.total_migrations
        ));
        // Gap is a Peak with no Option: 0 means "no notable gap", so omit it
        // rather than print a noisy gap=0ms on every quiet cgroup.
        if pcg.max_gap_ms > 0 {
            out.push_str(&format!(
                " | gap={}ms@cpu{}",
                pcg.max_gap_ms, pcg.max_gap_cpu
            ));
        }
        out.push('\n');
    }
    let total = per_cgroup.len();
    if total > MAX_RENDERED_CGROUPS {
        let dropped = total - MAX_RENDERED_CGROUPS;
        let noun = if dropped == 1 { "cgroup" } else { "cgroups" };
        out.push_str(&format!("    (+{dropped} more {noun})\n"));
    }
}

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

/// Reduce a phase's monitor samples to [`PhaseMetrics`].
///
/// `preemption_threshold_ns` is the vCPU-preemption exemption window for
/// stall detection: a non-advancing `rq_clock` on a CPU whose
/// `vcpu_cpu_time_ns` advanced by less than this is a host-preemption
/// artifact, not a scheduler stall, and is exempt. Pass `0` to derive it
/// from the guest kernel's `CONFIG_HZ` via
/// `crate::monitor::vcpu_preemption_threshold_ns` — the same resolution
/// [`MonitorSummary::from_samples_with_threshold`](crate::monitor::MonitorSummary::from_samples_with_threshold)
/// applies, so the per-phase `stall_count` applies the SAME per-(CPU,
/// window) `is_cpu_stuck` predicate as the run-level
/// `MonitorSummary::stuck_count` (run-level `>=` Σ per-phase: it also
/// windows across phase boundaries).
pub(crate) fn compute_metrics(
    samples: &[&MonitorSample],
    preemption_threshold_ns: u64,
) -> 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.
    // Route through `is_cpu_stuck` (the shared predicate the run-level
    // `MonitorSummary` path also uses) so the per-phase stall count and the
    // run-level stuck count apply the identical NOHZ-idle and
    // vCPU-preemption exemptions — the per-phase count uses the SAME
    // predicate as the run-level one (run-level `>=` Σ per-phase: it also
    // counts the boundary-straddling window pair and out-of-phase samples).
    let threshold = if preemption_threshold_ns > 0 {
        preemption_threshold_ns
    } else {
        crate::monitor::vcpu_preemption_threshold_ns(None)
    };
    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 {
            if crate::monitor::reader::is_cpu_stuck(&prev.cpus[cpu], &curr.cpus[cpu], threshold) {
                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 valid_count = valid.len();
    let n = valid_count as f64;
    // None when no valid samples — avoids a 0.0/0.0 NaN and keeps "no
    // data" distinct from a real zero (the detector skips None sides).
    PhaseMetrics {
        sample_count: valid_count,
        avg_imbalance: (valid_count > 0).then(|| total_imbalance / n),
        max_imbalance: (valid_count > 0).then_some(max_imbalance),
        avg_dsq_depth: (valid_count > 0).then(|| 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};

    /// `StimulusEvent::phase()` maps the 1-indexed wire `step_index` onto the
    /// canonical [`crate::assert::Phase`] (StepStart and StepEnd of step `k`
    /// both -> `Phase::step(k)`); the scenario-end terminal seeds no phase.
    #[test]
    fn stimulus_event_phase_maps_wire_step_index_to_phase() {
        use crate::assert::Phase;
        // Wire step_index 1 (Step 0) -> Phase::step(0); 2 (Step 1) -> step(1).
        assert_eq!(
            StimulusEvent::from_wire(&wire_event(0, 1, 0)).phase(),
            Some(Phase::step(0)),
        );
        assert_eq!(
            StimulusEvent::from_wire(&wire_event(100, 2, 50)).phase(),
            Some(Phase::step(1)),
        );
        // StepEnd carries the same step_index -> same Phase as its StepStart.
        assert_eq!(
            StimulusEvent::from_step_end(&wire_event(200, 2, 90)).phase(),
            Some(Phase::step(1)),
        );
        // The terminal boundary is not a step -> no Phase.
        assert_eq!(StimulusEvent::terminal(300, 100).phase(), None);
    }

    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,
            step_index: None,
            is_terminal: false,
            is_step_end: false,
        }
    }

    #[test]
    fn empty_inputs_empty_timeline() {
        let t = Timeline::build(&[], &[], 0);
        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, 0);
        assert!(t.phases.is_empty());
    }

    #[test]
    fn no_monitor_empty_timeline() {
        let events = vec![stimulus(0, "ScenarioStart")];
        let t = Timeline::build(&events, &[], 0);
        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, 0);
        assert_eq!(t.phases.len(), 1);
        // Both samples — including the one AT last_monitor_ms (700) —
        // must fall inside the single phase's [start, last_monitor_ms+1)
        // window. A > 0 check passes even if the last-sample-inclusion
        // off-by-one (end = last_monitor_ms+1) regressed to +0, dropping
        // the 700 sample. Pin the exact count.
        assert_eq!(t.phases[0].metrics.sample_count, 2);
    }

    #[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, 0);
        assert_eq!(t.phases.len(), 2);
        // Pin WHERE the boundary fell, not just non-emptiness: 60 samples
        // at i*100 (i in 5..65 → 500..6400); the >500 warmup drops the
        // 500 sample (i=5), leaving 59. The StepStart[0]@3000 boundary
        // (offset-adjusted) splits them 30/29. A > 0 check passes even if
        // the offset/boundary math shifted the split point while leaving
        // samples on both sides.
        assert_eq!(t.phases[0].metrics.sample_count, 30);
        assert_eq!(t.phases[1].metrics.sample_count, 29);
        assert_eq!(
            t.phases[0].metrics.sample_count + t.phases[1].metrics.sample_count,
            59,
            "59 = 60 samples minus the 500ms sample dropped by the >500 warmup",
        );
    }

    #[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, 0);
        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, 0);
        let formatted = t.format_with_context(&TimelineContext::default());
        assert!(formatted.contains("BASELINE"));
        assert!(formatted.contains("Phase 1"));
        assert!(formatted.contains("imbalance"));
    }

    /// A synthesized zero-capture step (sample_count==0) still renders its
    /// stimulus-derived throughput in the formatted timeline, not only
    /// "[no samples]". Pins the synthesized-step visibility in format_phases. The
    /// BASELINE bucket holds step_index 0 (its phase index, the settle
    /// render) so the synthesized step lands at a Phase index that takes
    /// the metric path.
    #[test]
    fn format_renders_synthesized_step_throughput() {
        let buckets = vec![
            crate::assert::PhaseBucket {
                per_cgroup: Default::default(),
                step_index: 0,
                label: "BASELINE".to_string(),
                start_ms: 0,
                end_ms: 1000,
                sample_count: 2,
                metrics: std::collections::BTreeMap::new(),
            },
            crate::assert::PhaseBucket {
                per_cgroup: Default::default(),
                step_index: 1,
                label: "Step[0]".to_string(),
                start_ms: 1000,
                end_ms: 2000,
                sample_count: 0, // synthesized zero-capture step
                metrics: std::collections::BTreeMap::from([("iteration_rate".to_string(), 1500.0)]),
            },
        ];
        let t = Timeline::from_phase_buckets(&buckets, &[], &TimelineContext::default());
        let formatted = t.format_with_context(&TimelineContext::default());
        assert!(
            formatted.contains("throughput: 1500 iter/s (stimulus-derived)"),
            "a synthesized step must render its stimulus-derived throughput, \
             not only '[no samples]'; got:\n{formatted}",
        );
    }

    // -- orphan not-measured marker + per-cgroup sub-block render --

    /// An orphan bucket — the unique (0,0)-window + empty-metrics +
    /// non-empty-per_cgroup shape from the fold's orphan arm — renders "window
    /// not measured", NOT a misleading "0ms".
    #[test]
    fn format_renders_orphan_window_as_not_measured() {
        let mut per_cgroup = std::collections::BTreeMap::new();
        per_cgroup.insert(
            "cg".to_string(),
            crate::assert::PhaseCgroupStats {
                off_cpu_pcts: vec![80.0],
                total_iterations: 900_000,
                ..Default::default()
            },
        );
        let buckets = vec![crate::assert::PhaseBucket {
            per_cgroup,
            step_index: 1,
            label: "Step[0]".to_string(),
            start_ms: 0,
            end_ms: 0,
            sample_count: 0,
            metrics: std::collections::BTreeMap::new(),
        }];
        let t = Timeline::from_phase_buckets(&buckets, &[], &TimelineContext::default());
        let formatted = t.format_with_context(&TimelineContext::default());
        assert!(
            formatted.contains("window not measured"),
            "orphan window must render not-measured; got:\n{formatted}",
        );
        assert!(
            !formatted.contains("(0ms,"),
            "orphan must NOT render as 0ms; got:\n{formatted}",
        );
        // The whole point of routing orphan carriers through the render (vs the
        // pre-fold path that dropped them) is that their per-cgroup telemetry —
        // an orphan's ONLY payload — SURFACES alongside the not-measured marker.
        assert!(
            formatted.contains("per-cgroup:"),
            "orphan's per-cgroup sub-block must render; got:\n{formatted}",
        );
        assert!(
            formatted.contains("cg: off-cpu avg=80.0%"),
            "orphan carrier's off-cpu reduction must render; got:\n{formatted}",
        );
        assert!(
            formatted.contains("iters=900000"),
            "orphan carrier's counters must render; got:\n{formatted}",
        );
        // The cg carrier has off-cpu but NO wake/run-delay pools — pin the
        // per-line omit when off-cpu is the sole reduction present (the most
        // likely real orphan-carrier shape).
        let cg_line = formatted
            .lines()
            .find(|l| l.contains("cg: off-cpu"))
            .expect("cg line");
        assert!(!cg_line.contains("wake p99="), "got:\n{formatted}");
        assert!(!cg_line.contains("run-delay mean="), "got:\n{formatted}");
    }

    /// Orphan-adjacency suppression: an orphan phase (all-None
    /// PhaseMetrics) BETWEEN two real phases makes detect_boundary_changes
    /// compare each real neighbor against the orphan's gated-away None metrics,
    /// NOT across the unmeasured window — so a step1->step3 throughput collapse
    /// is NOT flagged on step3 (no data for the intervening orphan step).
    /// INTENTIONAL + render-only; pins the documented from_phase_buckets behavior.
    #[test]
    fn format_orphan_phase_suppresses_cross_orphan_change_detection() {
        let real = |step: u16, rate: f64| crate::assert::PhaseBucket {
            per_cgroup: Default::default(),
            step_index: step,
            label: format!("Step[{}]", step - 1),
            start_ms: step as u64 * 1000,
            end_ms: step as u64 * 1000 + 500,
            sample_count: 5,
            metrics: std::collections::BTreeMap::from([("iteration_rate".to_string(), rate)]),
        };
        let mut orphan_pc = std::collections::BTreeMap::new();
        orphan_pc.insert(
            "cg".to_string(),
            crate::assert::PhaseCgroupStats {
                off_cpu_pcts: vec![50.0],
                ..Default::default()
            },
        );
        let orphan = crate::assert::PhaseBucket {
            per_cgroup: orphan_pc,
            step_index: 2,
            label: "Step[1]".to_string(),
            start_ms: 0,
            end_ms: 0,
            sample_count: 0,
            metrics: std::collections::BTreeMap::new(),
        };
        // CONTROL: step1 (rate 10000) adjacent to step3 (rate 1000) — a 90%
        // collapse (> the 30% rel threshold) IS flagged as a change on step3.
        let ctrl = Timeline::from_phase_buckets(
            &[real(1, 10000.0), real(3, 1000.0)],
            &[],
            &TimelineContext::default(),
        );
        let ctrl_step3 = ctrl.phases.iter().find(|p| p.index == 3).expect("step3");
        assert!(
            !ctrl_step3.changes.is_empty(),
            "control: adjacent step1->step3 throughput collapse IS flagged; got: {:?}",
            ctrl_step3.changes,
        );
        // With the orphan between, step3 is compared against the orphan's None
        // metrics -> the change is suppressed (no data for the gap).
        let t = Timeline::from_phase_buckets(
            &[real(1, 10000.0), orphan, real(3, 1000.0)],
            &[],
            &TimelineContext::default(),
        );
        let step3 = t.phases.iter().find(|p| p.index == 3).expect("step3");
        assert!(
            step3.changes.is_empty(),
            "orphan between suppresses the cross-orphan change; got: {:?}",
            step3.changes,
        );
        assert!(
            t.format_with_context(&TimelineContext::default())
                .contains("window not measured"),
            "the orphan still renders not-measured",
        );
    }

    /// The orphan signature is GUARDED on empty metrics: a (0,0)-window bucket
    /// that carries metrics is a captured bucket (or a measured zero-duration
    /// step), NOT an orphan — it renders 0ms, never "window not measured".
    #[test]
    fn format_does_not_mark_not_measured_when_metrics_present() {
        let mut per_cgroup = std::collections::BTreeMap::new();
        per_cgroup.insert(
            "cg".to_string(),
            crate::assert::PhaseCgroupStats {
                off_cpu_pcts: vec![50.0],
                ..Default::default()
            },
        );
        let buckets = vec![crate::assert::PhaseBucket {
            per_cgroup,
            step_index: 1,
            label: "Step[0]".to_string(),
            start_ms: 0,
            end_ms: 0,
            sample_count: 1,
            metrics: std::collections::BTreeMap::from([("avg_imbalance_ratio".to_string(), 1.0)]),
        }];
        let t = Timeline::from_phase_buckets(&buckets, &[], &TimelineContext::default());
        let formatted = t.format_with_context(&TimelineContext::default());
        assert!(
            !formatted.contains("window not measured"),
            "a (0,0) window WITH metrics is captured, not an orphan; got:\n{formatted}",
        );
        assert!(
            formatted.contains("(0ms,"),
            "a measured zero-duration window renders 0ms; got:\n{formatted}",
        );
    }

    /// The per-cgroup sub-block renders one line per cgroup (name order), with
    /// the None-vs-Some discipline: a not-measured off-CPU reduction renders
    /// `n/a` (not `0.0%`), and an empty wake/run-delay pool omits that segment.
    #[test]
    fn format_renders_per_cgroup_subblock_none_aware() {
        let mut per_cgroup = std::collections::BTreeMap::new();
        per_cgroup.insert(
            "cg_a".to_string(),
            crate::assert::PhaseCgroupStats {
                off_cpu_pcts: vec![80.0, 84.0],
                wake_latencies_ns: vec![100_000, 120_000],
                run_delays_ns: vec![45_000],
                total_iterations: 900_000,
                total_migrations: 12,
                max_gap_ms: 8,
                max_gap_cpu: 3,
                ..Default::default()
            },
        );
        per_cgroup.insert(
            "cg_b".to_string(),
            crate::assert::PhaseCgroupStats {
                off_cpu_pcts: vec![],
                total_iterations: 80_000,
                ..Default::default()
            },
        );
        let buckets = vec![crate::assert::PhaseBucket {
            per_cgroup,
            step_index: 1,
            label: "Step[0]".to_string(),
            start_ms: 1000,
            end_ms: 6000,
            sample_count: 10,
            metrics: std::collections::BTreeMap::from([("avg_imbalance_ratio".to_string(), 1.5)]),
        }];
        let t = Timeline::from_phase_buckets(&buckets, &[], &TimelineContext::default());
        let formatted = t.format_with_context(&TimelineContext::default());
        assert!(
            formatted.contains("per-cgroup:"),
            "sub-block header; got:\n{formatted}"
        );
        assert!(
            formatted.contains("cg_a: off-cpu avg=82.0% min=80.0% max=84.0% spread=4.0%"),
            "got:\n{formatted}",
        );
        assert!(
            formatted.contains("wake p99="),
            "wake segment present; got:\n{formatted}"
        );
        assert!(
            formatted.contains("run-delay mean="),
            "run-delay segment present; got:\n{formatted}",
        );
        assert!(
            formatted.contains("iters=900000 migrations=12"),
            "counters present; got:\n{formatted}",
        );
        assert!(
            formatted.contains("cg_b: off-cpu n/a"),
            "not-measured off-cpu is n/a, NOT 0.0%; got:\n{formatted}",
        );
        // cg_a has a coupled gap (8ms @ cpu 3) -> rendered with its cpu.
        assert!(
            formatted.contains("gap=8ms@cpu3"),
            "coupled gap renders ms@cpu; got:\n{formatted}",
        );
        // cg_b has no wake/run-delay pools AND max_gap_ms==0 -> those segments
        // omitted on its line (gap is omitted at 0, not printed as gap=0ms).
        let cg_b_line = formatted
            .lines()
            .find(|l| l.contains("cg_b:"))
            .expect("cg_b line");
        assert!(!cg_b_line.contains("wake p99="), "got:\n{formatted}");
        assert!(
            !cg_b_line.contains("gap="),
            "gap omitted at 0; got:\n{formatted}"
        );
        // BTreeMap name order: cg_a before cg_b.
        assert!(
            formatted.find("cg_a:").unwrap() < formatted.find("cg_b:").unwrap(),
            "cgroups render in name order; got:\n{formatted}",
        );
    }

    /// A stripped carrier (raw sample vectors dropped to fit the size-limited
    /// bulk frame) renders "samples stripped (size limit)" — distinct from a
    /// not-measured carrier's "off-cpu n/a" — so an operator does not read a
    /// size-limit drop as a quiet cgroup. The surviving counters still render.
    #[test]
    fn format_renders_stripped_carrier_distinctly() {
        let mut per_cgroup = std::collections::BTreeMap::new();
        per_cgroup.insert(
            "cg_stripped".to_string(),
            crate::assert::PhaseCgroupStats {
                stripped: true,
                total_iterations: 500_000,
                total_migrations: 9,
                ..Default::default()
            },
        );
        per_cgroup.insert(
            "cg_quiet".to_string(),
            // Genuinely measured nothing (NOT stripped).
            crate::assert::PhaseCgroupStats {
                total_iterations: 1_000,
                ..Default::default()
            },
        );
        let buckets = vec![crate::assert::PhaseBucket {
            per_cgroup,
            step_index: 1,
            label: "Step[0]".to_string(),
            start_ms: 1000,
            end_ms: 6000,
            sample_count: 10,
            metrics: std::collections::BTreeMap::from([("avg_imbalance_ratio".to_string(), 1.5)]),
        }];
        let t = Timeline::from_phase_buckets(&buckets, &[], &TimelineContext::default());
        let formatted = t.format_with_context(&TimelineContext::default());
        assert!(
            formatted.contains("cg_stripped: samples stripped (size limit)"),
            "stripped carrier shows the size-limit marker; got:\n{formatted}",
        );
        assert!(
            formatted.contains("iters=500000 migrations=9"),
            "stripped carrier still renders its surviving counters; got:\n{formatted}",
        );
        assert!(
            formatted.contains("cg_quiet: off-cpu n/a"),
            "a not-stripped, not-measured carrier stays n/a; got:\n{formatted}",
        );
        assert!(
            !formatted.contains("cg_quiet: samples stripped"),
            "a not-stripped carrier must NOT show the stripped marker; got:\n{formatted}",
        );
    }

    /// Measured-zero off-CPU renders `0.0%` (a real zero), distinct from the
    /// `n/a` not-measured state — the kind-specific None-vs-Some(0.0) boundary.
    #[test]
    fn format_renders_measured_zero_off_cpu_as_zero_not_na() {
        let mut per_cgroup = std::collections::BTreeMap::new();
        per_cgroup.insert(
            "cg".to_string(),
            crate::assert::PhaseCgroupStats {
                off_cpu_pcts: vec![0.0, 0.0],
                ..Default::default()
            },
        );
        let buckets = vec![crate::assert::PhaseBucket {
            per_cgroup,
            step_index: 1,
            label: "Step[0]".to_string(),
            start_ms: 1000,
            end_ms: 2000,
            sample_count: 1,
            metrics: std::collections::BTreeMap::from([("avg_imbalance_ratio".to_string(), 1.0)]),
        }];
        let t = Timeline::from_phase_buckets(&buckets, &[], &TimelineContext::default());
        let formatted = t.format_with_context(&TimelineContext::default());
        assert!(
            formatted.contains("off-cpu avg=0.0%"),
            "measured zero off-cpu is 0.0%, not n/a; got:\n{formatted}",
        );
        assert!(!formatted.contains("off-cpu n/a"), "got:\n{formatted}");
    }

    /// Symmetric with the off-cpu measured-zero test for wake + run-delay: a
    /// NON-empty pool of zeros is a measured zero (Some) and renders `0µs`, NOT
    /// omitted (which only an EMPTY pool -> None does). Guards against a refactor
    /// that special-cased a zero reduction to None (collapsing measured-zero
    /// into not-measured), silently omitting a real zero-latency reading.
    #[test]
    fn format_renders_measured_zero_wake_and_run_delay_not_omitted() {
        let mut per_cgroup = std::collections::BTreeMap::new();
        per_cgroup.insert(
            "cg".to_string(),
            crate::assert::PhaseCgroupStats {
                off_cpu_pcts: vec![5.0],
                wake_latencies_ns: vec![0],
                run_delays_ns: vec![0],
                ..Default::default()
            },
        );
        let buckets = vec![crate::assert::PhaseBucket {
            per_cgroup,
            step_index: 1,
            label: "Step[0]".to_string(),
            start_ms: 1000,
            end_ms: 2000,
            sample_count: 1,
            metrics: std::collections::BTreeMap::from([("avg_imbalance_ratio".to_string(), 1.0)]),
        }];
        let t = Timeline::from_phase_buckets(&buckets, &[], &TimelineContext::default());
        let formatted = t.format_with_context(&TimelineContext::default());
        assert!(
            formatted.contains("wake p99=0\u{00b5}s"),
            "measured-zero wake renders 0µs, not omitted; got:\n{formatted}",
        );
        assert!(
            formatted.contains("run-delay mean=0\u{00b5}s"),
            "measured-zero run-delay renders 0µs, not omitted; got:\n{formatted}",
        );
    }

    /// The per-cgroup sub-block caps at MAX_RENDERED_CGROUPS (16) by name
    /// order, with a "+J more" note — bounding failure-message size.
    #[test]
    fn format_caps_per_cgroup_subblock() {
        let mut per_cgroup = std::collections::BTreeMap::new();
        for i in 0..20 {
            per_cgroup.insert(
                format!("cg{i:02}"),
                crate::assert::PhaseCgroupStats {
                    off_cpu_pcts: vec![10.0],
                    ..Default::default()
                },
            );
        }
        let buckets = vec![crate::assert::PhaseBucket {
            per_cgroup,
            step_index: 1,
            label: "Step[0]".to_string(),
            start_ms: 1000,
            end_ms: 2000,
            sample_count: 1,
            metrics: std::collections::BTreeMap::from([("avg_imbalance_ratio".to_string(), 1.0)]),
        }];
        let t = Timeline::from_phase_buckets(&buckets, &[], &TimelineContext::default());
        let formatted = t.format_with_context(&TimelineContext::default());
        assert!(
            formatted.contains("(+4 more cgroups)"),
            "20 cgroups capped at 16 -> +4 more; got:\n{formatted}",
        );
        assert!(
            formatted.contains("cg15:"),
            "first 16 rendered; got:\n{formatted}"
        );
        assert!(
            !formatted.contains("cg16:"),
            "cg16 is beyond the cap; got:\n{formatted}",
        );
    }

    /// No per-cgroup sub-block when the phase carries no carriers (the
    /// monitor-only path, or a phase with empty per_cgroup).
    #[test]
    fn format_omits_per_cgroup_subblock_when_empty() {
        let buckets = vec![crate::assert::PhaseBucket {
            per_cgroup: Default::default(),
            step_index: 1,
            label: "Step[0]".to_string(),
            start_ms: 1000,
            end_ms: 2000,
            sample_count: 1,
            metrics: std::collections::BTreeMap::from([("avg_imbalance_ratio".to_string(), 1.0)]),
        }];
        let t = Timeline::from_phase_buckets(&buckets, &[], &TimelineContext::default());
        let formatted = t.format_with_context(&TimelineContext::default());
        assert!(
            !formatted.contains("per-cgroup:"),
            "no sub-block when carriers absent; got:\n{formatted}",
        );
    }

    /// Cap off-by-one boundary: EXACTLY 16 cgroups render all 16 with NO
    /// "more cgroups" note (total > MAX is false); 17 render 16 + "(+1 more
    /// cgroups)". Pins the `>` vs `>=` / `take(N)` edge against a silent drop
    /// (one cgroup gone, no note) or a "(+0 more)" lie.
    #[test]
    fn format_per_cgroup_cap_boundary_16_and_17() {
        let mk = |n: usize| {
            let mut per_cgroup = std::collections::BTreeMap::new();
            for i in 0..n {
                per_cgroup.insert(
                    format!("cg{i:02}"),
                    crate::assert::PhaseCgroupStats {
                        off_cpu_pcts: vec![10.0],
                        ..Default::default()
                    },
                );
            }
            let buckets = vec![crate::assert::PhaseBucket {
                per_cgroup,
                step_index: 1,
                label: "Step[0]".to_string(),
                start_ms: 1000,
                end_ms: 2000,
                sample_count: 1,
                metrics: std::collections::BTreeMap::from([(
                    "avg_imbalance_ratio".to_string(),
                    1.0,
                )]),
            }];
            Timeline::from_phase_buckets(&buckets, &[], &TimelineContext::default())
                .format_with_context(&TimelineContext::default())
        };
        let at16 = mk(16);
        assert!(at16.contains("cg15:"), "16th cgroup renders; got:\n{at16}");
        assert!(
            !at16.contains("more cgroups"),
            "exactly 16 has NO truncation note; got:\n{at16}",
        );
        let at17 = mk(17);
        assert!(at17.contains("cg15:"), "got:\n{at17}");
        assert!(
            !at17.contains("cg16:"),
            "17th is past the cap; got:\n{at17}"
        );
        assert!(
            at17.contains("(+1 more cgroup)") && !at17.contains("(+1 more cgroups)"),
            "17 cgroups -> exactly +1 more (singular 'cgroup'); got:\n{at17}",
        );
    }

    /// A synthesized (sample_count==0) bucket carrying monitor-derived
    /// metrics renders the imbalance/dsq block, not just "[no samples]" —
    /// the render-layer half of the monitor-parity handling. format_phases
    /// gates that block on `has_monitor_metrics`, not `sample_count > 0`.
    #[test]
    fn format_renders_synthesized_step_monitor_metrics() {
        let buckets = vec![
            crate::assert::PhaseBucket {
                per_cgroup: Default::default(),
                step_index: 0,
                label: "BASELINE".to_string(),
                start_ms: 0,
                end_ms: 1000,
                sample_count: 2,
                metrics: std::collections::BTreeMap::new(),
            },
            crate::assert::PhaseBucket {
                per_cgroup: Default::default(),
                step_index: 1,
                label: "Step[0]".to_string(),
                start_ms: 1000,
                end_ms: 2000,
                sample_count: 0, // synthesized zero-capture step
                metrics: std::collections::BTreeMap::from([
                    ("avg_imbalance_ratio".to_string(), 2.0),
                    ("max_imbalance_ratio".to_string(), 3.0),
                    ("avg_dsq_depth".to_string(), 5.0),
                    ("max_dsq_depth".to_string(), 7.0),
                ]),
            },
        ];
        let t = Timeline::from_phase_buckets(&buckets, &[], &TimelineContext::default());
        let formatted = t.format_with_context(&TimelineContext::default());
        assert!(
            formatted.contains("imbalance: avg=2.0 max=3.0"),
            "synthesized bucket's folded imbalance must render, not \
             '[no samples]'; got:\n{formatted}",
        );
        assert!(
            formatted.contains("dsq: avg=5 max=7"),
            "synthesized bucket's folded dsq must render; got:\n{formatted}",
        );
    }

    /// A from_phase_buckets render whose first bucket is a Step (no
    /// BASELINE bucket — e.g. under --cell-parent-cgroup where BASELINE
    /// captured nothing) must NOT mislabel that Step as "BASELINE".
    /// phase.index is the bucket's step_index, so format_phases' index==0
    /// BASELINE check fires only for a real BASELINE bucket.
    #[test]
    fn format_no_baseline_bucket_does_not_mislabel_first_step() {
        let buckets = vec![
            crate::assert::PhaseBucket {
                per_cgroup: Default::default(),
                step_index: 1, // scenario Step 0; NO BASELINE bucket present
                label: "Step[0]".to_string(),
                start_ms: 1000,
                end_ms: 2000,
                sample_count: 3,
                metrics: std::collections::BTreeMap::from([(
                    "avg_imbalance_ratio".to_string(),
                    1.0,
                )]),
            },
            crate::assert::PhaseBucket {
                per_cgroup: Default::default(),
                step_index: 2,
                label: "Step[1]".to_string(),
                start_ms: 2000,
                end_ms: 3000,
                sample_count: 3,
                metrics: std::collections::BTreeMap::from([(
                    "avg_imbalance_ratio".to_string(),
                    1.0,
                )]),
            },
        ];
        let t = Timeline::from_phase_buckets(&buckets, &[], &TimelineContext::default());
        let formatted = t.format_with_context(&TimelineContext::default());
        assert!(
            !formatted.contains("BASELINE"),
            "no BASELINE bucket -> no BASELINE label; the first Step must not \
             be mislabeled as BASELINE; got:\n{formatted}",
        );
        assert!(
            formatted.contains("Phase 1"),
            "the first Step renders as 'Phase 1' (its step_index), not \
             'Phase 0'/BASELINE; got:\n{formatted}",
        );
    }

    /// Sparse / non-contiguous step_index renders the TRUE step number:
    /// BASELINE(0) + a Step at step_index 3 renders "Phase 3", not the
    /// enumerate-position "Phase 1". Pins index == step_index for a
    /// non-zero, non-contiguous label — the case that distinguishes
    /// step_index from the old enumerate index (a revert to enumerate
    /// would pass the contiguous tests but fail this one).
    #[test]
    fn format_sparse_step_index_renders_true_step_number() {
        let buckets = vec![
            crate::assert::PhaseBucket {
                per_cgroup: Default::default(),
                step_index: 0,
                label: "BASELINE".to_string(),
                start_ms: 0,
                end_ms: 100,
                sample_count: 2,
                metrics: std::collections::BTreeMap::new(),
            },
            crate::assert::PhaseBucket {
                per_cgroup: Default::default(),
                step_index: 3, // sparse: Steps 0/1 absent, only step_index 3
                label: "Step[2]".to_string(),
                start_ms: 200,
                end_ms: 300,
                sample_count: 2,
                metrics: std::collections::BTreeMap::from([(
                    "avg_imbalance_ratio".to_string(),
                    1.0,
                )]),
            },
        ];
        let t = Timeline::from_phase_buckets(&buckets, &[], &TimelineContext::default());
        let formatted = t.format_with_context(&TimelineContext::default());
        assert!(
            formatted.contains("Phase 3"),
            "a sparse Step at step_index 3 must render 'Phase 3' (its \
             step_index), not the enumerate-position 'Phase 1'; got:\n{formatted}",
        );
        assert!(
            !formatted.contains("Phase 1"),
            "the enumerate-position 'Phase 1' must NOT appear for a \
             step_index-3 bucket; got:\n{formatted}",
        );
    }

    #[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, 0);
        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, 0);
        assert_eq!(t.phases[0].metrics.stall_count, 1);
    }

    #[test]
    fn compute_metrics_stall_count_accumulates_across_windows() {
        // cpu0 frozen across TWO consecutive windows -> stall_count == 2.
        // Pins that the per-phase path counts every stuck (CPU, window),
        // mirroring the run-level accumulation (both breaks removed).
        let s1 = sample(100, vec![(1, 0, 5000), (1, 0, 6000)]);
        let s2 = sample(200, vec![(1, 0, 5000), (1, 0, 6100)]);
        let s3 = sample(300, vec![(1, 0, 5000), (1, 0, 6200)]);
        let refs: Vec<&MonitorSample> = vec![&s1, &s2, &s3];
        let m = compute_metrics(&refs, 0);
        assert_eq!(m.stall_count, 2);
    }

    #[test]
    fn run_level_stuck_count_ge_sum_of_per_phase() {
        // Same is_cpu_stuck predicate, different windowing domains: the
        // run-level path windows(2) over the FULL stream and counts the
        // pair straddling a phase split; partitioning the stream into
        // per-phase subsets drops that boundary pair. So run-level >= Σ
        // per-phase (strict here) — pins the documented inequality.
        let s1 = sample(100, vec![(1, 0, 5000), (1, 0, 6000)]);
        let s2 = sample(200, vec![(1, 0, 5000), (1, 0, 6100)]);
        let s3 = sample(300, vec![(1, 0, 5000), (1, 0, 6200)]);
        let samples = vec![s1, s2, s3];
        // Run-level: windows(2) over all 3 -> cpu0 frozen in both windows.
        let run_level = crate::monitor::MonitorSummary::from_samples(&samples).stuck_count;
        assert_eq!(run_level, 2);
        // Partition after s2 (phase A = [s1,s2], phase B = [s3]): the
        // (s2,s3) boundary pair is in neither phase's windows(2).
        let phase_a: Vec<&MonitorSample> = vec![&samples[0], &samples[1]];
        let phase_b: Vec<&MonitorSample> = vec![&samples[2]];
        let sum_per_phase =
            compute_metrics(&phase_a, 0).stall_count + compute_metrics(&phase_b, 0).stall_count;
        assert_eq!(
            sum_per_phase, 1,
            "the (s2,s3) boundary pair is in neither phase"
        );
        assert!(
            run_level > sum_per_phase,
            "run-level counts the boundary-straddling pair the per-phase partition drops"
        );
    }

    #[test]
    fn compute_metrics_empty() {
        let m = compute_metrics(&[], 0);
        assert_eq!(m.sample_count, 0);
        // No samples -> no measurement, not a false 0.0 (the sentinel fix).
        assert_eq!(m.avg_imbalance, None);
        assert_eq!(m.max_imbalance, None);
        assert_eq!(m.avg_dsq_depth, None);
        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,
            step_index: None,
            is_terminal: false,
            is_step_end: false,
        };
        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, 0);
        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, 0);
        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, 0);
        assert_eq!(m.sample_count, 2);
        assert!((m.avg_imbalance.unwrap() - 2.5).abs() < 0.01); // (4+1)/2
        assert!((m.max_imbalance.unwrap() - 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, 0);
        // 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, 0);
        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, 0);
        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, 0);
        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, 0);
        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, 0);
        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, 0);
        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, 0);
        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, 0);
        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, 0);
        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, 0);
        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, 0);
        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.unwrap() - 1.0).abs() < 0.5,
            "baseline imbalance should be ~1.0, got {:?}",
            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, 0);
        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, 0);
        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());
    }

    // -- detect_boundary_changes: throughput across synthesized steps --

    /// Throughput is flagged across a SYNTHESIZED zero-capture phase
    /// (sample_count 0) — its iteration_rate is stimulus-derived and
    /// real — while monitor-derived metrics stay gated on samples, so
    /// the synthesized side never paints a phantom imbalance change.
    /// This is the collapse-suppression invariant: a throughput collapse entering or
    /// leaving a capture-free step must not be silently dropped.
    #[test]
    fn detect_boundary_changes_flags_throughput_across_synthesized_phase() {
        let before = PhaseMetrics {
            sample_count: 30,
            iteration_rate: Some(1000.0),
            avg_imbalance: Some(1.0),
            ..Default::default()
        };
        let after = PhaseMetrics {
            sample_count: 0,             // synthesized zero-capture step
            iteration_rate: Some(300.0), // 70% collapse
            avg_imbalance: Some(99.0),   // partial/default — must be ignored
            ..Default::default()
        };
        let changes = detect_boundary_changes(&before, &after);
        let throughput: Vec<_> = changes
            .iter()
            .filter(|c| c.metric == "throughput")
            .collect();
        assert_eq!(
            throughput.len(),
            1,
            "throughput collapse across a synthesized step must be flagged: {changes:?}",
        );
        assert_eq!(throughput[0].direction, ChangeDirection::Degraded);
        assert!(
            !changes.iter().any(|c| c.metric == "imbalance"),
            "monitor metrics must stay gated when a side has 0 samples: {changes:?}",
        );
    }

    /// The monitor-metric gate only suppresses a ZERO-sample side: when
    /// both phases captured samples, monitor-derived changes still
    /// surface. Guards the collapse suppression from over-suppressing the normal
    /// captured-to-captured boundary.
    #[test]
    fn detect_boundary_changes_reports_monitor_metrics_when_both_sampled() {
        let before = PhaseMetrics {
            sample_count: 30,
            iteration_rate: Some(1000.0),
            avg_imbalance: Some(1.0),
            ..Default::default()
        };
        let after = PhaseMetrics {
            sample_count: 30,
            iteration_rate: Some(1000.0), // unchanged throughput
            avg_imbalance: Some(5.0),     // imbalance jump > IMBALANCE_THRESHOLD
            ..Default::default()
        };
        let changes = detect_boundary_changes(&before, &after);
        assert!(
            changes.iter().any(|c| c.metric == "imbalance"),
            "imbalance change must surface when both sides sampled: {changes:?}",
        );
        assert!(
            !changes.iter().any(|c| c.metric == "throughput"),
            "unchanged throughput must not be flagged: {changes:?}",
        );
    }

    /// Throughput uses a STRICT relative threshold: a change of exactly
    /// ITERATION_RATE_REL_THRESHOLD (30%) is NOT flagged. Pins the `>`
    /// boundary so a future `>` -> `>=` flip is caught (the absolute
    /// detect_change gate is a different code path, tested separately).
    #[test]
    fn detect_boundary_changes_throughput_exactly_at_threshold_not_flagged() {
        let before = PhaseMetrics {
            sample_count: 30,
            iteration_rate: Some(1000.0),
            ..Default::default()
        };
        let after = PhaseMetrics {
            sample_count: 30,
            iteration_rate: Some(700.0), // rel = -0.3 exactly
            ..Default::default()
        };
        assert_eq!((700.0 - 1000.0) / 1000.0, -ITERATION_RATE_REL_THRESHOLD);
        assert!(
            !detect_boundary_changes(&before, &after)
                .iter()
                .any(|c| c.metric == "throughput"),
            "an exactly-30% relative change must not flag (strict >)",
        );
    }

    /// synthesized -> synthesized boundary (BOTH sides sample_count 0,
    /// both carrying a real stimulus-derived rate): throughput is still
    /// compared (the gate is symmetric and sample_count-independent for
    /// throughput) and monitor metrics stay suppressed on both sides.
    /// Pins the zero->zero cell of the transition matrix against a future
    /// edit that re-couples throughput to a per-side sample_count check.
    #[test]
    fn detect_boundary_changes_synthesized_to_synthesized_flags_throughput() {
        let before = PhaseMetrics {
            sample_count: 0,
            iteration_rate: Some(1000.0),
            avg_imbalance: Some(1.0),
            ..Default::default()
        };
        let after = PhaseMetrics {
            sample_count: 0,
            iteration_rate: Some(300.0), // 70% collapse
            avg_imbalance: Some(99.0),   // wild — must stay gated
            ..Default::default()
        };
        let changes = detect_boundary_changes(&before, &after);
        let throughput: Vec<_> = changes
            .iter()
            .filter(|c| c.metric == "throughput")
            .collect();
        assert_eq!(
            throughput.len(),
            1,
            "zero->zero throughput collapse must flag: {changes:?}"
        );
        assert_eq!(throughput[0].direction, ChangeDirection::Degraded);
        assert!(
            !changes.iter().any(|c| c.metric == "imbalance"),
            "monitor metrics gated on both zero-sample sides: {changes:?}",
        );
    }

    // -- 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),
            step_index: None,
            is_terminal: false,
            is_step_end: false,
        }
    }

    #[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, 0);
        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, 0);
        assert!(t.phases[0].metrics.iteration_rate.is_none());
        assert!(t.phases[1].metrics.iteration_rate.is_none());
    }

    /// Build a wire `StimulusEvent` so tests can drive the FULL
    /// `from_wire` path (the production conversion) rather than
    /// constructing the timeline event directly — the latter bypassed
    /// the `total_iterations == 0` sentinel.
    fn wire_event(
        elapsed_ms: u32,
        step_index: u16,
        total_iterations: u64,
    ) -> crate::vmm::wire::StimulusEvent {
        crate::vmm::wire::StimulusEvent {
            elapsed_ms,
            step_index,
            op_count: 0,
            op_kinds: 0,
            cgroup_count: 0,
            worker_count: 1,
            total_iterations,
        }
    }

    #[test]
    fn from_wire_zero_iterations_is_some_baseline() {
        // total_iterations is a cumulative counter, so a
        // start-of-window 0 is a legitimate baseline, NOT a missing
        // sample. from_wire must carry Some(0), never collapse it to
        // None.
        let te = StimulusEvent::from_wire(&wire_event(0, 1, 0));
        assert_eq!(te.total_iterations, Some(0));
        assert_eq!(te.step_index, Some(1));
        assert!(!te.is_terminal);
        assert!(
            !te.is_step_end,
            "a StepStart-derived event is not a StepEnd"
        );
    }

    #[test]
    fn from_step_end_carries_step_index_and_marks_step_end() {
        // A StepEnd frame reuses the StimulusEvent wire body.
        // from_step_end must carry the same 1-indexed step_index and the
        // step's end-of-hold total_iterations, flag is_step_end, and leave
        // is_terminal off (it is a real per-step boundary, not the
        // scenario terminal).
        let te = StimulusEvent::from_step_end(&wire_event(1_900, 1, 9_000));
        assert_eq!(te.step_index, Some(1));
        assert_eq!(te.total_iterations, Some(9_000));
        assert!(te.is_step_end, "StepEnd-derived event must set is_step_end");
        assert!(
            !te.is_terminal,
            "StepEnd is a per-step boundary, not the scenario terminal",
        );
    }

    #[test]
    fn from_wire_first_step_zero_baseline_yields_rate() {
        // First-step zero-baseline regression, driven through the FULL from_wire path
        // (unit tests previously injected Some(0) directly, masking the
        // wire 0->None collapse). First step frame reads 0 cumulative
        // iterations, the second reads 3000; the first phase must get a
        // rate rather than a silent None.
        let events: Vec<StimulusEvent> = [wire_event(0, 1, 0), wire_event(3000, 2, 3000)]
            .iter()
            .map(StimulusEvent::from_wire)
            .collect();
        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, 0);
        assert!(
            t.phases[0].metrics.iteration_rate.is_some(),
            "first phase must get a rate from the 0 baseline",
        );
    }

    #[test]
    fn terminal_event_gives_last_step_rate_without_phantom_phase() {
        // The last step has no successor step event, so its
        // iteration_rate needs the terminal scenario-end boundary. The
        // terminal must supply that boundary WITHOUT adding a phantom
        // trailing phase.
        let mut events: Vec<StimulusEvent> = [wire_event(0, 1, 0), wire_event(2000, 2, 4000)]
            .iter()
            .map(StimulusEvent::from_wire)
            .collect();
        events.push(StimulusEvent::terminal(4000, 10000));
        let samples: Vec<MonitorSample> = (5..45)
            .map(|i| sample(i * 100, vec![(2, 1, i * 1000)]))
            .collect();
        let t = Timeline::build(&events, &samples, 0);
        assert_eq!(
            t.phases.len(),
            2,
            "two step events -> two phases; terminal seeds none",
        );
        assert!(
            t.phases[1].metrics.iteration_rate.is_some(),
            "last step must get a rate from the terminal boundary",
        );
    }

    #[test]
    fn build_filters_step_end_events_no_phantom_phase() {
        // A StepEnd must be filtered from the PHASE-LAYOUT set
        // (it is an end-of-hold marker, not a step boundary) so it neither
        // adds a phantom phase nor misaligns the dense phase index. Two
        // StepStart events with an interleaved StepEnd still yield exactly
        // two phases. (StepEnd is still consumed for the step-local RATE —
        // see build_pairs_step_local_when_step_end_events_present.)
        let events: Vec<StimulusEvent> = vec![
            StimulusEvent::from_wire(&wire_event(0, 1, 0)),
            StimulusEvent::from_step_end(&wire_event(1_900, 1, 9_000)),
            StimulusEvent::from_wire(&wire_event(2_000, 2, 9_000)),
        ];
        let samples: Vec<MonitorSample> = (5..35)
            .map(|i| sample(i * 100, vec![(2, 1, i * 1000)]))
            .collect();
        let t = Timeline::build(&events, &samples, 0);
        assert_eq!(
            t.phases.len(),
            2,
            "two StepStart events -> two phases; the interleaved StepEnd seeds none",
        );
    }

    #[test]
    fn build_pairs_step_local_when_step_end_events_present() {
        // The monitor-only Timeline::build fallback must ALSO
        // use step-local StepStart[k] -> StepEnd[k] pairing when StepEnd
        // events are present (they are emitted independent of snapshot
        // captures), NOT the cross-step StepStart[k] -> StepStart[k+1]
        // pairing that reads 0 -> 0 for respawned-per-step workers. Two
        // fresh-per-step steps (each StepStart reads ~0); without
        // step-local pairing phase 0 would be None (0 -> 0 cross-step).
        // With it, both phases get a positive rate.
        let events: Vec<StimulusEvent> = vec![
            StimulusEvent::from_wire(&wire_event(0, 1, 0)), // StepStart[0], iters 0
            StimulusEvent::from_step_end(&wire_event(1_000, 1, 5_000)), // StepEnd[0], iters 5000
            StimulusEvent::from_wire(&wire_event(1_100, 2, 0)), // StepStart[1] respawned, iters 0
            StimulusEvent::from_step_end(&wire_event(2_100, 2, 3_000)), // StepEnd[1], iters 3000
        ];
        let samples: Vec<MonitorSample> = (1..30)
            .map(|i| sample(i * 100, vec![(2, 1, i * 1000)]))
            .collect();
        let t = Timeline::build(&events, &samples, 0);
        assert_eq!(
            t.phases.len(),
            2,
            "two StepStart events -> two phases (each StepEnd seeds none)",
        );
        assert!(
            t.phases[0].metrics.iteration_rate.is_some(),
            "phase 0 must get a step-local rate from StepStart[0] -> StepEnd[0], \
             not the cross-step 0 -> 0 None (the old cross-step fallback bug)",
        );
        assert!(
            t.phases[1].metrics.iteration_rate.is_some(),
            "phase 1 (respawned workers) must get its own step-local rate",
        );
    }

    #[test]
    fn build_stalled_step_with_step_end_reports_measured_zero_not_cross_step() {
        // Monitor-only path: a step that HAS a StepEnd but
        // stalled (StepEnd[k] == StepStart[k]) reports its MEASURED-ZERO
        // step-local rate (Some(0.0)) — its StepEnd lookup hits, so the
        // cross-step fallback must NOT run. Mirrors the snapshot path's
        // build_phase_buckets_with_stimulus_stalled_step_reports_measured_zero.
        // Step 0 stalls (0 -> 0); a persistent population reads 500 at
        // StepStart[1], so a cross-step StepStart[0] -> StepStart[1] leak
        // would be ~454/s. Step 1 advances 500 -> 5500 (5000/s).
        let events: Vec<StimulusEvent> = vec![
            StimulusEvent::from_wire(&wire_event(0, 1, 0)), // StepStart[0], iters 0
            StimulusEvent::from_step_end(&wire_event(1_000, 1, 0)), // StepEnd[0], STALLED 0
            StimulusEvent::from_wire(&wire_event(1_100, 2, 500)), // StepStart[1], persistent 500
            StimulusEvent::from_step_end(&wire_event(2_100, 2, 5_500)), // StepEnd[1], iters 5500
        ];
        let samples: Vec<MonitorSample> = (1..30)
            .map(|i| sample(i * 100, vec![(2, 1, i * 1000)]))
            .collect();
        let t = Timeline::build(&events, &samples, 0);
        assert_eq!(t.phases.len(), 2);
        assert_eq!(
            t.phases[0].metrics.iteration_rate,
            Some(0.0),
            "a stalled step reports measured-zero throughput, not the \
             cross-step StepStart[0] -> StepStart[1] persistent-leak rate",
        );
        assert!(
            t.phases[1].metrics.iteration_rate.is_some(),
            "step 1 still reports its own step-local rate",
        );
    }

    #[test]
    fn terminal_event_single_step_rate() {
        // Boundary case: a one-step scenario (first == last). With the
        // 0 baseline and the terminal boundary        // the single step still gets a rate, and the terminal adds no
        // phase.
        let mut events: Vec<StimulusEvent> = [wire_event(0, 1, 0)]
            .iter()
            .map(StimulusEvent::from_wire)
            .collect();
        events.push(StimulusEvent::terminal(3000, 9000));
        let samples: Vec<MonitorSample> = (5..35)
            .map(|i| sample(i * 100, vec![(2, 1, i * 1000)]))
            .collect();
        let t = Timeline::build(&events, &samples, 0);
        assert_eq!(
            t.phases.len(),
            1,
            "single step -> one phase; terminal adds none"
        );
        assert!(
            t.phases[0].metrics.iteration_rate.is_some(),
            "single step gets a rate (first == last)",
        );
    }

    #[test]
    fn terminal_event_stalled_last_step_reports_measured_zero() {
        // Boundary case: the last step's counter did not advance
        // (terminal count == last step-start count): e == s. That is
        // MEASURED ZERO throughput — a real value (the strongest
        // degradation signal), not "unmeasured" — so rate_to returns
        // Some(0.0), and the zero surfaces to the degradation detector.
        // Only a counter DECREASE (e < s) is unmeasurable -> None.
        let mut events: Vec<StimulusEvent> = [wire_event(0, 1, 0), wire_event(2000, 2, 4000)]
            .iter()
            .map(StimulusEvent::from_wire)
            .collect();
        events.push(StimulusEvent::terminal(4000, 4000)); // no advance
        let samples: Vec<MonitorSample> = (5..45)
            .map(|i| sample(i * 100, vec![(2, 1, i * 1000)]))
            .collect();
        let t = Timeline::build(&events, &samples, 0);
        assert_eq!(t.phases.len(), 2);
        assert_eq!(
            t.phases[1].metrics.iteration_rate,
            Some(0.0),
            "stalled last step (e == s) reports measured-zero, not None",
        );
    }

    #[test]
    fn iteration_rate_counter_decrease_yields_no_rate() {
        // A counter DECREASE between consecutive step frames (e.g. a
        // step-local worker population reset) is unmeasurable and must NOT
        // produce a negative or conflated rate — the `e < s` guard drops
        // the pair, returning None (distinct from `e == s`, which is a
        // measured-zero Some(0.0)). Pin it so a future change that loosens
        // the guard to allow a negative delta fails here.
        let events: Vec<StimulusEvent> = [
            wire_event(0, 1, 0),
            wire_event(2000, 2, 5000),
            wire_event(3000, 3, 1000), // counter dropped 5000 -> 1000
        ]
        .iter()
        .map(StimulusEvent::from_wire)
        .collect();
        let samples: Vec<MonitorSample> = (5..35)
            .map(|i| sample(i * 100, vec![(2, 1, i * 1000)]))
            .collect();
        let t = Timeline::build(&events, &samples, 0);
        // phase 1 is step 2 (frame iters 5000 -> next 1000): decrease.
        assert!(
            t.phases[1].metrics.iteration_rate.is_none(),
            "a counter decrease must not manufacture a (negative) rate",
        );
    }

    #[test]
    fn iteration_rate_zero_duration_yields_no_rate() {
        // Two consecutive frames with identical elapsed_ms -> the rate
        // denominator is 0; the duration==0 guard must drop the pair
        // rather than divide and produce inf/NaN.
        let events: Vec<StimulusEvent> = [wire_event(1000, 1, 0), wire_event(1000, 2, 2000)]
            .iter()
            .map(StimulusEvent::from_wire)
            .collect();
        let samples: Vec<MonitorSample> = (5..35)
            .map(|i| sample(i * 100, vec![(2, 1, i * 1000)]))
            .collect();
        let t = Timeline::build(&events, &samples, 0);
        assert!(
            t.phases[0].metrics.iteration_rate.is_none(),
            "zero-duration pair must not divide; rate stays None",
        );
    }

    #[test]
    fn terminal_not_last_does_not_misalign_or_misattribute() {
        // Robustness: even if a corrupt/out-of-order elapsed_ms made the
        // terminal sort BEFORE a real step, the explicit is_terminal
        // extraction (not positional) must keep the step phases aligned
        // and attribute the early step's rate correctly. A corrupt
        // terminal contributes no spurious rate (its position can't
        // shift the dense phase index).
        let mut events: Vec<StimulusEvent> = [wire_event(0, 1, 0), wire_event(2000, 2, 4000)]
            .iter()
            .map(StimulusEvent::from_wire)
            .collect();
        // Terminal with elapsed_ms BEFORE step 2 (simulated corruption).
        events.push(StimulusEvent::terminal(500, 9000));
        let samples: Vec<MonitorSample> = (5..45)
            .map(|i| sample(i * 100, vec![(2, 1, i * 1000)]))
            .collect();
        let t = Timeline::build(&events, &samples, 0);
        // Two step events -> two phases regardless of terminal position.
        assert_eq!(
            t.phases.len(),
            2,
            "terminal position must not change phase count"
        );
        // Phase 0 (step 1) still gets its correct rate (0 -> 4000 over
        // 2s = 2000/s): the misordered terminal did not misalign it.
        assert_eq!(
            t.phases[0].metrics.iteration_rate,
            Some(2000.0),
            "early step rate must be correct despite a misordered terminal",
        );
    }

    #[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, 0);
        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_collapse_to_zero_is_flagged() {
        // A phase that collapses to ZERO throughput (e == s, measured
        // zero) must be flagged as a degradation — it is the strongest
        // degradation signal. Previously the zero phase's rate_to returned
        // None, so the detector's Some/Some gate dropped it and the worst
        // degradation went silently unreported.
        let events = vec![
            stimulus_with_iters(0, "ScenarioStart", 0),
            stimulus_with_iters(2000, "StepStart[0]", 10000), // phase 0: ~5000/s
            stimulus_with_iters(4000, "StepStart[1]", 10000), // phase 1: 0/s (stalled)
        ];
        let samples: Vec<MonitorSample> = (5..45)
            .map(|i| sample(i * 100, vec![(2, 1, i * 1000)]))
            .collect();
        let t = Timeline::build(&events, &samples, 0);
        assert_eq!(
            t.phases[1].metrics.iteration_rate,
            Some(0.0),
            "the collapsed phase must report measured-zero throughput",
        );
        let degs: Vec<_> = t
            .degradations()
            .into_iter()
            .filter(|(p, c)| p.index == 1 && c.metric == "throughput")
            .collect();
        assert!(
            !degs.is_empty(),
            "a collapse to zero throughput must be flagged as a degradation",
        );
        assert_eq!(degs[0].1.direction, ChangeDirection::Degraded);
        assert_eq!(degs[0].1.after, 0.0);
    }

    #[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, 0);
        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, 0);
        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 from_phase_buckets_maps_known_metrics_and_renders_phase_block() {
        use crate::assert::PhaseBucket;
        use std::collections::BTreeMap;
        let mut s0_metrics = BTreeMap::new();
        s0_metrics.insert("max_dsq_depth".to_string(), 7.0);
        s0_metrics.insert("avg_dsq_depth".to_string(), 2.5);
        s0_metrics.insert("max_imbalance_ratio".to_string(), 3.5);
        s0_metrics.insert("avg_imbalance_ratio".to_string(), 1.8);
        s0_metrics.insert("total_fallback".to_string(), 200.0);
        let buckets = vec![
            PhaseBucket {
                per_cgroup: Default::default(),
                step_index: 0,
                label: "BASELINE".to_string(),
                start_ms: 0,
                end_ms: 1000,
                sample_count: 5,
                metrics: BTreeMap::new(),
            },
            PhaseBucket {
                per_cgroup: Default::default(),
                step_index: 1,
                label: "Step[0]".to_string(),
                start_ms: 1000,
                end_ms: 6000,
                sample_count: 20,
                metrics: s0_metrics,
            },
        ];
        let t = Timeline::from_phase_buckets(&buckets, &[], &TimelineContext::default());
        assert_eq!(t.phases.len(), 2);
        // Phase 0 (BASELINE) — no stimulus, no metrics.
        assert!(t.phases[0].stimulus.is_none());
        assert_eq!(t.phases[0].metrics.sample_count, 5);
        assert_eq!(t.phases[0].metrics.max_dsq_depth, 0);
        // Phase 1 (Step[0]) — stimulus set, metrics projected from
        // the bucket map.
        assert!(t.phases[1].stimulus.is_some());
        assert_eq!(t.phases[1].stimulus.as_ref().unwrap().label, "Step[0]");
        assert_eq!(t.phases[1].metrics.sample_count, 20);
        assert_eq!(t.phases[1].metrics.max_dsq_depth, 7);
        assert!((t.phases[1].metrics.avg_dsq_depth.unwrap() - 2.5).abs() < f64::EPSILON);
        assert!((t.phases[1].metrics.max_imbalance.unwrap() - 3.5).abs() < f64::EPSILON);
        assert!((t.phases[1].metrics.avg_imbalance.unwrap() - 1.8).abs() < f64::EPSILON);
        // fallback_rate = 200 / (5000 / 1000) = 40.0 events/s
        assert_eq!(t.phases[1].metrics.fallback_rate, Some(40.0));
        // keep_last_rate absent → None (no total_keep_last in metrics map)
        assert_eq!(t.phases[1].metrics.keep_last_rate, None);
        // avg_dsq_depth + avg_imbalance are now both wired
        // (per the doc table). iteration_rate is the only field
        // PhaseBucket cannot supply directly (depends on stimulus
        // event totals, not a per-Sample reading).
        assert_eq!(t.phases[1].metrics.iteration_rate, None);
        // Render produces a non-empty timeline block.
        let formatted = t.format_with_context(&TimelineContext::default());
        assert!(formatted.contains("--- timeline ---"));
        assert!(formatted.contains("BASELINE"));
        assert!(formatted.contains("Step[0]"));
    }

    /// Collapse-suppression production path: a throughput collapse INTO a synthesized
    /// zero-capture step surfaces through from_phase_buckets (the path
    /// `evaluate_vm_result` prefers), not only via the helper unit test.
    /// BASELINE captures samples + an iteration_rate; Step[0] is
    /// synthesized (sample_count 0) with a collapsed iteration_rate AND a
    /// divergent imbalance that must stay gated. Re-adding the old
    /// `if sample_count==0 { continue }` gate at this call site makes
    /// phases[1].changes empty, so this test fails — it is the
    /// regression pin for the removed gate. The producer half — that
    /// build_phase_buckets_with_stimulus actually populates a synthesized
    /// bucket's iteration_rate from stimulus deltas — is pinned by
    /// assert::tests_phase_bucket::build_phase_buckets_with_stimulus_synthesizes_zero_capture_step_bucket;
    /// together they pin the full producer->consumer chain.
    #[test]
    fn from_phase_buckets_flags_throughput_into_synthesized_step() {
        use crate::assert::PhaseBucket;
        use std::collections::BTreeMap;
        let mut baseline_metrics = BTreeMap::new();
        baseline_metrics.insert("iteration_rate".to_string(), 1000.0);
        baseline_metrics.insert("avg_imbalance_ratio".to_string(), 1.0);
        let mut step_metrics = BTreeMap::new();
        step_metrics.insert("iteration_rate".to_string(), 300.0); // 70% collapse
        step_metrics.insert("avg_imbalance_ratio".to_string(), 99.0); // must stay gated
        let buckets = vec![
            PhaseBucket {
                per_cgroup: Default::default(),
                step_index: 0,
                label: "BASELINE".to_string(),
                start_ms: 0,
                end_ms: 1000,
                sample_count: 5,
                metrics: baseline_metrics,
            },
            PhaseBucket {
                per_cgroup: Default::default(),
                step_index: 1,
                label: "Step[0]".to_string(),
                start_ms: 1000,
                end_ms: 6000,
                sample_count: 0, // synthesized zero-capture step
                metrics: step_metrics,
            },
        ];
        let t = Timeline::from_phase_buckets(&buckets, &[], &TimelineContext::default());
        assert_eq!(t.phases[1].metrics.sample_count, 0);
        assert_eq!(t.phases[1].metrics.iteration_rate, Some(300.0));
        let throughput: Vec<_> = t.phases[1]
            .changes
            .iter()
            .filter(|c| c.metric == "throughput")
            .collect();
        assert_eq!(
            throughput.len(),
            1,
            "throughput collapse into a synthesized step must surface via \
             from_phase_buckets: {:?}",
            t.phases[1].changes,
        );
        assert_eq!(throughput[0].direction, ChangeDirection::Degraded);
        assert!(
            !t.phases[1].changes.iter().any(|c| c.metric == "imbalance"),
            "monitor metrics must stay gated for a zero-sample side: {:?}",
            t.phases[1].changes,
        );
    }

    /// Boundary change-detection on the from_phase_buckets path — the
    /// PRODUCTION success path (`evaluate_vm_result` prefers
    /// from_phase_buckets over `build`). Two adjacent metric-bearing
    /// buckets whose avg_imbalance / avg_dsq_depth cross the thresholds
    /// in the worsening direction must record Degraded changes on the
    /// ENTERED phase (phases[1]), and the BASELINE phase records none.
    /// Without this, the 821-869 detection loop ships unverified (a
    /// wrong threshold, inverted direction, wrong-phase recording, or
    /// wrong metric field would all slip past the other
    /// from_phase_buckets tests, which never trigger the loop).
    #[test]
    fn from_phase_buckets_detects_boundary_degradation() {
        use crate::assert::PhaseBucket;
        use std::collections::BTreeMap;
        let mut base = BTreeMap::new();
        base.insert("avg_imbalance_ratio".to_string(), 1.0);
        base.insert("avg_dsq_depth".to_string(), 1.0);
        let mut step = BTreeMap::new();
        step.insert("avg_imbalance_ratio".to_string(), 2.0); // +1.0 > 0.5 threshold
        step.insert("avg_dsq_depth".to_string(), 6.0); // +5.0 > 3.0 threshold
        let buckets = vec![
            PhaseBucket {
                per_cgroup: Default::default(),
                step_index: 0,
                label: "BASELINE".to_string(),
                start_ms: 0,
                end_ms: 1000,
                sample_count: 5,
                metrics: base,
            },
            PhaseBucket {
                per_cgroup: Default::default(),
                step_index: 1,
                label: "Step[0]".to_string(),
                start_ms: 1000,
                end_ms: 6000,
                sample_count: 20,
                metrics: step,
            },
        ];
        let t = Timeline::from_phase_buckets(&buckets, &[], &TimelineContext::default());
        // Change recorded on the ENTERED phase, never the prior one.
        assert!(
            t.phases[0].changes.is_empty(),
            "BASELINE has no prior phase to diff; changes belong to the entered phase",
        );
        let changes = &t.phases[1].changes;
        let imb = changes
            .iter()
            .find(|c| c.metric == "imbalance")
            .expect("imbalance change must fire (1.0 -> 2.0 crosses 0.5)");
        assert_eq!(imb.direction, ChangeDirection::Degraded);
        assert!((imb.before - 1.0).abs() < f64::EPSILON);
        assert!((imb.after - 2.0).abs() < f64::EPSILON);
        let dsq = changes
            .iter()
            .find(|c| c.metric == "dsq_depth")
            .expect("dsq_depth change must fire (1.0 -> 6.0 crosses 3.0)");
        assert_eq!(dsq.direction, ChangeDirection::Degraded);
        assert!((dsq.before - 1.0).abs() < f64::EPSILON);
        assert!((dsq.after - 6.0).abs() < f64::EPSILON);
    }

    /// Sub-threshold deltas record NO change — guards a dropped/zeroed
    /// threshold that would fabricate spurious boundary changes.
    #[test]
    fn from_phase_buckets_subthreshold_records_no_change() {
        use crate::assert::PhaseBucket;
        use std::collections::BTreeMap;
        let mut base = BTreeMap::new();
        base.insert("avg_imbalance_ratio".to_string(), 1.0);
        base.insert("avg_dsq_depth".to_string(), 1.0);
        let mut step = BTreeMap::new();
        step.insert("avg_imbalance_ratio".to_string(), 1.2); // +0.2 < 0.5
        step.insert("avg_dsq_depth".to_string(), 2.0); // +1.0 < 3.0
        let buckets = vec![
            PhaseBucket {
                per_cgroup: Default::default(),
                step_index: 0,
                label: "BASELINE".to_string(),
                start_ms: 0,
                end_ms: 1000,
                sample_count: 5,
                metrics: base,
            },
            PhaseBucket {
                per_cgroup: Default::default(),
                step_index: 1,
                label: "Step[0]".to_string(),
                start_ms: 1000,
                end_ms: 6000,
                sample_count: 20,
                metrics: step,
            },
        ];
        let t = Timeline::from_phase_buckets(&buckets, &[], &TimelineContext::default());
        assert!(
            t.phases[1].changes.is_empty(),
            "sub-threshold deltas must not record a boundary change",
        );
    }

    /// Decreasing imbalance across the boundary records an IMPROVEMENT —
    /// locks the higher_is_worse direction so an inverted flag cannot
    /// report a regression as an improvement (or vice versa).
    #[test]
    fn from_phase_buckets_detects_boundary_improvement() {
        use crate::assert::PhaseBucket;
        use std::collections::BTreeMap;
        let mut base = BTreeMap::new();
        base.insert("avg_imbalance_ratio".to_string(), 2.0);
        let mut step = BTreeMap::new();
        step.insert("avg_imbalance_ratio".to_string(), 1.0); // -1.0, |delta|>0.5, after<before
        let buckets = vec![
            PhaseBucket {
                per_cgroup: Default::default(),
                step_index: 0,
                label: "BASELINE".to_string(),
                start_ms: 0,
                end_ms: 1000,
                sample_count: 5,
                metrics: base,
            },
            PhaseBucket {
                per_cgroup: Default::default(),
                step_index: 1,
                label: "Step[0]".to_string(),
                start_ms: 1000,
                end_ms: 6000,
                sample_count: 20,
                metrics: step,
            },
        ];
        let t = Timeline::from_phase_buckets(&buckets, &[], &TimelineContext::default());
        let imb = t.phases[1]
            .changes
            .iter()
            .find(|c| c.metric == "imbalance")
            .expect("imbalance change must fire (2.0 -> 1.0 crosses 0.5)");
        assert_eq!(
            imb.direction,
            ChangeDirection::Improved,
            "a decreasing imbalance is an improvement, not a degradation",
        );
    }

    /// from_phase_buckets must CORRELATE a real stimulus event into the
    /// phase header, carrying its op_kind + detail. Every other
    /// from_phase_buckets test passes `&[]`, so only the synthetic
    /// None-placeholder arm ran and the `Some(ev) => (*ev).clone()`
    /// correlation arm (added to stop headers degrading to "Step[N]: ?")
    /// was untested. A wrong interval bound or cloning the wrong event
    /// would drop the operator-facing op/detail with no failure.
    #[test]
    fn from_phase_buckets_correlates_real_stimulus_op_and_detail() {
        use crate::assert::PhaseBucket;
        use std::collections::BTreeMap;
        let event = StimulusEvent {
            elapsed_ms: 1000,
            label: "Step[0]".to_string(),
            op_kind: Some("SetCpuset".to_string()),
            detail: Some("4 cpus".to_string()),
            total_iterations: None,
            step_index: Some(1),
            is_terminal: false,
            is_step_end: false,
        };
        let buckets = vec![
            PhaseBucket {
                per_cgroup: Default::default(),
                step_index: 0,
                label: "BASELINE".to_string(),
                start_ms: 0,
                end_ms: 1000,
                sample_count: 5,
                metrics: BTreeMap::new(),
            },
            PhaseBucket {
                per_cgroup: Default::default(),
                step_index: 1,
                label: "Step[0]".to_string(),
                start_ms: 1000,
                end_ms: 6000,
                sample_count: 20,
                metrics: BTreeMap::new(),
            },
        ];
        let t = Timeline::from_phase_buckets(&buckets, &[event], &TimelineContext::default());
        let stim = t.phases[1]
            .stimulus
            .as_ref()
            .expect("Step[0] phase carries a stimulus");
        assert_eq!(
            stim.op_kind.as_deref(),
            Some("SetCpuset"),
            "the correlated event's op_kind must be carried, not the None placeholder",
        );
        assert_eq!(stim.detail.as_deref(), Some("4 cpus"));
    }

    #[test]
    fn from_phase_buckets_zero_duration_window_emits_no_rate() {
        use crate::assert::PhaseBucket;
        use std::collections::BTreeMap;
        let mut metrics = BTreeMap::new();
        metrics.insert("total_fallback".to_string(), 100.0);
        let bucket = PhaseBucket {
            per_cgroup: Default::default(),
            step_index: 1,
            label: "Step[0]".to_string(),
            start_ms: 500,
            end_ms: 500,
            sample_count: 1,
            metrics,
        };
        let t = Timeline::from_phase_buckets(&[bucket], &[], &TimelineContext::default());
        // Degenerate window (start == end) yields duration_s == 0,
        // so rate divisions stay None rather than producing
        // spurious infinities.
        assert_eq!(t.phases[0].metrics.fallback_rate, None);
    }

    #[test]
    fn from_phase_buckets_absent_imbalance_metric_is_none_not_zero() {
        // A bucket carrying no avg_imbalance_ratio / avg_dsq_depth
        // metric must yield None (no data), NOT Some(0.0) — so the change
        // detector skips it instead of comparing a false zero-imbalance.
        use crate::assert::PhaseBucket;
        use std::collections::BTreeMap;
        let bucket = PhaseBucket {
            per_cgroup: Default::default(),
            step_index: 1,
            label: "Step[0]".to_string(),
            start_ms: 100,
            end_ms: 600,
            sample_count: 3,
            metrics: BTreeMap::new(),
        };
        let t = Timeline::from_phase_buckets(&[bucket], &[], &TimelineContext::default());
        assert_eq!(t.phases[0].metrics.avg_imbalance, None);
        assert_eq!(t.phases[0].metrics.max_imbalance, None);
        assert_eq!(t.phases[0].metrics.avg_dsq_depth, None);
    }

    #[test]
    fn from_phase_buckets_sorts_by_step_index() {
        use crate::assert::PhaseBucket;
        use std::collections::BTreeMap;
        // Out-of-order input; from_phase_buckets must sort by
        // step_index so the rendered phase block walks BASELINE
        // → Step[0] → Step[1] in time order regardless of
        // how the caller arranged the input vec.
        let buckets = vec![
            PhaseBucket {
                per_cgroup: Default::default(),
                step_index: 2,
                label: "Step[1]".to_string(),
                start_ms: 2000,
                end_ms: 3000,
                sample_count: 5,
                metrics: BTreeMap::new(),
            },
            PhaseBucket {
                per_cgroup: Default::default(),
                step_index: 0,
                label: "BASELINE".to_string(),
                start_ms: 0,
                end_ms: 500,
                sample_count: 2,
                metrics: BTreeMap::new(),
            },
            PhaseBucket {
                per_cgroup: Default::default(),
                step_index: 1,
                label: "Step[0]".to_string(),
                start_ms: 500,
                end_ms: 2000,
                sample_count: 5,
                metrics: BTreeMap::new(),
            },
        ];
        let t = Timeline::from_phase_buckets(&buckets, &[], &TimelineContext::default());
        assert_eq!(t.phases.len(), 3);
        assert_eq!(t.phases[0].start_ms, 0);
        assert_eq!(t.phases[1].start_ms, 500);
        assert_eq!(t.phases[2].start_ms, 2000);
    }

    #[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, 0);
        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"));
    }
}