ktstr 0.23.0

//! Host-side guest memory monitor and BPF map introspection.
//!
//! Reads per-CPU runqueue structures from guest VM memory via BTF-resolved
//! offsets. Observes scheduler state without instrumenting the guest
//! kernel or the scheduler under test.
//!
//! The [`bpf_map`] module provides host-side discovery and read/write
//! access to BPF maps in guest memory. The [`bpf_prog`] module provides
//! host-side enumeration of BPF programs and their verifier/runtime stats.
//! Both locate kernel objects by walking IDR xarrays (shared infrastructure
//! in the [`idr`] module) through page table translation. No guest
//! cooperation is needed.
//!
//! See the [Monitor](https://ktstr.dev/guide/architecture/monitor.html)
//! chapter of the guide.

pub mod arena;
pub mod bpf_map;
pub mod bpf_prog;
pub mod bpf_syscall;
pub mod btf_offsets;
pub mod btf_render;
pub(crate) mod cast_analysis;
pub mod cgroup_walk;
pub mod dmesg_scx;
pub mod dump;
pub mod guest;
pub mod idr;
pub(crate) mod kva_io;
pub mod live_host_kernel;
pub mod perf_counters;
pub mod reader;
pub mod runnable_scan;
pub mod scx_static_alloc;
pub mod scx_walker;
pub mod sdt_alloc;
pub mod symbols;
pub mod task_enrichment;
pub mod timeline;

#[cfg(test)]
pub(crate) mod test_util;

#[cfg(test)]
mod enforce_tests;
#[cfg(test)]
mod evaluate_helpers_tests;
#[cfg(test)]
mod event_rates_tests;
#[cfg(test)]
mod kernel_hz_tests;
#[cfg(test)]
mod schedstat_tests;
#[cfg(test)]
mod stall_detection_tests;
#[cfg(test)]
mod summary_tests;
#[cfg(test)]
mod sustained_tracker_tests;
#[cfg(test)]
mod thresholds_tests;
#[cfg(test)]
mod validity_tests;

/// Guest physical address of the top-level page-table page (CR3 on x86,
/// TTBR1 on aarch64). Newtype around `u64` so address kinds can't
/// accidentally mix — passing a `PageOffset` where a `Cr3Pa` is
/// expected fails to compile instead of silently walking the wrong
/// tree.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub(crate) struct Cr3Pa(pub u64);

/// Kernel direct-map base (x86-64 `PAGE_OFFSET`, aarch64 linear map
/// base). Adding this to a DRAM offset yields a KVA; subtracting it
/// from a KVA yields the DRAM offset that `GuestMem` reads use.
/// Newtype around `u64`; see [`Cr3Pa`] for the rationale.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub(crate) struct PageOffset(pub u64);

/// Kernel virtual address. Newtype around `u64` so a KVA can't be
/// mistaken for a guest DRAM offset (`PA`) or for the base values
/// [`Cr3Pa`]/[`PageOffset`] at any page-walk call site. `Display`
/// renders as `0x<hex>` for tracing output.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
pub(crate) struct Kva(pub u64);

impl std::fmt::Display for Kva {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{:#x}", self.0)
    }
}

/// DSQ depth above this value indicates uninitialized guest memory.
/// Real kernels never queue this many tasks on a single CPU's local DSQ.
pub const DSQ_PLAUSIBILITY_CEILING: u32 = 10_000;

/// Non-negative delta between two samples of a monotonic counter.
///
/// BPF / kernel counters can reset mid-run (scheduler restart, counter
/// re-init) so the raw `last - first` can be negative. Callers want a
/// rate-computation-safe value, so clamp to zero.
///
/// Shared with [`crate::timeline`] so both derivation paths agree on
/// the clamp semantics instead of each reinventing the same closure.
pub(crate) fn counter_delta(last: i64, first: i64) -> i64 {
    (last - first).max(0)
}

/// Number of tick periods a vCPU must have run before rq_clock is expected
/// to have advanced. 10 ticks gives the scheduler tick multiple chances
/// to fire and update rq_clock.
const PREEMPTION_TICK_MULTIPLE: u64 = 10;

/// Default HZ when CONFIG_HZ cannot be determined from the kernel.
/// 250 is the most conservative common value (longest tick period =
/// highest threshold), avoiding false stall detection.
const DEFAULT_HZ: u64 = 250;

/// Compute the vCPU preemption threshold for a given kernel.
///
/// Tries to determine CONFIG_HZ from (in order):
/// 1. Embedded IKCONFIG in the vmlinux (gzip blob after `IKCFG_ST` marker)
/// 2. `.config` next to the kernel image (build directory)
/// 3. Host `/boot/config-$(uname -r)` (covers virtme default: host == guest)
/// 4. Falls back to 250 (40ms threshold)
///
/// Returns the threshold in nanoseconds: `(1e9 / HZ) * 10`.
pub(crate) fn vcpu_preemption_threshold_ns(kernel_path: Option<&std::path::Path>) -> u64 {
    let hz = guest_kernel_hz(kernel_path);
    let tick_ns = 1_000_000_000u64 / hz;
    tick_ns * PREEMPTION_TICK_MULTIPLE
}

/// Determine CONFIG_HZ for the guest kernel.
///
/// The host's `/boot/config-$(uname -r)` is only consulted when the
/// caller passed no explicit `kernel_path` (virtme default: host ==
/// guest). A cached kernel whose IKCONFIG was stripped and whose
/// build `.config` is unavailable must NOT inherit the host's
/// CONFIG_HZ — the cached kernel was built with its own config, and
/// using the host's CONFIG_HZ would silently misscale every
/// tick-dependent threshold. Instead, such kernels fall through to
/// [`DEFAULT_HZ`], which is the conservative default.
pub(crate) fn guest_kernel_hz(kernel_path: Option<&std::path::Path>) -> u64 {
    if let Some(kp) = kernel_path {
        // Try embedded IKCONFIG in the vmlinux.
        if let Some(vmlinux) = find_vmlinux(kp)
            && let Some(hz) = read_hz_from_ikconfig(&vmlinux)
        {
            return hz;
        }

        // Try .config next to the kernel image.
        if let Some(hz) = read_hz_from_kernel_dir(kp) {
            return hz;
        }

        // Explicit kernel_path whose config can't be recovered —
        // don't fall through to the host's /boot/config; host HZ
        // is unrelated to the cached/built guest kernel's HZ.
        tracing::warn!(
            kernel = %kp.display(),
            default_hz = DEFAULT_HZ,
            "guest_kernel_hz: no IKCONFIG or .config alongside \
             kernel; falling back to DEFAULT_HZ rather than host \
             /boot/config (tick-dependent thresholds may be \
             conservative)"
        );
        return DEFAULT_HZ;
    }

    // No kernel path given → virtme-style run where guest kernel ==
    // host kernel. Host boot config is authoritative.
    if let Some(hz) = read_hz_from_boot_config() {
        return hz;
    }

    DEFAULT_HZ
}

use crate::vmm::find_vmlinux;

/// IKCONFIG marker: gzip data starts immediately after this 8-byte sequence.
const IKCONFIG_MAGIC: &[u8] = b"IKCFG_ST";

/// ELF sections [`read_hz_from_ikconfig`] reads.
///
/// The cached-vmlinux strip pipeline
/// ([`crate::cache::strip_vmlinux_debug`]) preserves these bytes
/// verbatim via its keep-list predicate.
pub(crate) const VMLINUX_KEEP_SECTIONS: &[&[u8]] = &[
    b".rodata", // IKCONFIG gzip blob, bracketed by IKCFG_ST / IKCFG_ED markers
];

/// Extract CONFIG_HZ from the embedded IKCONFIG blob in a vmlinux ELF.
///
/// The kernel (when built with CONFIG_IKCONFIG) embeds a gzip-compressed
/// copy of `.config` in `.rodata`, bracketed by `IKCFG_ST` / `IKCFG_ED`
/// markers. This function scans the raw bytes for the marker, decompresses
/// the gzip data, and parses CONFIG_HZ from the result.
fn read_hz_from_ikconfig(vmlinux_path: &std::path::Path) -> Option<u64> {
    let data = std::fs::read(vmlinux_path).ok()?;
    // vmlinux images are tens of MB; the old
    // `windows(8).position(|w| w == IKCFG_ST)` was a naive O(n)
    // byte-wise scan. memchr's two-way matcher uses the available
    // SIMD path (x86_64 AVX2 / aarch64 Neon) and cuts scan time by
    // a constant factor on every host we care about.
    let pos = memchr::memmem::find(&data, IKCONFIG_MAGIC)?;
    let gz_start = pos + IKCONFIG_MAGIC.len();
    if gz_start >= data.len() {
        return None;
    }
    let cursor = std::io::Cursor::new(&data[gz_start..]);
    let mut decoder = flate2::read::GzDecoder::new(cursor);
    let mut config = String::new();
    std::io::Read::read_to_string(&mut decoder, &mut config).ok()?;
    parse_config_hz(&config)
}

/// Look for a `.config` file in the kernel's build directory and parse
/// CONFIG_HZ from it. Walks up from the kernel image path (e.g.
/// `<root>/arch/x86/boot/bzImage`) toward the build root.
fn read_hz_from_kernel_dir(kernel_path: &std::path::Path) -> Option<u64> {
    let mut dir = kernel_path.parent()?;
    // Walk up at most 4 levels (arch/x86/boot/bzImage -> root).
    for _ in 0..4 {
        let config = dir.join(".config");
        if config.exists() {
            let contents = std::fs::read_to_string(&config).ok()?;
            return parse_config_hz(&contents);
        }
        dir = dir.parent()?;
    }
    None
}

/// Parse CONFIG_HZ=N from `/boot/config-$(uname -r)`.
fn read_hz_from_boot_config() -> Option<u64> {
    let uname = rustix::system::uname();
    let release = uname.release().to_str().ok()?;
    let path = format!("/boot/config-{release}");
    let contents = std::fs::read_to_string(path).ok()?;
    parse_config_hz(&contents)
}

/// Extract `CONFIG_HZ=N` from kernel config text.
fn parse_config_hz(config: &str) -> Option<u64> {
    for line in config.lines() {
        let line = line.trim();
        if let Some(val) = line.strip_prefix("CONFIG_HZ=") {
            return val.parse().ok();
        }
    }
    None
}

/// Check whether a single monitor sample contains plausible data.
///
/// Returns false when any CPU's local_dsq_depth exceeds the plausibility
/// ceiling, indicating uninitialized guest memory rather than real
/// scheduler state.
pub fn sample_looks_valid(sample: &MonitorSample) -> bool {
    sample
        .cpus
        .iter()
        .all(|cpu| cpu.local_dsq_depth <= DSQ_PLAUSIBILITY_CEILING)
}

/// Find a vmlinux for tests.
///
/// Routes the `KTSTR_KERNEL` read through [`crate::ktstr_kernel_env`]
/// so the empty/whitespace normalization matches every other reader.
/// When the env value is a [`crate::kernel_path::KernelId::Version`]
/// or [`KernelId::CacheKey`], the cache is consulted via
/// [`crate::cli::resolve_cached_kernel`] and the resolved entry dir is
/// passed to [`crate::kernel_path::resolve_btf`]; when it is a
/// [`KernelId::Path`], the path flows through `resolve_btf` directly
/// as before. Unset env or unresolved Version/CacheKey falls through
/// to `resolve_btf(None)` so the local-tree / sysfs fallbacks still
/// apply. See `resolve_btf` for the full resolution order.
#[cfg(test)]
pub fn find_test_vmlinux() -> Option<std::path::PathBuf> {
    use crate::kernel_path::KernelId;
    let raw = crate::ktstr_kernel_env();
    let resolved_dir: Option<String> = match raw.as_deref().map(KernelId::parse) {
        Some(KernelId::Path(p)) => p.into_os_string().into_string().ok(),
        Some(id @ (KernelId::Version(_) | KernelId::CacheKey(_))) => {
            // Cache lookup. On failure, fall through to `None` so
            // `resolve_btf(None)` still tries the local-tree / sysfs
            // fallbacks — a test running with a stale env pointer
            // shouldn't be any worse off than a test with no env set.
            crate::cli::resolve_cached_kernel(&id, "ktstr test", None)
                .ok()
                .and_then(|p| p.into_os_string().into_string().ok())
        }
        // Range and git specs (`A..B` ranges, `git+URL#branch=main`) cannot
        // resolve to a single BTF source — there is no dispatch
        // loop here, just a one-shot lookup feeding `resolve_btf`.
        // Treat as "no env hint" and let the local-tree / sysfs
        // fallbacks pick a vmlinux if one exists; the env value
        // would have surfaced a hard error at the actual VM-boot
        // entry point instead.
        Some(KernelId::Range { .. }) | Some(KernelId::Git { .. }) => None,
        None => None,
    };
    let result = crate::kernel_path::resolve_btf(resolved_dir.as_deref());
    if result.is_none() {
        crate::report::test_skip(format!("no vmlinux found; {}", crate::KTSTR_KERNEL_HINT));
    }
    result
}

