ktstr 0.4.12

//! Guest physical memory access and monitor sampling loop.
//!
//! [`GuestMem`] wraps host pointers to guest DRAM regions and
//! provides bounds-checked volatile reads and writes for scalar types;
//! `read_bytes` uses `copy_nonoverlapping` for bulk copies. Multi-region
//! NUMA layouts are supported: each read/write resolves the target
//! region via binary search. It also implements 4-level and 5-level
//! x86-64 page table walks and 3-level aarch64 walks (64KB granule)
//! for vmalloc'd addresses.
//!
//! The monitor loop (`monitor_loop`) periodically reads per-CPU
//! runqueue state from guest memory and collects `MonitorSample`s.

use super::btf_offsets::{
    CPU_MAX_IDLE_TYPES, KernelOffsets, SchedDomainOffsets, SchedDomainStatsOffsets,
    SchedstatOffsets, ScxEventOffsets,
};
use super::{
    CpuSnapshot, Kva, MonitorSample, RqSchedstat, SchedDomainSnapshot, SchedDomainStats,
    ScxEventCounters,
};
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::{Duration, Instant};

/// Per-NUMA-node host memory region within a GuestMem.
#[derive(Debug, Clone, Copy)]
pub(crate) struct MemRegion {
    /// Host pointer to the start of this region's mapping.
    pub(crate) host_ptr: *mut u8,
    /// DRAM-relative offset where this region starts.
    pub(crate) offset: u64,
    /// Size in bytes.
    pub(crate) size: u64,
}

/// Host pointer to the start of guest DRAM. Offsets passed to read/write
/// methods are DRAM-relative (x86_64: GPA 0, aarch64: GPA DRAM_START).
///
/// Carries per-NUMA-node region info (one region for single-node,
/// multiple for multi-node topologies). Each read/write resolves
/// the target region via binary search. With contiguous MAP_FIXED VA
/// (the current allocation strategy), the resolved pointer is
/// identical to `base.add(offset)`.
///
/// SAFETY: The pointer is valid for the lifetime of the KVM VM.
/// `ReservationGuard` owns the VA reservation (munmaps on drop);
/// per-node `MmapRegion`s have `owned=false` and do not munmap.
/// The guard outlives all threads that hold a `GuestMem`.
pub struct GuestMem {
    size: u64,
    regions: Vec<MemRegion>,
}

// SAFETY: `MemRegion::host_ptr` values point into KVM mmap'd
// regions whose lifetime is guaranteed by `ReservationGuard`. Reads
// and writes use volatile ops; concurrent access is acceptable
// because the monitor is a best-effort sampler of guest-owned data.
unsafe impl Send for GuestMem {}
unsafe impl Sync for GuestMem {}

impl GuestMem {
    /// Wrap the host-mapped guest DRAM region.
    ///
    /// # Safety
    ///
    /// `base` must point to the start of a valid, readable memory
    /// mapping at least `size` bytes long. The mapping MUST outlive
    /// every `GuestMem` access (the type holds no lifetime tying
    /// itself to the backing allocation). The caller is also
    /// responsible for ensuring concurrent writers do not shrink
    /// the mapping out from under the reader (e.g. via `ftruncate`
    /// on an underlying SHM fd).
    ///
    /// Marked `unsafe` because the raw-pointer contract is not
    /// expressible in the type system — `base: *mut u8` could be
    /// dangling, null, or into unmapped memory, and every subsequent
    /// `read_u64` / `read_slice` would miscompute or SIGSEGV. Every
    /// internal caller (including tests) constructs `base` from a
    /// live allocation whose lifetime is proven at the call site.
    ///
    /// # Memory ordering
    ///
    /// Reads go through `std::ptr::read_volatile` (see
    /// `read_volatile_bytes`), which
    /// disables compiler reordering and caching of the load but
    /// provides no hardware fence. Consequently, `GuestMem` offers
    /// no happens-before relationship with guest-side writes: a
    /// reader may observe torn writes, stale values, or partial
    /// updates from a concurrent guest mutator. Callers that
    /// require atomic snapshot semantics (e.g. double-check a
    /// CRC or re-read to confirm a stable value) must layer that
    /// logic themselves.
    pub unsafe fn new(base: *mut u8, size: u64) -> Self {
        Self {
            size,
            regions: vec![MemRegion {
                host_ptr: base,
                offset: 0,
                size,
            }],
        }
    }

    /// Test-only constructor: build a `GuestMem` from explicit
    /// `MemRegion` entries.
    ///
    /// `regions` must be sorted by ascending `offset` and each region's
    /// host pointer must address a live, readable mapping of at least
    /// `region.size` bytes. The reported `size()` is the largest
    /// `offset + size` across all regions (the DRAM-relative end).
    /// The lifetime of every backing mapping must outlive the
    /// returned `GuestMem`.
    ///
    /// # Safety
    /// Same constraints as [`GuestMem::new`], applied per region.
    #[cfg(test)]
    pub(crate) unsafe fn from_regions_for_test(regions: Vec<MemRegion>) -> Self {
        assert!(!regions.is_empty(), "at least one region required");
        let size = regions
            .iter()
            .map(|r| r.offset + r.size)
            .max()
            .expect("non-empty");
        Self { size, regions }
    }

    /// Build a multi-region GuestMem from a NUMA memory layout.
    ///
    /// Each `NodeRegion` in the layout becomes a `MemRegion` with its
    /// host pointer resolved via `GuestMemoryMmap::get_host_address`.
    /// DRAM-relative offsets are computed by subtracting
    /// `layout.dram_base()` from each region's `gpa_start`.
    ///
    /// # Panics
    ///
    /// Panics if `get_host_address` fails for any region in the
    /// layout. This indicates the `GuestMemoryMmap` was not built
    /// from the same layout — a programming error in the caller.
    pub(crate) fn from_layout(
        layout: &crate::vmm::numa_mem::NumaMemoryLayout,
        guest_mem: &vm_memory::GuestMemoryMmap,
    ) -> Self {
        use vm_memory::GuestMemory;

        let dram_base = layout.dram_base();
        let total_size = layout.total_bytes();
        let mut regions = Vec::with_capacity(layout.regions().len());
        for nr in layout.regions() {
            let host_ptr = guest_mem
                .get_host_address(vm_memory::GuestAddress(nr.gpa_start))
                .unwrap();
            regions.push(MemRegion {
                host_ptr,
                offset: nr.gpa_start - dram_base,
                size: nr.size,
            });
        }

        Self {
            size: total_size,
            regions,
        }
    }

    /// Resolve a DRAM-relative byte offset to a host pointer plus the
    /// number of bytes remaining in the resolved region (i.e. how far
    /// the returned pointer can be advanced before leaving the mmap).
    ///
    /// Binary-searches the sorted region list. Returns `None` if the
    /// offset falls outside all regions.
    fn resolve_ptr(&self, offset: u64) -> Option<(*mut u8, u64)> {
        let idx = self
            .regions
            .partition_point(|r| r.offset <= offset)
            .checked_sub(1)?;
        let r = &self.regions[idx];
        let local = offset - r.offset;
        if local < r.size {
            // SAFETY: `local < r.size` ensures the offset is within
            // the mmap'd region that `host_ptr` points to.
            let ptr = unsafe { r.host_ptr.add(local as usize) };
            Some((ptr, r.size - local))
        } else {
            None
        }
    }

    /// Read `N` volatile bytes from `ptr`. Each byte is read via
    /// `read_volatile` so the compiler cannot cache or elide across
    /// the loop (the guest writes to this memory concurrently and
    /// those writes are invisible to Rust's model). Returning a
    /// `[u8; N]` lets callers recompose the fundamental integer via
    /// `from_ne_bytes` without needing pointer alignment to match.
    ///
    /// Handles the alignment case: even if `ptr` is not aligned for
    /// `T`, per-byte `read_volatile` has 1-byte alignment and is
    /// always safe.
    ///
    /// # Safety
    /// `ptr..ptr+N` must be a valid, readable range in the mapped
    /// guest region. The caller (`read_u32`/`read_u64`/etc.) bounds
    /// checks before resolving the pointer.
    #[inline]
    unsafe fn read_volatile_bytes<const N: usize>(ptr: *const u8) -> [u8; N] {
        let mut bytes = [0u8; N];
        for (i, slot) in bytes.iter_mut().enumerate() {
            // SAFETY: ptr..ptr+N is in-bounds per caller's check.
            *slot = unsafe { std::ptr::read_volatile(ptr.add(i)) };
        }
        bytes
    }

    /// Write `N` volatile bytes to `ptr`. Mirror of
    /// [`read_volatile_bytes`] for the store path.
    ///
    /// # Safety
    /// `ptr..ptr+N` must be a valid, writable range in the mapped
    /// guest region.
    #[inline]
    unsafe fn write_volatile_bytes<const N: usize>(ptr: *mut u8, bytes: [u8; N]) {
        for (i, &byte) in bytes.iter().enumerate() {
            // SAFETY: ptr..ptr+N is in-bounds per caller's check.
            unsafe { std::ptr::write_volatile(ptr.add(i), byte) };
        }
    }

    /// Bounds-checked volatile read of `N` little/native-endian bytes at
    /// DRAM offset `pa + offset`. Returns `[0; N]` if the range falls
    /// outside the mapped region, straddles a region boundary in a
    /// multi-region (NUMA) layout, or if the address arithmetic overflows
    /// (`pa` may be derived from an attacker-controlled guest page-table
    /// entry).
    ///
    /// `N` must match the width of the scalar caller. `read_volatile_bytes`
    /// reads byte-by-byte, so the access does not require `N`-alignment.
    fn read_scalar<const N: usize>(&self, pa: u64, offset: usize) -> [u8; N] {
        let Some(addr) = pa.checked_add(offset as u64) else {
            return [0; N];
        };
        let Some(end) = addr.checked_add(N as u64) else {
            return [0; N];
        };
        if end > self.size {
            return [0; N];
        }
        match self.resolve_ptr(addr) {
            Some((ptr, region_avail)) => {
                // Reject reads that would walk past the end of the
                // resolved region's mmap. Multi-region GuestMems can
                // have non-contiguous host mappings; reading off the
                // end of one region's mmap is undefined behavior even
                // if `addr + N <= self.size` overall.
                if (N as u64) > region_avail {
                    return [0; N];
                }
                // SAFETY: bounds checked above; resolve_ptr returned a
                // valid pointer into the mapped region and the read of
                // N bytes stays within `region_avail`.
                unsafe { Self::read_volatile_bytes::<N>(ptr as *const u8) }
            }
            None => [0; N],
        }
    }

    /// Bounds-checked volatile write of `bytes` at DRAM offset `pa + offset`.
    /// Silently no-ops if the range falls outside the mapped region,
    /// straddles a region boundary in a multi-region (NUMA) layout, or if
    /// the address arithmetic overflows (`pa` may be derived from an
    /// attacker-controlled guest page-table entry).
    fn write_scalar<const N: usize>(&self, pa: u64, offset: usize, bytes: [u8; N]) {
        let Some(addr) = pa.checked_add(offset as u64) else {
            return;
        };
        let Some(end) = addr.checked_add(N as u64) else {
            return;
        };
        if end > self.size {
            return;
        }
        if let Some((ptr, region_avail)) = self.resolve_ptr(addr) {
            // Reject writes that would walk past the end of the
            // resolved region's mmap (see `read_scalar` for the
            // rationale on multi-region GuestMems).
            if (N as u64) > region_avail {
                return;
            }
            // SAFETY: bounds checked above; the write of N bytes stays
            // within `region_avail`.
            unsafe { Self::write_volatile_bytes::<N>(ptr, bytes) };
        }
    }

    /// Read a u8 at DRAM offset `pa + offset`.
    pub fn read_u8(&self, pa: u64, offset: usize) -> u8 {
        u8::from_ne_bytes(self.read_scalar::<1>(pa, offset))
    }

    /// Read a u32 at DRAM offset `pa + offset`.
    pub fn read_u32(&self, pa: u64, offset: usize) -> u32 {
        u32::from_ne_bytes(self.read_scalar::<4>(pa, offset))
    }

    /// Read a u64 at DRAM offset `pa + offset`.
    pub fn read_u64(&self, pa: u64, offset: usize) -> u64 {
        u64::from_ne_bytes(self.read_scalar::<8>(pa, offset))
    }

    /// Read an i64 at DRAM offset `pa + offset`.
    pub fn read_i64(&self, pa: u64, offset: usize) -> i64 {
        self.read_u64(pa, offset) as i64
    }

    /// Write a u8 at DRAM offset `pa + offset`.
    pub fn write_u8(&self, pa: u64, offset: usize, val: u8) {
        self.write_scalar::<1>(pa, offset, val.to_ne_bytes());
    }

    /// Write a u64 at DRAM offset `pa + offset`.
    pub fn write_u64(&self, pa: u64, offset: usize, val: u64) {
        self.write_scalar::<8>(pa, offset, val.to_ne_bytes());
    }

    /// Read `len` bytes from DRAM offset `pa` into `buf`.
    /// Returns the number of bytes actually read (may be less than `len`
    /// if the read would go past the end of guest memory or the end of
    /// the resolved region — multi-region NUMA layouts can have
    /// non-contiguous host mappings, so the copy must not extend past
    /// the region containing `pa`).
    pub fn read_bytes(&self, pa: u64, buf: &mut [u8]) -> usize {
        let len = buf.len() as u64;
        if pa >= self.size {
            return 0;
        }
        let avail = (self.size - pa).min(len) as usize;
        match self.resolve_ptr(pa) {
            Some((ptr, region_avail)) => {
                let copy_len = avail.min(region_avail as usize);
                // SAFETY: `copy_len <= region_avail`, so the read stays
                // within the mmap that `ptr` points into.
                unsafe {
                    std::ptr::copy_nonoverlapping(ptr, buf.as_mut_ptr(), copy_len);
                }
                copy_len
            }
            None => 0,
        }
    }

    /// Test helper — write a u32 at DRAM offset `pa + offset`.
    #[cfg(test)]
    pub fn write_u32(&self, pa: u64, offset: usize, val: u32) {
        self.write_scalar::<4>(pa, offset, val.to_ne_bytes());
    }

    /// Translate a kernel virtual address to guest physical address via
    /// page table walk.
    ///
    /// x86-64: supports 4-level (PGD -> PUD -> PMD -> PTE) and 5-level
    /// (PML5 -> P4D -> PUD -> PMD -> PTE) paging.
    ///
    /// aarch64: 3-level walk with AArch64 translation table descriptors
    /// (64KB granule, 48-bit VA). `l5` is ignored.
    ///
    /// `cr3_pa` is the physical address of the top-level page table.
    /// `l5` selects 5-level paging (x86 LA57); use `resolve_pgtable_l5`
    /// to detect the guest's mode at runtime.
    /// Returns `None` if any level is not present or the address is
    /// out of guest memory bounds.
    pub(crate) fn translate_kva(&self, cr3_pa: u64, kva: Kva, l5: bool) -> Option<u64> {
        #[cfg(target_arch = "x86_64")]
        {
            if l5 {
                self.walk_5level(cr3_pa, kva)
            } else {
                self.walk_4level(cr3_pa, kva)
            }
        }
        #[cfg(target_arch = "aarch64")]
        {
            let _ = l5; // x86-only flag; aarch64 here always uses the 3-level 64KB granule path
            self.walk_3level_aarch64_64k(cr3_pa, kva)
        }
    }

    /// 4-level page table walk (x86-64).
    ///
    /// CR3 -> PGD -> PUD -> PMD -> PTE. Uses PS bit (bit 7) for
    /// huge pages, OA in bits \[51:12\].
    #[cfg(target_arch = "x86_64")]
    fn walk_4level(&self, cr3_pa: u64, kva: Kva) -> Option<u64> {
        const PRESENT: u64 = 1;
        const PS: u64 = 1 << 7;
        const ADDR_MASK: u64 = 0x000F_FFFF_FFFF_F000;

        let kva_bits = kva.0;
        let pgd_idx = (kva_bits >> 39) & 0x1FF;
        let pud_idx = (kva_bits >> 30) & 0x1FF;
        let pmd_idx = (kva_bits >> 21) & 0x1FF;
        let pte_idx = (kva_bits >> 12) & 0x1FF;
        let page_off = kva_bits & 0xFFF;

        // PGD
        let pgd_pa = (cr3_pa & ADDR_MASK) + pgd_idx * 8;
        let pgde = self.read_u64(pgd_pa, 0);
        if pgde & PRESENT == 0 {
            return None;
        }

        // PUD
        let pud_pa = (pgde & ADDR_MASK) + pud_idx * 8;
        let pude = self.read_u64(pud_pa, 0);
        if pude & PRESENT == 0 {
            return None;
        }
        if pude & PS != 0 {
            let base = pude & 0x000F_FFFF_C000_0000;
            return Some(base | (kva_bits & 0x3FFF_FFFF));
        }

        // PMD
        let pmd_pa = (pude & ADDR_MASK) + pmd_idx * 8;
        let pmde = self.read_u64(pmd_pa, 0);
        if pmde & PRESENT == 0 {
            return None;
        }
        if pmde & PS != 0 {
            let base = pmde & 0x000F_FFFF_FFE0_0000;
            return Some(base | (kva_bits & 0x1F_FFFF));
        }

        // PTE
        let pte_pa = (pmde & ADDR_MASK) + pte_idx * 8;
        let ptee = self.read_u64(pte_pa, 0);
        if ptee & PRESENT == 0 {
            return None;
        }

        Some((ptee & ADDR_MASK) | page_off)
    }