/// Outcome of the freeze coordinator's pre-resolution boot-complete
/// wait (the 5 s epoll on the sys_rdy eventfd + the kill eventfd).
/// Surfaced on [`MonitorReport`] so the sys_rdy-regression test can
/// tell a real "sys_rdy fired but the monitor never woke on it"
/// regression apart from an inconclusive slow boot (sys_rdy never
/// emitted within the ceiling).
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub enum BootWaitOutcome {
    /// Not a confirmed sampled-run wake: no sys_rdy eventfd was
    /// configured (the in-`monitor_loop` test path), the monitor was
    /// killed during setup before reaching the sample loop, or the
    /// outcome was not captured. The test treats this like `TimedOut`
    /// — inconclusive, skip. Default.
    #[default]
    NotConfigured,
    /// The sys_rdy eventfd fired before the 5 s ceiling → the monitor
    /// proceeded into the sample loop on the wake. The regression-
    /// detecting state: if this holds and the first sample is past its
    /// budget, sys_rdy delivery to the monitor is broken.
    Fired,
    /// The 5 s ceiling elapsed without sys_rdy firing → the wait fell
    /// through. Inconclusive for the sys_rdy contract — a slow guest
    /// boot, not necessarily a delivery regression.
    TimedOut,
}

/// Collected monitor data from a VM run.
#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
#[non_exhaustive]
pub struct MonitorReport {
    /// Periodic snapshots of per-CPU state.
    pub samples: Vec<MonitorSample>,
    /// Aggregated summary statistics.
    pub summary: MonitorSummary,
    /// vCPU preemption threshold (ns) derived from the guest kernel's
    /// CONFIG_HZ at the time the VM ran. Used by evaluate() to gate
    /// stall detection. 0 means use a default.
    pub preemption_threshold_ns: u64,
    /// Post-write readback of the scx_sched.watchdog_timeout field.
    /// Framework-internal regression guard that the host-side override
    /// actually lands in guest memory; populated once per VM run after
    /// the first successful deref.
    #[doc(hidden)]
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub watchdog_observation: Option<WatchdogObservation>,
    /// Live `PAGE_OFFSET` value used by the monitor for KVA→PA
    /// translation, captured at the moment the per-iteration
    /// `DATA_VALID` latch fired. On KASLR-randomized kernels
    /// (`CONFIG_RANDOMIZE_MEMORY`) the guest publishes a
    /// randomized base into `page_offset_base` only after early
    /// boot completes, so this records the value the monitor
    /// actually used to read `struct rq` rather than the static
    /// fallback (`0xffff_8880_0000_0000` on x86_64). 0 means the
    /// latch never fired (guest never finished boot, or the
    /// monitor was not started).
    #[doc(hidden)]
    #[serde(default)]
    pub page_offset: u64,
    /// Outcome of the freeze coordinator's boot-complete wait — see
    /// [`BootWaitOutcome`]. `Fired` = the monitor woke on the sys_rdy
    /// eventfd and proceeded to sample; `TimedOut` = the 5 s ceiling
    /// elapsed (slow boot); `NotConfigured` = sys_rdy not wired, or
    /// the monitor was killed before sampling. The sys_rdy-regression
    /// test asserts first-sample timing only when this is `Fired`, and
    /// skips otherwise (`TimedOut`/`NotConfigured` are inconclusive).
    #[doc(hidden)]
    #[serde(default)]
    pub boot_wait_outcome: BootWaitOutcome,
    /// Whether the running guest kernel structurally supports the
    /// `SCX_EV_*` sched_ext event counters, decided by whether the
    /// monitor resolved their BTF offsets (`event_offsets.is_some()`).
    /// The counters — and the `struct scx_sched` / `scx_root` machinery
    /// the monitor walks to reach them — are a 6.16-cycle addition, so a
    /// kernel older than 6.16 (e.g. 6.14) resolves no offsets and this
    /// is `false`. A BTF-capability probe rather than a version compare,
    /// so it stays correct across backports. `false` means "no event
    /// counters exist to capture"; `true` means the kernel exposes them,
    /// so an empty `event_counters` across the whole run is a capture
    /// regression rather than an absent feature. Event-counter
    /// regression tests skip when this is `false` and assert only when
    /// it is `true`.
    #[doc(hidden)]
    #[serde(default)]
    pub scx_event_counters_supported: bool,
}

/// Observation of the `scx_sched.watchdog_timeout` override,
/// recorded once by the monitor loop after the first successful
/// write to the runtime-allocated scx_sched struct.
///
/// Regression guard for the host-write mechanism: when the kernel
/// refactors the location of watchdog_timeout (as happened before
/// the runtime scx_root deref was introduced), `observed_jiffies`
/// will diverge from `expected_jiffies` and the test will fail.
#[doc(hidden)]
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct WatchdogObservation {
    /// Jiffies value the override was configured to write.
    pub expected_jiffies: u64,
    /// Jiffies value read back from guest memory after the write.
    pub observed_jiffies: u64,
}

/// Tracks consecutive threshold violations and records the worst run.
///
/// Used by `MonitorThresholds::evaluate` (post-hoc, from the collected
/// sample vector) and the reactive `monitor_loop` dump path. Both paths
/// share the tracker so "sustained for N samples" means exactly the
/// same thing to the inline SysRq-D trigger and the after-the-fact
/// verdict. Call `record(true)` on violation, `record(false)` on pass.
#[derive(Debug, Clone, Default)]
pub(crate) struct SustainedViolationTracker {
    consecutive: usize,
    worst_run: usize,
    worst_value: f64,
    worst_at: usize,
}

impl SustainedViolationTracker {
    /// Record a sample. `violated`: whether the threshold was exceeded.
    /// `value`: the metric value for this sample. `at`: sample index.
    pub(crate) fn record(&mut self, violated: bool, value: f64, at: usize) {
        if violated {
            self.consecutive += 1;
            if self.consecutive > self.worst_run {
                self.worst_run = self.consecutive;
                self.worst_value = value;
                self.worst_at = at;
            }
        } else {
            self.consecutive = 0;
        }
    }

    /// Whether the worst run met or exceeded the sustained threshold.
    pub(crate) fn sustained(&self, threshold: usize) -> bool {
        self.worst_run >= threshold
    }
}

/// Point-in-time snapshot of all CPUs.
#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
pub struct MonitorSample {
    /// Milliseconds since VM start.
    pub elapsed_ms: u64,
    /// Per-CPU state at this instant.
    pub cpus: Vec<CpuSnapshot>,
    /// Per-program BPF runtime stats (summed across CPUs).
    /// None when no struct_ops programs are loaded.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub prog_stats: Option<Vec<bpf_prog::ProgRuntimeStats>>,
    /// System-wide PSI-irq pressure at this sample, host-walked from
    /// the global `psi_system` (the cgroup-hierarchy root — system-wide, the
    /// `/proc/pressure/irq` framing; per-cgroup IRQ pressure is a separate
    /// axis). `None` when `psi_system` / `PSI_IRQ_FULL` is absent
    /// (`CONFIG_IRQ_TIME_ACCOUNTING` off) — loud-absent. Raw kernel values;
    /// decoded at the [`MonitorSummary`] fold.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub psi_irq: Option<PsiIrqSample>,
    /// Watched scheduler BPF-map fields read at this sample (one entry per
    /// declared [`crate::test_support::WatchBpfMap`] target that has resolved).
    /// Empty until the scheduler attaches and its maps appear (lazy
    /// resolution) and when no targets are declared. Folded run-level at
    /// [`MonitorSummary`].
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub bpf_map_fields: Vec<BpfMapFieldSample>,
}

/// System-wide PSI-irq pressure read from `psi_system` at one monitor sample.
/// RAW kernel values, decoded at the [`MonitorSummary`] fold via
/// [`crate::monitor::btf_offsets::decode_avg10_percent`] /
/// [`crate::monitor::btf_offsets::decode_total_us`]: the run-level
/// `psi_irq_full_avg10` is the mean of the decoded `avg10` (Gauge), and
/// `total_irq_pressure_us` is the end-start delta of the decoded `total`
/// (Counter).
#[derive(Debug, Clone, Copy, serde::Serialize, serde::Deserialize)]
pub struct PsiIrqSample {
    /// Raw `psi_system.avg[PSI_IRQ_FULL][0]` — the 10s EWMA in fixed-point
    /// (percent × `FIXED_1`=2048); decode via `decode_avg10_percent`.
    pub avg10_raw: u64,
    /// Raw `psi_system.total[PSI_AVGS][PSI_IRQ_FULL]` — cumulative IRQ stall
    /// ns; the run-level metric is its end-start delta (decode via
    /// `decode_total_us`).
    pub total_ns: u64,
}

/// One watched scheduler BPF-map field ([`crate::test_support::WatchBpfMap`])
/// captured at a single monitor sample, read observer-effect-free (the live
/// host reads the running guest's map memory; no VM freeze). Values are
/// decoded to `f64` at read time (the field's BTF width — u16/u32 — is
/// resolved once when the map first appears). Exactly one of `scalar` /
/// `per_cpu` is `Some`, matching the target's [`crate::test_support::BpfMapAgg`].
/// Folded run-level at [`MonitorSummary`]: a gauge `scalar` -> mean across
/// reporting samples; a counter `scalar` (`scalar_counter`) -> the value at
/// the last reporting sample; a gauge `per_cpu` -> cross-(CPU, sample) mean
/// (`_avg`) + spatial max (`_max`); a counter `per_cpu` (`per_cpu_counter`)
/// -> the cross-CPU SUM at the last reporting sample (one key, the total).
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct BpfMapFieldSample {
    /// Final metric-key base `<scheduler-obj>_<label>` (e.g.
    /// `scx_lavd_lat_headroom`). The fold appends `_avg`/`_max` for per-CPU;
    /// scalar uses the base verbatim.
    pub key_base: String,
    /// Scalar field value (a `.bss` global struct member) at this sample.
    /// Mutually exclusive with `per_cpu`.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub scalar: Option<f64>,
    /// Per-CPU field values (a `PERCPU_ARRAY` value member), reporting CPUs
    /// only. Mutually exclusive with `scalar`.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub per_cpu: Option<Vec<f64>>,
    /// True when `scalar` is a monotonic counter
    /// ([`crate::test_support::BpfMapAgg::ScalarCounter`]): the run-level fold
    /// takes the value at the last reporting sample (the accumulated total)
    /// instead of the mean. `false` (the serde default) folds `scalar` as the
    /// mean (a gauge). Applies only to `scalar`; for a per-CPU counter use
    /// `per_cpu_counter`.
    #[serde(default)]
    pub scalar_counter: bool,
    /// True when `per_cpu` is a monotonic counter
    /// ([`crate::test_support::BpfMapAgg::PerCpuCounter`]): the run-level fold
    /// takes the CROSS-CPU SUM at the last reporting sample (the accumulated
    /// total across reporting CPUs) and emits ONE key. `false` (the serde
    /// default) folds `per_cpu` as the gauge mean (`_avg`) + spatial max
    /// (`_max`). Ignored when `scalar` is `Some`.
    #[serde(default)]
    pub per_cpu_counter: bool,
}

impl MonitorSample {
    /// Create a sample with no prog_stats / psi_irq.
    pub fn new(elapsed_ms: u64, cpus: Vec<CpuSnapshot>) -> Self {
        Self {
            elapsed_ms,
            cpus,
            prog_stats: None,
            psi_irq: None,
            bpf_map_fields: Vec::new(),
        }
    }

    /// Compute the imbalance ratio for this sample: max(nr_running) / max(1, min(nr_running)).
    /// Returns 1.0 for empty samples, 0.0 when all CPUs have nr_running=0.
    /// Delegates to [`imbalance_ratio_of`] so the monitor loop's reactive
    /// dump-trigger check can compute the identical ratio from a borrowed
    /// `&[CpuSnapshot]` without cloning into a throwaway `MonitorSample`.
    pub fn imbalance_ratio(&self) -> f64 {
        imbalance_ratio_of(&self.cpus)
    }

    /// Sum a field from event counters across all CPUs.
    /// Returns `None` if no CPU has event counters.
    pub fn sum_event_field(&self, f: fn(&ScxEventCounters) -> i64) -> Option<i64> {
        let mut total = 0i64;
        let mut any = false;
        for cpu in &self.cpus {
            if let Some(ev) = &cpu.event_counters {
                total += f(ev);
                any = true;
            }
        }
        any.then_some(total)
    }
}

/// `max(nr_running) / max(1, min(nr_running))` over a slice of per-CPU
/// snapshots — 1.0 for an empty slice, 0.0 when every CPU is idle
/// (`max_nr == 0`). Extracted from [`MonitorSample::imbalance_ratio`]
/// so the monitor loop's reactive dump-trigger check shares the EXACT
/// computation the post-hoc verdict uses (any drift would let the
/// reactive SysRq-D trigger fire on conditions the verdict rejects)
/// while reading a borrowed `&[CpuSnapshot]` directly — no clone into
/// a throwaway `MonitorSample`.
pub(crate) fn imbalance_ratio_of(cpus: &[CpuSnapshot]) -> f64 {
    if cpus.is_empty() {
        return 1.0;
    }
    let mut min_nr = u32::MAX;
    let mut max_nr = 0u32;
    for cpu in cpus {
        min_nr = min_nr.min(cpu.nr_running);
        max_nr = max_nr.max(cpu.nr_running);
    }
    max_nr as f64 / min_nr.max(1) as f64
}