    /// aarch64 page table walk (64KB granule, 3-level, 48-bit VA).
    ///
    /// TTBR_EL1 -> PGD -> PMD -> PTE.
    /// With 64KB pages and 48-bit VA, the kernel uses 3 levels:
    ///   PGD: bits [47:42] = 6 bits, 64 entries
    ///   PMD: bits [41:29] = 13 bits, 8192 entries
    ///   PTE: bits [28:16] = 13 bits, 8192 entries
    ///   page offset: bits [15:0] = 16 bits
    ///
    /// Descriptor format (ARMv8 D5.3):
    /// - bits [1:0] = 0b00: invalid
    /// - bits [1:0] = 0b01: block descriptor (PGD/PMD levels)
    /// - bits [1:0] = 0b11: table descriptor (PGD/PMD) or page (PTE)
    /// - bits [47:16]: output address for 64KB granule
    ///
    /// Page table entries contain guest physical addresses (GPAs). Since
    /// GuestMem is mapped at DRAM_START, all GPAs are adjusted by
    /// subtracting DRAM_START to produce offsets into the memory region.
    #[cfg(target_arch = "aarch64")]
    fn walk_3level_aarch64_64k(&self, ttbr_pa: u64, kva: Kva) -> Option<u64> {
        use crate::vmm::kvm::DRAM_START;

        const VALID: u64 = 1;
        // 0b11 means "table descriptor" at PGD/PMD levels and "page
        // descriptor" at the PTE level. Same encoding, role-dependent
        // interpretation per ARMv8-A.
        const TABLE: u64 = 0b11;
        const BLOCK: u64 = 0b01;
        const DESC_MASK: u64 = 0b11;
        // OA mask for 64KB granule: bits [47:16]
        const ADDR_MASK: u64 = 0x0000_FFFF_FFFF_0000;

        // Convert a guest physical address to a DRAM-relative offset.
        // `checked_sub` rejects descriptors whose payload addresses fall
        // below DRAM_START — a malicious or corrupted descriptor that
        // wraps would otherwise produce a near-u64::MAX offset and cause
        // out-of-bounds reads. Treat underflow as "not present" (None).
        let to_offset = |gpa: u64| -> Option<u64> { gpa.checked_sub(DRAM_START) };

        // 3-level walk for 64KB granule, 48-bit VA.
        let kva_bits = kva.0;
        let pgd_idx = (kva_bits >> 42) & 0x3F; // bits [47:42], 6 bits
        let pmd_idx = (kva_bits >> 29) & 0x1FFF; // bits [41:29], 13 bits
        let pte_idx = (kva_bits >> 16) & 0x1FFF; // bits [28:16], 13 bits
        let page_off = kva_bits & 0xFFFF; // bits [15:0], 16 bits

        // PGD — ttbr_pa is already a GuestMem offset.
        let pgd_off = (ttbr_pa & ADDR_MASK) + pgd_idx * 8;
        let pgde = self.read_u64(pgd_off, 0);
        if pgde & VALID == 0 {
            return None;
        }
        // PGD block: 4TB region (unlikely but spec-allowed)
        if pgde & DESC_MASK == BLOCK {
            let base = pgde & 0x0000_FC00_0000_0000;
            return Some(to_offset(base)? | (kva_bits & 0x3FF_FFFF_FFFF));
        }

        // PMD
        let pmd_off = to_offset(pgde & ADDR_MASK)? + pmd_idx * 8;
        let pmde = self.read_u64(pmd_off, 0);
        if pmde & VALID == 0 {
            return None;
        }
        // PMD block: 512MB region
        if pmde & DESC_MASK == BLOCK {
            let base = pmde & 0x0000_FFFF_E000_0000;
            return Some(to_offset(base)? | (kva_bits & 0x1FFF_FFFF));
        }

        // PTE — page descriptor (bits [1:0] = 0b11)
        let pte_off = to_offset(pmde & ADDR_MASK)? + pte_idx * 8;
        let ptee = self.read_u64(pte_off, 0);
        if ptee & VALID == 0 {
            return None;
        }
        if ptee & DESC_MASK != TABLE {
            return None;
        }

        Some(to_offset(ptee & ADDR_MASK)? | page_off)
    }

    /// 5-level page table walk: CR3 -> PML5 -> P4D -> PUD -> PMD -> PTE.
    /// x86-64 only; aarch64 does not use 5-level paging.
    #[cfg(target_arch = "x86_64")]
    fn walk_5level(&self, cr3_pa: u64, kva: Kva) -> Option<u64> {
        const PRESENT: u64 = 1;
        const ADDR_MASK: u64 = 0x000F_FFFF_FFFF_F000;

        // PML5 index: bits 56:48.
        let pml5_idx = (kva.0 >> 48) & 0x1FF;

        let pml5_pa = (cr3_pa & ADDR_MASK) + pml5_idx * 8;
        let pml5e = self.read_u64(pml5_pa, 0);
        if pml5e & PRESENT == 0 {
            return None;
        }

        // P4D is the next level; continue with 4-level walk from there.
        let p4d_pa = pml5e & ADDR_MASK;
        self.walk_4level(p4d_pa, kva)
    }

    /// Guest memory size in bytes.
    pub fn size(&self) -> u64 {
        self.size
    }
}

/// Read scheduler stats from one CPU's struct rq at the given physical address.
///
/// Populates the core rq fields: `nr_running`, `scx_nr_running`,
/// `local_dsq_depth`, `rq_clock`, `scx_flags`. Leaves `event_counters`,
/// `schedstat`, `vcpu_cpu_time_ns`, and `sched_domains` as `None` —
/// those are filled in separately by `read_event_stats`,
/// `read_rq_schedstat`, vCPU stats collection, and sched_domain
/// traversal respectively.
pub(crate) fn read_rq_stats(mem: &GuestMem, rq_pa: u64, offsets: &KernelOffsets) -> CpuSnapshot {
    CpuSnapshot {
        nr_running: mem.read_u32(rq_pa, offsets.rq_nr_running),
        scx_nr_running: mem.read_u32(rq_pa, offsets.rq_scx + offsets.scx_rq_nr_running),
        local_dsq_depth: mem.read_u32(
            rq_pa,
            offsets.rq_scx + offsets.scx_rq_local_dsq + offsets.dsq_nr,
        ),
        rq_clock: mem.read_u64(rq_pa, offsets.rq_clock),
        scx_flags: mem.read_u32(rq_pa, offsets.rq_scx + offsets.scx_rq_flags),
        event_counters: None,
        schedstat: None,
        vcpu_cpu_time_ns: None,
        sched_domains: None,
    }
}

/// Read scx event counters from one CPU's per-CPU event stats struct.
/// On 6.18+ (and 6.17.7+ stable), `pcpu_pa` points to `scx_sched_pcpu`;
/// on 6.16 through 6.17.6, it points directly to `scx_event_stats`
/// (`event_stats_off` = 0).
///
/// Version boundaries are approximate; see [`resolve_event_pcpu_pas`]
/// for the detection logic.
pub(crate) fn read_event_stats(
    mem: &GuestMem,
    pcpu_pa: u64,
    ev: &ScxEventOffsets,
) -> ScxEventCounters {
    let base = pcpu_pa + ev.event_stats_off as u64;
    let read_opt = |off: Option<usize>| off.map(|o| mem.read_i64(base, o)).unwrap_or(0);
    ScxEventCounters {
        select_cpu_fallback: mem.read_i64(base, ev.ev_select_cpu_fallback),
        dispatch_local_dsq_offline: mem.read_i64(base, ev.ev_dispatch_local_dsq_offline),
        dispatch_keep_last: mem.read_i64(base, ev.ev_dispatch_keep_last),
        enq_skip_exiting: mem.read_i64(base, ev.ev_enq_skip_exiting),
        enq_skip_migration_disabled: mem.read_i64(base, ev.ev_enq_skip_migration_disabled),
        reenq_immed: read_opt(ev.ev_reenq_immed),
        reenq_local_repeat: read_opt(ev.ev_reenq_local_repeat),
        refill_slice_dfl: read_opt(ev.ev_refill_slice_dfl),
        bypass_duration: read_opt(ev.ev_bypass_duration),
        bypass_dispatch: read_opt(ev.ev_bypass_dispatch),
        bypass_activate: read_opt(ev.ev_bypass_activate),
        insert_not_owned: read_opt(ev.ev_insert_not_owned),
        sub_bypass_dispatch: read_opt(ev.ev_sub_bypass_dispatch),
    }
}

/// Read schedstat fields from one CPU's struct rq at the given physical address.
///
/// Reads CONFIG_SCHEDSTATS counters: `run_delay` and `pcount` from the
/// embedded `sched_info` substruct, plus `yld_count`, `sched_count`,
/// `sched_goidle`, `ttwu_count`, and `ttwu_local` from the rq itself.
pub(crate) fn read_rq_schedstat(mem: &GuestMem, rq_pa: u64, ss: &SchedstatOffsets) -> RqSchedstat {
    let sched_info_pa = rq_pa + ss.rq_sched_info as u64;
    RqSchedstat {
        run_delay: mem.read_u64(sched_info_pa, ss.sched_info_run_delay),
        pcount: mem.read_u64(sched_info_pa, ss.sched_info_pcount),
        yld_count: mem.read_u32(rq_pa, ss.rq_yld_count),
        sched_count: mem.read_u32(rq_pa, ss.rq_sched_count),
        sched_goidle: mem.read_u32(rq_pa, ss.rq_sched_goidle),
        ttwu_count: mem.read_u32(rq_pa, ss.rq_ttwu_count),
        ttwu_local: mem.read_u32(rq_pa, ss.rq_ttwu_local),
    }
}

/// Read a u32 array of `CPU_MAX_IDLE_TYPES` elements from guest memory.
fn read_u32_array(mem: &GuestMem, pa: u64, base_offset: usize) -> [u32; CPU_MAX_IDLE_TYPES] {
    std::array::from_fn(|i| mem.read_u32(pa, base_offset + i * 4))
}

/// Read CONFIG_SCHEDSTATS fields from one sched_domain.
///
/// Each load-balance counter (`lb_*`) is a per-idle-type array with
/// `CPU_MAX_IDLE_TYPES` u32 elements (indexed by `enum cpu_idle_type`);
/// scalar counters (`alb_*`, `sbe_*`, `sbf_*`, `ttwu_*`) are single
/// u32 fields on the sched_domain itself.
fn read_sd_stats(mem: &GuestMem, sd_pa: u64, so: &SchedDomainStatsOffsets) -> SchedDomainStats {
    SchedDomainStats {
        lb_count: read_u32_array(mem, sd_pa, so.sd_lb_count),
        lb_failed: read_u32_array(mem, sd_pa, so.sd_lb_failed),
        lb_balanced: read_u32_array(mem, sd_pa, so.sd_lb_balanced),
        lb_imbalance_load: read_u32_array(mem, sd_pa, so.sd_lb_imbalance_load),
        lb_imbalance_util: read_u32_array(mem, sd_pa, so.sd_lb_imbalance_util),
        lb_imbalance_task: read_u32_array(mem, sd_pa, so.sd_lb_imbalance_task),
        lb_imbalance_misfit: read_u32_array(mem, sd_pa, so.sd_lb_imbalance_misfit),
        lb_gained: read_u32_array(mem, sd_pa, so.sd_lb_gained),
        lb_hot_gained: read_u32_array(mem, sd_pa, so.sd_lb_hot_gained),
        lb_nobusyg: read_u32_array(mem, sd_pa, so.sd_lb_nobusyg),
        lb_nobusyq: read_u32_array(mem, sd_pa, so.sd_lb_nobusyq),
        alb_count: mem.read_u32(sd_pa, so.sd_alb_count),
        alb_failed: mem.read_u32(sd_pa, so.sd_alb_failed),
        alb_pushed: mem.read_u32(sd_pa, so.sd_alb_pushed),
        sbe_count: mem.read_u32(sd_pa, so.sd_sbe_count),
        sbe_balanced: mem.read_u32(sd_pa, so.sd_sbe_balanced),
        sbe_pushed: mem.read_u32(sd_pa, so.sd_sbe_pushed),
        sbf_count: mem.read_u32(sd_pa, so.sd_sbf_count),
        sbf_balanced: mem.read_u32(sd_pa, so.sd_sbf_balanced),
        sbf_pushed: mem.read_u32(sd_pa, so.sd_sbf_pushed),
        ttwu_wake_remote: mem.read_u32(sd_pa, so.sd_ttwu_wake_remote),
        ttwu_move_affine: mem.read_u32(sd_pa, so.sd_ttwu_move_affine),
        ttwu_move_balance: mem.read_u32(sd_pa, so.sd_ttwu_move_balance),
    }
}

/// Read the `sd->name` string from guest memory.
///
/// `sd->name` is a `char *` pointer to a static string in kernel rodata.
/// Rodata lives in the text mapping (`__START_KERNEL_map`), so
/// `text_kva_to_pa` is tried first. Falls back to direct mapping
/// (`kva_to_pa`) for kernels that place topology name strings
/// differently. Returns an empty string if the pointer is null or
/// translation fails.
fn read_sd_name(mem: &GuestMem, sd_pa: u64, name_offset: usize, page_offset: u64) -> String {
    let name_kva = mem.read_u64(sd_pa, name_offset);
    if name_kva == 0 {
        return String::new();
    }
    // Try text mapping first (rodata), then direct mapping.
    let text_pa = super::symbols::text_kva_to_pa(name_kva);
    let name_pa = if text_pa < mem.size() {
        text_pa
    } else {
        let direct_pa = super::symbols::kva_to_pa(name_kva, page_offset);
        if direct_pa >= mem.size() {
            return String::new();
        }
        direct_pa
    };
    // Domain names are short static strings ("SMT", "MC", "DIE", "NUMA",
    // "PKG", "BOOK", "DRAWER"). Read up to 16 bytes.
    let mut buf = [0u8; 16];
    let n = mem.read_bytes(name_pa, &mut buf);
    let end = buf[..n].iter().position(|&b| b == 0).unwrap_or(n);
    String::from_utf8_lossy(&buf[..end]).into_owned()
}

/// Read the sched_domain tree for one CPU.
///
/// Starts at `rq->sd` (the lowest-level domain), walks `sd->parent`
/// until NULL. Each domain is kmalloc'd and lives in the direct mapping.
///
/// `page_offset` is the runtime `PAGE_OFFSET` for direct-mapping translation.
///
/// Returns `None` if `rq->sd` is null (domain not yet built, or CPU
/// offline). Returns an empty `Vec` if the first domain pointer cannot
/// be translated.
///
/// Maximum depth is bounded to 8 levels and a visited-set of domain
/// KVAs breaks `sd->parent` cycles — a corrupted or self-referential
/// chain would otherwise emit the same domain up to MAX_DEPTH times.
pub(crate) fn read_sched_domain_tree(
    mem: &GuestMem,
    rq_pa: u64,
    sd_offsets: &SchedDomainOffsets,
    page_offset: u64,
) -> Option<Vec<SchedDomainSnapshot>> {
    const MAX_DEPTH: usize = 8;

    // rq->sd is a pointer (KVA).
    let sd_kva = mem.read_u64(rq_pa, sd_offsets.rq_sd);
    if sd_kva == 0 {
        return None;
    }

    let mut domains = Vec::new();
    let mut current_kva = sd_kva;
    let mut visited: std::collections::HashSet<u64> = std::collections::HashSet::new();

    for _ in 0..MAX_DEPTH {
        if current_kva == 0 {
            break;
        }
        if !visited.insert(current_kva) {
            // Cycle or self-reference: same KVA already emitted. Stop
            // so we do not inflate the tree with duplicate snapshots.
            tracing::warn!(
                sd_kva = format_args!("{current_kva:#x}"),
                "sched_domain cycle detected; truncating tree"
            );
            break;
        }

        // sched_domain is kmalloc'd — lives in direct mapping.
        let sd_pa = super::symbols::kva_to_pa(current_kva, page_offset);
        if sd_pa >= mem.size() {
            break;
        }

        let level = mem.read_u32(sd_pa, sd_offsets.sd_level) as i32;
        let name = read_sd_name(mem, sd_pa, sd_offsets.sd_name, page_offset);
        let flags = mem.read_u32(sd_pa, sd_offsets.sd_flags) as i32;
        let span_weight = mem.read_u32(sd_pa, sd_offsets.sd_span_weight);

        let stats = sd_offsets
            .stats_offsets
            .as_ref()
            .map(|so| read_sd_stats(mem, sd_pa, so));

        let snap = SchedDomainSnapshot {
            level,
            name,
            flags,
            span_weight,
            balance_interval: mem.read_u32(sd_pa, sd_offsets.sd_balance_interval),
            nr_balance_failed: mem.read_u32(sd_pa, sd_offsets.sd_nr_balance_failed),
            newidle_call: sd_offsets
                .sd_newidle_call
                .map(|off| mem.read_u32(sd_pa, off)),
            newidle_success: sd_offsets
                .sd_newidle_success
                .map(|off| mem.read_u32(sd_pa, off)),
            newidle_ratio: sd_offsets
                .sd_newidle_ratio
                .map(|off| mem.read_u32(sd_pa, off)),
            max_newidle_lb_cost: mem.read_u64(sd_pa, sd_offsets.sd_max_newidle_lb_cost),
            stats,
        };

        domains.push(snap);

        // Follow sd->parent.
        current_kva = mem.read_u64(sd_pa, sd_offsets.sd_parent);
    }

    Some(domains)
}

/// Resolve per-CPU physical addresses for event counter reads.
///
/// Reads `*scx_root` to find the active `scx_sched` struct, then reads
/// the percpu pointer at `percpu_ptr_off` within it. On 6.18+ (and
/// 6.17.7+ stable) this is `scx_sched.pcpu` (pointing to `scx_sched_pcpu`);
/// on 6.16 through 6.17.6 it is `scx_sched.event_stats_cpu` (pointing
/// directly to `scx_event_stats`). Computes each CPU's PA via
/// `__per_cpu_offset`.
///
/// Returns None if `scx_root` is null (no scheduler loaded).
pub(crate) fn resolve_event_pcpu_pas(
    mem: &GuestMem,
    scx_root_pa: u64,
    ev: &ScxEventOffsets,
    per_cpu_offsets: &[u64],
    page_offset: u64,
) -> Option<Vec<u64>> {
    let scx_sched_kva = mem.read_u64(scx_root_pa, 0);
    if scx_sched_kva == 0 {
        return None;
    }

    let scx_sched_pa = super::symbols::kva_to_pa(scx_sched_kva, page_offset);
    let pcpu_kva = mem.read_u64(scx_sched_pa, ev.percpu_ptr_off);
    if pcpu_kva == 0 {
        return None;
    }

    let pas: Vec<u64> = per_cpu_offsets
        .iter()
        .map(|&cpu_off| super::symbols::kva_to_pa(pcpu_kva.wrapping_add(cpu_off), page_offset))
        .collect();

    Some(pas)
}

/// Per-vCPU host thread timing info for gating stall detection.
///
/// When the host is loaded, vCPU threads get preempted and rq_clock
/// cannot advance. Reading per-thread CPU time distinguishes real
/// stalls (vCPU running but clock stuck) from host preemption
/// (vCPU not scheduled, clock can't advance).
pub(crate) struct VcpuTiming {
    /// pthread_t handles for each vCPU, indexed by vCPU ID.
    /// Used with `pthread_getcpuclockid()` + `clock_gettime()`.
    pub pthreads: Vec<libc::pthread_t>,
}

impl VcpuTiming {
    /// Read CPU time for each vCPU thread. Returns `Some(ns)` per vCPU
    /// on success, `None` when the per-thread clock could not be read.
    ///
    /// `None` propagates through `CpuSnapshot::vcpu_cpu_time_ns`.
    /// Downstream stall detection (`evaluate_preempted`) treats a
    /// `None` on either side of a pair as `preempted=false` — the
    /// stall check falls through to `rq_clock` comparison and fires
    /// if progress isn't observed there. This deliberately prefers
    /// spurious alerts (better visibility) over missed stalls (silent
    /// failure) when clock reads are unavailable. The previous bug
    /// did the opposite: a silent `0` collided with `saturating_sub`
    /// to fabricate "no delta", which looked like preemption and
    /// suppressed every stall after the first clock-read failure.
    ///
    /// Emits a one-shot `tracing::warn` per vCPU (debounced via
    /// `reported_err`) naming the failing syscall + errno so a user
    /// can diagnose why stall gating has degraded to "no data".
    fn read_cpu_times(&self, reported_err: &mut [bool]) -> Vec<Option<u64>> {
        self.pthreads
            .iter()
            .enumerate()
            .map(|(vcpu, &pt)| {
                let mut clk: libc::clockid_t = 0;
                let ret = unsafe { libc::pthread_getcpuclockid(pt, &mut clk) };
                if ret != 0 {
                    if let Some(slot) = reported_err.get_mut(vcpu)
                        && !*slot
                    {
                        tracing::warn!(
                            vcpu,
                            ret,
                            errno = std::io::Error::last_os_error().raw_os_error(),
                            "pthread_getcpuclockid failed; stall gating unavailable for this vCPU"
                        );
                        *slot = true;
                    }
                    return None;
                }
                let mut ts = libc::timespec {
                    tv_sec: 0,
                    tv_nsec: 0,
                };
                let ret = unsafe { libc::clock_gettime(clk, &mut ts) };
                if ret != 0 {
                    if let Some(slot) = reported_err.get_mut(vcpu)
                        && !*slot
                    {
                        tracing::warn!(
                            vcpu,
                            ret,
                            errno = std::io::Error::last_os_error().raw_os_error(),
                            "clock_gettime on pthread clock failed; stall gating unavailable for this vCPU"
                        );
                        *slot = true;
                    }
                    return None;
                }
                // CPU-time pthread clocks are cumulative nanoseconds
                // and thus always non-negative; guard anyway so a
                // negative tv_sec or tv_nsec from a hypothetical clock
                // bug doesn't silently wrap through `as u64`.
                if ts.tv_sec < 0 || ts.tv_nsec < 0 {
                    if let Some(slot) = reported_err.get_mut(vcpu)
                        && !*slot
                    {
                        tracing::warn!(
                            vcpu,
                            tv_sec = ts.tv_sec,
                            tv_nsec = ts.tv_nsec,
                            "negative clock_gettime result; stall gating unavailable for this vCPU"
                        );
                        *slot = true;
                    }
                    return None;
                }
                // Re-arm the error latch on a successful read so a
                // transient failure doesn't permanently mute the log.
                if let Some(slot) = reported_err.get_mut(vcpu) {
                    *slot = false;
                }
                Some(ts.tv_sec as u64 * 1_000_000_000 + ts.tv_nsec as u64)
            })
            .collect()
    }
}

/// Decide whether a vCPU was preempted between two consecutive samples.
///
/// Returns `true` only when BOTH samples produced a valid reading
/// (`Some`) and the delta falls strictly below `threshold_ns`. Any
/// missing reading (`None` on either side) is treated as "no data" and
/// returns `false` — the stall path must NEVER infer preemption from
/// absent data, otherwise a clock-read failure would silently suppress
/// every subsequent stall (the original bug).
///
/// Uses `saturating_sub` to tolerate non-monotonic reads across clock
/// resolution edges; a non-monotonic sample yields delta=0, which is
/// below any positive threshold, so `preempted=true` — matching the
/// semantics of "the vCPU made no measurable progress".
pub(crate) fn evaluate_preempted(prev: Option<u64>, curr: Option<u64>, threshold_ns: u64) -> bool {
    match (prev, curr) {
        (Some(p), Some(c)) => c.saturating_sub(p) < threshold_ns,
        _ => false,
    }
}

/// Decide whether a CPU stalled between two consecutive samples.
///
/// A stall means the scheduler made no progress on this CPU: `rq_clock`
/// did not advance AND the CPU was not legitimately quiescent. The two
/// legitimate exemptions — NOHZ idle (both samples show `nr_running==0`)
/// and vCPU preemption (host scheduled the vCPU thread off-CPU, so the
/// vCPU couldn't tick the clock) — are recognized here so callers don't
/// re-derive the predicate.
///
/// This helper exists to keep the post-hoc `MonitorSummary::from_samples`
/// path and the reactive `MonitorThresholds::evaluate` path in lock-step:
/// previously each site re-implemented the same four-condition conjunction
/// and drifting one half would let the SysRq-D trigger fire on conditions
/// the post-hoc verdict accepted (or vice versa). Both callers now agree
/// on a single definition of "stall" by construction.
///
/// `rq_clock == 0` is treated as "never sampled" and returns false —
/// the first sample interval typically reads zero before the kernel
/// writes rq_clock, and a zero-to-zero comparison must not fire a stall.
pub(crate) fn is_cpu_stalled(
    prev: &super::CpuSnapshot,
    curr: &super::CpuSnapshot,
    preemption_threshold_ns: u64,
) -> bool {
    if curr.rq_clock == 0 || curr.rq_clock != prev.rq_clock {
        return false;
    }
    let idle = curr.nr_running == 0 && prev.nr_running == 0;
    if idle {
        return false;
    }
    let preempted = evaluate_preempted(
        prev.vcpu_cpu_time_ns,
        curr.vcpu_cpu_time_ns,
        preemption_threshold_ns,
    );
    !preempted
}

/// Configuration for reactive SysRq-D dump triggering.
///
/// When provided to `monitor_loop`, the monitor evaluates thresholds inline
/// and writes the dump request flag to guest SHM on sustained violation.
pub(crate) struct DumpTrigger {
    /// Physical address of the SHM region base in guest memory.
    pub shm_base_pa: u64,
    /// Thresholds for violation detection.
    pub thresholds: super::MonitorThresholds,
}

/// Override for the scheduler watchdog timeout, written every monitor
/// iteration.
///
/// Two write paths are supported:
/// - 7.1+ (`ScxSched`): deref `*scx_root` to find the runtime
///   `scx_sched` struct, then write at the BTF-resolved offset.
///   Re-derefs each iteration because `scx_sched` is reallocated on
///   scheduler (re)load.
/// - pre-7.1 (`StaticGlobal`): write directly to the PA of the
///   `scx_watchdog_timeout` static global. No deref needed — the
///   address is fixed for the kernel's lifetime.
pub(crate) enum WatchdogOverride {
    /// 7.1+ path: deref `scx_root` -> `scx_sched` -> write at offset.
    ScxSched {
        /// PA of the `scx_root` global pointer (text mapping).
        scx_root_pa: u64,
        /// Byte offset of `watchdog_timeout` within `struct scx_sched`.
        watchdog_offset: usize,
        /// Jiffies value to write.
        jiffies: u64,
        /// Runtime `PAGE_OFFSET` for KVA-to-PA translation.
        page_offset: u64,
    },
    /// Pre-7.1 path: write directly to the static global's PA.
    StaticGlobal {
        /// PA of the `scx_watchdog_timeout` static global (text mapping).
        watchdog_timeout_pa: u64,
        /// Jiffies value to write.
        jiffies: u64,
    },
}

/// Pre-resolved BPF program stats context for the monitor loop.
pub(crate) struct ProgStatsCtx {
    /// Cached per-program info (name + stats_percpu_kva) resolved
    /// once at startup to avoid re-walking `prog_idr` each sample.
    pub cached: Vec<super::bpf_prog::CachedProgInfo>,
    /// Per-CPU offset table (`__per_cpu_offset[]`) used to translate
    /// each program's percpu stats pointer into a concrete KVA.
    pub per_cpu_offsets: Vec<u64>,
    /// Runtime `PAGE_OFFSET` used for direct-mapping KVA translation.
    pub page_offset: u64,
    /// BTF offsets for the `bpf_prog` + related struct fields read
    /// while summing stats.
    pub offsets: super::btf_offsets::BpfProgOffsets,
}

/// Samples, SHM drain, and optional watchdog observation returned by
/// [`monitor_loop`].
pub(crate) struct MonitorLoopResult {
    /// Per-interval `MonitorSample`s collected across the run.
    pub(crate) samples: Vec<MonitorSample>,
    /// Final drain of the guest-to-host SHM ring (stimulus events,
    /// test results, any residual messages).
    pub(crate) drain: crate::vmm::shm_ring::ShmDrainResult,
    /// Watchdog read-back, when a watchdog override was installed.
    pub(crate) watchdog_observation: Option<super::WatchdogObservation>,
}

/// Configuration for the monitor sampling loop.
///
/// Bundles the parameters that `monitor_loop` needs beyond the
/// required `mem`, `rq_pas`, `offsets`, `interval`, `kill`, and `run_start`.
pub(crate) struct MonitorConfig<'a> {
    /// Per-CPU physical addresses of `scx_sched_pcpu`. When present (and
    /// `event_offsets` exist), each sample includes event counters.
    pub event_pcpu_pas: Option<&'a [u64]>,
    /// Reactive dump configuration. When a sustained threshold violation is
    /// detected, writes the dump request flag to guest SHM to trigger a
    /// SysRq-D dump inside the guest.
    pub dump_trigger: Option<&'a DumpTrigger>,
    /// Optional watchdog timeout override to install before sampling
    /// begins; read back into `WatchdogObservation` after the loop.
    pub watchdog_override: Option<&'a WatchdogOverride>,
    /// Optional per-vCPU timing context for preemption accounting.
    pub vcpu_timing: Option<&'a VcpuTiming>,
    /// Preemption threshold in nanoseconds used for stall detection.
    /// Pass 0 to derive it from the guest kernel's CONFIG_HZ.
    pub preemption_threshold_ns: u64,
    /// Physical address of the guest-side SHM ring region; enables
    /// mid-flight drain when present.
    pub shm_base_pa: Option<u64>,
    /// Optional BPF program statistics context; when present, each
    /// sample includes per-program exec counters.
    pub prog_stats_ctx: Option<&'a ProgStatsCtx>,
    /// Runtime `PAGE_OFFSET` for direct-mapping KVA translation. Used by
    /// sched_domain tree walking to translate `rq->sd` and `sd->parent`
    /// pointers.
    pub page_offset: u64,
}