/// Per-CPU state read from guest VM memory.
#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
pub struct CpuSnapshot {
    /// Total runnable tasks on this CPU (`rq.nr_running`).
    pub nr_running: u32,
    /// Tasks managed by the sched_ext scheduler (`scx_rq.nr_running`).
    pub scx_nr_running: u32,
    /// Depth of the scx local dispatch queue (`scx_rq.local_dsq.nr`).
    pub local_dsq_depth: u32,
    /// Runqueue clock value (`rq.clock`). Non-advancing clock indicates a stall.
    pub rq_clock: u64,
    /// sched_ext flags for this CPU (`scx_rq.flags`).
    pub scx_flags: u32,
    /// PELT IRQ load average (`rq.avg_irq.util_avg`), an INSTANTANEOUS gauge in
    /// `[0, SCHED_CAPACITY_SCALE=1024]`. `None` when CONFIG_HAVE_SCHED_AVG_IRQ
    /// is off (the field is absent from BTF, so the offset resolves to None).
    /// Sampled per periodic monitor tick (the dense gauge axis), NOT a freeze
    /// counter; the run-level `avg_irq_util` / `max_avg_irq_util` metrics are
    /// folded from this in `MonitorSummary`.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub avg_irq_util: Option<u64>,
    /// scx event counters (cumulative). None when event counter
    /// offsets are unavailable or scx_root is not set.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub event_counters: Option<ScxEventCounters>,
    /// Runqueue schedstat fields (cumulative). None when CONFIG_SCHEDSTATS
    /// is not enabled (schedstat offsets unavailable).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub schedstat: Option<RqSchedstat>,
    /// Cumulative CPU time (ns) of the vCPU thread hosting this CPU.
    /// Used by evaluate() to distinguish real stalls from host preemption.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub vcpu_cpu_time_ns: Option<u64>,
    /// Host-side hardware perf counters for the vCPU thread that owns
    /// this guest CPU. Captured via `perf_event_open(2)` with
    /// `exclude_host=1` so the PMU only ticks while the vCPU is
    /// running guest code. `None` when perf is unavailable on the
    /// host (paranoid policy, missing CAP_PERFMON, hardware lacks
    /// the requested counter), when no TID is registered for this
    /// vCPU, or before the per-vCPU counter set was opened on the
    /// first sample. See [`crate::monitor::perf_counters`].
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub vcpu_perf: Option<perf_counters::VcpuPerfSample>,
    /// Sched domain tree for this CPU. Each entry is one domain level,
    /// ordered from lowest (e.g. SMT) to highest (e.g. NUMA). None when
    /// sched_domain offsets are unavailable or `rq->sd` is null.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub sched_domains: Option<Vec<SchedDomainSnapshot>>,
}

/// Per-CPU runqueue schedstat fields read from guest memory.
///
/// Matches kernel `struct rq` schedstat fields (guarded by CONFIG_SCHEDSTATS).
/// `run_delay` and `pcount` come from the embedded `struct sched_info`;
/// the remaining fields are direct members of `struct rq`.
#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
pub struct RqSchedstat {
    /// Cumulative scheduling delay (ns) on this CPU (`rq.rq_sched_info.run_delay`).
    pub run_delay: u64,
    /// Count of non-idle task arrivals on this CPU (`rq.rq_sched_info.pcount`).
    pub pcount: u64,
    /// Yield count (`rq.yld_count`).
    pub yld_count: u32,
    /// Context switch count (`rq.sched_count`).
    pub sched_count: u32,
    /// Go-idle count (`rq.sched_goidle`).
    pub sched_goidle: u32,
    /// Try-to-wake-up count (`rq.ttwu_count`).
    pub ttwu_count: u32,
    /// Try-to-wake-up local count (`rq.ttwu_local`).
    pub ttwu_local: u32,
}

/// Snapshot of one `struct sched_domain` level for a single CPU.
///
/// Domains are ordered from lowest (e.g. SMT, level 0) to highest
/// (e.g. NUMA, level N) following the kernel's `sd->parent` chain.
/// `newidle_call`, `newidle_success`, and `newidle_ratio` are `None`
/// when the kernel lacks these fields (added in 7.0; backported to
/// 6.18.5+, 6.12.65+; not present on 6.16-6.18.4).
/// CONFIG_SCHEDSTATS load balancing stats are in the optional `stats`
/// field.
#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
pub struct SchedDomainSnapshot {
    /// Domain level number (`sd->level`). 0 = innermost (e.g. SMT).
    pub level: i32,
    /// Domain name from `sd->name` (e.g. "SMT", "MC", "DIE", "NUMA").
    pub name: String,
    /// Domain flags (`sd->flags`). SD_* values.
    pub flags: i32,
    /// Number of CPUs in this domain's span (`sd->span_weight`).
    pub span_weight: u32,

    // -- Runtime fields --
    /// Current balance interval in ms (`sd->balance_interval`).
    pub balance_interval: u32,
    /// Consecutive load balance failures (`sd->nr_balance_failed`).
    pub nr_balance_failed: u32,
    /// Number of newidle balance calls (`sd->newidle_call`).
    /// None when BTF lacks this field (added in 7.0; backported to
    /// 6.18.5+, 6.12.65+). Not present on 6.16-6.18.4.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub newidle_call: Option<u32>,
    /// Successful newidle balance calls (`sd->newidle_success`).
    /// None when BTF lacks this field (added in 7.0; backported to
    /// 6.18.5+, 6.12.65+). Not present on 6.16-6.18.4.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub newidle_success: Option<u32>,
    /// Newidle balance ratio (`sd->newidle_ratio`).
    /// None when BTF lacks this field (added in 7.0; backported to
    /// 6.18.5+, 6.12.65+). Not present on 6.16-6.18.4.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub newidle_ratio: Option<u32>,
    /// Max cost of newidle load balancing in ns (`sd->max_newidle_lb_cost`).
    pub max_newidle_lb_cost: u64,

    /// CONFIG_SCHEDSTATS load balancing stats. None when
    /// CONFIG_SCHEDSTATS is not enabled.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub stats: Option<SchedDomainStats>,
}

/// CONFIG_SCHEDSTATS load balancing stats for one `struct sched_domain`.
///
/// Array fields have `CPU_MAX_IDLE_TYPES` (3) elements indexed by
/// `cpu_idle_type`: \[0\] = CPU_NOT_IDLE, \[1\] = CPU_IDLE,
/// \[2\] = CPU_NEWLY_IDLE. All counters are cumulative — compute
/// deltas between samples to get rates.
#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
pub struct SchedDomainStats {
    /// `sd->lb_count[CPU_MAX_IDLE_TYPES]`: number of load balance calls.
    pub lb_count: [u32; btf_offsets::CPU_MAX_IDLE_TYPES],
    /// `sd->lb_failed[CPU_MAX_IDLE_TYPES]`: load balance calls that found
    /// imbalance but failed to move any task.
    pub lb_failed: [u32; btf_offsets::CPU_MAX_IDLE_TYPES],
    /// `sd->lb_balanced[CPU_MAX_IDLE_TYPES]`: load balance calls that
    /// found no imbalance.
    pub lb_balanced: [u32; btf_offsets::CPU_MAX_IDLE_TYPES],
    /// `sd->lb_imbalance_load[CPU_MAX_IDLE_TYPES]`: cumulative LOAD-based
    /// imbalance MAGNITUDE (capacity-scaled load), summed via
    /// `schedstat_add(env->imbalance)` over `migrate_load` balance attempts —
    /// an accumulated magnitude, NOT a count of attempts.
    pub lb_imbalance_load: [u32; btf_offsets::CPU_MAX_IDLE_TYPES],
    /// `sd->lb_imbalance_util[CPU_MAX_IDLE_TYPES]`: cumulative UTILIZATION-based
    /// imbalance MAGNITUDE (PELT util), summed via `schedstat_add` over
    /// `migrate_util` attempts — accumulated magnitude, not a count.
    pub lb_imbalance_util: [u32; btf_offsets::CPU_MAX_IDLE_TYPES],
    /// `sd->lb_imbalance_task[CPU_MAX_IDLE_TYPES]`: cumulative TASK-COUNT
    /// imbalance, summed via `schedstat_add` over `migrate_task` attempts —
    /// accumulated task-count delta, not a count of attempts.
    pub lb_imbalance_task: [u32; btf_offsets::CPU_MAX_IDLE_TYPES],
    /// `sd->lb_imbalance_misfit[CPU_MAX_IDLE_TYPES]`: cumulative MISFIT
    /// imbalance, summed via `schedstat_add` over `migrate_misfit` attempts
    /// (1 per misfit migration) — accumulated magnitude, not a count.
    pub lb_imbalance_misfit: [u32; btf_offsets::CPU_MAX_IDLE_TYPES],
    /// `sd->lb_gained[CPU_MAX_IDLE_TYPES]`: tasks pulled during load balance.
    pub lb_gained: [u32; btf_offsets::CPU_MAX_IDLE_TYPES],
    /// `sd->lb_hot_gained[CPU_MAX_IDLE_TYPES]`: cache-hot tasks pulled.
    pub lb_hot_gained: [u32; btf_offsets::CPU_MAX_IDLE_TYPES],
    /// `sd->lb_nobusyg[CPU_MAX_IDLE_TYPES]`: times no busy group was found.
    pub lb_nobusyg: [u32; btf_offsets::CPU_MAX_IDLE_TYPES],
    /// `sd->lb_nobusyq[CPU_MAX_IDLE_TYPES]`: times no busy queue was found.
    pub lb_nobusyq: [u32; btf_offsets::CPU_MAX_IDLE_TYPES],

    /// `sd->alb_count`: active load balance attempts.
    pub alb_count: u32,
    /// `sd->alb_failed`: active load balance failures.
    pub alb_failed: u32,
    /// `sd->alb_pushed`: tasks pushed via active load balancing.
    pub alb_pushed: u32,

    /// `sd->sbe_count`: exec balance attempts.
    pub sbe_count: u32,
    /// `sd->sbe_balanced`: exec balance found no imbalance.
    pub sbe_balanced: u32,
    /// `sd->sbe_pushed`: tasks pushed via exec balancing.
    pub sbe_pushed: u32,

    /// `sd->sbf_count`: fork balance attempts.
    pub sbf_count: u32,
    /// `sd->sbf_balanced`: fork balance found no imbalance.
    pub sbf_balanced: u32,
    /// `sd->sbf_pushed`: tasks pushed via fork balancing.
    pub sbf_pushed: u32,

    /// `sd->ttwu_wake_remote`: wakeups targeting a remote CPU.
    pub ttwu_wake_remote: u32,
    /// `sd->ttwu_move_affine`: wakeups moved to an affine CPU.
    pub ttwu_move_affine: u32,
    /// `sd->ttwu_move_balance`: wakeups moved for load balance.
    pub ttwu_move_balance: u32,
}

/// Cumulative scx event counter values for a single CPU.
/// These are s64 in the kernel but always non-negative; stored as i64.
#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
pub struct ScxEventCounters {
    /// `SCX_EV_SELECT_CPU_FALLBACK`: scheduler's `ops.select_cpu()` failed to find a CPU.
    pub select_cpu_fallback: i64,
    /// `SCX_EV_DISPATCH_LOCAL_DSQ_OFFLINE`: dispatch to an offline CPU's local DSQ.
    pub dispatch_local_dsq_offline: i64,
    /// `SCX_EV_DISPATCH_KEEP_LAST`: CPU re-dispatched the previously running task.
    pub dispatch_keep_last: i64,
    /// `SCX_EV_ENQ_SKIP_EXITING`: enqueue skipped because the task is exiting.
    pub enq_skip_exiting: i64,
    /// `SCX_EV_ENQ_SKIP_MIGRATION_DISABLED`: enqueue skipped because migration is disabled.
    pub enq_skip_migration_disabled: i64,
    /// `SCX_EV_REENQ_IMMED`: task re-enqueued because CPU was unavailable for immediate execution.
    pub reenq_immed: i64,
    /// `SCX_EV_REENQ_LOCAL_REPEAT`: recursive local DSQ re-enqueue from `SCX_ENQ_IMMED` race.
    pub reenq_local_repeat: i64,
    /// `SCX_EV_REFILL_SLICE_DFL`: time slice refilled with `SCX_SLICE_DFL`.
    pub refill_slice_dfl: i64,
    /// `SCX_EV_BYPASS_DURATION`: total bypass mode duration in nanoseconds.
    pub bypass_duration: i64,
    /// `SCX_EV_BYPASS_DISPATCH`: tasks dispatched during bypass mode.
    pub bypass_dispatch: i64,
    /// `SCX_EV_BYPASS_ACTIVATE`: bypass mode activations.
    pub bypass_activate: i64,
    /// `SCX_EV_INSERT_NOT_OWNED`: attempts to insert a non-owned task into a DSQ.
    pub insert_not_owned: i64,
    /// `SCX_EV_SUB_BYPASS_DISPATCH`: tasks from bypassing descendants scheduled from sub_bypass_dsq.
    pub sub_bypass_dispatch: i64,
}

/// Aggregated monitor statistics from a set of samples.
#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
pub struct MonitorSummary {
    /// Number of samples collected.
    pub total_samples: usize,
    /// Peak imbalance ratio across all samples: `max(nr_running) / max(1, min(nr_running))`.
    pub max_imbalance_ratio: f64,
    /// Peak local DSQ depth across all CPUs and samples.
    pub max_local_dsq_depth: u32,
    /// Count of (CPU, consecutive-sample-pair) observations where the
    /// CPU's `rq_clock` failed to advance. Idle CPUs (`nr_running == 0`
    /// in both samples) and host-preempted vCPUs are exempt (see
    /// `reader::is_cpu_stuck`). This is the run-level analog of the
    /// per-phase `PhaseMetrics::stall_count`: both apply the SAME
    /// per-(CPU, window) `is_cpu_stuck` predicate, but this counts over
    /// the full sample stream while the per-phase path windows within
    /// each phase, so run-level `>=` the sum of per-phase counts (it
    /// additionally counts cross-phase-boundary window pairs and
    /// out-of-phase samples — boot-settle warmup, inter-step gaps).
    pub stuck_count: usize,
    /// Average imbalance ratio across valid samples.
    pub avg_imbalance_ratio: f64,
    /// Average nr_running per CPU across valid samples.
    pub avg_nr_running: f64,
    /// Average local DSQ depth per CPU across valid samples.
    pub avg_local_dsq_depth: f64,
    /// Mean PELT IRQ load (`rq.avg_irq.util_avg`, 0..=1024) across the CPUs
    /// and valid samples that REPORTED it. `None` when no sample carried an
    /// avg_irq reading (CONFIG_HAVE_SCHED_AVG_IRQ off) — loud-absent, never a
    /// false 0.0. The divisor is the reporting-CPU-reading count, NOT
    /// all CPU readings.
    pub avg_irq_util: Option<f64>,
    /// Peak (max across CPUs and samples) PELT IRQ load. `None` when no sample
    /// reported avg_irq (same gate as `avg_irq_util`).
    pub max_avg_irq_util: Option<f64>,
    /// Mean system-wide PSI-irq `full` avg10 pressure (percent, 0..=100) across
    /// the monitor samples that REPORTED it (host-walked from `psi_system`).
    /// `None` when no sample carried a PSI-irq reading (CONFIG_PSI /
    /// CONFIG_IRQ_TIME_ACCOUNTING off, or no `psi_system` symbol) — loud-absent,
    /// never a false 0.0. The run-level `psi_irq_full_avg10` gauge.
    pub psi_irq_full_avg10: Option<f64>,
    /// Cumulative system-wide PSI-irq `full` stall over the monitoring window
    /// (microseconds): the end-start delta of `total[PSI_AVGS][PSI_IRQ_FULL]`
    /// across the first/last samples that reported PSI-irq. `None` under the
    /// same gate as `psi_irq_full_avg10`. The run-level `total_irq_pressure_us`
    /// counter.
    pub total_irq_pressure_us: Option<f64>,
    /// Aggregate event counter deltas over the monitoring window.
    /// None when event counters are not available.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub event_deltas: Option<ScxEventDeltas>,
    /// Aggregate schedstat deltas over the monitoring window.
    /// None when CONFIG_SCHEDSTATS is not enabled.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub schedstat_deltas: Option<SchedstatDeltas>,
    /// Per-domain-level CFS load-balance counter deltas over the monitoring
    /// window, summed across CPUs by domain level name. None when no sample
    /// carried sched_domains data (CONFIG_SCHEDSTATS off, or a pre-domain-
    /// capture kernel). See [`SchedDomainLbDelta`] for the scx-marginal caveat.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub sched_domain_lb: Option<Vec<SchedDomainLbDelta>>,
    /// Per-program BPF callback profile over the monitoring window.
    /// None when no struct_ops programs are loaded.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub prog_stats_deltas: Option<Vec<ProgStatsDelta>>,
    /// Folded watched-scheduler-BPF-map-field metrics over the monitoring
    /// window: one entry per Dynamic run-level metric key (a scalar target
    /// emits one key; a per-CPU target emits `_avg` + `_max`). None when no
    /// [`crate::test_support::WatchBpfMap`] target was declared OR none
    /// resolved (the scheduler never attached / the map never appeared) —
    /// loud-absent, so `run_metric` returns None and an assertion fails
    /// rather than reading a false 0.0. See [`BpfMapFieldValue`].
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub bpf_map_fields: Option<Vec<BpfMapFieldValue>>,
}

/// Per-program BPF callback profile computed from first/last monitor samples.
#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
pub struct ProgStatsDelta {
    /// Program name.
    pub name: String,
    /// Total invocations over the monitoring window.
    pub cnt: u64,
    /// Total CPU time in nanoseconds over the monitoring window.
    pub nsecs: u64,
    /// Average nanoseconds per call (nsecs / cnt). 0 when cnt is 0.
    pub nsecs_per_call: f64,
}

/// A folded watched-scheduler-BPF-map-field metric: a Dynamic run-level metric
/// key and its value, ready to insert into the run-level ext map. Built by
/// [`MonitorSummary::from_samples`] from the per-sample [`BpfMapFieldSample`]s
/// and surfaced via [`MonitorSummary::fold_run_level_ext`] (and thus
/// [`crate::vmm::result::VmResult::run_metric`]). A scalar gauge target yields
/// one value (`<prefix>_<label>`, the mean over reporting samples); a scalar
/// counter target yields one value (`<prefix>_<label>`, the last reporting
/// sample's value); a per-CPU target yields two (`_avg` = cross-(CPU, sample)
/// mean, `_max` = spatial max).
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct BpfMapFieldValue {
    /// The Dynamic run-level metric key (e.g. `scx_lavd_avg_lat_cri`,
    /// `scx_lavd_lat_headroom_avg`).
    pub key: String,
    /// The folded value.
    pub value: f64,
    /// True when this key is a monotonic counter (a `ScalarCounter` scalar
    /// target). Counter keys SUM-fold ACROSS runs (matching registered
    /// counters); gauge / per-CPU keys mean-fold. Carried into
    /// [`MonitorSummary::fold_run_level_ext`]'s counter-key set so the
    /// cross-run aggregator (`stats::group::fold_ext_metrics`) sums them
    /// instead of averaging. `#[serde(default)]` matches the module's other
    /// evolving serialized fields (`scalar_counter`, `page_offset`, …): a stale
    /// sidecar lacking the field degrades to `false` (mean-fold, the pre-
    /// counter-fold behavior) rather than hard-failing the whole deserialize.
    #[serde(default)]
    pub is_counter: bool,
}

/// Aggregate schedstat deltas computed from first/last monitor samples.
///
/// All values are summed across CPUs and represent the delta over the
/// monitoring window. `total_schedstat_wall_sec` carries that window's span in
/// seconds — the denominator for the per-second schedstat Rate metrics
/// (`*_per_sec`, derived in the metric registry).
#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
pub struct SchedstatDeltas {
    /// Total scheduling delay increase (ns) across all CPUs.
    pub total_run_delay: u64,
    /// Total pcount increase across all CPUs.
    pub total_pcount: u64,
    /// Total schedule() invocation increase (`rq.sched_count`) across all CPUs.
    pub total_sched_count: u64,
    /// Total yield count increase across all CPUs.
    pub total_yld_count: u64,
    /// Total go-idle count increase across all CPUs.
    pub total_sched_goidle: u64,
    /// Total ttwu count increase across all CPUs.
    pub total_ttwu_count: u64,
    /// Total ttwu_local count increase across all CPUs.
    pub total_ttwu_local: u64,
    /// Monitor-window span in seconds (last-first schedstat-bearing sample) —
    /// the denominator for the per-second schedstat Rate metrics; 0.0 on a
    /// degenerate single-sample window.
    pub total_schedstat_wall_sec: f64,
}

/// Per-domain-level CFS load-balance counter deltas over the monitoring window.
///
/// `struct sched_domain` is per-CPU (the kernel allocates one per CPU per
/// topology level — kernel/sched/topology.c `__sdt_alloc`), so each counter
/// here is SUMMED across every CPU's domain at the given level: the total CFS
/// load-balance activity at that level, analogous to the per-CPU rq schedstat
/// sum. Levels are keyed by NAME (SMT/CLS/MC/PKG/NUMA), not the CPU-relative
/// domain index. Each value is the first→last delta over the window; the
/// per-idle-type arrays (CPU_NOT_IDLE/CPU_IDLE/CPU_NEWLY_IDLE) are summed
/// across all three.
///
/// All fields are monotonic counters. NOTE (scx-marginal): under a switch-all
/// sched_ext scheduler the fair runqueues are near-empty, so `lb_count` keeps
/// ticking while the move counters (`lb_gained` / `lb_imbalance_*` / `alb_pushed`)
/// stay ~0 — these observe CFS-class balancing, not the scx scheduler's own
/// (BPF) placement. Useful on stock-CFS scenarios and residual fair-class tasks.
///
/// The four `lb_imbalance_*` accumulators are kept SEPARATE (not summed into
/// one) because the kernel routes `env->imbalance` into them by migration type
/// and each carries a different, incommensurable unit — capacity-scaled load /
/// PELT utilization / task count / misfit. Summing them would be dimensionally
/// meaningless (dominated by the load/util magnitudes, with task/misfit lost),
/// so a delta on a combined value would have no consistent interpretation.
#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
pub struct SchedDomainLbDelta {
    /// Domain level name (`sd->name`): SMT / CLS / MC / PKG / NUMA.
    pub level: String,
    /// `sd->lb_count` summed over idle types + CPUs: load-balance attempts.
    pub lb_count: u64,
    /// `sd->lb_failed` summed: attempts that found imbalance but moved no task
    /// (balance friction; higher is worse).
    pub lb_failed: u64,
    /// `sd->lb_gained` summed: tasks pulled during load balance.
    pub lb_gained: u64,
    /// `sd->lb_imbalance_load` summed (idle types + CPUs): cumulative
    /// capacity-scaled LOAD imbalance magnitude. Same-unit accumulator — see
    /// the struct doc for why the four variants are not summed together.
    pub lb_imbalance_load: u64,
    /// `sd->lb_imbalance_util` summed: cumulative PELT-UTILIZATION imbalance.
    pub lb_imbalance_util: u64,
    /// `sd->lb_imbalance_task` summed: cumulative TASK-COUNT imbalance.
    pub lb_imbalance_task: u64,
    /// `sd->lb_imbalance_misfit` summed: cumulative MISFIT imbalance.
    pub lb_imbalance_misfit: u64,
    /// `sd->alb_count` summed: active-load-balance (migration-thread) attempts.
    pub alb_count: u64,
    /// `sd->alb_pushed` summed: tasks pushed via active load balancing.
    pub alb_pushed: u64,
}

/// Aggregate event counter statistics computed from first/last samples.
#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
pub struct ScxEventDeltas {
    /// Total select_cpu_fallback events across all CPUs over the window.
    pub total_fallback: i64,
    /// Fallback events per second (total_fallback / duration_secs).
    pub fallback_rate: f64,
    /// Max single-sample delta of fallback across all CPUs.
    pub max_fallback_burst: i64,
    /// Total dispatch_local_dsq_offline events.
    pub total_dispatch_offline: i64,
    /// Total dispatch_keep_last events.
    pub total_dispatch_keep_last: i64,
    /// Keep-last events per second (total_dispatch_keep_last / duration_secs).
    pub keep_last_rate: f64,
    /// Total enq_skip_exiting events.
    pub total_enq_skip_exiting: i64,
    /// Total enq_skip_migration_disabled events.
    pub total_enq_skip_migration_disabled: i64,
    /// Total reenq_immed events.
    pub total_reenq_immed: i64,
    /// Total reenq_local_repeat events.
    pub total_reenq_local_repeat: i64,
    /// Total refill_slice_dfl events.
    pub total_refill_slice_dfl: i64,
    /// Total bypass_duration in nanoseconds.
    pub total_bypass_duration: i64,
    /// Total bypass_dispatch events.
    pub total_bypass_dispatch: i64,
    /// Total bypass_activate events.
    pub total_bypass_activate: i64,
    /// Total insert_not_owned events.
    pub total_insert_not_owned: i64,
    /// Total sub_bypass_dispatch events.
    pub total_sub_bypass_dispatch: i64,
}

impl MonitorSummary {
    /// Summarize a run's monitor samples using the derived default
    /// preemption threshold (equivalent to
    /// [`from_samples_with_threshold`](Self::from_samples_with_threshold)
    /// with `0`).
    pub fn from_samples(samples: &[MonitorSample]) -> Self {
        Self::from_samples_with_threshold(samples, 0)
    }