/// Run the monitor loop, sampling all CPUs at the given interval
/// until `kill` is set. Returns a [`MonitorLoopResult`] containing
/// the collected per-interval samples, a final drain of the
/// guest-to-host SHM ring, and — when a watchdog override was
/// installed — the post-run `WatchdogObservation` read-back.
pub(crate) fn monitor_loop(
    mem: &GuestMem,
    rq_pas: &[u64],
    offsets: &KernelOffsets,
    interval: Duration,
    kill: &AtomicBool,
    run_start: Instant,
    cfg: &MonitorConfig<'_>,
) -> MonitorLoopResult {
    let event_pcpu_pas = cfg.event_pcpu_pas;
    let dump_trigger = cfg.dump_trigger;
    let watchdog_override = cfg.watchdog_override;
    let vcpu_timing = cfg.vcpu_timing;
    let preemption_threshold_ns = cfg.preemption_threshold_ns;
    let shm_base_pa = cfg.shm_base_pa;
    let prog_stats_ctx = cfg.prog_stats_ctx;
    let page_offset = cfg.page_offset;
    let preemption_threshold_ns = if preemption_threshold_ns > 0 {
        preemption_threshold_ns
    } else {
        super::vcpu_preemption_threshold_ns(None)
    };
    let mut samples: Vec<MonitorSample> = Vec::new();
    // Reactive threshold trackers — reuse the post-hoc
    // `SustainedViolationTracker` so "sustained for N samples"
    // means the same thing to the reactive SysRq-D dump and to
    // `MonitorThresholds::evaluate` running over the full sample vec.
    let mut imbalance_tracker = super::SustainedViolationTracker::default();
    let mut dsq_tracker = super::SustainedViolationTracker::default();
    let mut stall_trackers: Vec<super::SustainedViolationTracker> =
        vec![super::SustainedViolationTracker::default(); rq_pas.len()];
    let mut dump_requested = false;
    let mut cpus: Vec<CpuSnapshot> = Vec::with_capacity(rq_pas.len());
    let mut vcpu_timing_err_reported: Vec<bool> = vcpu_timing
        .map(|vt| vec![false; vt.pthreads.len()])
        .unwrap_or_default();
    let mut shm_entries: Vec<crate::vmm::shm_ring::ShmEntry> = Vec::new();
    let mut shm_drops: u64 = 0;
    let mut watchdog_observation: Option<super::WatchdogObservation> = None;

    loop {
        if kill.load(Ordering::Acquire) {
            break;
        }
        if let Some(wd) = watchdog_override {
            let (write_pa, write_offset, wd_jiffies) = match wd {
                WatchdogOverride::ScxSched {
                    scx_root_pa,
                    watchdog_offset,
                    jiffies,
                    page_offset,
                } => {
                    let sch_kva = mem.read_u64(*scx_root_pa, 0);
                    if sch_kva == 0 {
                        (None, 0, *jiffies)
                    } else {
                        let sch_pa = super::symbols::kva_to_pa(sch_kva, *page_offset);
                        (Some(sch_pa), *watchdog_offset, *jiffies)
                    }
                }
                WatchdogOverride::StaticGlobal {
                    watchdog_timeout_pa,
                    jiffies,
                } => (Some(*watchdog_timeout_pa), 0, *jiffies),
            };
            if let Some(pa) = write_pa {
                mem.write_u64(pa, write_offset, wd_jiffies);
                if watchdog_observation.is_none() {
                    let observed = mem.read_u64(pa, write_offset);
                    watchdog_observation = Some(super::WatchdogObservation {
                        expected_jiffies: wd_jiffies,
                        observed_jiffies: observed,
                    });
                }
            }
        }
        cpus.clear();
        cpus.extend(rq_pas.iter().map(|&pa| read_rq_stats(mem, pa, offsets)));

        // Overlay event counters if available.
        if let (Some(pcpu_pas), Some(ev)) = (event_pcpu_pas, &offsets.event_offsets) {
            for (i, cpu) in cpus.iter_mut().enumerate() {
                if let Some(&pcpu_pa) = pcpu_pas.get(i) {
                    cpu.event_counters = Some(read_event_stats(mem, pcpu_pa, ev));
                }
            }
        }

        // Overlay schedstat fields if available.
        if let Some(ss) = &offsets.schedstat_offsets {
            for (i, cpu) in cpus.iter_mut().enumerate() {
                if let Some(&rq_pa) = rq_pas.get(i) {
                    cpu.schedstat = Some(read_rq_schedstat(mem, rq_pa, ss));
                }
            }
        }

        // Overlay sched domain tree if available.
        if let Some(sd) = &offsets.sched_domain_offsets {
            for (i, cpu) in cpus.iter_mut().enumerate() {
                if let Some(&rq_pa) = rq_pas.get(i) {
                    cpu.sched_domains = read_sched_domain_tree(mem, rq_pa, sd, page_offset);
                }
            }
        }

        // Stamp vCPU CPU times into the per-CPU snapshots. Reactive
        // stall detection below reads these via `is_cpu_stalled`; the
        // post-hoc `MonitorThresholds::evaluate` path reads them off
        // the pushed samples.
        if let Some(vt) = vcpu_timing {
            let times = vt.read_cpu_times(&mut vcpu_timing_err_reported);
            for (i, cpu) in cpus.iter_mut().enumerate() {
                if let Some(&t) = times.get(i) {
                    cpu.vcpu_cpu_time_ns = t;
                }
            }
        }

        // Inline threshold evaluation for reactive dump. Each check
        // mirrors `MonitorThresholds::evaluate`: the same
        // `SustainedViolationTracker`, the same `is_cpu_stalled`
        // predicate, the same `imbalance_ratio`/`local_dsq_depth`
        // reads. Any drift would let the reactive SysRq-D trigger
        // fire on conditions the post-hoc verdict accepts (or vice
        // versa).
        if let Some(trigger) = dump_trigger
            && !dump_requested
            && !cpus.is_empty()
        {
            let t = &trigger.thresholds;
            let sample_idx = samples.len();

            // Imbalance check — use the shared sample method so the
            // min_nr.max(1)/max_nr calculation matches post-hoc.
            let tmp_sample = MonitorSample {
                elapsed_ms: 0,
                cpus: cpus.clone(),
                prog_stats: None,
            };
            let ratio = tmp_sample.imbalance_ratio();
            imbalance_tracker.record(ratio > t.max_imbalance_ratio, ratio, sample_idx);

            // DSQ depth check.
            let worst_dsq = cpus.iter().map(|c| c.local_dsq_depth).max().unwrap_or(0);
            dsq_tracker.record(
                worst_dsq > t.max_local_dsq_depth,
                worst_dsq as f64,
                sample_idx,
            );

            // Stall check: per-CPU sustained window. Delegate to
            // `is_cpu_stalled` so reactive and post-hoc stall paths
            // cannot drift — the predicate owns the idle + preempted
            // exemptions. `vcpu_cpu_time_ns` is already stamped into
            // `cpus[i]` (and into the last pushed sample) above, so the
            // helper sees the same vCPU timing the post-hoc path sees.
            if t.fail_on_stall
                && let Some(prev) = samples.last()
            {
                let n = prev.cpus.len().min(cpus.len()).min(stall_trackers.len());
                for i in 0..n {
                    let is_stall = is_cpu_stalled(&prev.cpus[i], &cpus[i], preemption_threshold_ns);
                    stall_trackers[i].record(is_stall, cpus[i].rq_clock as f64, sample_idx);
                }
            }
            let sustained = imbalance_tracker.sustained(t.sustained_samples)
                || dsq_tracker.sustained(t.sustained_samples)
                || stall_trackers
                    .iter()
                    .any(|s| s.sustained(t.sustained_samples));

            if sustained {
                mem.write_u8(
                    trigger.shm_base_pa,
                    crate::vmm::shm_ring::DUMP_REQ_OFFSET,
                    crate::vmm::shm_ring::DUMP_REQ_SYSRQ_D,
                );
                dump_requested = true;
            }
        }

        let prog_stats = prog_stats_ctx.map(|ctx| {
            super::bpf_prog::read_prog_runtime_stats(
                mem,
                &ctx.cached,
                &ctx.per_cpu_offsets,
                ctx.page_offset,
                &ctx.offsets,
            )
        });

        samples.push(MonitorSample {
            elapsed_ms: run_start.elapsed().as_millis() as u64,
            cpus: cpus.clone(),
            prog_stats,
        });

        // Mid-flight SHM drain: advance read_ptr so the guest can
        // reclaim ring space. Accumulate drained entries for the
        // caller to merge with the post-mortem drain.
        if let Some(shm_pa) = shm_base_pa {
            let drain = crate::vmm::shm_ring::shm_drain_live(mem, shm_pa);
            shm_drops = shm_drops.max(drain.drops);
            // Check for scheduler death signal before accumulating.
            // The guest init writes MSG_TYPE_SCHED_EXIT when the
            // scheduler process exits during test execution.
            if drain
                .entries
                .iter()
                .any(|e| e.msg_type == crate::vmm::shm_ring::MSG_TYPE_SCHED_EXIT && e.crc_ok)
            {
                shm_entries.extend(drain.entries);
                kill.store(true, Ordering::Release);
                break;
            }
            shm_entries.extend(drain.entries);
        }

        std::thread::sleep(interval);
    }
    let shm_result = crate::vmm::shm_ring::ShmDrainResult {
        entries: shm_entries,
        drops: shm_drops,
    };
    MonitorLoopResult {
        samples,
        drain: shm_result,
        watchdog_observation,
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::os::unix::thread::JoinHandleExt;

    const THRESHOLD_NS: u64 = 10_000_000;

    #[test]
    fn evaluate_preempted_both_none_is_not_preempted() {
        assert!(!evaluate_preempted(None, None, THRESHOLD_NS));
    }

    #[test]
    fn evaluate_preempted_first_read_failed_is_not_preempted() {
        // Prev missing: we have no baseline, so the stall path must
        // fire if other conditions match. Treating this as preempted
        // would mask every first-sample stall.
        assert!(!evaluate_preempted(None, Some(1_000_000_000), THRESHOLD_NS));
    }

    #[test]
    fn evaluate_preempted_current_read_failed_is_not_preempted() {
        // Curr missing: likewise no evidence of preemption. The bug
        // this replaces would have returned `saturating_sub(big, 0)
        // < threshold` = false, or `saturating_sub(0, big)` = 0 <
        // threshold = true — both wrong.
        assert!(!evaluate_preempted(Some(1_000_000_000), None, THRESHOLD_NS));
    }

    #[test]
    fn evaluate_preempted_delta_below_threshold_is_preempted() {
        // 1ms delta with 10ms threshold: vCPU barely ran — preempted.
        assert!(evaluate_preempted(
            Some(1_000_000_000),
            Some(1_001_000_000),
            THRESHOLD_NS,
        ));
    }

    #[test]
    fn evaluate_preempted_delta_at_threshold_is_not_preempted() {
        // Exactly 10ms delta: not below threshold, so running, not preempted.
        assert!(!evaluate_preempted(
            Some(1_000_000_000),
            Some(1_010_000_000),
            THRESHOLD_NS,
        ));
    }

    #[test]
    fn evaluate_preempted_delta_above_threshold_is_not_preempted() {
        assert!(!evaluate_preempted(
            Some(1_000_000_000),
            Some(2_000_000_000),
            THRESHOLD_NS,
        ));
    }

    #[test]
    fn evaluate_preempted_non_monotonic_treated_as_no_progress() {
        // saturating_sub of a reverse-going clock yields 0 < threshold,
        // so "no measurable progress" maps to preempted=true. Documented
        // invariant: never unwinds into a false not-preempted when the
        // clock read jitters backwards.
        assert!(evaluate_preempted(
            Some(1_000_000_000),
            Some(999_000_000),
            THRESHOLD_NS,
        ));
    }

    #[test]
    fn evaluate_preempted_zero_threshold_never_preempted() {
        // Degenerate case: with threshold=0, nothing is strictly below,
        // so preempted is always false (stall path always fires).
        assert!(!evaluate_preempted(Some(100), Some(100), 0));
        assert!(!evaluate_preempted(Some(100), Some(200), 0));
    }

    fn test_config() -> MonitorConfig<'static> {
        MonitorConfig {
            event_pcpu_pas: None,
            dump_trigger: None,
            watchdog_override: None,
            vcpu_timing: None,
            preemption_threshold_ns: 0,
            shm_base_pa: None,
            prog_stats_ctx: None,
            page_offset: 0,
        }
    }

    fn test_offsets() -> KernelOffsets {
        KernelOffsets {
            rq_nr_running: 8,
            rq_clock: 16,
            rq_scx: 100,
            scx_rq_nr_running: 4,
            scx_rq_local_dsq: 20,
            scx_rq_flags: 8,
            dsq_nr: 0,
            event_offsets: None,
            schedstat_offsets: None,
            sched_domain_offsets: None,
            watchdog_offsets: None,
        }
    }

    /// Build a byte buffer simulating a struct rq with the given field values.
    fn make_rq_buffer(
        offsets: &KernelOffsets,
        nr_running: u32,
        scx_nr: u32,
        dsq_nr: u32,
        clock: u64,
        flags: u32,
    ) -> Vec<u8> {
        let size = offsets.rq_scx + offsets.scx_rq_local_dsq + offsets.dsq_nr + 8;
        let mut buf = vec![0u8; size];

        buf[offsets.rq_nr_running..offsets.rq_nr_running + 4]
            .copy_from_slice(&nr_running.to_ne_bytes());
        buf[offsets.rq_clock..offsets.rq_clock + 8].copy_from_slice(&clock.to_ne_bytes());

        let scx_base = offsets.rq_scx;
        buf[scx_base + offsets.scx_rq_nr_running..scx_base + offsets.scx_rq_nr_running + 4]
            .copy_from_slice(&scx_nr.to_ne_bytes());
        buf[scx_base + offsets.scx_rq_flags..scx_base + offsets.scx_rq_flags + 4]
            .copy_from_slice(&flags.to_ne_bytes());

        let dsq_base = scx_base + offsets.scx_rq_local_dsq;
        buf[dsq_base + offsets.dsq_nr..dsq_base + offsets.dsq_nr + 4]
            .copy_from_slice(&dsq_nr.to_ne_bytes());
        buf
    }

    #[test]
    fn read_rq_stats_known_values() {
        let offsets = test_offsets();
        let buf = make_rq_buffer(&offsets, 5, 3, 7, 999_000, 0x1);
        // SAFETY: buf is a live local buffer (Vec<u8> or stack array)
        // whose backing storage outlives the GuestMem use.
        let mem = unsafe { GuestMem::new(buf.as_ptr() as *mut u8, buf.len() as u64) };
        let snap = read_rq_stats(&mem, 0, &offsets);
        assert_eq!(snap.nr_running, 5);
        assert_eq!(snap.scx_nr_running, 3);
        assert_eq!(snap.local_dsq_depth, 7);
        assert_eq!(snap.rq_clock, 999_000);
        assert_eq!(snap.scx_flags, 0x1);
    }

    #[test]
    fn read_rq_stats_all_zeros() {
        let offsets = test_offsets();
        let buf = make_rq_buffer(&offsets, 0, 0, 0, 0, 0);
        // SAFETY: buf is a live local buffer (Vec<u8> or stack array)
        // whose backing storage outlives the GuestMem use.
        let mem = unsafe { GuestMem::new(buf.as_ptr() as *mut u8, buf.len() as u64) };
        let snap = read_rq_stats(&mem, 0, &offsets);
        assert_eq!(snap.nr_running, 0);
        assert_eq!(snap.scx_nr_running, 0);
        assert_eq!(snap.local_dsq_depth, 0);
        assert_eq!(snap.rq_clock, 0);
        assert_eq!(snap.scx_flags, 0);
    }

    #[test]
    fn read_rq_stats_max_values() {
        let offsets = test_offsets();
        let buf = make_rq_buffer(&offsets, u32::MAX, u32::MAX, u32::MAX, u64::MAX, u32::MAX);
        // SAFETY: buf is a live local buffer (Vec<u8> or stack array)
        // whose backing storage outlives the GuestMem use.
        let mem = unsafe { GuestMem::new(buf.as_ptr() as *mut u8, buf.len() as u64) };
        let snap = read_rq_stats(&mem, 0, &offsets);
        assert_eq!(snap.nr_running, u32::MAX);
        assert_eq!(snap.scx_nr_running, u32::MAX);
        assert_eq!(snap.local_dsq_depth, u32::MAX);
        assert_eq!(snap.rq_clock, u64::MAX);
        assert_eq!(snap.scx_flags, u32::MAX);
    }

    #[test]
    fn read_u32_out_of_bounds() {
        let buf = [0xFFu8; 8];
        // SAFETY: buf is a live local buffer (Vec<u8> or stack array)
        // whose backing storage outlives the GuestMem use.
        let mem = unsafe { GuestMem::new(buf.as_ptr() as *mut u8, buf.len() as u64) };
        // PA 6 + 4 bytes = 10 > 8, out of bounds
        assert_eq!(mem.read_u32(6, 0), 0);
        // Exactly at boundary: PA 4, offset 0 => addr 4, 4+4=8 == size, not >
        assert_eq!(mem.read_u32(4, 0), u32::from_ne_bytes([0xFF; 4]));
        // One past: PA 5, offset 0 => addr 5, 5+4=9 > 8
        assert_eq!(mem.read_u32(5, 0), 0);
    }

    #[test]
    fn read_u64_out_of_bounds() {
        let buf = [0xFFu8; 16];
        // SAFETY: buf is a live local buffer (Vec<u8> or stack array)
        // whose backing storage outlives the GuestMem use.
        let mem = unsafe { GuestMem::new(buf.as_ptr() as *mut u8, buf.len() as u64) };
        // PA 10 + 8 = 18 > 16
        assert_eq!(mem.read_u64(10, 0), 0);
        // Exactly at boundary: PA 8, 8+8=16 == size
        assert_eq!(mem.read_u64(8, 0), u64::from_ne_bytes([0xFF; 8]));
        // One past
        assert_eq!(mem.read_u64(9, 0), 0);
    }

    #[test]
    fn monitor_loop_kill_immediately() {
        let offsets = test_offsets();
        let buf = make_rq_buffer(&offsets, 1, 1, 1, 100, 0);
        // SAFETY: buf is a live local buffer (Vec<u8> or stack array)
        // whose backing storage outlives the GuestMem use.
        let mem = unsafe { GuestMem::new(buf.as_ptr() as *mut u8, buf.len() as u64) };
        let kill = AtomicBool::new(true);
        let MonitorLoopResult { samples, .. } = monitor_loop(
            &mem,
            &[0],
            &offsets,
            Duration::from_millis(10),
            &kill,
            Instant::now(),
            &test_config(),
        );
        assert!(samples.is_empty());
    }

    #[test]
    fn monitor_loop_one_iteration() {
        let offsets = test_offsets();
        let buf = make_rq_buffer(&offsets, 2, 1, 3, 500, 0);
        // SAFETY: buf is a live local buffer (Vec<u8> or stack array)
        // whose backing storage outlives the GuestMem use.
        let mem = unsafe { GuestMem::new(buf.as_ptr() as *mut u8, buf.len() as u64) };
        let kill = std::sync::Arc::new(AtomicBool::new(false));

        let handle = {
            let kill = std::sync::Arc::clone(&kill);
            std::thread::spawn(move || {
                std::thread::sleep(Duration::from_millis(50));
                kill.store(true, Ordering::Release);
            })
        };

        let MonitorLoopResult { samples, .. } = monitor_loop(
            &mem,
            &[0],
            &offsets,
            Duration::from_millis(10),
            &kill,
            Instant::now(),
            &test_config(),
        );
        handle.join().unwrap();

        assert!(!samples.is_empty());
        assert_eq!(samples[0].cpus.len(), 1);
        assert_eq!(samples[0].cpus[0].nr_running, 2);
        assert_eq!(samples[0].cpus[0].scx_nr_running, 1);
        assert_eq!(samples[0].cpus[0].local_dsq_depth, 3);
        assert_eq!(samples[0].cpus[0].rq_clock, 500);
    }

    #[test]
    fn two_cpu_independent_reads() {
        let offsets = test_offsets();
        let buf0 = make_rq_buffer(&offsets, 10, 5, 2, 1000, 0x1);
        let buf1 = make_rq_buffer(&offsets, 20, 15, 8, 2000, 0x2);

        // Concatenate into a single memory region; CPU 1's rq starts after CPU 0's.
        let pa1 = buf0.len() as u64;
        let mut combined = buf0;
        combined.extend_from_slice(&buf1);

        // SAFETY: combined is a live local buffer (Vec<u8> or stack
        // array) whose backing storage outlives the GuestMem use.
        let mem = unsafe { GuestMem::new(combined.as_ptr() as *mut u8, combined.len() as u64) };

        let snap0 = read_rq_stats(&mem, 0, &offsets);
        let snap1 = read_rq_stats(&mem, pa1, &offsets);

        assert_eq!(snap0.nr_running, 10);
        assert_eq!(snap0.scx_nr_running, 5);
        assert_eq!(snap0.local_dsq_depth, 2);
        assert_eq!(snap0.rq_clock, 1000);
        assert_eq!(snap0.scx_flags, 0x1);

        assert_eq!(snap1.nr_running, 20);
        assert_eq!(snap1.scx_nr_running, 15);
        assert_eq!(snap1.local_dsq_depth, 8);
        assert_eq!(snap1.rq_clock, 2000);
        assert_eq!(snap1.scx_flags, 0x2);
    }

    #[test]
    fn read_u32_nonzero_pa_and_offset() {
        // Check that PA + offset are combined correctly.
        let mut buf = [0u8; 32];
        // Place 0xDEADBEEF at byte 20 (PA=12, offset=8).
        buf[20..24].copy_from_slice(&0xDEADBEEFu32.to_ne_bytes());
        // SAFETY: buf is a live local buffer (Vec<u8> or stack array)
        // whose backing storage outlives the GuestMem use.
        let mem = unsafe { GuestMem::new(buf.as_ptr() as *mut u8, buf.len() as u64) };
        assert_eq!(mem.read_u32(12, 8), 0xDEADBEEF);
    }

    #[test]
    fn read_u64_nonzero_pa_and_offset() {
        let mut buf = [0u8; 32];
        // Place value at byte 16 (PA=10, offset=6).
        buf[16..24].copy_from_slice(&0x0123456789ABCDEFu64.to_ne_bytes());
        // SAFETY: buf is a live local buffer (Vec<u8> or stack array)
        // whose backing storage outlives the GuestMem use.
        let mem = unsafe { GuestMem::new(buf.as_ptr() as *mut u8, buf.len() as u64) };
        assert_eq!(mem.read_u64(10, 6), 0x0123456789ABCDEF);
    }

    #[test]
    fn monitor_loop_multi_cpu() {
        let offsets = test_offsets();
        let buf0 = make_rq_buffer(&offsets, 3, 2, 1, 100, 0);
        let buf1 = make_rq_buffer(&offsets, 7, 5, 4, 200, 0);
        let pa1 = buf0.len() as u64;
        let mut combined = buf0;
        combined.extend_from_slice(&buf1);

        // SAFETY: combined is a live local buffer (Vec<u8> or stack
        // array) whose backing storage outlives the GuestMem use.
        let mem = unsafe { GuestMem::new(combined.as_ptr() as *mut u8, combined.len() as u64) };
        let kill = std::sync::Arc::new(AtomicBool::new(false));

        let handle = {
            let kill = std::sync::Arc::clone(&kill);
            std::thread::spawn(move || {
                std::thread::sleep(Duration::from_millis(50));
                kill.store(true, Ordering::Release);
            })
        };

        let MonitorLoopResult { samples, .. } = monitor_loop(
            &mem,
            &[0, pa1],
            &offsets,
            Duration::from_millis(10),
            &kill,
            Instant::now(),
            &test_config(),
        );
        handle.join().unwrap();

        assert!(!samples.is_empty());
        // Each sample should have 2 CPUs.
        for s in &samples {
            assert_eq!(s.cpus.len(), 2);
        }
        // CPU 0 values
        assert_eq!(samples[0].cpus[0].nr_running, 3);
        assert_eq!(samples[0].cpus[0].scx_nr_running, 2);
        // CPU 1 values
        assert_eq!(samples[0].cpus[1].nr_running, 7);
        assert_eq!(samples[0].cpus[1].scx_nr_running, 5);
    }

    #[test]
    fn monitor_loop_elapsed_ms_progresses() {
        let offsets = test_offsets();
        let buf = make_rq_buffer(&offsets, 1, 1, 1, 100, 0);
        // SAFETY: buf is a live local buffer (Vec<u8> or stack array)
        // whose backing storage outlives the GuestMem use.
        let mem = unsafe { GuestMem::new(buf.as_ptr() as *mut u8, buf.len() as u64) };
        let kill = std::sync::Arc::new(AtomicBool::new(false));

        let handle = {
            let kill = std::sync::Arc::clone(&kill);
            std::thread::spawn(move || {
                std::thread::sleep(Duration::from_millis(200));
                kill.store(true, Ordering::Release);
            })
        };

        let MonitorLoopResult { samples, .. } = monitor_loop(
            &mem,
            &[0],
            &offsets,
            Duration::from_millis(10),
            &kill,
            Instant::now(),
            &test_config(),
        );
        handle.join().unwrap();

        assert!(
            samples.len() >= 2,
            "need at least 2 samples, got {}",
            samples.len()
        );
        // elapsed_ms must be monotonically non-decreasing.
        for w in samples.windows(2) {
            assert!(
                w[1].elapsed_ms >= w[0].elapsed_ms,
                "elapsed_ms went backwards: {} -> {}",
                w[0].elapsed_ms,
                w[1].elapsed_ms
            );
        }
        // Last sample should have elapsed > 0.
        assert!(samples.last().unwrap().elapsed_ms > 0);
    }

    fn test_event_offsets() -> ScxEventOffsets {
        ScxEventOffsets {
            percpu_ptr_off: 0,
            event_stats_off: 0,
            ev_select_cpu_fallback: 0,
            ev_dispatch_local_dsq_offline: 8,
            ev_dispatch_keep_last: 16,
            ev_enq_skip_exiting: 24,
            ev_enq_skip_migration_disabled: 32,
            ev_reenq_immed: None,
            ev_reenq_local_repeat: None,
            ev_refill_slice_dfl: None,
            ev_bypass_duration: None,
            ev_bypass_dispatch: None,
            ev_bypass_activate: None,
            ev_insert_not_owned: None,
            ev_sub_bypass_dispatch: None,
        }
    }

    /// Build a byte buffer simulating a scx_sched_pcpu with event_stats.
    fn make_event_stats_buffer(
        ev: &ScxEventOffsets,
        fallback: i64,
        offline: i64,
        keep_last: i64,
        skip_exit: i64,
        skip_mig: i64,
    ) -> Vec<u8> {
        let size = ev.event_stats_off + ev.ev_enq_skip_migration_disabled + 8;
        let mut buf = vec![0u8; size];
        let base = ev.event_stats_off;
        buf[base + ev.ev_select_cpu_fallback..base + ev.ev_select_cpu_fallback + 8]
            .copy_from_slice(&fallback.to_ne_bytes());
        buf[base + ev.ev_dispatch_local_dsq_offline..base + ev.ev_dispatch_local_dsq_offline + 8]
            .copy_from_slice(&offline.to_ne_bytes());
        buf[base + ev.ev_dispatch_keep_last..base + ev.ev_dispatch_keep_last + 8]
            .copy_from_slice(&keep_last.to_ne_bytes());
        buf[base + ev.ev_enq_skip_exiting..base + ev.ev_enq_skip_exiting + 8]
            .copy_from_slice(&skip_exit.to_ne_bytes());
        buf[base + ev.ev_enq_skip_migration_disabled..base + ev.ev_enq_skip_migration_disabled + 8]
            .copy_from_slice(&skip_mig.to_ne_bytes());
        buf
    }

    #[test]
    fn read_event_stats_known_values() {
        let ev = test_event_offsets();
        let buf = make_event_stats_buffer(&ev, 42, 7, 100, 3, 5);
        // SAFETY: buf is a live local buffer (Vec<u8> or stack array)
        // whose backing storage outlives the GuestMem use.
        let mem = unsafe { GuestMem::new(buf.as_ptr() as *mut u8, buf.len() as u64) };
        let stats = read_event_stats(&mem, 0, &ev);
        assert_eq!(stats.select_cpu_fallback, 42);
        assert_eq!(stats.dispatch_local_dsq_offline, 7);
        assert_eq!(stats.dispatch_keep_last, 100);
        assert_eq!(stats.enq_skip_exiting, 3);
        assert_eq!(stats.enq_skip_migration_disabled, 5);
    }

    #[test]
    fn read_event_stats_zeros() {
        let ev = test_event_offsets();
        let buf = make_event_stats_buffer(&ev, 0, 0, 0, 0, 0);
        // SAFETY: buf is a live local buffer (Vec<u8> or stack array)
        // whose backing storage outlives the GuestMem use.
        let mem = unsafe { GuestMem::new(buf.as_ptr() as *mut u8, buf.len() as u64) };
        let stats = read_event_stats(&mem, 0, &ev);
        assert_eq!(stats.select_cpu_fallback, 0);
        assert_eq!(stats.dispatch_local_dsq_offline, 0);
    }

    #[test]
    fn read_event_stats_optional_fields() {
        let mut ev = test_event_offsets();
        // Place bypass_activate at offset 40 (after the 5 mandatory fields).
        ev.ev_bypass_activate = Some(40);
        let mut buf = [0u8; 48];
        let val: i64 = 999;
        buf[40..48].copy_from_slice(&val.to_ne_bytes());
        // SAFETY: buf is a live local buffer (Vec<u8> or stack array)
        // whose backing storage outlives the GuestMem use.
        let mem = unsafe { GuestMem::new(buf.as_ptr() as *mut u8, buf.len() as u64) };
        let stats = read_event_stats(&mem, 0, &ev);
        assert_eq!(stats.bypass_activate, 999);
        // Fields without offsets remain 0.
        assert_eq!(stats.reenq_immed, 0);
        assert_eq!(stats.bypass_duration, 0);
        assert_eq!(stats.sub_bypass_dispatch, 0);
    }

    #[test]
    fn read_i64_roundtrip() {
        let val: i64 = -12345;
        let buf = val.to_ne_bytes();
        // SAFETY: buf is a live local buffer (Vec<u8> or stack array)
        // whose backing storage outlives the GuestMem use.
        let mem = unsafe { GuestMem::new(buf.as_ptr() as *mut u8, buf.len() as u64) };
        assert_eq!(mem.read_i64(0, 0), -12345);
    }

    #[test]
    fn write_u8_and_read_u8() {
        let mut buf = [0u8; 16];
        // SAFETY: buf is a live local buffer (Vec<u8> or stack array)
        // whose backing storage outlives the GuestMem use.
        let mem = unsafe { GuestMem::new(buf.as_mut_ptr(), buf.len() as u64) };
        mem.write_u8(0, 5, 0xAB);
        assert_eq!(mem.read_u8(0, 5), 0xAB);
        assert_eq!(buf[5], 0xAB);
    }

    #[test]
    fn write_u8_out_of_bounds() {
        let mut buf = [0u8; 4];
        // SAFETY: buf is a live local buffer (Vec<u8> or stack array)
        // whose backing storage outlives the GuestMem use.
        let mem = unsafe { GuestMem::new(buf.as_mut_ptr(), buf.len() as u64) };
        // Should not panic or write.
        mem.write_u8(4, 0, 0xFF);
        assert_eq!(buf, [0u8; 4]);
    }

    #[test]
    fn write_u64_and_read_u64() {
        let mut buf = [0u8; 32];
        // SAFETY: buf is a live local buffer (Vec<u8> or stack array)
        // whose backing storage outlives the GuestMem use.
        let mem = unsafe { GuestMem::new(buf.as_mut_ptr(), buf.len() as u64) };
        mem.write_u64(0, 8, 0xDEAD_BEEF_CAFE_1234);
        assert_eq!(mem.read_u64(0, 8), 0xDEAD_BEEF_CAFE_1234);
        assert_eq!(
            u64::from_ne_bytes(buf[8..16].try_into().unwrap()),
            0xDEAD_BEEF_CAFE_1234
        );
    }

    #[test]
    fn write_u64_out_of_bounds() {
        let mut buf = [0u8; 8];
        // SAFETY: buf is a live local buffer (Vec<u8> or stack array)
        // whose backing storage outlives the GuestMem use.
        let mem = unsafe { GuestMem::new(buf.as_mut_ptr(), buf.len() as u64) };
        // addr 1 + 8 = 9 > 8, out of bounds
        mem.write_u64(1, 0, 0xFF);
        assert_eq!(buf, [0u8; 8]);
    }

    #[test]
    fn write_u64_at_boundary() {
        let mut buf = [0u8; 16];
        // SAFETY: buf is a live local buffer (Vec<u8> or stack array)
        // whose backing storage outlives the GuestMem use.
        let mem = unsafe { GuestMem::new(buf.as_mut_ptr(), buf.len() as u64) };
        // PA 8 + 8 = 16 == size, should succeed
        mem.write_u64(8, 0, 0x0123_4567_89AB_CDEF);
        assert_eq!(mem.read_u64(8, 0), 0x0123_4567_89AB_CDEF);
    }

    #[test]
    fn read_u8_out_of_bounds() {
        let buf = [0xFFu8; 4];
        // SAFETY: buf is a live local buffer (Vec<u8> or stack array)
        // whose backing storage outlives the GuestMem use.
        let mem = unsafe { GuestMem::new(buf.as_ptr() as *mut u8, buf.len() as u64) };
        assert_eq!(mem.read_u8(4, 0), 0);
        assert_eq!(mem.read_u8(3, 0), 0xFF);
    }

    #[test]
    fn read_rq_stats_has_no_event_counters() {
        let offsets = test_offsets();
        let buf = make_rq_buffer(&offsets, 1, 1, 1, 100, 0);
        // SAFETY: buf is a live local buffer (Vec<u8> or stack array)
        // whose backing storage outlives the GuestMem use.
        let mem = unsafe { GuestMem::new(buf.as_ptr() as *mut u8, buf.len() as u64) };
        let snap = read_rq_stats(&mem, 0, &offsets);
        assert!(snap.event_counters.is_none());
    }

    #[test]
    fn monitor_loop_with_event_counters() {
        let ev = test_event_offsets();
        let mut offsets = test_offsets();
        offsets.event_offsets = Some(ev.clone());

        let rq_buf = make_rq_buffer(&offsets, 1, 1, 1, 100, 0);
        let ev_buf = make_event_stats_buffer(&ev, 10, 20, 30, 40, 50);

        let rq_pa = 0u64;
        let ev_pa = rq_buf.len() as u64;
        let mut combined = rq_buf;
        combined.extend_from_slice(&ev_buf);

        // SAFETY: combined is a live local buffer (Vec<u8> or stack
        // array) whose backing storage outlives the GuestMem use.
        let mem = unsafe { GuestMem::new(combined.as_ptr() as *mut u8, combined.len() as u64) };
        let kill = std::sync::Arc::new(AtomicBool::new(false));

        let handle = {
            let kill = std::sync::Arc::clone(&kill);
            std::thread::spawn(move || {
                std::thread::sleep(Duration::from_millis(30));
                kill.store(true, Ordering::Release);
            })
        };

        let ev_pas = vec![ev_pa];
        let cfg = MonitorConfig {
            event_pcpu_pas: Some(&ev_pas),
            ..test_config()
        };
        let MonitorLoopResult { samples, .. } = monitor_loop(
            &mem,
            &[rq_pa],
            &offsets,
            Duration::from_millis(10),
            &kill,
            Instant::now(),
            &cfg,
        );
        handle.join().unwrap();

        assert!(!samples.is_empty());
        let counters = samples[0].cpus[0].event_counters.as_ref().unwrap();
        assert_eq!(counters.select_cpu_fallback, 10);
        assert_eq!(counters.dispatch_local_dsq_offline, 20);
        assert_eq!(counters.dispatch_keep_last, 30);
        assert_eq!(counters.enq_skip_exiting, 40);
        assert_eq!(counters.enq_skip_migration_disabled, 50);
    }

    #[test]
    fn monitor_loop_no_event_counters_when_none() {
        let offsets = test_offsets();
        let buf = make_rq_buffer(&offsets, 1, 1, 1, 100, 0);
        // SAFETY: buf is a live local buffer (Vec<u8> or stack array)
        // whose backing storage outlives the GuestMem use.
        let mem = unsafe { GuestMem::new(buf.as_ptr() as *mut u8, buf.len() as u64) };
        let kill = std::sync::Arc::new(AtomicBool::new(false));

        let handle = {
            let kill = std::sync::Arc::clone(&kill);
            std::thread::spawn(move || {
                std::thread::sleep(Duration::from_millis(30));
                kill.store(true, Ordering::Release);
            })
        };

        let MonitorLoopResult { samples, .. } = monitor_loop(
            &mem,
            &[0],
            &offsets,
            Duration::from_millis(10),
            &kill,
            Instant::now(),
            &test_config(),
        );
        handle.join().unwrap();

        assert!(!samples.is_empty());
        assert!(samples[0].cpus[0].event_counters.is_none());
    }

    #[test]
    fn resolve_event_pcpu_pas_null_scx_root() {
        let ev = test_event_offsets();
        // scx_root pointer is 0 (null) — no scheduler loaded.
        let buf = [0u8; 64];
        // SAFETY: buf is a live local buffer (Vec<u8> or stack array)
        // whose backing storage outlives the GuestMem use.
        let mem = unsafe { GuestMem::new(buf.as_ptr() as *mut u8, buf.len() as u64) };
        let result = resolve_event_pcpu_pas(&mem, 0, &ev, &[0, 0x4000], 0);
        assert!(result.is_none());
    }

    #[test]
    fn monitor_loop_with_watchdog_override() {
        let offsets = test_offsets();
        // Layout:
        //   [rq_buf]
        //   [scx_root pointer slot @ scx_root_pa] (holds scx_sched KVA)
        //   [scx_sched struct @ sch_pa, with watchdog_timeout at watchdog_offset]
        // The monitor derefs *scx_root_pa -> KVA, translates via PAGE_OFFSET -> PA,
        // then writes jiffies at sch_pa + watchdog_offset.
        let rq_buf = make_rq_buffer(&offsets, 1, 1, 1, 100, 0);
        let scx_root_pa = rq_buf.len() as u64;
        let sch_pa = scx_root_pa + 8;
        let watchdog_offset: usize = 16;
        let page_offset = super::super::symbols::DEFAULT_PAGE_OFFSET;
        let scx_sched_kva = page_offset.wrapping_add(sch_pa);

        // Buffer = rq_buf | 8 bytes (scx_root slot) | 64 bytes (scx_sched stub).
        let mut combined = rq_buf;
        combined.extend_from_slice(&scx_sched_kva.to_ne_bytes());
        combined.extend_from_slice(&[0u8; 64]);

        // SAFETY: combined is a live local buffer (Vec<u8> or stack
        // array) whose backing storage outlives the GuestMem use.
        let mem = unsafe { GuestMem::new(combined.as_mut_ptr(), combined.len() as u64) };
        let kill = std::sync::Arc::new(AtomicBool::new(false));

        let wd = WatchdogOverride::ScxSched {
            scx_root_pa,
            watchdog_offset,
            jiffies: 99999,
            page_offset,
        };

        let handle = {
            let kill = std::sync::Arc::clone(&kill);
            std::thread::spawn(move || {
                std::thread::sleep(Duration::from_millis(30));
                kill.store(true, Ordering::Release);
            })
        };

        let cfg = MonitorConfig {
            watchdog_override: Some(&wd),
            ..test_config()
        };
        let MonitorLoopResult {
            samples,
            watchdog_observation,
            ..
        } = monitor_loop(
            &mem,
            &[0],
            &offsets,
            Duration::from_millis(10),
            &kill,
            Instant::now(),
            &cfg,
        );
        handle.join().unwrap();

        assert!(!samples.is_empty());
        // Check the watchdog value was written at sch_pa + watchdog_offset.
        let write_pa = sch_pa as usize + watchdog_offset;
        let written = u64::from_ne_bytes(combined[write_pa..write_pa + 8].try_into().unwrap());
        assert_eq!(written, 99999);
        // Check monitor_loop recorded the observation.
        let obs = watchdog_observation.expect("watchdog_observation should be Some after write");
        assert_eq!(obs.expected_jiffies, 99999);
        assert_eq!(obs.observed_jiffies, 99999);
    }

    #[test]
    fn monitor_loop_watchdog_override_skipped_when_scx_root_null() {
        let offsets = test_offsets();
        // Layout: rq_buf | scx_root slot = 0 (no scheduler loaded).
        let rq_buf = make_rq_buffer(&offsets, 1, 1, 1, 100, 0);
        let scx_root_pa = rq_buf.len() as u64;
        let mut combined = rq_buf;
        combined.extend_from_slice(&[0u8; 8]); // scx_root = null
        // Extra space in case of accidental write via garbage deref.
        combined.extend_from_slice(&[0u8; 128]);

        // SAFETY: combined is a live local buffer (Vec<u8> or stack
        // array) whose backing storage outlives the GuestMem use.
        let mem = unsafe { GuestMem::new(combined.as_mut_ptr(), combined.len() as u64) };
        let kill = std::sync::Arc::new(AtomicBool::new(false));
        let wd = WatchdogOverride::ScxSched {
            scx_root_pa,
            watchdog_offset: 16,
            jiffies: 0xDEADBEEF,
            page_offset: super::super::symbols::DEFAULT_PAGE_OFFSET,
        };

        let handle = {
            let kill = std::sync::Arc::clone(&kill);
            std::thread::spawn(move || {
                std::thread::sleep(Duration::from_millis(30));
                kill.store(true, Ordering::Release);
            })
        };

        let cfg = MonitorConfig {
            watchdog_override: Some(&wd),
            ..test_config()
        };
        let MonitorLoopResult {
            watchdog_observation,
            ..
        } = monitor_loop(
            &mem,
            &[0],
            &offsets,
            Duration::from_millis(10),
            &kill,
            Instant::now(),
            &cfg,
        );
        handle.join().unwrap();

        // No write should have happened: buffer is all zeros past rq_buf.
        assert!(
            combined[scx_root_pa as usize..].iter().all(|&b| b == 0),
            "no write should occur when scx_root is null"
        );
        // No observation should have been recorded.
        assert!(
            watchdog_observation.is_none(),
            "watchdog_observation should be None when scx_root is null"
        );
    }

    #[test]
    fn monitor_loop_watchdog_static_global_writes_directly() {
        let offsets = test_offsets();
        let rq_buf = make_rq_buffer(&offsets, 1, 1, 1, 100, 0);
        let watchdog_pa = rq_buf.len() as u64;

        let mut combined = rq_buf;
        combined.extend_from_slice(&[0u8; 8]);

        // SAFETY: combined is a live local buffer (Vec<u8> or stack
        // array) whose backing storage outlives the GuestMem use.
        let mem = unsafe { GuestMem::new(combined.as_mut_ptr(), combined.len() as u64) };
        let kill = std::sync::Arc::new(AtomicBool::new(false));

        let wd = WatchdogOverride::StaticGlobal {
            watchdog_timeout_pa: watchdog_pa,
            jiffies: 77777,
        };

        let handle = {
            let kill = std::sync::Arc::clone(&kill);
            std::thread::spawn(move || {
                std::thread::sleep(Duration::from_millis(30));
                kill.store(true, Ordering::Release);
            })
        };

        let cfg = MonitorConfig {
            watchdog_override: Some(&wd),
            ..test_config()
        };
        let MonitorLoopResult {
            samples,
            watchdog_observation,
            ..
        } = monitor_loop(
            &mem,
            &[0],
            &offsets,
            Duration::from_millis(10),
            &kill,
            Instant::now(),
            &cfg,
        );
        handle.join().unwrap();

        assert!(!samples.is_empty());
        let written = u64::from_ne_bytes(
            combined[watchdog_pa as usize..watchdog_pa as usize + 8]
                .try_into()
                .unwrap(),
        );
        assert_eq!(written, 77777);
        let obs = watchdog_observation.expect("watchdog_observation should be Some");
        assert_eq!(obs.expected_jiffies, 77777);
        assert_eq!(obs.observed_jiffies, 77777);
    }

    #[test]
    fn monitor_loop_dump_trigger_fires_on_imbalance() {
        let offsets = test_offsets();
        // Two rq buffers: CPU0 = 1 task, CPU1 = 20 tasks -> ratio=20 >> threshold.
        let buf0 = make_rq_buffer(&offsets, 1, 1, 1, 100, 0);
        let buf1 = make_rq_buffer(&offsets, 20, 20, 1, 200, 0);
        let pa1 = buf0.len() as u64;
        let mut combined = buf0;
        combined.extend_from_slice(&buf1);
        // Append SHM region (64 bytes minimum for dump req offset).
        let shm_pa = combined.len() as u64;
        combined.extend(vec![0u8; 64]);

        // SAFETY: combined is a live local buffer (Vec<u8> or stack
        // array) whose backing storage outlives the GuestMem use.
        let mem = unsafe { GuestMem::new(combined.as_mut_ptr(), combined.len() as u64) };
        let kill = std::sync::Arc::new(AtomicBool::new(false));

        let trigger = DumpTrigger {
            shm_base_pa: shm_pa,
            thresholds: super::super::MonitorThresholds {
                max_imbalance_ratio: 2.0,
                sustained_samples: 2,
                fail_on_stall: false,
                ..Default::default()
            },
        };

        let handle = {
            let kill = std::sync::Arc::clone(&kill);
            std::thread::spawn(move || {
                std::thread::sleep(Duration::from_millis(200));
                kill.store(true, Ordering::Release);
            })
        };

        let cfg = MonitorConfig {
            dump_trigger: Some(&trigger),
            ..test_config()
        };
        let MonitorLoopResult { samples, .. } = monitor_loop(
            &mem,
            &[0, pa1],
            &offsets,
            Duration::from_millis(10),
            &kill,
            Instant::now(),
            &cfg,
        );
        handle.join().unwrap();

        assert!(!samples.is_empty());
        // Check dump request was written to SHM.
        let dump_byte = combined[shm_pa as usize + crate::vmm::shm_ring::DUMP_REQ_OFFSET];
        assert_eq!(
            dump_byte,
            crate::vmm::shm_ring::DUMP_REQ_SYSRQ_D,
            "dump request should have been written to SHM"
        );
    }

    #[test]
    fn monitor_loop_dump_trigger_stall_with_sustained_window() {
        // Reactive stall path: stuck rq_clock with nr_running>0 triggers
        // dump after sustained_samples consecutive stall pairs.
        let offsets = test_offsets();
        // Single CPU: nr_running=2 (busy), rq_clock stuck at 5000.
        // Need a second CPU with a different clock value so samples
        // differ (otherwise all-same-clock triggers the uninitialized
        // check in from_samples, though monitor_loop's reactive path
        // doesn't use from_samples — it checks inline).
        let buf = make_rq_buffer(&offsets, 2, 1, 1, 5000, 0);
        let shm_pa = buf.len() as u64;
        let mut combined = buf;
        combined.extend(vec![0u8; 64]);

        // SAFETY: combined is a live local buffer (Vec<u8> or stack
        // array) whose backing storage outlives the GuestMem use.
        let mem = unsafe { GuestMem::new(combined.as_mut_ptr(), combined.len() as u64) };
        let kill = std::sync::Arc::new(AtomicBool::new(false));

        let trigger = DumpTrigger {
            shm_base_pa: shm_pa,
            thresholds: super::super::MonitorThresholds {
                max_imbalance_ratio: 100.0,
                max_local_dsq_depth: 10000,
                fail_on_stall: true,
                sustained_samples: 2,
                ..Default::default()
            },
        };

        let handle = {
            let kill = std::sync::Arc::clone(&kill);
            std::thread::spawn(move || {
                std::thread::sleep(Duration::from_millis(200));
                kill.store(true, Ordering::Release);
            })
        };

        let cfg = MonitorConfig {
            dump_trigger: Some(&trigger),
            ..test_config()
        };
        let MonitorLoopResult { samples, .. } = monitor_loop(
            &mem,
            &[0],
            &offsets,
            Duration::from_millis(10),
            &kill,
            Instant::now(),
            &cfg,
        );
        handle.join().unwrap();

        // Should have enough samples for 2+ stall pairs.
        assert!(
            samples.len() >= 3,
            "need >= 3 samples for 2 stall pairs, got {}",
            samples.len()
        );
        // Dump should have fired due to sustained stall.
        let dump_byte = combined[shm_pa as usize + crate::vmm::shm_ring::DUMP_REQ_OFFSET];
        assert_eq!(
            dump_byte,
            crate::vmm::shm_ring::DUMP_REQ_SYSRQ_D,
            "stall should trigger dump after sustained_samples=2"
        );
    }

    #[test]
    fn monitor_loop_dump_trigger_idle_cpu_no_stall() {
        // Reactive path: nr_running==0 (idle) with stuck rq_clock should
        // NOT trigger the dump, even with fail_on_stall=true.
        let offsets = test_offsets();
        // CPU idle: nr_running=0, rq_clock stuck at 5000.
        let buf = make_rq_buffer(&offsets, 0, 0, 0, 5000, 0);
        let shm_pa = buf.len() as u64;
        let mut combined = buf;
        combined.extend(vec![0u8; 64]);

        // SAFETY: combined is a live local buffer (Vec<u8> or stack
        // array) whose backing storage outlives the GuestMem use.
        let mem = unsafe { GuestMem::new(combined.as_mut_ptr(), combined.len() as u64) };
        let kill = std::sync::Arc::new(AtomicBool::new(false));

        let trigger = DumpTrigger {
            shm_base_pa: shm_pa,
            thresholds: super::super::MonitorThresholds {
                max_imbalance_ratio: 100.0,
                max_local_dsq_depth: 10000,
                fail_on_stall: true,
                sustained_samples: 1,
                ..Default::default()
            },
        };

        let handle = {
            let kill = std::sync::Arc::clone(&kill);
            std::thread::spawn(move || {
                std::thread::sleep(Duration::from_millis(200));
                kill.store(true, Ordering::Release);
            })
        };

        let cfg = MonitorConfig {
            dump_trigger: Some(&trigger),
            ..test_config()
        };
        let MonitorLoopResult { samples, .. } = monitor_loop(
            &mem,
            &[0],
            &offsets,
            Duration::from_millis(10),
            &kill,
            Instant::now(),
            &cfg,
        );
        handle.join().unwrap();

        assert!(
            samples.len() >= 2,
            "need >= 2 samples, got {}",
            samples.len()
        );
        // Dump should NOT have fired — idle CPU is exempt.
        let dump_byte = combined[shm_pa as usize + crate::vmm::shm_ring::DUMP_REQ_OFFSET];
        assert_eq!(dump_byte, 0, "idle CPU should not trigger stall dump");
    }

    #[test]
    fn monitor_loop_vcpu_timing_preempted_no_stall() {
        // Sleeping thread: CPU time stays near zero between samples.
        // rq_clock stuck + CPU time not advancing = preempted, suppress stall.
        // 30ms interval gives margin on loaded hosts. Explicit threshold
        // (10ms) avoids host CONFIG_HZ dependency.
        let offsets = test_offsets();
        let buf = make_rq_buffer(&offsets, 2, 1, 1, 5000, 0);
        let shm_pa = buf.len() as u64;
        let mut combined = buf;
        combined.extend(vec![0u8; 64]);

        // SAFETY: combined is a live local buffer (Vec<u8> or stack
        // array) whose backing storage outlives the GuestMem use.
        let mem = unsafe { GuestMem::new(combined.as_mut_ptr(), combined.len() as u64) };
        let kill = std::sync::Arc::new(AtomicBool::new(false));

        let sleeper_kill = std::sync::Arc::new(AtomicBool::new(false));
        let sleeper_kill_clone = sleeper_kill.clone();
        let sleeper = std::thread::Builder::new()
            .name("vcpu-sleeper".into())
            .spawn(move || {
                while !sleeper_kill_clone.load(Ordering::Acquire) {
                    std::thread::sleep(Duration::from_millis(100));
                }
            })
            .unwrap();

        let pt = sleeper.as_pthread_t() as libc::pthread_t;
        let vcpu_timing = VcpuTiming { pthreads: vec![pt] };

        let trigger = DumpTrigger {
            shm_base_pa: shm_pa,
            thresholds: super::super::MonitorThresholds {
                max_imbalance_ratio: 100.0,
                max_local_dsq_depth: 10000,
                fail_on_stall: true,
                sustained_samples: 1,
                ..Default::default()
            },
        };

        let handle = {
            let kill = std::sync::Arc::clone(&kill);
            std::thread::spawn(move || {
                std::thread::sleep(Duration::from_millis(150));
                kill.store(true, Ordering::Release);
            })
        };

        let cfg = MonitorConfig {
            dump_trigger: Some(&trigger),
            vcpu_timing: Some(&vcpu_timing),
            preemption_threshold_ns: 10_000_000,
            ..test_config()
        };
        let MonitorLoopResult { samples, .. } = monitor_loop(
            &mem,
            &[0],
            &offsets,
            Duration::from_millis(30),
            &kill,
            Instant::now(),
            &cfg,
        );
        handle.join().unwrap();
        sleeper_kill.store(true, Ordering::Release);
        let _ = sleeper.join();

        assert!(
            samples.len() >= 2,
            "need >= 2 samples, got {}",
            samples.len()
        );
        let dump_byte = combined[shm_pa as usize + crate::vmm::shm_ring::DUMP_REQ_OFFSET];
        assert_eq!(dump_byte, 0, "preempted vCPU should not trigger stall dump");
    }

    #[test]
    fn monitor_loop_vcpu_timing_running_stall_fires() {
        // Busy-spinning thread: accumulates CPU time every interval.
        // 30ms interval ensures spinner clears the 10ms preemption
        // threshold with margin.
        // rq_clock stuck + CPU time advancing = real stall. Explicit
        // threshold (10ms) avoids host CONFIG_HZ dependency (CONFIG_HZ=250
        // gives 40ms threshold, which would mask 30ms of spin time).
        let offsets = test_offsets();
        let buf = make_rq_buffer(&offsets, 2, 1, 1, 5000, 0);
        let shm_pa = buf.len() as u64;
        let mut combined = buf;
        combined.extend(vec![0u8; 64]);

        // SAFETY: combined is a live local buffer (Vec<u8> or stack
        // array) whose backing storage outlives the GuestMem use.
        let mem = unsafe { GuestMem::new(combined.as_mut_ptr(), combined.len() as u64) };
        let kill = std::sync::Arc::new(AtomicBool::new(false));

        let spinner_kill = std::sync::Arc::new(AtomicBool::new(false));
        let spinner_kill_clone = spinner_kill.clone();
        let spinner = std::thread::Builder::new()
            .name("vcpu-spinner".into())
            .spawn(move || {
                while !spinner_kill_clone.load(Ordering::Relaxed) {
                    std::hint::spin_loop();
                }
            })
            .unwrap();

        let pt = spinner.as_pthread_t() as libc::pthread_t;
        let vcpu_timing = VcpuTiming { pthreads: vec![pt] };

        let trigger = DumpTrigger {
            shm_base_pa: shm_pa,
            thresholds: super::super::MonitorThresholds {
                max_imbalance_ratio: 100.0,
                max_local_dsq_depth: 10000,
                fail_on_stall: true,
                sustained_samples: 2,
                ..Default::default()
            },
        };

        let handle = {
            let kill = std::sync::Arc::clone(&kill);
            std::thread::spawn(move || {
                std::thread::sleep(Duration::from_millis(200));
                kill.store(true, Ordering::Release);
            })
        };

        let cfg = MonitorConfig {
            dump_trigger: Some(&trigger),
            vcpu_timing: Some(&vcpu_timing),
            preemption_threshold_ns: 10_000_000,
            ..test_config()
        };
        let MonitorLoopResult { samples, .. } = monitor_loop(
            &mem,
            &[0],
            &offsets,
            Duration::from_millis(30),
            &kill,
            Instant::now(),
            &cfg,
        );
        handle.join().unwrap();
        spinner_kill.store(true, Ordering::Release);
        let _ = spinner.join();

        assert!(
            samples.len() >= 3,
            "need >= 3 samples for 2 stall pairs, got {}",
            samples.len()
        );
        let dump_byte = combined[shm_pa as usize + crate::vmm::shm_ring::DUMP_REQ_OFFSET];
        assert_eq!(
            dump_byte,
            crate::vmm::shm_ring::DUMP_REQ_SYSRQ_D,
            "real stall (vCPU running, clock stuck, nr_running>0) should trigger dump"
        );
    }

    #[test]
    fn reactive_and_evaluate_stall_consistency() {
        // Check that the reactive path (monitor_loop with dump_trigger)
        // and the post-hoc path (evaluate) agree on stall detection.
        // Build a scenario where stall fires: stuck rq_clock, nr_running>0,
        // sustained_samples=2.
        // Two CPUs: cpu0 stuck (rq_clock=5000), cpu1 has a different
        // clock value from cpu0 in each sample, so data_looks_valid
        // sees non-identical clocks.
        let offsets = test_offsets();
        let buf0 = make_rq_buffer(&offsets, 2, 1, 1, 5000, 0);
        let buf1 = make_rq_buffer(&offsets, 1, 1, 1, 9000, 0);
        let pa1 = buf0.len() as u64;
        let mut combined = buf0;
        combined.extend_from_slice(&buf1);
        let shm_pa = combined.len() as u64;
        combined.extend(vec![0u8; 64]);

        // SAFETY: combined is a live local buffer (Vec<u8> or stack
        // array) whose backing storage outlives the GuestMem use.
        let mem = unsafe { GuestMem::new(combined.as_mut_ptr(), combined.len() as u64) };
        let kill = std::sync::Arc::new(AtomicBool::new(false));

        let thresholds = super::super::MonitorThresholds {
            max_imbalance_ratio: 100.0,
            max_local_dsq_depth: 10000,
            fail_on_stall: true,
            sustained_samples: 2,
            ..Default::default()
        };

        let trigger = DumpTrigger {
            shm_base_pa: shm_pa,
            thresholds,
        };

        let handle = {
            let kill = std::sync::Arc::clone(&kill);
            std::thread::spawn(move || {
                std::thread::sleep(Duration::from_millis(200));
                kill.store(true, Ordering::Release);
            })
        };

        let cfg = MonitorConfig {
            dump_trigger: Some(&trigger),
            ..test_config()
        };
        let MonitorLoopResult { samples, .. } = monitor_loop(
            &mem,
            &[0, pa1],
            &offsets,
            Duration::from_millis(10),
            &kill,
            Instant::now(),
            &cfg,
        );
        handle.join().unwrap();

        assert!(
            samples.len() >= 3,
            "need >= 3 samples, got {}",
            samples.len()
        );

        // Reactive path result: check if dump fired.
        let dump_byte = combined[shm_pa as usize + crate::vmm::shm_ring::DUMP_REQ_OFFSET];
        let reactive_stall = dump_byte == crate::vmm::shm_ring::DUMP_REQ_SYSRQ_D;

        // Post-hoc evaluate path on the same samples.
        let summary = super::super::MonitorSummary::from_samples(&samples);
        let report = super::super::MonitorReport {
            samples,
            summary,
            ..Default::default()
        };
        let verdict = thresholds.evaluate(&report);

        // Both paths should agree: stall detected on cpu0.
        assert!(reactive_stall, "reactive path should detect stall");
        assert!(
            !verdict.passed,
            "evaluate should detect stall: {:?}",
            verdict.details
        );
        assert!(
            verdict.details.iter().any(|d| d.contains("rq_clock stall")),
            "evaluate details should mention stall: {:?}",
            verdict.details
        );
    }

    #[test]
    fn reactive_and_evaluate_idle_consistency() {
        // Both reactive and evaluate should agree: idle CPU is exempt.
        let offsets = test_offsets();
        // nr_running=0, rq_clock stuck.
        let buf = make_rq_buffer(&offsets, 0, 0, 0, 5000, 0);
        let shm_pa = buf.len() as u64;
        let mut combined = buf;
        combined.extend(vec![0u8; 64]);

        // SAFETY: combined is a live local buffer (Vec<u8> or stack
        // array) whose backing storage outlives the GuestMem use.
        let mem = unsafe { GuestMem::new(combined.as_mut_ptr(), combined.len() as u64) };
        let kill = std::sync::Arc::new(AtomicBool::new(false));

        let thresholds = super::super::MonitorThresholds {
            max_imbalance_ratio: 100.0,
            max_local_dsq_depth: 10000,
            fail_on_stall: true,
            sustained_samples: 1,
            ..Default::default()
        };

        let trigger = DumpTrigger {
            shm_base_pa: shm_pa,
            thresholds,
        };

        let handle = {
            let kill = std::sync::Arc::clone(&kill);
            std::thread::spawn(move || {
                std::thread::sleep(Duration::from_millis(200));
                kill.store(true, Ordering::Release);
            })
        };

        let cfg = MonitorConfig {
            dump_trigger: Some(&trigger),
            ..test_config()
        };
        let MonitorLoopResult { samples, .. } = monitor_loop(
            &mem,
            &[0],
            &offsets,
            Duration::from_millis(10),
            &kill,
            Instant::now(),
            &cfg,
        );
        handle.join().unwrap();

        assert!(
            samples.len() >= 2,
            "need >= 2 samples, got {}",
            samples.len()
        );

        // Reactive: dump should NOT fire.
        let dump_byte = combined[shm_pa as usize + crate::vmm::shm_ring::DUMP_REQ_OFFSET];
        assert_eq!(
            dump_byte, 0,
            "reactive: idle CPU should not trigger stall dump"
        );

        // Evaluate: from_samples should not detect stall.
        let summary = super::super::MonitorSummary::from_samples(&samples);
        assert!(
            !summary.stall_detected,
            "from_samples: idle CPU should not flag stall"
        );

        // Evaluate verdict: should pass (no stall on idle CPU).
        // Note: evaluate may pass via data_looks_valid returning false
        // (all-same clocks with single CPU) — that's consistent behavior.
        let report = super::super::MonitorReport {
            samples,
            summary,
            ..Default::default()
        };
        let verdict = thresholds.evaluate(&report);
        assert!(
            verdict.passed,
            "evaluate: idle CPU should pass: {:?}",
            verdict.details
        );
    }

    fn test_schedstat_offsets() -> super::super::btf_offsets::SchedstatOffsets {
        super::super::btf_offsets::SchedstatOffsets {
            rq_sched_info: 200,
            sched_info_run_delay: 8,
            sched_info_pcount: 0,
            rq_yld_count: 300,
            rq_sched_count: 304,
            rq_sched_goidle: 308,
            rq_ttwu_count: 312,
            rq_ttwu_local: 316,
        }
    }

    /// Build a byte buffer simulating a struct rq with schedstat fields.
    #[allow(clippy::too_many_arguments)]
    fn make_schedstat_buffer(
        ss: &super::super::btf_offsets::SchedstatOffsets,
        run_delay: u64,
        pcount: u64,
        yld_count: u32,
        sched_count: u32,
        sched_goidle: u32,
        ttwu_count: u32,
        ttwu_local: u32,
    ) -> Vec<u8> {
        let size = ss.rq_ttwu_local + 4 + 8;
        let mut buf = vec![0u8; size];

        let si_base = ss.rq_sched_info;
        buf[si_base + ss.sched_info_pcount..si_base + ss.sched_info_pcount + 8]
            .copy_from_slice(&pcount.to_ne_bytes());
        buf[si_base + ss.sched_info_run_delay..si_base + ss.sched_info_run_delay + 8]
            .copy_from_slice(&run_delay.to_ne_bytes());

        buf[ss.rq_yld_count..ss.rq_yld_count + 4].copy_from_slice(&yld_count.to_ne_bytes());
        buf[ss.rq_sched_count..ss.rq_sched_count + 4].copy_from_slice(&sched_count.to_ne_bytes());
        buf[ss.rq_sched_goidle..ss.rq_sched_goidle + 4]
            .copy_from_slice(&sched_goidle.to_ne_bytes());
        buf[ss.rq_ttwu_count..ss.rq_ttwu_count + 4].copy_from_slice(&ttwu_count.to_ne_bytes());
        buf[ss.rq_ttwu_local..ss.rq_ttwu_local + 4].copy_from_slice(&ttwu_local.to_ne_bytes());
        buf
    }

    #[test]
    fn read_rq_schedstat_known_values() {
        let ss = test_schedstat_offsets();
        let buf = make_schedstat_buffer(&ss, 50000, 10, 3, 100, 20, 80, 40);
        // SAFETY: buf is a live local buffer (Vec<u8> or stack array)
        // whose backing storage outlives the GuestMem use.
        let mem = unsafe { GuestMem::new(buf.as_ptr() as *mut u8, buf.len() as u64) };
        let stats = read_rq_schedstat(&mem, 0, &ss);
        assert_eq!(stats.run_delay, 50000);
        assert_eq!(stats.pcount, 10);
        assert_eq!(stats.yld_count, 3);
        assert_eq!(stats.sched_count, 100);
        assert_eq!(stats.sched_goidle, 20);
        assert_eq!(stats.ttwu_count, 80);
        assert_eq!(stats.ttwu_local, 40);
    }

    #[test]
    fn read_rq_schedstat_zeros() {
        let ss = test_schedstat_offsets();
        let buf = make_schedstat_buffer(&ss, 0, 0, 0, 0, 0, 0, 0);
        // SAFETY: buf is a live local buffer (Vec<u8> or stack array)
        // whose backing storage outlives the GuestMem use.
        let mem = unsafe { GuestMem::new(buf.as_ptr() as *mut u8, buf.len() as u64) };
        let stats = read_rq_schedstat(&mem, 0, &ss);
        assert_eq!(stats.run_delay, 0);
        assert_eq!(stats.pcount, 0);
        assert_eq!(stats.yld_count, 0);
        assert_eq!(stats.sched_count, 0);
        assert_eq!(stats.sched_goidle, 0);
        assert_eq!(stats.ttwu_count, 0);
        assert_eq!(stats.ttwu_local, 0);
    }

    #[test]
    fn monitor_loop_with_schedstat_overlay() {
        let ss = test_schedstat_offsets();
        let mut offsets = test_offsets();
        offsets.schedstat_offsets = Some(ss.clone());

        // Build a buffer that contains both rq fields and schedstat fields.
        // The rq buffer must be large enough to cover schedstat offsets.
        let rq_size = ss.rq_ttwu_local + 4 + 8;
        let mut buf = vec![0u8; rq_size];

        // Write rq base fields.
        buf[offsets.rq_nr_running..offsets.rq_nr_running + 4].copy_from_slice(&2u32.to_ne_bytes());
        buf[offsets.rq_clock..offsets.rq_clock + 8].copy_from_slice(&500u64.to_ne_bytes());

        // Write schedstat fields.
        let si_base = ss.rq_sched_info;
        buf[si_base + ss.sched_info_run_delay..si_base + ss.sched_info_run_delay + 8]
            .copy_from_slice(&12345u64.to_ne_bytes());
        buf[si_base + ss.sched_info_pcount..si_base + ss.sched_info_pcount + 8]
            .copy_from_slice(&7u64.to_ne_bytes());
        buf[ss.rq_sched_count..ss.rq_sched_count + 4].copy_from_slice(&42u32.to_ne_bytes());

        // SAFETY: buf is a live local buffer (Vec<u8> or stack array)
        // whose backing storage outlives the GuestMem use.
        let mem = unsafe { GuestMem::new(buf.as_ptr() as *mut u8, buf.len() as u64) };
        let kill = std::sync::Arc::new(AtomicBool::new(false));

        let handle = {
            let kill = std::sync::Arc::clone(&kill);
            std::thread::spawn(move || {
                std::thread::sleep(Duration::from_millis(30));
                kill.store(true, Ordering::Release);
            })
        };

        let MonitorLoopResult { samples, .. } = monitor_loop(
            &mem,
            &[0],
            &offsets,
            Duration::from_millis(10),
            &kill,
            Instant::now(),
            &test_config(),
        );
        handle.join().unwrap();

        assert!(!samples.is_empty());
        let ss_snap = samples[0].cpus[0].schedstat.as_ref().unwrap();
        assert_eq!(ss_snap.run_delay, 12345);
        assert_eq!(ss_snap.pcount, 7);
        assert_eq!(ss_snap.sched_count, 42);
    }

    #[test]
    fn monitor_loop_no_schedstat_when_none() {
        let offsets = test_offsets();
        assert!(offsets.schedstat_offsets.is_none());

        let buf = make_rq_buffer(&offsets, 1, 1, 1, 100, 0);
        // SAFETY: buf is a live local buffer (Vec<u8> or stack array)
        // whose backing storage outlives the GuestMem use.
        let mem = unsafe { GuestMem::new(buf.as_ptr() as *mut u8, buf.len() as u64) };
        let kill = std::sync::Arc::new(AtomicBool::new(false));

        let handle = {
            let kill = std::sync::Arc::clone(&kill);
            std::thread::spawn(move || {
                std::thread::sleep(Duration::from_millis(30));
                kill.store(true, Ordering::Release);
            })
        };

        let MonitorLoopResult { samples, .. } = monitor_loop(
            &mem,
            &[0],
            &offsets,
            Duration::from_millis(10),
            &kill,
            Instant::now(),
            &test_config(),
        );
        handle.join().unwrap();

        assert!(!samples.is_empty());
        assert!(samples[0].cpus[0].schedstat.is_none());
    }

    fn test_sched_domain_offsets() -> SchedDomainOffsets {
        // Synthetic offsets for a sched_domain struct.
        // Layout: parent(0) level(8) flags(12) name(16) span_weight(24)
        //         balance_interval(28) nr_balance_failed(32)
        //         newidle_call(36) newidle_success(40) newidle_ratio(44)
        //         max_newidle_lb_cost(48)
        //         [stats at 56+]
        SchedDomainOffsets {
            rq_sd: 400,
            sd_parent: 0,
            sd_level: 8,
            sd_flags: 12,
            sd_name: 16,
            sd_span_weight: 24,
            sd_balance_interval: 28,
            sd_nr_balance_failed: 32,
            sd_newidle_call: Some(36),
            sd_newidle_success: Some(40),
            sd_newidle_ratio: Some(44),
            sd_max_newidle_lb_cost: 48,
            stats_offsets: Some(test_sd_stats_offsets()),
        }
    }

    fn test_sd_stats_offsets() -> SchedDomainStatsOffsets {
        SchedDomainStatsOffsets {
            sd_lb_count: 56,
            sd_lb_failed: 68,
            sd_lb_balanced: 80,
            sd_lb_imbalance_load: 92,
            sd_lb_imbalance_util: 104,
            sd_lb_imbalance_task: 116,
            sd_lb_imbalance_misfit: 128,
            sd_lb_gained: 140,
            sd_lb_hot_gained: 152,
            sd_lb_nobusyg: 164,
            sd_lb_nobusyq: 176,
            sd_alb_count: 188,
            sd_alb_failed: 192,
            sd_alb_pushed: 196,
            sd_sbe_count: 200,
            sd_sbe_balanced: 204,
            sd_sbe_pushed: 208,
            sd_sbf_count: 212,
            sd_sbf_balanced: 216,
            sd_sbf_pushed: 220,
            sd_ttwu_wake_remote: 224,
            sd_ttwu_move_affine: 228,
            sd_ttwu_move_balance: 232,
        }
    }

    /// Build a synthetic sched_domain buffer with known values.
    /// `parent_kva`: KVA of parent domain (0 = no parent).
    /// `name_kva`: KVA of name string (0 = no name).
    /// Returns a buffer representing one sched_domain struct.
    #[allow(clippy::too_many_arguments)]
    fn make_sd_buffer(
        sd: &SchedDomainOffsets,
        parent_kva: u64,
        level: i32,
        flags: i32,
        name_kva: u64,
        span_weight: u32,
        balance_interval: u32,
        newidle_call: u32,
        lb_count_0: u32,
        alb_pushed: u32,
        ttwu_wake_remote: u32,
    ) -> Vec<u8> {
        // Size must cover the highest offset used.
        let so = sd.stats_offsets.as_ref().unwrap();
        let size = so.sd_ttwu_move_balance + 4 + 8;
        let mut buf = vec![0u8; size];

        buf[sd.sd_parent..sd.sd_parent + 8].copy_from_slice(&parent_kva.to_ne_bytes());
        buf[sd.sd_level..sd.sd_level + 4].copy_from_slice(&level.to_ne_bytes());
        buf[sd.sd_flags..sd.sd_flags + 4].copy_from_slice(&flags.to_ne_bytes());
        buf[sd.sd_name..sd.sd_name + 8].copy_from_slice(&name_kva.to_ne_bytes());
        buf[sd.sd_span_weight..sd.sd_span_weight + 4].copy_from_slice(&span_weight.to_ne_bytes());
        buf[sd.sd_balance_interval..sd.sd_balance_interval + 4]
            .copy_from_slice(&balance_interval.to_ne_bytes());
        if let Some(off) = sd.sd_newidle_call {
            buf[off..off + 4].copy_from_slice(&newidle_call.to_ne_bytes());
        }
        buf[so.sd_lb_count..so.sd_lb_count + 4].copy_from_slice(&lb_count_0.to_ne_bytes());
        buf[so.sd_alb_pushed..so.sd_alb_pushed + 4].copy_from_slice(&alb_pushed.to_ne_bytes());
        buf[so.sd_ttwu_wake_remote..so.sd_ttwu_wake_remote + 4]
            .copy_from_slice(&ttwu_wake_remote.to_ne_bytes());
        buf
    }

    #[test]
    fn read_sched_domain_tree_null_sd() {
        // rq->sd is null — should return None.
        let sd_off = test_sched_domain_offsets();
        let buf = vec![0u8; 512];
        // SAFETY: buf is a live local buffer (Vec<u8> or stack array)
        // whose backing storage outlives the GuestMem use.
        let mem = unsafe { GuestMem::new(buf.as_ptr() as *mut u8, buf.len() as u64) };
        let result = read_sched_domain_tree(&mem, 0, &sd_off, 0);
        assert!(result.is_none());
    }

    #[test]
    fn read_sched_domain_tree_single_domain() {
        let sd_off = test_sched_domain_offsets();

        // Build: rq at PA 0 with rq->sd pointing to a domain.
        // Domain at some offset in the buffer, parent=0 (no parent).
        // page_offset=0 so KVA == PA for testing.
        let sd_pa: u64 = 1024;
        let name_pa: u64 = 2048;

        let sd_buf = make_sd_buffer(&sd_off, 0, 0, 0x42, name_pa, 4, 64, 15, 10, 3, 7);
        let name_bytes = b"SMT\0";

        // Build combined buffer: rq region + sd region + name region.
        let total_size = (name_pa as usize) + 16;
        let mut buf = vec![0u8; total_size];

        // Write rq->sd pointer (KVA == PA since page_offset=0).
        buf[sd_off.rq_sd..sd_off.rq_sd + 8].copy_from_slice(&sd_pa.to_ne_bytes());

        // Write sched_domain at sd_pa.
        buf[sd_pa as usize..sd_pa as usize + sd_buf.len()].copy_from_slice(&sd_buf);

        // Write name string.
        buf[name_pa as usize..name_pa as usize + name_bytes.len()].copy_from_slice(name_bytes);

        // SAFETY: buf is a live local buffer (Vec<u8> or stack array)
        // whose backing storage outlives the GuestMem use.
        let mem = unsafe { GuestMem::new(buf.as_ptr() as *mut u8, buf.len() as u64) };
        let domains = read_sched_domain_tree(&mem, 0, &sd_off, 0).unwrap();

        assert_eq!(domains.len(), 1);
        assert_eq!(domains[0].level, 0);
        assert_eq!(domains[0].name, "SMT");
        assert_eq!(domains[0].flags, 0x42);
        assert_eq!(domains[0].span_weight, 4);
        assert_eq!(domains[0].balance_interval, 64);
        assert_eq!(domains[0].newidle_call, Some(15));
        let stats = domains[0].stats.as_ref().unwrap();
        assert_eq!(stats.lb_count[0], 10);
        assert_eq!(stats.alb_pushed, 3);
        assert_eq!(stats.ttwu_wake_remote, 7);
    }

    #[test]
    fn read_sched_domain_tree_two_levels() {
        let sd_off = test_sched_domain_offsets();

        // page_offset=0 so KVA == PA.
        let sd0_pa: u64 = 1024;
        let sd1_pa: u64 = 2048;
        let name0_pa: u64 = 3072;
        let name1_pa: u64 = 3088;

        // Domain 0 (SMT, level 0) -> parent = Domain 1
        let sd0_buf = make_sd_buffer(&sd_off, sd1_pa, 0, 0x10, name0_pa, 2, 32, 8, 5, 1, 2);
        // Domain 1 (MC, level 1) -> parent = 0 (top)
        let sd1_buf = make_sd_buffer(&sd_off, 0, 1, 0x20, name1_pa, 8, 128, 22, 20, 4, 10);

        let total_size = 3104;
        let mut buf = vec![0u8; total_size];

        // rq->sd -> domain 0
        buf[sd_off.rq_sd..sd_off.rq_sd + 8].copy_from_slice(&sd0_pa.to_ne_bytes());
        buf[sd0_pa as usize..sd0_pa as usize + sd0_buf.len()].copy_from_slice(&sd0_buf);
        buf[sd1_pa as usize..sd1_pa as usize + sd1_buf.len()].copy_from_slice(&sd1_buf);
        buf[name0_pa as usize..name0_pa as usize + 4].copy_from_slice(b"SMT\0");
        buf[name1_pa as usize..name1_pa as usize + 3].copy_from_slice(b"MC\0");

        // SAFETY: buf is a live local buffer (Vec<u8> or stack array)
        // whose backing storage outlives the GuestMem use.
        let mem = unsafe { GuestMem::new(buf.as_ptr() as *mut u8, buf.len() as u64) };
        let domains = read_sched_domain_tree(&mem, 0, &sd_off, 0).unwrap();

        assert_eq!(domains.len(), 2);
        // First = lowest level (SMT).
        assert_eq!(domains[0].level, 0);
        assert_eq!(domains[0].name, "SMT");
        assert_eq!(domains[0].span_weight, 2);
        assert_eq!(domains[0].balance_interval, 32);
        assert_eq!(domains[0].newidle_call, Some(8));
        let s0 = domains[0].stats.as_ref().unwrap();
        assert_eq!(s0.lb_count[0], 5);
        // Second = higher level (MC).
        assert_eq!(domains[1].level, 1);
        assert_eq!(domains[1].name, "MC");
        assert_eq!(domains[1].span_weight, 8);
        assert_eq!(domains[1].balance_interval, 128);
        assert_eq!(domains[1].newidle_call, Some(22));
        let s1 = domains[1].stats.as_ref().unwrap();
        assert_eq!(s1.lb_count[0], 20);
        assert_eq!(s1.alb_pushed, 4);
        assert_eq!(s1.ttwu_wake_remote, 10);
    }

    #[test]
    fn read_sched_domain_tree_self_reference_breaks_cycle() {
        let sd_off = test_sched_domain_offsets();

        // Self-referential: sd->parent == sd. With the visited-set
        // cycle check, the walker emits sd exactly once and stops on
        // the next iteration rather than emitting the same domain
        // MAX_DEPTH times.
        let sd_pa: u64 = 1024;
        let sd_buf = make_sd_buffer(&sd_off, sd_pa, 0, 0, 0, 1, 0, 0, 0, 0, 0);

        let total_size = sd_pa as usize + sd_buf.len();
        let mut buf = vec![0u8; total_size];
        buf[sd_off.rq_sd..sd_off.rq_sd + 8].copy_from_slice(&sd_pa.to_ne_bytes());
        buf[sd_pa as usize..sd_pa as usize + sd_buf.len()].copy_from_slice(&sd_buf);

        // SAFETY: buf is a live local buffer (Vec<u8> or stack array)
        // whose backing storage outlives the GuestMem use.
        let mem = unsafe { GuestMem::new(buf.as_ptr() as *mut u8, buf.len() as u64) };
        let domains = read_sched_domain_tree(&mem, 0, &sd_off, 0).unwrap();

        assert_eq!(
            domains.len(),
            1,
            "self-referential sd should produce exactly one snapshot"
        );
    }

    #[test]
    fn read_sched_domain_tree_max_depth_bound_on_long_chain() {
        let sd_off = test_sched_domain_offsets();
        // Per-struct size covers make_sd_buffer's layout
        // (sd_ttwu_move_balance=232 + 4 bytes for that u32 + 8 bytes
        // guard = 244) rounded up to the next multiple of 8. The
        // readers go through `GuestMem::read_u32` / `read_u64`, which
        // call `read_volatile_bytes<N>` — per-byte volatile reads that
        // are always 1-aligned and recompose the integer via
        // `from_ne_bytes`, so misaligned PAs are safe. The 8-alignment
        // baked into this stride is therefore no longer load-bearing
        // for correctness; it remains because every real kernel
        // sched_domain is `__randomize_layout`-aligned past u64 and
        // matching that stride keeps the fixture realistic.
        //
        // Byte-wise volatile has no atomicity guarantee: a concurrent
        // guest-side write mid-read can produce a torn integer. The
        // monitor treats its samples as best-effort, so tearing is
        // acceptable here. `from_ne_bytes` uses host endianness;
        // x86_64/aarch64 guests and hosts share little-endian, so
        // the recomposed value matches the guest's stored value.
        //
        // Each sched_domain lives at a distinct PA and points at the
        // next via sd->parent, forming an acyclic chain longer than
        // MAX_DEPTH so the depth bound — not the visited set — is
        // what stops the walk.
        const SD_SIZE: u64 = 248;
        const CHAIN_LEN: usize = 10;
        let first_pa: u64 = 1024;

        let pa = |i: usize| first_pa + (i as u64) * SD_SIZE;
        let total_size = pa(CHAIN_LEN) as usize;
        let mut buf = vec![0u8; total_size];

        // rq->sd -> first domain.
        buf[sd_off.rq_sd..sd_off.rq_sd + 8].copy_from_slice(&pa(0).to_ne_bytes());

        for i in 0..CHAIN_LEN {
            let parent_kva = if i + 1 == CHAIN_LEN { 0 } else { pa(i + 1) };
            let sd_buf = make_sd_buffer(&sd_off, parent_kva, i as i32, 0, 0, 1, 0, 0, 0, 0, 0);
            let start = pa(i) as usize;
            buf[start..start + sd_buf.len()].copy_from_slice(&sd_buf);
        }

        // SAFETY: buf is a live local buffer (Vec<u8> or stack array)
        // whose backing storage outlives the GuestMem use.
        let mem = unsafe { GuestMem::new(buf.as_ptr() as *mut u8, buf.len() as u64) };
        let domains = read_sched_domain_tree(&mem, 0, &sd_off, 0).unwrap();

        assert_eq!(
            domains.len(),
            8,
            "acyclic chain of {CHAIN_LEN} levels must truncate at MAX_DEPTH=8"
        );
        // Sanity: the emitted levels are the first 8 in order.
        for (i, snap) in domains.iter().enumerate() {
            assert_eq!(snap.level, i as i32, "level mismatch at index {i}");
        }
    }

    #[test]
    fn read_sched_domain_tree_out_of_bounds_pa() {
        let sd_off = test_sched_domain_offsets();

        // rq->sd points to a KVA that translates to a PA beyond guest memory.
        let bad_kva: u64 = 0xFFFF_FFFF_FFFF_0000;
        let mut buf = vec![0u8; 512];
        buf[sd_off.rq_sd..sd_off.rq_sd + 8].copy_from_slice(&bad_kva.to_ne_bytes());

        // SAFETY: buf is a live local buffer (Vec<u8> or stack array)
        // whose backing storage outlives the GuestMem use.
        let mem = unsafe { GuestMem::new(buf.as_ptr() as *mut u8, buf.len() as u64) };
        // page_offset=0 -> PA = bad_kva which is > buf.len().
        let domains = read_sched_domain_tree(&mem, 0, &sd_off, 0);

        // Should return Some(empty vec) — non-null sd but untranslatable.
        assert!(domains.is_some());
        assert!(domains.unwrap().is_empty());
    }

    #[test]
    fn read_sched_domain_tree_newidle_none() {
        // 6.16 kernel: newidle_call/success/ratio are absent.
        // Other fields (level, name, span_weight, balance_interval) must
        // still populate correctly.
        let mut sd_off = test_sched_domain_offsets();
        sd_off.sd_newidle_call = None;
        sd_off.sd_newidle_success = None;
        sd_off.sd_newidle_ratio = None;

        let sd_pa: u64 = 1024;
        let name_pa: u64 = 2048;

        let sd_buf = make_sd_buffer(&sd_off, 0, 0, 0x42, name_pa, 4, 64, 0, 10, 3, 7);
        let name_bytes = b"SMT\0";

        let total_size = (name_pa as usize) + 16;
        let mut buf = vec![0u8; total_size];

        buf[sd_off.rq_sd..sd_off.rq_sd + 8].copy_from_slice(&sd_pa.to_ne_bytes());
        buf[sd_pa as usize..sd_pa as usize + sd_buf.len()].copy_from_slice(&sd_buf);
        buf[name_pa as usize..name_pa as usize + name_bytes.len()].copy_from_slice(name_bytes);

        // SAFETY: buf is a live local buffer (Vec<u8> or stack array)
        // whose backing storage outlives the GuestMem use.
        let mem = unsafe { GuestMem::new(buf.as_ptr() as *mut u8, buf.len() as u64) };
        let domains = read_sched_domain_tree(&mem, 0, &sd_off, 0).unwrap();

        assert_eq!(domains.len(), 1);
        assert_eq!(domains[0].level, 0);
        assert_eq!(domains[0].name, "SMT");
        assert_eq!(domains[0].flags, 0x42);
        assert_eq!(domains[0].span_weight, 4);
        assert_eq!(domains[0].balance_interval, 64);
        assert_eq!(domains[0].newidle_call, None);
        assert_eq!(domains[0].newidle_success, None);
        assert_eq!(domains[0].newidle_ratio, None);
        let stats = domains[0].stats.as_ref().unwrap();
        assert_eq!(stats.lb_count[0], 10);
        assert_eq!(stats.alb_pushed, 3);
        assert_eq!(stats.ttwu_wake_remote, 7);
    }

    #[test]
    fn read_u32_array_known_values() {
        let mut buf = [0u8; 16];
        buf[0..4].copy_from_slice(&10u32.to_ne_bytes());
        buf[4..8].copy_from_slice(&20u32.to_ne_bytes());
        buf[8..12].copy_from_slice(&30u32.to_ne_bytes());
        // SAFETY: buf is a live local buffer (Vec<u8> or stack array)
        // whose backing storage outlives the GuestMem use.
        let mem = unsafe { GuestMem::new(buf.as_ptr() as *mut u8, buf.len() as u64) };
        let arr = read_u32_array(&mem, 0, 0);
        assert_eq!(arr, [10, 20, 30]);
    }

    /// End-to-end BTF-resolved offsets → reader readback.
    ///
    /// Parses `KernelOffsets` from the real test vmlinux (skipping when
    /// no cached test kernel is available), writes known values at the
    /// BTF-resolved field offsets in a synthetic byte buffer, and asserts
    /// `read_rq_stats` returns exactly those values. Catches drift
    /// between BTF parsing and reader field arithmetic.
    #[test]
    fn btf_offsets_couple_with_rq_reader() {
        let path = match crate::monitor::find_test_vmlinux() {
            Some(p) => p,
            None => skip!("no test vmlinux available"),
        };
        let offsets = crate::test_support::require_kernel_offsets(&path);

        let max_scalar_off = offsets.rq_clock + 8;
        let max_scx_off = offsets.rq_scx + offsets.scx_rq_local_dsq + offsets.dsq_nr + 4;
        let max_flags_off = offsets.rq_scx + offsets.scx_rq_flags + 4;
        let size = max_scalar_off.max(max_scx_off).max(max_flags_off) + 64;
        let mut buf = vec![0u8; size];

        let nr_running: u32 = 0xDEAD_BEEF;
        let scx_nr: u32 = 0x1234_5678;
        let dsq_depth: u32 = 0x0BAD_F00D;
        let clock: u64 = 0xCAFE_BABE_1357_9BDF;
        let flags: u32 = 0xA5A5_A5A5;

        buf[offsets.rq_nr_running..offsets.rq_nr_running + 4]
            .copy_from_slice(&nr_running.to_ne_bytes());
        buf[offsets.rq_clock..offsets.rq_clock + 8].copy_from_slice(&clock.to_ne_bytes());
        let scx_nr_off = offsets.rq_scx + offsets.scx_rq_nr_running;
        buf[scx_nr_off..scx_nr_off + 4].copy_from_slice(&scx_nr.to_ne_bytes());
        let scx_flags_off = offsets.rq_scx + offsets.scx_rq_flags;
        buf[scx_flags_off..scx_flags_off + 4].copy_from_slice(&flags.to_ne_bytes());
        let dsq_off = offsets.rq_scx + offsets.scx_rq_local_dsq + offsets.dsq_nr;
        buf[dsq_off..dsq_off + 4].copy_from_slice(&dsq_depth.to_ne_bytes());

        // SAFETY: buf is a live local buffer (Vec<u8> or stack array)
        // whose backing storage outlives the GuestMem use.
        let mem = unsafe { GuestMem::new(buf.as_ptr() as *mut u8, buf.len() as u64) };
        let snap = read_rq_stats(&mem, 0, &offsets);
        assert_eq!(snap.nr_running, nr_running);
        assert_eq!(snap.scx_nr_running, scx_nr);
        assert_eq!(snap.local_dsq_depth, dsq_depth);
        assert_eq!(snap.rq_clock, clock);
        assert_eq!(snap.scx_flags, flags);
    }

    // ---- GuestMem write/resolve coverage --------------------------
    //
    // Pin the bounds-check semantics of `write_scalar` (the sole
    // gateway for `write_u8` / `write_u32` / `write_u64`), the
    // size-vs-offset interaction documented on `GuestMem::new`, and
    // the multi-region `resolve_ptr` routing that backs every read
    // and write on a NUMA layout.

    #[test]
    fn write_u32_at_boundary_writes_full_word() {
        let mut buf = [0u8; 16];
        // SAFETY: buf outlives the GuestMem use.
        let mem = unsafe { GuestMem::new(buf.as_mut_ptr(), buf.len() as u64) };
        // PA 12 + 4 = 16 == size: write_scalar's `>` bound is not
        // crossed, so the full word lands at the very end of the
        // mapping.
        mem.write_u32(12, 0, 0xDEAD_BEEF);
        assert_eq!(mem.read_u32(12, 0), 0xDEAD_BEEF);
        assert_eq!(
            u32::from_ne_bytes(buf[12..16].try_into().unwrap()),
            0xDEAD_BEEF
        );
    }

    #[test]
    fn write_u32_one_past_boundary_is_noop() {
        let mut buf = [0u8; 16];
        // SAFETY: buf outlives the GuestMem use.
        let mem = unsafe { GuestMem::new(buf.as_mut_ptr(), buf.len() as u64) };
        // PA 13 + 4 = 17 > 16: write must drop silently.
        mem.write_u32(13, 0, 0xFFFF_FFFF);
        assert_eq!(buf, [0u8; 16]);
    }

    #[test]
    fn write_u8_at_boundary_writes_last_byte() {
        let mut buf = [0u8; 4];
        // SAFETY: buf outlives the GuestMem use.
        let mem = unsafe { GuestMem::new(buf.as_mut_ptr(), buf.len() as u64) };
        // PA 3 + 1 = 4 == size.
        mem.write_u8(3, 0, 0xAB);
        assert_eq!(buf[3], 0xAB);
    }

    #[test]
    fn write_scalar_offset_arg_is_added_to_pa() {
        // `write_scalar` computes `addr = pa + offset`, so a write
        // through (pa, offset) must land at the same byte as a write
        // through (pa+offset, 0). Pin this so the offset path can
        // never silently drift from the pa path.
        let mut buf = [0u8; 32];
        // SAFETY: buf outlives the GuestMem use.
        let mem = unsafe { GuestMem::new(buf.as_mut_ptr(), buf.len() as u64) };
        mem.write_u64(8, 16, 0x0123_4567_89AB_CDEF);
        assert_eq!(mem.read_u64(24, 0), 0x0123_4567_89AB_CDEF);
        assert_eq!(
            u64::from_ne_bytes(buf[24..32].try_into().unwrap()),
            0x0123_4567_89AB_CDEF
        );
    }

    #[test]
    fn write_scalar_offset_only_out_of_bounds_is_noop() {
        // pa fits but pa+offset crosses the boundary. The bounds
        // check must combine pa and offset, not check them
        // independently.
        let mut buf = [0xCCu8; 8];
        // SAFETY: buf outlives the GuestMem use.
        let mem = unsafe { GuestMem::new(buf.as_mut_ptr(), buf.len() as u64) };
        // pa=4 (in-bounds), offset=4, addr=8, addr+8=16 > 8.
        mem.write_u64(4, 4, 0xFFFF_FFFF_FFFF_FFFF);
        assert_eq!(buf, [0xCCu8; 8]);
    }

    #[test]
    fn guest_mem_new_size_smaller_than_write_offset_is_noop() {
        // Construct a GuestMem reporting size N over a backing
        // buffer of >= N bytes, then attempt a write at offset > N.
        // The bounds check uses the *declared* size, so the write
        // must drop and the backing bytes past `size` must remain
        // untouched.
        let mut backing = [0u8; 32];
        let declared_size: u64 = 8;
        // SAFETY: backing outlives the GuestMem use; declared_size
        // is <= backing.len() so reads/writes stay within the
        // allocated buffer when they are accepted.
        let mem = unsafe { GuestMem::new(backing.as_mut_ptr(), declared_size) };
        // Write at byte 16 (well past declared size 8) — must noop.
        mem.write_u64(16, 0, 0xDEAD_BEEF_CAFE_1234);
        assert_eq!(backing, [0u8; 32]);
        // The same write inside declared bounds must succeed.
        mem.write_u64(0, 0, 0xDEAD_BEEF_CAFE_1234);
        assert_eq!(
            u64::from_ne_bytes(backing[0..8].try_into().unwrap()),
            0xDEAD_BEEF_CAFE_1234
        );
        // Bytes past the declared size remain untouched.
        assert_eq!(backing[8..32], [0u8; 24]);
    }

    #[test]
    fn resolve_ptr_multi_region_routes_to_correct_region() {
        // Two distinct host buffers (separate allocations) wired
        // into a single GuestMem with a gap between their DRAM
        // offsets. Reads and writes must route to the right host
        // buffer based on offset; otherwise multi-region NUMA
        // layouts would silently corrupt or read stale data.
        let mut buf0 = [0xAAu8; 64];
        let mut buf1 = [0xBBu8; 64];
        let regions = vec![
            MemRegion {
                host_ptr: buf0.as_mut_ptr(),
                offset: 0,
                size: 64,
            },
            MemRegion {
                host_ptr: buf1.as_mut_ptr(),
                offset: 1024, // gap from 64..1024
                size: 64,
            },
        ];
        // SAFETY: buf0 and buf1 outlive the GuestMem use; each
        // region's host_ptr addresses a 64-byte mapping.
        let mem = unsafe { GuestMem::from_regions_for_test(regions) };

        // Reads within region 0 see buf0's contents.
        assert_eq!(mem.read_u8(0, 0), 0xAA);
        assert_eq!(mem.read_u8(63, 0), 0xAA);
        // Reads within region 1 see buf1's contents.
        assert_eq!(mem.read_u8(1024, 0), 0xBB);
        assert_eq!(mem.read_u8(1087, 0), 0xBB);
        // Reads in the gap return 0 (resolve_ptr -> None ->
        // read_scalar returns zeroed bytes).
        assert_eq!(mem.read_u8(64, 0), 0);
        assert_eq!(mem.read_u8(512, 0), 0);
        assert_eq!(mem.read_u8(1023, 0), 0);

        // A write in region 1 hits buf1 and leaves buf0 alone.
        mem.write_u32(1024, 0, 0x1234_5678);
        assert_eq!(buf0, [0xAAu8; 64]);
        assert_eq!(
            u32::from_ne_bytes(buf1[0..4].try_into().unwrap()),
            0x1234_5678
        );

        // A write in the gap is a no-op (resolve_ptr -> None).
        // Re-zero buf1[60..64] then attempt a write into the gap and
        // verify both buffers remain untouched at their unaffected
        // byte ranges.
        let buf0_snapshot = buf0;
        let buf1_snapshot = buf1;
        mem.write_u32(900, 0, 0xFFFF_FFFF);
        assert_eq!(buf0, buf0_snapshot);
        assert_eq!(buf1, buf1_snapshot);
    }

    #[test]
    fn resolve_ptr_multi_region_read_ring_volatile_routes_correctly() {
        // `read_ring_volatile` (in shm_ring.rs) reads byte-by-byte
        // via `mem.read_u8`. With a multi-region GuestMem where the
        // ring's data area sits inside region 1 (past the end of
        // region 0), each `read_u8` must resolve to region 1's host
        // pointer — not stale bytes past region 0's end.
        let mut buf0 = [0u8; 64];
        let mut buf1 = [0u8; 64];
        // Plant a known pattern at the start of region 1.
        buf1[0..8].copy_from_slice(&[1, 2, 3, 4, 5, 6, 7, 8]);
        let regions = vec![
            MemRegion {
                host_ptr: buf0.as_mut_ptr(),
                offset: 0,
                size: 64,
            },
            MemRegion {
                host_ptr: buf1.as_mut_ptr(),
                offset: 1024,
                size: 64,
            },
        ];
        // SAFETY: backing buffers outlive the GuestMem use.
        let mem = unsafe { GuestMem::from_regions_for_test(regions) };

        // Byte-by-byte reads from region 1's first 8 bytes must
        // return the planted pattern, not bytes from region 0.
        for (i, expected) in [1, 2, 3, 4, 5, 6, 7, 8].iter().enumerate() {
            assert_eq!(mem.read_u8(1024 + i as u64, 0), *expected);
        }
    }

    #[test]
    fn resolve_ptr_offset_at_exact_region_end_is_out_of_region() {
        // resolve_ptr's `local < r.size` check is strict: an offset
        // equal to a region's end must fall outside that region.
        // For a single-region GuestMem this means offset == size
        // resolves to None.
        let mut buf = [0xCCu8; 16];
        // SAFETY: buf outlives the GuestMem use.
        let mem = unsafe { GuestMem::new(buf.as_mut_ptr(), buf.len() as u64) };
        // read_scalar's bounds check (`addr + N > size`) catches
        // this for >= 1-byte reads, returning zero. The exact-end
        // u8 read at offset 16 has addr=16, addr+1=17 > 16, so the
        // outer check returns 0 before resolve_ptr is reached.
        assert_eq!(mem.read_u8(16, 0), 0);
    }

    /// Unified bounds-check pin for `write_u8` and `read_u8`: a
    /// single GuestMem of declared size N is exercised at three
    /// load-bearing positions:
    /// 1. last valid offset (N-1) — write must land and a read
    ///    back must observe the written byte;
    /// 2. one past the end (N) — write must be a silent no-op
    ///    (no panic, no out-of-bounds write to the backing buffer);
    /// 3. one past the end (N) — read must return 0 (the
    ///    `read_scalar` `addr + N as u64 > self.size` arm fires
    ///    before `resolve_ptr` is consulted).
    ///
    /// `write_scalar`'s bound is `addr + N as u64 > self.size`. For
    /// the 1-byte path (N=1) the boundary is `addr == size - 1`
    /// inclusive, `addr == size` exclusive. Pinning all three
    /// positions on one fixture catches a regression that flips
    /// the `>` to `>=` (which would reject the last valid byte) or
    /// drops the bound entirely (which would scribble past the
    /// declared mapping).
    #[test]
    fn write_u8_and_read_u8_bounds_at_declared_size() {
        const SIZE: u64 = 8;
        let mut buf = [0u8; SIZE as usize];
        // SAFETY: buf outlives the GuestMem use; declared_size
        // matches the backing allocation so even an accepted write
        // at the last valid offset stays inside `buf`.
        let mem = unsafe { GuestMem::new(buf.as_mut_ptr(), SIZE) };

        // (1) Last valid offset (SIZE - 1). pa=SIZE-1, offset=0,
        //     addr=SIZE-1, addr+1 = SIZE which is NOT > SIZE, so
        //     the write is accepted.
        mem.write_u8(SIZE - 1, 0, 0xAB);
        assert_eq!(
            mem.read_u8(SIZE - 1, 0),
            0xAB,
            "write at last valid offset must round-trip via read_u8"
        );
        assert_eq!(
            buf[(SIZE - 1) as usize],
            0xAB,
            "write at last valid offset must land in the backing byte"
        );

        // (2) Past the end (SIZE). pa=SIZE, offset=0, addr=SIZE,
        //     addr+1 = SIZE+1 > SIZE → bound trips, write is a
        //     silent no-op. Snapshot the buffer to confirm no other
        //     bytes moved either.
        let snapshot = buf;
        mem.write_u8(SIZE, 0, 0xFF);
        assert_eq!(
            buf, snapshot,
            "write past the end must be a silent no-op — no byte of \
             the backing buffer may change"
        );

        // (3) Read past the end. addr=SIZE, addr+1 > SIZE → returns 0.
        assert_eq!(
            mem.read_u8(SIZE, 0),
            0,
            "read past the end must return 0, not stale memory"
        );
        // One byte further is also out of bounds.
        assert_eq!(
            mem.read_u8(SIZE + 1, 0),
            0,
            "read several bytes past the end must also return 0"
        );
    }
}