    /// Fold this summary's run-level ext-only monitor metrics into `ext`:
    /// `avg_nr_running`, the PELT IRQ load pair (`avg_irq_util` /
    /// `max_avg_irq_util`), and the PSI-irq pair (`psi_irq_full_avg10` /
    /// `total_irq_pressure_us`). A no-op when `total_samples == 0` (a 0-sample
    /// run carries no occupancy / IRQ signal — absent, never a false 0.0). The
    /// `Option` fields insert only on `Some` (loud-absent on a kernel without the
    /// source: non-HAVE_SCHED_AVG_IRQ, or no CONFIG_PSI / CONFIG_IRQ_TIME_ACCOUNTING).
    /// `entry().or_insert()` so a value already placed by an earlier populator
    /// wins. The fixed ext keys (avg_nr_running, the IRQ/PSI pair) are disjoint
    /// from the read_sample- and phase-folded keys; the Dynamic keys are
    /// scheduler-obj-prefixed (`<obj>_<label>`) or level-suffixed (`lb_*_<lvl>`)
    /// so collisions with the fixed set are not constructed. Shared by
    /// `stats::sidecar_to_row`
    /// (the perf-delta sidecar row) and `VmResult::run_metric` (the in-test
    /// accessor) so the key list + loud-absent guard cannot drift between the two
    /// surfaces.
    ///
    /// Value-only convenience for callers that don't need the cross-run
    /// counter-key set (`VmResult::run_metric`, unit tests). Delegates to
    /// [`Self::fold_run_level_ext_with_counter_keys`].
    pub(crate) fn fold_run_level_ext(&self, ext: &mut std::collections::BTreeMap<String, f64>) {
        let mut counter_keys = std::collections::BTreeSet::new();
        self.fold_run_level_ext_with_counter_keys(ext, &mut counter_keys);
    }

    /// Like [`Self::fold_run_level_ext`] but also collects the emitted Dynamic
    /// monotonic-counter keys into `counter_keys` (the level-suffixed
    /// `lb_*`/`alb_*` schedstat deltas + any `ScalarCounter` bpf field); gauge
    /// keys (`avg_nr_running`, the IRQ / PSI pair, per-CPU `_avg`/`_max`) are
    /// omitted. The cross-run aggregator (`stats::group::fold_ext_metrics`)
    /// SUM-folds the counter keys instead of averaging them, matching the
    /// registered-Counter cross-run convention. Used by `stats::sidecar_to_row`
    /// to populate `GauntletRow::ext_counter_keys`.
    pub(crate) fn fold_run_level_ext_with_counter_keys(
        &self,
        ext: &mut std::collections::BTreeMap<String, f64>,
        counter_keys: &mut std::collections::BTreeSet<String>,
    ) {
        if self.total_samples == 0 {
            return;
        }
        ext.entry("avg_nr_running".to_string())
            .or_insert(self.avg_nr_running);
        if let Some(v) = self.avg_irq_util {
            ext.entry("avg_irq_util".to_string()).or_insert(v);
        }
        if let Some(v) = self.max_avg_irq_util {
            ext.entry("max_avg_irq_util".to_string()).or_insert(v);
        }
        if let Some(v) = self.psi_irq_full_avg10 {
            ext.entry("psi_irq_full_avg10".to_string()).or_insert(v);
        }
        if let Some(v) = self.total_irq_pressure_us {
            ext.entry("total_irq_pressure_us".to_string()).or_insert(v);
        }
        // Per-domain-level CFS load-balance counters: one key set per topology
        // level present in the run, level-suffixed (e.g. lb_failed_mc). These
        // are Dynamic ext keys — the level set is host-topology-dependent
        // (SMT/CLS/MC/PKG/NUMA), so a fixed typed-metric enum would be
        // incomplete; absent levels emit nothing. All Counter-kind. See
        // [`SchedDomainLbDelta`] for the scx-marginal caveat.
        if let Some(domains) = &self.sched_domain_lb {
            for d in domains {
                let lvl = d.level.to_ascii_lowercase();
                // All sched_domain load-balance fields are deltas of kernel
                // monotonic counters (see [`SchedDomainLbDelta`]) -> Dynamic
                // Counter keys that SUM-fold across runs. The four imbalance
                // accumulators stay SEPARATE keys — each a distinct,
                // incommensurable unit (load / util / task / misfit), never
                // summed INTO one key — but each individually SUM-folds across
                // runs like the other counters.
                for (suffix, val) in [
                    ("lb_count", d.lb_count),
                    ("lb_failed", d.lb_failed),
                    ("lb_gained", d.lb_gained),
                    ("lb_imbalance_load", d.lb_imbalance_load),
                    ("lb_imbalance_util", d.lb_imbalance_util),
                    ("lb_imbalance_task", d.lb_imbalance_task),
                    ("lb_imbalance_misfit", d.lb_imbalance_misfit),
                    ("alb_count", d.alb_count),
                    ("alb_pushed", d.alb_pushed),
                ] {
                    let k = format!("{suffix}_{lvl}");
                    counter_keys.insert(k.clone());
                    ext.entry(k).or_insert(val as f64);
                }
            }
        }
        // Watched scheduler BPF-map fields: Dynamic, scheduler-obj-prefixed
        // keys built at fold time (the target/scheduler set is runtime, so a
        // fixed typed-metric enum cannot enumerate them — same rationale as
        // the level-suffixed sched_domain_lb keys above). Absent targets emit
        // nothing (loud-absent).
        if let Some(fields) = &self.bpf_map_fields {
            for f in fields {
                if f.is_counter {
                    counter_keys.insert(f.key.clone());
                }
                ext.entry(f.key.clone()).or_insert(f.value);
            }
        }
    }

    /// Like [`from_samples`](Self::from_samples) but uses an explicit
    /// preemption threshold (ns) for stall detection. Pass 0 to derive
    /// the threshold from the guest kernel's `CONFIG_HZ` by calling
    /// [`vcpu_preemption_threshold_ns`], which tries (in order) the
    /// embedded IKCONFIG in the guest `vmlinux`, a `.config` beside
    /// the kernel image, the host's `/boot/config-$(uname -r)`, and
    /// finally the built-in `DEFAULT_HZ`.
    pub fn from_samples_with_threshold(
        samples: &[MonitorSample],
        preemption_threshold_ns: u64,
    ) -> Self {
        if samples.is_empty() {
            return Self::default();
        }

        let mut max_imbalance_ratio: f64 = 1.0;
        let mut max_local_dsq_depth: u32 = 0;
        let mut sum_imbalance_ratio: f64 = 0.0;
        let mut sum_nr_running: f64 = 0.0;
        let mut sum_local_dsq_depth: f64 = 0.0;
        let mut valid_sample_count: usize = 0;
        let mut total_cpu_readings: usize = 0;
        // PELT IRQ load: mean + peak over the CPUs/samples that REPORTED
        // avg_irq_util (a gate-aware Option — absent on non-HAVE_SCHED_AVG_IRQ
        // kernels). Divisor is `avg_irq_readings`, NOT total_cpu_readings, so a
        // kernel where only some/no CPUs report it is not diluted/false-zeroed.
        let mut sum_avg_irq: f64 = 0.0;
        let mut max_avg_irq: f64 = 0.0;
        let mut avg_irq_readings: usize = 0;

        for sample in samples {
            if sample.cpus.is_empty() || !sample_looks_valid(sample) {
                continue;
            }
            valid_sample_count += 1;
            for cpu in &sample.cpus {
                max_local_dsq_depth = max_local_dsq_depth.max(cpu.local_dsq_depth);
                sum_nr_running += cpu.nr_running as f64;
                sum_local_dsq_depth += cpu.local_dsq_depth as f64;
                total_cpu_readings += 1;
                if let Some(u) = cpu.avg_irq_util {
                    let u = u as f64;
                    sum_avg_irq += u;
                    if u > max_avg_irq {
                        max_avg_irq = u;
                    }
                    avg_irq_readings += 1;
                }
            }
            let ratio = sample.imbalance_ratio();
            sum_imbalance_ratio += ratio;
            if ratio > max_imbalance_ratio {
                max_imbalance_ratio = ratio;
            }
        }

        let avg_imbalance_ratio = if valid_sample_count > 0 {
            sum_imbalance_ratio / valid_sample_count as f64
        } else {
            0.0
        };
        let avg_nr_running = if total_cpu_readings > 0 {
            sum_nr_running / total_cpu_readings as f64
        } else {
            0.0
        };
        let avg_local_dsq_depth = if total_cpu_readings > 0 {
            sum_local_dsq_depth / total_cpu_readings as f64
        } else {
            0.0
        };
        // Loud-absent: None (not 0.0) when no CPU reported avg_irq, preserving
        // the absent-vs-measured-zero distinction.
        let avg_irq_util = (avg_irq_readings > 0).then(|| sum_avg_irq / avg_irq_readings as f64);
        let max_avg_irq_util = (avg_irq_readings > 0).then_some(max_avg_irq);

        // Stuck count: number of (CPU, consecutive-sample-pair)
        // observations whose rq_clock did not advance. Skip invalid samples.
        // Counts EVERY hit across all CPUs and all window pairs (no early
        // break). Windowed over the full sample stream, so this is `>=` the
        // sum of per-phase `PhaseMetrics::stall_count` (both route through
        // `is_cpu_stuck`; the per-phase path windows within each phase and
        // drops cross-boundary pairs + out-of-phase samples).
        // Exempt idle CPUs: nr_running==0 in both samples means the tick
        // is stopped (NOHZ) and rq_clock legitimately does not advance.
        // Exempt preempted vCPUs: vcpu_cpu_time_ns didn't advance enough
        // means the host preempted the vCPU thread.
        let threshold = if preemption_threshold_ns > 0 {
            preemption_threshold_ns
        } else {
            vcpu_preemption_threshold_ns(None)
        };
        let mut stuck_count: usize = 0;
        let valid_samples: Vec<&MonitorSample> = samples
            .iter()
            .filter(|s| !s.cpus.is_empty() && sample_looks_valid(s))
            .collect();
        for w in valid_samples.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 reader::is_cpu_stuck(&prev.cpus[cpu], &curr.cpus[cpu], threshold) {
                    stuck_count += 1;
                }
            }
        }

        // System-wide PSI-irq pressure (host-walked from `psi_system`, one
        // reading per sample). `psi_irq_full_avg10` is the MEAN of the decoded
        // avg10 EWMA (percent) across the samples that reported PSI-irq — a
        // Gauge. `total_irq_pressure_us` is the end-start delta of the
        // cumulative `total` ns (decoded to µs) — a Counter; `saturating_sub`
        // guards a counter reset (the accumulator is monotonic, but a PSI /
        // scheduler reset would rewind it). Both `None` when no sample reported
        // PSI-irq (loud-absent), preserving the absent-vs-measured-zero
        // distinction. Decoupled from the cpu-validity gate above: the reading
        // is a singleton `.data` global, captured only on `data_valid` samples.
        let psi_samples: Vec<&PsiIrqSample> =
            samples.iter().filter_map(|s| s.psi_irq.as_ref()).collect();
        let psi_irq_full_avg10 = (!psi_samples.is_empty()).then(|| {
            let sum: f64 = psi_samples
                .iter()
                .map(|p| btf_offsets::decode_avg10_percent(p.avg10_raw))
                .sum();
            sum / psi_samples.len() as f64
        });
        let total_irq_pressure_us =
            psi_samples
                .first()
                .zip(psi_samples.last())
                .map(|(first, last)| {
                    btf_offsets::decode_total_us(last.total_ns.saturating_sub(first.total_ns))
                });

        let event_deltas = Self::compute_event_deltas(samples);
        let schedstat_deltas = Self::compute_schedstat_deltas(samples);
        let sched_domain_lb = Self::compute_sched_domain_deltas(samples);
        let prog_stats_deltas = Self::compute_prog_stats_deltas(samples);
        let bpf_map_fields = Self::compute_bpf_map_field_deltas(samples);

        Self {
            total_samples: samples.len(),
            max_imbalance_ratio,
            max_local_dsq_depth,
            stuck_count,
            avg_imbalance_ratio,
            avg_nr_running,
            avg_local_dsq_depth,
            avg_irq_util,
            max_avg_irq_util,
            psi_irq_full_avg10,
            total_irq_pressure_us,
            event_deltas,
            schedstat_deltas,
            sched_domain_lb,
            prog_stats_deltas,
            bpf_map_fields,
        }
    }

    /// Compute event counter deltas from the sample series.
    /// Returns None if no samples have event counters.
    fn compute_event_deltas(samples: &[MonitorSample]) -> Option<ScxEventDeltas> {
        // Find first and last samples that have event counters on any CPU.
        let has_events = |s: &MonitorSample| s.cpus.iter().any(|c| c.event_counters.is_some());
        let first = samples.iter().find(|s| has_events(s))?;
        let last = samples.iter().rev().find(|s| has_events(s))?;

        let total_fallback = 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 total_keep_last = 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),
        );

        // Compute rates.
        let duration_ms = last.elapsed_ms.saturating_sub(first.elapsed_ms);
        let duration_secs = duration_ms as f64 / 1000.0;
        let fallback_rate = if duration_secs > 0.0 {
            total_fallback as f64 / duration_secs
        } else {
            0.0
        };
        let keep_last_rate = if duration_secs > 0.0 {
            total_keep_last as f64 / duration_secs
        } else {
            0.0
        };

        // Max per-sample fallback burst: largest delta between consecutive
        // samples, summed across all CPUs. A counter reset between
        // samples yields a negative raw delta — ignore it rather than
        // letting it decrease the running max.
        let mut max_fallback_burst: i64 = 0;
        for w in samples.windows(2) {
            let prev_sum = w[0].sum_event_field(|e| e.select_cpu_fallback).unwrap_or(0);
            let curr_sum = w[1].sum_event_field(|e| e.select_cpu_fallback).unwrap_or(0);
            let delta = counter_delta(curr_sum, prev_sum);
            if delta > max_fallback_burst {
                max_fallback_burst = delta;
            }
        }

        let delta = |f: fn(&ScxEventCounters) -> i64| -> i64 {
            counter_delta(
                last.sum_event_field(f).unwrap_or(0),
                first.sum_event_field(f).unwrap_or(0),
            )
        };

        Some(ScxEventDeltas {
            total_fallback,
            fallback_rate,
            max_fallback_burst,
            total_dispatch_offline: delta(|e| e.dispatch_local_dsq_offline),
            total_dispatch_keep_last: total_keep_last,
            keep_last_rate,
            total_enq_skip_exiting: delta(|e| e.enq_skip_exiting),
            total_enq_skip_migration_disabled: delta(|e| e.enq_skip_migration_disabled),
            total_reenq_immed: delta(|e| e.reenq_immed),
            total_reenq_local_repeat: delta(|e| e.reenq_local_repeat),
            total_refill_slice_dfl: delta(|e| e.refill_slice_dfl),
            total_bypass_duration: delta(|e| e.bypass_duration),
            total_bypass_dispatch: delta(|e| e.bypass_dispatch),
            total_bypass_activate: delta(|e| e.bypass_activate),
            total_insert_not_owned: delta(|e| e.insert_not_owned),
            total_sub_bypass_dispatch: delta(|e| e.sub_bypass_dispatch),
        })
    }

    /// Compute schedstat deltas from the sample series.
    /// Returns None if no samples have schedstat data on any CPU.
    fn compute_schedstat_deltas(samples: &[MonitorSample]) -> Option<SchedstatDeltas> {
        let has_schedstat = |s: &MonitorSample| s.cpus.iter().any(|c| c.schedstat.is_some());
        let first = samples.iter().find(|s| has_schedstat(s))?;
        let last = samples.iter().rev().find(|s| has_schedstat(s))?;

        let sum_field = |s: &MonitorSample, f: fn(&RqSchedstat) -> u64| -> u64 {
            s.cpus
                .iter()
                .filter_map(|c| c.schedstat.as_ref().map(&f))
                .sum()
        };
        let sum_field_u32 = |s: &MonitorSample, f: fn(&RqSchedstat) -> u32| -> u64 {
            s.cpus
                .iter()
                .filter_map(|c| c.schedstat.as_ref().map(|ss| f(ss) as u64))
                .sum()
        };

        let total_run_delay =
            sum_field(last, |ss| ss.run_delay).saturating_sub(sum_field(first, |ss| ss.run_delay));
        let total_pcount =
            sum_field(last, |ss| ss.pcount).saturating_sub(sum_field(first, |ss| ss.pcount));
        let total_sched_count = sum_field_u32(last, |ss| ss.sched_count)
            .saturating_sub(sum_field_u32(first, |ss| ss.sched_count));
        let total_yld_count = sum_field_u32(last, |ss| ss.yld_count)
            .saturating_sub(sum_field_u32(first, |ss| ss.yld_count));
        let total_sched_goidle = sum_field_u32(last, |ss| ss.sched_goidle)
            .saturating_sub(sum_field_u32(first, |ss| ss.sched_goidle));
        let total_ttwu_count = sum_field_u32(last, |ss| ss.ttwu_count)
            .saturating_sub(sum_field_u32(first, |ss| ss.ttwu_count));
        let total_ttwu_local = sum_field_u32(last, |ss| ss.ttwu_local)
            .saturating_sub(sum_field_u32(first, |ss| ss.ttwu_local));

        // Window span over the SAME first/last schedstat-bearing samples the
        // total_* deltas span — the provenance-correct per-second denominator
        // (a different window than the IRQ total_phase_wall_sec, so this carries
        // its own). The registry per-second Rates divide the total_* numerators
        // by this; 0.0 on a single-sample window disables the rate (Rate
        // derivation skips a zero/non-finite denominator).
        let duration_ms = last.elapsed_ms.saturating_sub(first.elapsed_ms);
        let duration_secs = duration_ms as f64 / 1000.0;

        Some(SchedstatDeltas {
            total_run_delay,
            total_pcount,
            total_sched_count,
            total_yld_count,
            total_sched_goidle,
            total_ttwu_count,
            total_ttwu_local,
            total_schedstat_wall_sec: duration_secs,
        })
    }

    /// Compute per-domain-level load-balance counter deltas from the sample
    /// series. Returns None if no sample carries sched_domains data.
    ///
    /// `struct sched_domain` is per-CPU, so the counters for a given level
    /// NAME are summed across every CPU's domain at that level (the total
    /// balancing activity at the level), keyed by name because the domain
    /// INDEX is CPU-relative and not stable across CPUs. The per-idle-type
    /// arrays are summed across all three idle types. Only domains whose
    /// CONFIG_SCHEDSTATS `stats` block is present contribute.
    fn compute_sched_domain_deltas(samples: &[MonitorSample]) -> Option<Vec<SchedDomainLbDelta>> {
        let has_domains = |s: &MonitorSample| {
            s.cpus.iter().any(|c| {
                c.sched_domains
                    .as_ref()
                    .is_some_and(|ds| ds.iter().any(|d| d.stats.is_some()))
            })
        };
        let first = samples.iter().find(|s| has_domains(s))?;
        let last = samples.iter().rev().find(|s| has_domains(s))?;

        // Sum the curated counters per level NAME across all CPUs' domains in
        // one sample.
        fn per_level(
            sample: &MonitorSample,
        ) -> std::collections::BTreeMap<String, SchedDomainLbDelta> {
            let sum3 = |a: &[u32; btf_offsets::CPU_MAX_IDLE_TYPES]| -> u64 {
                a.iter().map(|&v| v as u64).sum()
            };
            let mut acc: std::collections::BTreeMap<String, SchedDomainLbDelta> =
                std::collections::BTreeMap::new();
            for cpu in &sample.cpus {
                let Some(domains) = cpu.sched_domains.as_ref() else {
                    continue;
                };
                for d in domains {
                    let Some(st) = d.stats.as_ref() else {
                        continue;
                    };
                    let e = acc
                        .entry(d.name.clone())
                        .or_insert_with(|| SchedDomainLbDelta {
                            level: d.name.clone(),
                            ..SchedDomainLbDelta::default()
                        });
                    e.lb_count += sum3(&st.lb_count);
                    e.lb_failed += sum3(&st.lb_failed);
                    e.lb_gained += sum3(&st.lb_gained);
                    // The four imbalance accumulators stay separate — each is a
                    // different unit (load / util / task / misfit); summing them
                    // would be dimensionally meaningless.
                    e.lb_imbalance_load += sum3(&st.lb_imbalance_load);
                    e.lb_imbalance_util += sum3(&st.lb_imbalance_util);
                    e.lb_imbalance_task += sum3(&st.lb_imbalance_task);
                    e.lb_imbalance_misfit += sum3(&st.lb_imbalance_misfit);
                    e.alb_count += st.alb_count as u64;
                    e.alb_pushed += st.alb_pushed as u64;
                }
            }
            acc
        }

        let first_lv = per_level(first);
        let last_lv = per_level(last);
        if last_lv.is_empty() {
            return None;
        }

        // first→last delta per level. Topology is static so a level in `last`
        // is normally in `first` too; a missing baseline is treated as 0.
        // saturating_sub guards a counter reset / first-absent.
        let out: Vec<SchedDomainLbDelta> = last_lv
            .into_iter()
            .map(|(name, last_v)| {
                let f = first_lv.get(&name);
                let base = |sel: fn(&SchedDomainLbDelta) -> u64| f.map(sel).unwrap_or(0);
                SchedDomainLbDelta {
                    lb_count: last_v.lb_count.saturating_sub(base(|d| d.lb_count)),
                    lb_failed: last_v.lb_failed.saturating_sub(base(|d| d.lb_failed)),
                    lb_gained: last_v.lb_gained.saturating_sub(base(|d| d.lb_gained)),
                    lb_imbalance_load: last_v
                        .lb_imbalance_load
                        .saturating_sub(base(|d| d.lb_imbalance_load)),
                    lb_imbalance_util: last_v
                        .lb_imbalance_util
                        .saturating_sub(base(|d| d.lb_imbalance_util)),
                    lb_imbalance_task: last_v
                        .lb_imbalance_task
                        .saturating_sub(base(|d| d.lb_imbalance_task)),
                    lb_imbalance_misfit: last_v
                        .lb_imbalance_misfit
                        .saturating_sub(base(|d| d.lb_imbalance_misfit)),
                    alb_count: last_v.alb_count.saturating_sub(base(|d| d.alb_count)),
                    alb_pushed: last_v.alb_pushed.saturating_sub(base(|d| d.alb_pushed)),
                    level: name,
                }
            })
            .collect();
        Some(out)
    }

    /// Compute per-program callback profile from first/last samples
    /// that contain prog_stats.
    fn compute_prog_stats_deltas(samples: &[MonitorSample]) -> Option<Vec<ProgStatsDelta>> {
        let first = samples.iter().find(|s| s.prog_stats.is_some())?;
        let last = samples.iter().rev().find(|s| s.prog_stats.is_some())?;

        let first_progs = first.prog_stats.as_ref()?;
        let last_progs = last.prog_stats.as_ref()?;

        let first_by_name: std::collections::HashMap<&str, &bpf_prog::ProgRuntimeStats> =
            first_progs.iter().map(|p| (p.name.as_str(), p)).collect();

        let deltas: Vec<ProgStatsDelta> = last_progs
            .iter()
            .map(|lp| {
                let fp = first_by_name.get(lp.name.as_str()).copied();
                let cnt = lp.cnt.saturating_sub(fp.map_or(0, |p| p.cnt));
                let nsecs = lp.nsecs.saturating_sub(fp.map_or(0, |p| p.nsecs));
                let nsecs_per_call = if cnt > 0 {
                    nsecs as f64 / cnt as f64
                } else {
                    0.0
                };
                ProgStatsDelta {
                    name: lp.name.clone(),
                    cnt,
                    nsecs,
                    nsecs_per_call,
                }
            })
            .collect();

        if deltas.is_empty() {
            None
        } else {
            Some(deltas)
        }
    }

    /// Fold the per-sample watched BPF-map field reads
    /// ([`MonitorSample::bpf_map_fields`]) into run-level Dynamic metrics.
    /// Grouped by `key_base`: a scalar GAUGE target yields one value (mean
    /// over the reporting samples, key `key_base`); a scalar COUNTER target
    /// (`scalar_counter`) yields one value (the LAST reporting sample's value,
    /// key `key_base` — the accumulated total at end-of-window); a per-CPU
    /// target yields two (`<key_base>_avg` = mean over all reporting (CPU,
    /// sample) readings, `<key_base>_max` = spatial max across them). The
    /// gauge/per-CPU divisor is the count of reporting readings, never the
    /// topology CPU/sample count, so a kernel/scheduler that never reported a
    /// target is loud-absent (no key) rather than a false 0.0. `None` when no
    /// target reported in any sample. `BTreeMap` keys give a deterministic
    /// metric ordering.
    fn compute_bpf_map_field_deltas(samples: &[MonitorSample]) -> Option<Vec<BpfMapFieldValue>> {
        use std::collections::BTreeMap;
        struct Acc {
            scalar_sum: f64,
            scalar_n: usize,
            scalar_last: f64,
            scalar_counter: bool,
            pc_sum: f64,
            pc_n: usize,
            pc_max: f64,
            pc_counter: bool,
            pc_last_sum: f64,
        }
        let mut accs: BTreeMap<&str, Acc> = BTreeMap::new();
        for s in samples {
            for f in &s.bpf_map_fields {
                let a = accs.entry(f.key_base.as_str()).or_insert(Acc {
                    scalar_sum: 0.0,
                    scalar_n: 0,
                    scalar_last: 0.0,
                    scalar_counter: false,
                    pc_sum: 0.0,
                    pc_n: 0,
                    pc_max: f64::NEG_INFINITY,
                    pc_counter: false,
                    pc_last_sum: 0.0,
                });
                if let Some(v) = f.scalar {
                    a.scalar_sum += v;
                    a.scalar_n += 1;
                    a.scalar_last = v;
                    a.scalar_counter = f.scalar_counter;
                }
                if let Some(vs) = &f.per_cpu {
                    if f.per_cpu_counter {
                        // Per-CPU monotonic counter: this sample's cross-CPU
                        // total; the last NON-EMPTY sample wins. Guard the empty
                        // case (the live reader never emits an empty per-CPU
                        // sample, but a deserialized sidecar could): an all-empty
                        // target emits no key (loud-absent, no phantom 0.0), and
                        // an empty last sample falls back to the prior non-empty
                        // sum (the last successful read, like the scalar path).
                        if !vs.is_empty() {
                            a.pc_counter = true;
                            a.pc_last_sum = vs.iter().copied().sum();
                        }
                    } else {
                        for &v in vs {
                            a.pc_sum += v;
                            a.pc_n += 1;
                            if v > a.pc_max {
                                a.pc_max = v;
                            }
                        }
                    }
                }
            }
        }
        let mut out: Vec<BpfMapFieldValue> = Vec::new();
        for (key_base, a) in accs {
            if a.scalar_n > 0 {
                // Counter -> last reporting sample's value (accumulated total);
                // gauge -> mean over the reporting samples.
                let value = if a.scalar_counter {
                    a.scalar_last
                } else {
                    a.scalar_sum / a.scalar_n as f64
                };
                out.push(BpfMapFieldValue {
                    key: key_base.to_string(),
                    value,
                    is_counter: a.scalar_counter,
                });
            }
            if a.pc_counter {
                // Per-CPU counter: the accumulated cross-CPU total at the last
                // non-empty reporting sample (ONE key, is_counter -> SUM-folds
                // across runs). pc_counter is set only by a non-empty per-CPU
                // sample (guarded above), so this is loud-absent — no key when
                // nothing reported.
                out.push(BpfMapFieldValue {
                    key: key_base.to_string(),
                    value: a.pc_last_sum,
                    is_counter: true,
                });
            } else if a.pc_n > 0 {
                // Per-CPU gauge keys are a mean (`_avg`) and a spatial peak
                // (`_max`), not counters — they mean-fold across runs.
                out.push(BpfMapFieldValue {
                    key: format!("{key_base}_avg"),
                    value: a.pc_sum / a.pc_n as f64,
                    is_counter: false,
                });
                out.push(BpfMapFieldValue {
                    key: format!("{key_base}_max"),
                    value: a.pc_max,
                    is_counter: false,
                });
            }
        }
        if out.is_empty() { None } else { Some(out) }
    }
}

/// Configurable thresholds for monitor-based pass/fail verdicts.
///
/// Default behaviour is REPORT-ONLY: violations populate
/// [`MonitorVerdict::details`] but [`MonitorVerdict::passed`] stays
/// `true`. To make violations fail the test, opt into enforcement via
/// [`crate::assert::Assert::with_monitor_defaults`] (which sets
/// `enforce = true`) or by constructing `MonitorThresholds` with
/// `enforce: true` explicitly.
///
/// The two-mode design lets a test attach monitor coverage for
/// diagnostic purposes without inheriting a five-axis failure
/// surface the test author did not opt into.
#[derive(Debug, Clone, Copy)]
pub struct MonitorThresholds {
    /// Max allowed imbalance ratio (max_nr_running / max(1, min_nr_running)).
    pub max_imbalance_ratio: f64,
    /// Max allowed local DSQ depth on any CPU in any sample.
    pub max_local_dsq_depth: u32,
    /// Flag when any CPU's rq_clock does not advance between consecutive samples.
    pub fail_on_stall: bool,
    /// Number of consecutive samples that must violate a threshold before flagging.
    pub sustained_samples: usize,
    /// Max sustained select_cpu_fallback events/s across all CPUs.
    pub max_fallback_rate: f64,
    /// Max sustained dispatch_keep_last events/s across all CPUs.
    pub max_keep_last_rate: f64,
    /// Promote threshold violations from report-only to pass/fail.
    /// When `false` (the default),
    /// [`MonitorThresholds::evaluate`] still walks every sample
    /// and records every violation in the verdict's `details`, but
    /// returns `passed: true` regardless. When `true`, any recorded
    /// violation also fails the verdict.
    ///
    /// `enforce` gates only the THRESHOLD-VIOLATION path. The
    /// no-signal Inconclusive arms — empty sample buffer
    /// ([`MonitorVerdict`] with `summary: "no monitor samples"`)
    /// and `data_looks_valid` rejection of uninitialized guest
    /// memory ([`MonitorVerdict`] with
    /// `summary: "monitor data not yet initialized"`) — fire
    /// BEFORE `enforce` is consulted and return
    /// `passed: false, inconclusive: true` regardless of
    /// `enforce`. "Couldn't evaluate" is not the same as
    /// "evaluated and OK", so the no-signal path always surfaces
    /// distinct from Pass.
    ///
    /// Test authors asserting `!verdict.passed` on threshold
    /// violations MUST opt into enforcement:
    /// - For inline `MonitorThresholds { ... }` literals: include
    ///   `enforce: true` before `..Default::default()`.
    /// - For [`Assert`](crate::assert::Assert) builder chains:
    ///   terminate with
    ///   [`.with_monitor_defaults()`](crate::assert::Assert::with_monitor_defaults)
    ///   which fills unset threshold fields with canonical defaults
    ///   AND sets the inner `enforce_monitor_thresholds` flag that
    ///   propagates to this field via
    ///   [`Assert::monitor_thresholds`](crate::assert::Assert::monitor_thresholds).
    ///
    /// Without this opt-in, a test that sets `fail_on_stall: true`
    /// and asserts `!verdict.passed` on a THRESHOLD VIOLATION
    /// will PASS despite the violation: the violation is recorded
    /// in `details` and the verdict's `summary` carries the
    /// report-only advisory ("monitor flagged N violation(s)
    /// (report-only; pass `Assert::with_monitor_defaults` to
    /// enforce)"), but `passed` stays `true`. The advisory is
    /// the operator-visible signal that flags missing enforcement;
    /// the test still passes against `!verdict.passed`. The
    /// no-signal arms above are not gated by this — use
    /// `verdict.is_fail()` (rather than `!verdict.passed`) on
    /// callers that want to ignore the Inconclusive case.
    pub enforce: bool,
}

impl MonitorThresholds {
    /// Build the default thresholds. `const fn` so it can sit in
    /// `static` / `const` initializers and in
    /// `..MonitorThresholds::new()` spread expressions.
    ///
    /// - imbalance 4.0: a scheduler that can't keep CPUs within 4x
    ///   load for `sustained_samples` consecutive reads has a real
    ///   balancing problem. Lower ratios (2-3) false-positive during
    ///   cpuset transitions when cgroups are being created/destroyed.
    /// - DSQ depth 50: local DSQ is a per-CPU overflow queue. Sustained
    ///   depth > 50 means the scheduler is not consuming dispatched tasks.
    ///   Transient spikes during cpuset changes are filtered by the
    ///   sustained_samples window.
    /// - fail_on_stall true: rq_clock not advancing on a CPU with
    ///   runnable tasks means the scheduler stalled. Idle CPUs
    ///   (nr_running==0 in both samples) are exempt because NOHZ
    ///   stops the tick. Preempted vCPUs are exempt when the vCPU
    ///   thread's CPU time didn't advance past the preemption
    ///   threshold. Uses the sustained_samples window.
    /// - sustained_samples 5: at ~100ms sample interval, requires ~500ms
    ///   of sustained violation. Filters transient spikes from cpuset
    ///   reconfiguration, cgroup creation, and scheduler restart.
    /// - max_fallback_rate 200.0: select_cpu_fallback fires when the
    ///   scheduler's ops.select_cpu() fails to find a CPU. Sustained
    ///   200/s across all CPUs indicates systematic select_cpu failure.
    /// - max_keep_last_rate 100.0: dispatch_keep_last fires when a CPU
    ///   re-dispatches the previously running task because the scheduler
    ///   provided nothing. Sustained 100/s indicates dispatch starvation.
    pub const fn new() -> Self {
        Self {
            max_imbalance_ratio: 4.0,
            max_local_dsq_depth: 50,
            fail_on_stall: true,
            sustained_samples: 5,
            max_fallback_rate: 200.0,
            max_keep_last_rate: 100.0,
            enforce: false,
        }
    }
}

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

/// Verdict from evaluating monitor data against thresholds.
///
/// Encodes a 3-state outcome under the same `Fail > Inconclusive >
/// Pass` precedence as [`crate::assert::AssertResult`]. The struct
/// uses paired bools rather than an enum so existing callers that
/// read [`Self::passed`] directly continue to compile; the
/// [`Self::inconclusive`] bit is added alongside under a strict
/// mutex (at most one of `is_pass`/`is_fail`/`is_inconclusive`
/// returns true).
///
/// - **Pass** (`passed=true, inconclusive=false`): the monitor
///   walked every sample, no threshold tripped.
/// - **Inconclusive** (`passed=false, inconclusive=true`): the
///   monitor had no signal to evaluate — empty sample buffer or
///   data that fails the
///   [`MonitorThresholds::data_looks_valid`] plausibility check
///   (uninitialized guest memory). Distinct from Pass to prevent
///   the silent-pass class of bug where a CI gate reading
///   [`Self::is_pass`] would treat "monitor never measured"
///   identically to "monitor measured and OK".
/// - **Fail** (`passed=false, inconclusive=false`): the monitor
///   evaluated samples and at least one threshold was both
///   tripped and enforced (`MonitorThresholds::enforce = true`).
#[derive(Debug, Clone)]
pub struct MonitorVerdict {
    /// `true` if all thresholds were met. False under both Fail
    /// and Inconclusive — disambiguate via [`Self::inconclusive`]
    /// or the [`Self::is_pass`] / [`Self::is_fail`] /
    /// [`Self::is_inconclusive`] strict-mutex accessors.
    pub passed: bool,
    /// `true` when the monitor lacked signal to evaluate (no
    /// samples or data that failed the plausibility check). When
    /// set, `passed` is always `false` and `details` is empty —
    /// the operator-visible narrative lives in `summary`. Pinned
    /// by the strict-mutex invariant: `(passed, inconclusive)`
    /// is one of `(true, false)`, `(false, true)`, or
    /// `(false, false)`; `(true, true)` is illegal.
    pub inconclusive: bool,
    /// Per-violation detail messages (empty when `passed` is true
    /// AND when `inconclusive` is true — the no-signal arms carry
    /// the operator-visible narrative in `summary`, not `details`).
    pub details: Vec<String>,
    /// One-line summary: "monitor OK" / "monitor FAILED: N
    /// violation(s)" / "no monitor samples" / "monitor data not yet
    /// initialized".
    pub summary: String,
}

impl MonitorVerdict {
    /// Construct an Inconclusive verdict (no signal to evaluate).
    /// The two no-signal arms in [`MonitorThresholds::evaluate`]
    /// (empty sample buffer; `data_looks_valid` rejection of
    /// uninitialized guest memory) both route through this
    /// constructor so the strict-mutex invariant
    /// `(passed=false, inconclusive=true)` is enforced at
    /// construction time rather than by inspection at each
    /// call site. `details` is always empty on the Inconclusive
    /// arm — the operator-visible narrative lives in `summary`.
    ///
    /// Name matches the variant-named constructor convention on
    /// sibling types ([`crate::assert::AssertResult::inconclusive`]).
    pub fn inconclusive(summary: impl Into<String>) -> Self {
        Self {
            passed: false,
            inconclusive: true,
            details: Vec::new(),
            summary: summary.into(),
        }
    }
    /// Convenience accessor mirroring [`crate::assert::AssertResult::is_pass`] so the
    /// `is_pass` / `is_fail` / `is_inconclusive` vocabulary applies
    /// uniformly across [`crate::assert::AssertResult`] /
    /// [`crate::test_support::SidecarResult`] / `GauntletRow` (in
    /// the `stats` module, which is `pub(crate)`) / `MonitorVerdict` /
    /// `Verdict` (in `assert::claim`, re-exported as
    /// [`crate::assert::Verdict`]) / [`crate::assert::Outcome`].
    /// The bare code spans for `GauntletRow` are intentional — the
    /// containing module is `pub(crate)` so a clickable intra-doc
    /// link would trip the rustdoc `private_intra_doc_links`
    /// warning, but the method itself is `pub` within the crate.
    /// Returns true only on the real-Pass arm — neither Fail nor
    /// Inconclusive satisfy it. CI gates that want "monitor
    /// verified OK" semantics call this method and only this
    /// method.
    ///
    /// Skip vocabulary asymmetry: `MonitorVerdict` exposes a
    /// 3-state surface (`is_pass` / `is_fail` / `is_inconclusive`)
    /// without an `is_skip` peer, because a monitor never skips by
    /// design — it either evaluates samples, fails on a threshold,
    /// or lacks signal (Inconclusive). The sibling 4-state types
    /// add `is_skip` for the "scenario didn't run" path which has
    /// no monitor analog.
    ///
    /// The accessor `debug_assert!`s the strict-mutex invariant
    /// `!(passed && inconclusive)` because the type's fields are
    /// `pub` for ergonomic reads — external code can struct-literal
    /// an illegal `(true, true)` shape that bypasses
    /// [`Self::inconclusive`]. The assertion catches the bug at
    /// test time without paying for the check in release builds.
    pub fn is_pass(&self) -> bool {
        debug_assert!(
            !(self.passed && self.inconclusive),
            "MonitorVerdict strict 3-state mutex: (passed=true, inconclusive=true) is illegal",
        );
        self.passed && !self.inconclusive
    }
    /// Convenience accessor mirroring [`crate::assert::AssertResult::is_fail`].
    /// Returns true only when the monitor evaluated and a
    /// threshold tripped — the Inconclusive arm (no signal to
    /// evaluate) is explicitly excluded so "couldn't measure"
    /// doesn't masquerade as "measured and broken". Carries the
    /// same `debug_assert!` strict-mutex guard as [`Self::is_pass`].
    pub fn is_fail(&self) -> bool {
        debug_assert!(
            !(self.passed && self.inconclusive),
            "MonitorVerdict strict 3-state mutex: (passed=true, inconclusive=true) is illegal",
        );
        !self.passed && !self.inconclusive
    }
    /// Convenience accessor mirroring
    /// [`crate::assert::AssertResult::is_inconclusive`]. True when
    /// the monitor lacked signal to evaluate (no samples or data
    /// that failed the plausibility check). Distinct from
    /// [`Self::is_pass`] (which means the monitor verified the
    /// scheduler) and [`Self::is_fail`] (which means a threshold
    /// tripped). CI gates that need "did monitor produce a
    /// verdict?" should test
    /// `v.is_pass() || v.is_fail()` and treat
    /// `is_inconclusive()` as "couldn't measure". Carries the
    /// same `debug_assert!` strict-mutex guard as [`Self::is_pass`].
    pub fn is_inconclusive(&self) -> bool {
        debug_assert!(
            !(self.passed && self.inconclusive),
            "MonitorVerdict strict 3-state mutex: (passed=true, inconclusive=true) is illegal",
        );
        self.inconclusive
    }
    /// Iterate the violation-detail messages. Naming mirrors
    /// [`crate::assert::AssertResult::failure_details`] so consumers can swap
    /// between the two verdict types with the same vocabulary;
    /// MonitorVerdict's details are `String` rather than
    /// `AssertDetail` because the monitor-thread surface doesn't
    /// carry kind tags.
    ///
    /// No `inconclusive_details` peer: the Inconclusive arms
    /// (no-signal paths via [`Self::inconclusive`]) carry the
    /// operator-visible narrative in `summary` rather than a
    /// per-violation `details` vec. There are no per-detail
    /// inconclusive payloads to iterate — the singular summary
    /// IS the narrative. `AssertResult` exposes an
    /// `inconclusive_details` iterator because each
    /// `record_inconclusive` push carries its own `AssertDetail`
    /// payload; the monitor surface produces at most one
    /// Inconclusive signal per evaluate() call, so the
    /// peer would always yield zero or one entry.
    #[allow(dead_code)]
    pub fn failure_details(&self) -> impl Iterator<Item = &String> {
        self.details.iter()
    }
}

impl MonitorThresholds {
    /// Evaluate a MonitorReport against these thresholds.
    ///
    /// Returns an INCONCLUSIVE verdict when samples are empty or
    /// when the monitor data appears to be uninitialized guest
    /// memory (all rq_clocks identical across every CPU and sample,
    /// or DSQ depths above a plausibility ceiling). The monitor
    /// thread reads raw guest memory via BTF offsets; in short-lived
    /// VMs the kernel may not have populated the per-CPU runqueue
    /// structures before the monitor starts sampling.
    ///
    /// The no-signal arms used to return `passed: true` — the
    /// silent-pass class of bug, where a CI gate keying off
    /// [`MonitorVerdict::is_pass`] couldn't distinguish "monitor
    /// verified OK" from "monitor never measured". Inconclusive
    /// surfaces this distinction explicitly so the operator triages
    /// it instead of shipping on a non-measurement.
    pub fn evaluate(&self, report: &MonitorReport) -> MonitorVerdict {
        if report.samples.is_empty() {
            return MonitorVerdict::inconclusive("no monitor samples");
        }
        // Validity check: detect uninitialized guest memory.
        // If all rq_clock values across every CPU in every sample are
        // identical, the kernel never wrote to these fields — the monitor
        // was reading zeroed or garbage memory.
        if !Self::data_looks_valid(&report.samples) {
            return MonitorVerdict::inconclusive("monitor data not yet initialized");
        }

        let mut details = Vec::new();
        let mut failed = false;

        let (imbalance, dsq, worst_dsq_cpu) = self.track_imbalance_and_dsq(&report.samples);
        if imbalance.sustained(self.sustained_samples) {
            failed = true;
            details.push(format!(
                "imbalance ratio {:.1} exceeded threshold {:.1} for {} consecutive samples (ending at sample {})",
                imbalance.worst_value,
                self.max_imbalance_ratio,
                imbalance.worst_run,
                imbalance.worst_at,
            ));
        }
        if dsq.sustained(self.sustained_samples) {
            failed = true;
            details.push(format!(
                "local DSQ depth {} on cpu{} exceeded threshold {} for {} consecutive samples (ending at sample {})",
                dsq.worst_value as u32,
                worst_dsq_cpu,
                self.max_local_dsq_depth,
                dsq.worst_run,
                dsq.worst_at,
            ));
        }

        if self.fail_on_stall {
            for (cpu, tracker) in self.track_stall(report).iter().enumerate() {
                if tracker.sustained(self.sustained_samples) {
                    failed = true;
                    details.push(format!(
                        "rq_clock stall on cpu{} for {} consecutive samples (ending at sample {}, clock={})",
                        cpu,
                        tracker.worst_run,
                        tracker.worst_at,
                        tracker.worst_value as u64,
                    ));
                }
            }
        }

        let (fallback_rate, keep_last_rate) = self.track_event_rates(&report.samples);
        if fallback_rate.sustained(self.sustained_samples) {
            failed = true;
            details.push(format!(
                "fallback rate {:.1}/s exceeded threshold {:.1}/s for {} consecutive intervals (ending at sample {})",
                fallback_rate.worst_value,
                self.max_fallback_rate,
                fallback_rate.worst_run,
                fallback_rate.worst_at,
            ));
        }
        if keep_last_rate.sustained(self.sustained_samples) {
            failed = true;
            details.push(format!(
                "keep_last rate {:.1}/s exceeded threshold {:.1}/s for {} consecutive intervals (ending at sample {})",
                keep_last_rate.worst_value,
                self.max_keep_last_rate,
                keep_last_rate.worst_run,
                keep_last_rate.worst_at,
            ));
        }

        MonitorVerdict {
            passed: !failed || !self.enforce,
            inconclusive: false,
            summary: Self::summarize(failed, self.enforce, details.len()),
            details,
        }
    }

    /// Per-sample loop that tracks imbalance ratio and worst-CPU DSQ
    /// depth against their thresholds. Returns the (imbalance, dsq,
    /// worst_dsq_cpu) trio in the same shape the inline form used —
    /// extracted for readability and direct unit-test coverage.
    fn track_imbalance_and_dsq(
        &self,
        samples: &[MonitorSample],
    ) -> (SustainedViolationTracker, SustainedViolationTracker, usize) {
        let mut imbalance = SustainedViolationTracker::default();
        let mut dsq = SustainedViolationTracker::default();
        let mut worst_dsq_cpu = 0usize;

        for (i, sample) in samples.iter().enumerate() {
            if sample.cpus.is_empty() {
                imbalance.record(false, 0.0, i);
                dsq.record(false, 0.0, i);
                continue;
            }

            let ratio = sample.imbalance_ratio();
            imbalance.record(ratio > self.max_imbalance_ratio, ratio, i);

            let mut dsq_violated = false;
            let mut sample_worst_depth = 0u32;
            let mut sample_worst_cpu = 0usize;
            for (cpu_idx, cpu) in sample.cpus.iter().enumerate() {
                if cpu.local_dsq_depth > self.max_local_dsq_depth
                    && cpu.local_dsq_depth > sample_worst_depth
                {
                    dsq_violated = true;
                    sample_worst_depth = cpu.local_dsq_depth;
                    sample_worst_cpu = cpu_idx;
                }
            }
            dsq.record(dsq_violated, sample_worst_depth as f64, i);
            if dsq_violated && dsq.worst_value == sample_worst_depth as f64 {
                worst_dsq_cpu = sample_worst_cpu;
            }
        }

        (imbalance, dsq, worst_dsq_cpu)
    }

    /// Per-CPU stall detection over consecutive sample pairs. Exempts
    /// idle CPUs (NOHZ stopped the tick so rq_clock legitimately doesn't
    /// advance) and preempted vCPUs (host stole the core, so the vCPU
    /// couldn't tick the clock) via [`reader::is_cpu_stuck`].
    ///
    /// Returns one [`SustainedViolationTracker`] per CPU. Sized to the
    /// max `cpus.len()` across the report so a sample with fewer CPUs
    /// doesn't truncate the per-CPU vector mid-walk.
    fn track_stall(&self, report: &MonitorReport) -> Vec<SustainedViolationTracker> {
        let threshold = if report.preemption_threshold_ns > 0 {
            report.preemption_threshold_ns
        } else {
            vcpu_preemption_threshold_ns(None)
        };

        let num_cpus = report
            .samples
            .iter()
            .map(|s| s.cpus.len())
            .max()
            .unwrap_or(0);
        let mut stall: Vec<SustainedViolationTracker> =
            vec![SustainedViolationTracker::default(); num_cpus];

        for i in 1..report.samples.len() {
            let prev = &report.samples[i - 1];
            let curr = &report.samples[i];
            let cpu_count = prev.cpus.len().min(curr.cpus.len());
            for (cpu, stall_tracker) in stall.iter_mut().enumerate().take(cpu_count) {
                let is_stall = reader::is_cpu_stuck(&prev.cpus[cpu], &curr.cpus[cpu], threshold);
                stall_tracker.record(is_stall, curr.cpus[cpu].rq_clock as f64, i);
            }
        }

        stall
    }

    /// Per-interval event-counter rate computation against fallback /
    /// keep_last thresholds. Returns `(fallback, keep_last)` trackers
    /// keyed by sample index. Intervals with zero or negative duration,
    /// or with missing event counters on either side, record a
    /// non-violation `0.0` so the sustained-window state machine
    /// continues to advance.
    fn track_event_rates(
        &self,
        samples: &[MonitorSample],
    ) -> (SustainedViolationTracker, SustainedViolationTracker) {
        let mut fallback_rate = SustainedViolationTracker::default();
        let mut keep_last_rate = SustainedViolationTracker::default();

        for i in 1..samples.len() {
            let prev = &samples[i - 1];
            let curr = &samples[i];
            let interval_s = curr.elapsed_ms.saturating_sub(prev.elapsed_ms) as f64 / 1000.0;
            if interval_s <= 0.0 {
                fallback_rate.record(false, 0.0, i);
                keep_last_rate.record(false, 0.0, i);
                continue;
            }

            if let (Some(prev_fb), Some(curr_fb)) = (
                prev.sum_event_field(|e| e.select_cpu_fallback),
                curr.sum_event_field(|e| e.select_cpu_fallback),
            ) {
                let rate = (curr_fb - prev_fb) as f64 / interval_s;
                fallback_rate.record(rate > self.max_fallback_rate, rate, i);
            } else {
                fallback_rate.record(false, 0.0, i);
            }

            if let (Some(prev_kl), Some(curr_kl)) = (
                prev.sum_event_field(|e| e.dispatch_keep_last),
                curr.sum_event_field(|e| e.dispatch_keep_last),
            ) {
                let rate = (curr_kl - prev_kl) as f64 / interval_s;
                keep_last_rate.record(rate > self.max_keep_last_rate, rate, i);
            } else {
                keep_last_rate.record(false, 0.0, i);
            }
        }

        (fallback_rate, keep_last_rate)
    }

    /// Render the verdict summary line. Three arms:
    /// - no failure → `"monitor OK"`
    /// - failure with `enforce=true` → terse `"monitor FAILED: N violation(s)"`
    /// - failure with `enforce=false` → advisory `"flagged ... (report-only; ...)"`
    fn summarize(failed: bool, enforce: bool, n_details: usize) -> String {
        if !failed {
            return "monitor OK".into();
        }
        if enforce {
            format!("monitor FAILED: {n_details} violation(s)")
        } else {
            format!(
                "monitor flagged {n_details} violation(s) (report-only; pass `Assert::with_monitor_defaults` to enforce)"
            )
        }
    }

    /// Check whether the monitor samples contain plausible data.
    ///
    /// Returns false when the data looks like uninitialized guest memory:
    /// - All rq_clock values across every CPU in every sample are identical
    ///   (the kernel never wrote to these fields).
    /// - Any local_dsq_depth exceeds a plausibility ceiling (real kernels
    ///   never queue millions of tasks on a single CPU's local DSQ).
    fn data_looks_valid(samples: &[MonitorSample]) -> bool {
        let mut first_clock: Option<u64> = None;
        let mut all_clocks_same = true;

        for sample in samples {
            if !sample_looks_valid(sample) {
                return false;
            }
            for cpu in &sample.cpus {
                match first_clock {
                    None => first_clock = Some(cpu.rq_clock),
                    Some(fc) => {
                        if cpu.rq_clock != fc {
                            all_clocks_same = false;
                        }
                    }
                }
            }
        }

        // If we saw at least 2 clock readings and they were all identical,
        // the data is uninitialized.
        if first_clock.is_some() && all_clocks_same {
            // Check we actually had multiple readings to compare.
            let total_readings: usize = samples.iter().map(|s| s.cpus.len()).sum();
            if total_readings > 1 {
                return false;
            }
        }

        true
    }
}