ktstr 0.4.9

/// Shared-memory ring buffer for guest-to-host data transfer.
///
/// The guest writes TLV-framed messages into a fixed region at the top of
/// guest physical memory. The host drains both mid-flight and after VM exit.
/// Multiple guest-side producers (step executor, sched-exit-mon) serialize
/// writes via `SHM_WRITE_LOCK`. Single consumer (host), no read-side locking.
///
/// Memory layout:
///   [ShmRingHeader (40 bytes)] [data (capacity bytes)]
///
/// The SHM region is excluded from usable RAM: on x86_64 via an E820 gap
/// (no E820 entry covers it), on aarch64 via FDT /reserved-memory and
/// /memreserve/. The guest init binary discovers the region via KTSTR_SHM_BASE
/// and KTSTR_SHM_SIZE parameters on the kernel command line.
use std::ptr;

use zerocopy::{FromBytes, IntoBytes};

/// Result of a successful `/dev/mem` mmap of the SHM region.
pub(crate) struct ShmMmap {
    /// Pointer to the start of the SHM region (page-offset adjusted).
    pub ptr: *mut u8,
    /// Base address passed to munmap (page-aligned).
    pub map_base: *mut libc::c_void,
    /// Size passed to munmap.
    pub map_size: usize,
}

/// Page-aligned mmap of a physical address range via an open `/dev/mem` fd.
/// Returns the adjusted pointer to `shm_base` within the mapping.
pub(crate) fn mmap_devmem(
    fd: std::os::unix::io::RawFd,
    shm_base: u64,
    shm_size: u64,
) -> Option<ShmMmap> {
    let page_size = unsafe { libc::sysconf(libc::_SC_PAGESIZE) } as u64;
    let aligned_base = shm_base & !(page_size - 1);
    let offset_in_page = (shm_base - aligned_base) as usize;
    let map_size = shm_size as usize + offset_in_page;

    let map_base = unsafe {
        libc::mmap(
            std::ptr::null_mut(),
            map_size,
            libc::PROT_READ | libc::PROT_WRITE,
            libc::MAP_SHARED,
            fd,
            aligned_base as libc::off_t,
        )
    };
    if map_base == libc::MAP_FAILED {
        return None;
    }

    let ptr = unsafe { (map_base as *mut u8).add(offset_in_page) };
    Some(ShmMmap {
        ptr,
        map_base,
        map_size,
    })
}

/// Magic value identifying a valid SHM ring header.
pub const SHM_RING_MAGIC: u32 = 0x5354_4d52; // "STMR"

/// Message type for stimulus events written by the guest step executor.
pub const MSG_TYPE_STIMULUS: u32 = 0x5354_494D; // "STIM"

/// Message type for scenario start marker.
pub const MSG_TYPE_SCENARIO_START: u32 = 0x5343_5354; // "SCST"

/// Message type for scenario end marker.
pub const MSG_TYPE_SCENARIO_END: u32 = 0x5343_454E; // "SCEN"

/// Message type for guest exit code (payload: 4-byte i32).
pub const MSG_TYPE_EXIT: u32 = 0x4558_4954; // "EXIT"

/// Message type for test result (payload: JSON-encoded AssertResult).
pub const MSG_TYPE_TEST_RESULT: u32 = 0x5445_5354; // "TEST"

/// Message type for scheduler process exit (payload: 4-byte i32 exit code).
/// Written by the guest init when the scheduler child process terminates
/// during test execution. The host monitor thread can detect this via
/// mid-flight SHM drain and terminate the VM early instead of waiting
/// for the full watchdog timeout.
pub const MSG_TYPE_SCHED_EXIT: u32 = 0x5343_4458; // "SCDX"

/// Message type for guest crash (payload: UTF-8 panic message + backtrace).
/// Written by the panic hook in rust_init.rs. SHM delivery is reliable
/// (memcpy to mapped memory) unlike serial which truncates large backtraces
/// because the UART cannot drain fast enough before reboot.
pub const MSG_TYPE_CRASH: u32 = 0x4352_5348; // "CRSH"

/// Message type for per-payload-invocation metrics (payload: JSON-encoded
/// [`PayloadMetrics`](crate::test_support::PayloadMetrics)). One entry
/// per terminal `.run()` / `.wait()` / `.kill()` / `.try_wait()` on a
/// [`PayloadRun`](crate::scenario::payload_run::PayloadRun) or
/// [`PayloadHandle`](crate::scenario::payload_run::PayloadHandle). The
/// host-side eval loop drains every `MSG_TYPE_PAYLOAD_METRICS` entry
/// in order and feeds the resulting `Vec<PayloadMetrics>` to the
/// sidecar writer so per-invocation provenance is preserved across
/// composed payload runs.
///
/// For [`OutputFormat::LlmExtract`](crate::test_support::OutputFormat::LlmExtract)
/// payloads, this carries an empty `metrics` vec — extraction runs
/// host-side post-VM-exit on the paired
/// [`MSG_TYPE_RAW_PAYLOAD_OUTPUT`] entry and replaces the empty vec
/// with the extracted metrics before sidecar write.
pub const MSG_TYPE_PAYLOAD_METRICS: u32 = 0x504d_4554; // "PMET"

/// Message type for raw stdout/stderr captured from a payload that
/// declared [`OutputFormat::LlmExtract`](crate::test_support::OutputFormat::LlmExtract).
/// Payload: JSON-encoded
/// [`RawPayloadOutput`](crate::test_support::RawPayloadOutput).
///
/// Emitted ALONGSIDE [`MSG_TYPE_PAYLOAD_METRICS`] (with empty
/// `metrics`) so the host can pair them by `payload_index`
/// equality. Both messages carry the same per-invocation
/// `payload_index` allocated from the guest's per-process
/// counter; the host's drain loop builds a
/// `HashMap<payload_index, slot>` over `PayloadMetrics` and looks
/// up each `RawPayloadOutput`'s index in O(1) — no reliance on SHM
/// emission order, which would conflate a `Json` payload that
/// produced zero numeric leaves with an LlmExtract placeholder.
/// The host then runs `extract_via_llm` on the captured text
/// stdout-primary with stderr-fallback, and replaces the matched
/// PayloadMetrics' empty `metrics` vec with the extracted result.
///
/// LLM extraction NEVER runs in the guest: the model (~2.4 GiB) does
/// not fit in the guest VM's available RAM, and the cache lives on
/// the host. The guest captures raw text only and ships it across
/// the SHM ring for host-side resolution.
pub const MSG_TYPE_RAW_PAYLOAD_OUTPUT: u32 = 0x5241_574f; // "RAWO"

/// Current header version.
pub const SHM_RING_VERSION: u32 = 1;

/// Upper bound on a valid ring capacity, in bytes.
///
/// `ShmRingHeader.capacity` is `u32` so a torn init or a guest writing
/// a wild value (e.g. `0xFFFF_FFFF`) can produce a capacity that
/// vastly exceeds the actual mapped SHM region. Using such a value
/// would make the drain path read beyond the mapped window (live: via
/// `mem.read_volatile` past the end of the region; snapshot: via
/// slice indexing into `buf` past `HEADER_SIZE + actual_capacity`),
/// aborting the host under `panic = "abort"`.
///
/// Chosen at 1 GiB — orders of magnitude above any realistic
/// `KTSTR_SHM_SIZE` (tens of KiB to a few MiB for metric / TLV
/// traffic) while leaving `u32::MAX` / the corrupt-pattern region
/// strictly rejected.
pub const MAX_SHM_CAPACITY: u32 = 1 << 30;

/// Byte offset within the SHM region for the host-to-guest dump request flag.
/// Occupies the first byte of the `control_bytes` field in ShmRingHeader (offset 12).
/// Host writes `DUMP_REQ_SYSRQ_D` to request a SysRq-D dump; guest polls
/// this byte, triggers the dump, and clears it back to 0.
pub const DUMP_REQ_OFFSET: usize = 12;

/// Value written to DUMP_REQ_OFFSET to request a SysRq-D dump.
pub const DUMP_REQ_SYSRQ_D: u8 = b'D';

/// Byte offset within the SHM region for the host-to-guest stall request flag.
/// Occupies the second byte of the `control_bytes` field in ShmRingHeader (offset 13).
/// Host writes `STALL_REQ_ACTIVATE` to request a scheduler stall; guest polls
/// this byte, creates /tmp/ktstr_stall, and clears it back to 0.
pub const STALL_REQ_OFFSET: usize = 13;

/// Value written to STALL_REQ_OFFSET to request a scheduler stall.
///
/// Consumed by the guest-side poll in `rust_init::shm_poll_loop`
/// (src/vmm/rust_init.rs). No production host-side writer exists yet
/// — the stall is activated by out-of-band pokes to the SHM byte
/// during debugging sessions. The const is kept in the lib crate (not
/// `#[cfg(test)]`) so the guest reader can reference it by name,
/// giving the wire protocol a single source of truth: a future
/// byte-value change at this definition automatically reaches the
/// reader without a hand-sync step.
pub const STALL_REQ_ACTIVATE: u8 = b'S';

/// Base offset within the SHM region for numbered signal slots.
/// Slots occupy bytes starting at offset 14 (third byte of `control_bytes`)
/// and extending into byte 15 (fourth byte of `control_bytes`), providing
/// 2 slots (0..1). AtomicU8 with Acquire/Release ordering.
pub const SIGNAL_SLOT_BASE: usize = 14;

/// Number of available signal slots.
const SIGNAL_SLOT_COUNT: usize = 2;

/// Value written to signal slot 0 by the host to request graceful shutdown.
/// Distinct from the BPF map write signal (value 1) so the guest poll loop
/// can differentiate.
pub const SIGNAL_SHUTDOWN_REQ: u8 = 0xDD;

/// Value written to slot 1 by the guest when probes are attached and the
/// scenario is about to start. The `start_bpf_map_write` thread polls for
/// this value before writing the crash trigger, ensuring probes capture
/// the crash rather than missing it.
pub const SIGNAL_PROBES_READY: u8 = 2;

/// Guest-side: poll SHM slot until non-zero or timeout.
/// Reads via AtomicU8 with Acquire ordering. The SHM mmap pointer
/// is cached in a OnceLock, initialized from /proc/cmdline during
/// the first call (or from `init_shm_ptr`).
pub fn wait_for(slot: u8, timeout: std::time::Duration) -> anyhow::Result<()> {
    assert!(
        (slot as usize) < SIGNAL_SLOT_COUNT,
        "signal slot {slot} out of range"
    );
    let (ptr, _) = shm_ptr()?;
    let offset = SIGNAL_SLOT_BASE + slot as usize;
    let atom = unsafe { &*(ptr.add(offset) as *const std::sync::atomic::AtomicU8) };
    let deadline = std::time::Instant::now() + timeout;
    while std::time::Instant::now() < deadline {
        if atom.load(std::sync::atomic::Ordering::Acquire) != 0 {
            return Ok(());
        }
        std::thread::sleep(std::time::Duration::from_millis(50));
    }
    anyhow::bail!("signal slot {slot} timed out after {timeout:?}")
}

/// Guest-side: set a slot to non-zero.
/// Writes via AtomicU8 with Release ordering.
pub fn signal(slot: u8) {
    signal_value(slot, 1);
}

/// Host-side: write 1 to a signal slot in guest memory.
/// `mem` provides direct access to guest DRAM;
/// `shm_base` is the DRAM-relative offset of the SHM region.
pub fn signal_guest(mem: &crate::monitor::reader::GuestMem, shm_base: u64, slot: u8) {
    signal_guest_value(mem, shm_base, slot, 1);
}

/// Host-side: write an arbitrary value to a signal slot in guest memory.
pub fn signal_guest_value(
    mem: &crate::monitor::reader::GuestMem,
    shm_base: u64,
    slot: u8,
    value: u8,
) {
    assert!(
        (slot as usize) < SIGNAL_SLOT_COUNT,
        "signal slot {slot} out of range"
    );
    mem.write_u8(shm_base, SIGNAL_SLOT_BASE + slot as usize, value);
}

/// Guest-side: read the current value of a signal slot.
pub fn read_signal(slot: u8) -> u8 {
    assert!(
        (slot as usize) < SIGNAL_SLOT_COUNT,
        "signal slot {slot} out of range"
    );
    let Ok((ptr, _)) = shm_ptr() else { return 0 };
    let offset = SIGNAL_SLOT_BASE + slot as usize;
    let atom = unsafe { &*(ptr.add(offset) as *const std::sync::atomic::AtomicU8) };
    atom.load(std::sync::atomic::Ordering::Acquire)
}

/// Guest-side: set a slot to a specific value.
pub fn signal_value(slot: u8, value: u8) {
    assert!(
        (slot as usize) < SIGNAL_SLOT_COUNT,
        "signal slot {slot} out of range"
    );
    let Ok((ptr, _)) = shm_ptr() else { return };
    let offset = SIGNAL_SLOT_BASE + slot as usize;
    let atom = unsafe { &*(ptr.add(offset) as *const std::sync::atomic::AtomicU8) };
    atom.store(value, std::sync::atomic::Ordering::Release);
}

/// Set the cached SHM base pointer and region size. Called from
/// `shm_poll_loop` (spawned by `start_shm_poll`) in the guest init
/// after the /dev/mem mmap succeeds.
pub fn init_shm_ptr(base: *mut u8, size: usize) {
    let _ = SHM_PTR.set(ShmPtr { ptr: base, size });
}

/// Guest-side: write a TLV message to the SHM ring using the cached
/// mmap pointer. No-op if SHM is not initialized.
///
/// Acquires `SHM_WRITE_LOCK` to serialize against concurrent writers
/// (sched-exit-mon thread and step executor). Operates on the raw
/// mmap pointer via volatile reads/writes — never materializes a
/// `&mut [u8]` over the SHM region. The host monitor thread reads
/// the same memory concurrently (`shm_drain_live`), so a `&mut [u8]`
/// would alias the host's `&` view and violate Rust's reference
/// rules, even though the lock serializes guest-side writers.
pub fn write_msg(msg_type: u32, payload: &[u8]) {
    let Ok((ptr, size)) = shm_ptr() else { return };
    // Safe to re-enter: write_ptr advances only after header + payload
    // land, so a mid-write panic cannot corrupt committed messages.
    let _guard = SHM_WRITE_LOCK.lock().unwrap_or_else(|e| e.into_inner());
    // SAFETY: `ptr` points to a `size`-byte mmap region that outlives
    // every guest thread (set once via `OnceLock` during init).
    unsafe { shm_write_raw(ptr, size, msg_type, payload) };
}

/// Guest-side: try to write a TLV message without blocking.
///
/// Uses `try_lock()` on `SHM_WRITE_LOCK`. If the lock is held (e.g.,
/// the panic occurred on the thread that holds it), silently returns
/// false so the caller can fall back to serial. No-op if SHM is not
/// initialized.
pub fn write_msg_nonblocking(msg_type: u32, payload: &[u8]) -> bool {
    let Ok((ptr, size)) = shm_ptr() else {
        return false;
    };
    // `try_lock` fails if the lock is contended OR poisoned. Both
    // map to "don't block, just drop this message" in the
    // non-blocking path — the caller is invariably an exit/dump
    // hook that cannot afford to wait for the writer.
    let Ok(_guard) = SHM_WRITE_LOCK.try_lock() else {
        return false;
    };
    // SAFETY: same invariant as `write_msg` — `ptr` is a valid
    // `size`-byte mmap region.
    unsafe { shm_write_raw(ptr, size, msg_type, payload) };
    true
}

/// Wrapper for a raw pointer + size that is Send+Sync.
/// SAFETY: The SHM pointer is set once via OnceLock during init and
/// points into a /dev/mem mmap that outlives all guest threads.
struct ShmPtr {
    ptr: *mut u8,
    size: usize,
}
unsafe impl Send for ShmPtr {}
unsafe impl Sync for ShmPtr {}

/// Cached SHM mmap pointer for guest-side signal operations.
static SHM_PTR: std::sync::OnceLock<ShmPtr> = std::sync::OnceLock::new();

/// Mutex serializing guest-side SHM ring writes. Prevents the sched-exit-mon
/// thread (write_msg) and the step executor (ShmWriter::write) from
/// concurrently modifying the ring's write_ptr.
pub static SHM_WRITE_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(());

/// Get the cached SHM mmap pointer and size, initializing from
/// /proc/cmdline if not already set.
fn shm_ptr() -> anyhow::Result<(*mut u8, usize)> {
    if let Some(p) = SHM_PTR.get() {
        return Ok((p.ptr, p.size));
    }
    // Lazy init from /proc/cmdline.
    let cmdline = std::fs::read_to_string("/proc/cmdline")
        .map_err(|e| anyhow::anyhow!("/proc/cmdline: {e}"))?;
    let (shm_base, shm_size) = parse_shm_params_from_str(&cmdline)
        .ok_or_else(|| anyhow::anyhow!("no SHM params in cmdline"))?;

    let fd = std::fs::OpenOptions::new()
        .read(true)
        .write(true)
        .open("/dev/mem")
        .map_err(|e| anyhow::anyhow!("/dev/mem open: {e}"))?;

    let m = mmap_devmem(
        std::os::unix::io::AsRawFd::as_raw_fd(&fd),
        shm_base,
        shm_size,
    )
    .ok_or_else(|| anyhow::anyhow!("/dev/mem mmap failed: {}", std::io::Error::last_os_error()))?;

    let size = shm_size as usize;
    let _ = SHM_PTR.set(ShmPtr { ptr: m.ptr, size });
    Ok((m.ptr, size))
}

/// Parse KTSTR_SHM_BASE and KTSTR_SHM_SIZE from a kernel command line string.
pub(crate) fn parse_shm_params_from_str(cmdline: &str) -> Option<(u64, u64)> {
    let base = cmdline
        .split_whitespace()
        .find(|s| s.starts_with("KTSTR_SHM_BASE="))?
        .strip_prefix("KTSTR_SHM_BASE=")?;
    let size = cmdline
        .split_whitespace()
        .find(|s| s.starts_with("KTSTR_SHM_SIZE="))?
        .strip_prefix("KTSTR_SHM_SIZE=")?;
    let base =
        u64::from_str_radix(base.trim_start_matches("0x").trim_start_matches("0X"), 16).ok()?;
    let size =
        u64::from_str_radix(size.trim_start_matches("0x").trim_start_matches("0X"), 16).ok()?;
    Some((base, size))
}

/// Ring buffer header at the start of the SHM region.
///
/// write_ptr and read_ptr are monotonically increasing byte offsets into
/// the data area. Actual position = ptr % capacity.
#[repr(C)]
#[derive(
    Clone, Copy, Default, FromBytes, IntoBytes, zerocopy::Immutable, zerocopy::KnownLayout,
)]
pub struct ShmRingHeader {
    pub magic: u32,
    pub version: u32,
    /// Data area size in bytes (region_size - sizeof(ShmRingHeader)).
    pub capacity: u32,
    /// Packed host→guest control bytes — NOT padding despite being
    /// declared as `u32` for alignment. Byte 0 = `DUMP_REQ_OFFSET`
    /// (SysRq-D dump trigger), byte 1 = `STALL_REQ_OFFSET` (scheduler
    /// stall trigger), bytes 2-3 = `SIGNAL_SLOT_BASE + {0, 1}` (2
    /// indexed `AtomicU8` signal slots). Read/written byte-wise with
    /// `Acquire`/`Release` ordering via the `*_OFFSET` / `*_BASE`
    /// constants above; the `u32` spelling exists only so the header
    /// remains a plain POD for zerocopy derive.
    pub control_bytes: u32,
    /// Total bytes written by the guest (monotonic).
    pub write_ptr: u64,
    /// Total bytes read by the host (monotonic).
    pub read_ptr: u64,
    /// Number of messages dropped by `shm_write`. Conflates three
    /// distinct failure modes:
    ///   1. ring-full — the common case, used-space + message would
    ///      exceed `capacity`;
    ///   2. total-size overflow — `MSG_HEADER_SIZE + payload.len()`
    ///      overflows `usize` (pathological payload, effectively
    ///      unreachable);
    ///   3. length-field overflow — `payload.len() > u32::MAX` so the
    ///      `ShmMessage.length` field cannot represent it
    ///      (unreachable in the current schema where `capacity: u32`
    ///      already caps payload size at ~4GB).
    ///
    /// Host telemetry readers treat all three as "producer lost a
    /// message"; the ring-full case dominates and the overflow cases
    /// exist only as defense-in-depth (see `shm_write`). Splitting
    /// this into separate counters would add header bytes and
    /// observable bytes for paths that should never fire in practice,
    /// so the single counter is the right tradeoff.
    pub drops: u64,
}

const _HEADER_SIZE: () = assert!(std::mem::size_of::<ShmRingHeader>() == 40);

impl ShmRingHeader {
    /// Build a fresh ring header for an SHM region of `shm_size` bytes.
    /// Saturating subtraction on `shm_size - HEADER_SIZE` means a mis-
    /// sized region (`shm_size < HEADER_SIZE`) surfaces as a zero-
    /// capacity ring rather than a panic — every `shm_write` then hits
    /// the ring-full branch and the operator sees the empty data area
    /// instead of losing the VMM to an arithmetic underflow before the
    /// layout error can surface.
    ///
    /// Single source of truth for magic/version/capacity field
    /// population: production `VmmState::init_shm_region`
    /// (src/vmm/mod.rs) and the `#[cfg(test)] shm_init` helper both
    /// call this, so a schema edit lands once.
    pub fn new(shm_size: usize) -> Self {
        let capacity = shm_size.saturating_sub(HEADER_SIZE);
        Self {
            magic: SHM_RING_MAGIC,
            version: SHM_RING_VERSION,
            capacity: capacity as u32,
            control_bytes: 0,
            write_ptr: 0,
            read_ptr: 0,
            drops: 0,
        }
    }
}

/// TLV message header preceding each payload in the ring.
///
/// CRC32 covers only the payload bytes (not this header).
#[repr(C)]
#[derive(
    Clone, Copy, Default, FromBytes, IntoBytes, zerocopy::Immutable, zerocopy::KnownLayout,
)]
pub struct ShmMessage {
    pub msg_type: u32,
    pub length: u32,
    pub crc32: u32,
    pub _pad: u32,
}

const _MSG_SIZE: () = assert!(std::mem::size_of::<ShmMessage>() == 16);

/// Size of the ShmRingHeader.
pub const HEADER_SIZE: usize = std::mem::size_of::<ShmRingHeader>();
/// Size of the ShmMessage TLV header.
pub const MSG_HEADER_SIZE: usize = std::mem::size_of::<ShmMessage>();

/// A parsed message from the ring buffer.
#[derive(Debug, Clone)]
#[allow(dead_code)]
pub struct ShmEntry {
    pub msg_type: u32,
    pub payload: Vec<u8>,
    /// True if the CRC32 matched.
    pub crc_ok: bool,
}

/// Result of draining the ring buffer.
#[derive(Debug, Clone, Default)]
#[allow(dead_code)]
pub struct ShmDrainResult {
    pub entries: Vec<ShmEntry>,
    pub drops: u64,
}

/// Payload for stimulus events written by the guest step executor.
///
/// Compact 24-byte struct describing the state after each step's ops
/// are applied. The host correlates these with monitor samples to map
/// scheduler telemetry to scenario phases.
#[repr(C)]
#[derive(Clone, Copy, Default, Debug, IntoBytes, zerocopy::Immutable, zerocopy::KnownLayout)]
pub struct StimulusPayload {
    /// Milliseconds since scenario start.
    pub elapsed_ms: u32,
    /// Index of the step that was just applied.
    pub step_index: u16,
    /// Number of ops applied in this step.
    pub op_count: u16,
    /// Bitmask of Op variant discriminants present in this step.
    pub op_kinds: u32,
    /// Number of live cgroups after this step: sum of step-local
    /// cgroups (from the current Step's `CgroupDef`s + `Op`s) and
    /// Backdrop-owned cgroups that persist across every Step.
    pub cgroup_count: u16,
    /// Total worker handles after this step: sum of step-local
    /// workers and Backdrop-spawned workers that persist across
    /// every Step.
    pub worker_count: u16,
    /// Sum of all workers' iteration counts at this step boundary.
    /// Read from shared MAP_SHARED counters in the step executor.
    pub total_iterations: u64,
}

const _STIMULUS_SIZE: () = assert!(std::mem::size_of::<StimulusPayload>() == 24);

/// Deserialized stimulus event from the SHM ring.
#[derive(Debug, Clone)]
#[allow(dead_code)]
pub struct StimulusEvent {
    pub elapsed_ms: u32,
    pub step_index: u16,
    pub op_count: u16,
    pub op_kinds: u32,
    pub cgroup_count: u16,
    pub worker_count: u16,
    pub total_iterations: u64,
}

impl StimulusEvent {
    /// Deserialize from raw payload bytes.
    pub fn from_payload(data: &[u8]) -> Option<Self> {
        if data.len() < std::mem::size_of::<StimulusPayload>() {
            return None;
        }
        Some(StimulusEvent {
            elapsed_ms: u32::from_ne_bytes(data[0..4].try_into().ok()?),
            step_index: u16::from_ne_bytes(data[4..6].try_into().ok()?),
            op_count: u16::from_ne_bytes(data[6..8].try_into().ok()?),
            op_kinds: u32::from_ne_bytes(data[8..12].try_into().ok()?),
            cgroup_count: u16::from_ne_bytes(data[12..14].try_into().ok()?),
            worker_count: u16::from_ne_bytes(data[14..16].try_into().ok()?),
            total_iterations: u64::from_ne_bytes(data[16..24].try_into().ok()?),
        })
    }
}

/// Test helper — initialize the SHM ring header at the given
/// offset in guest memory.
///
/// `buf` is the full guest memory slice. `shm_offset` is the byte offset
/// where the SHM region starts. `shm_size` is the total region size.
///
/// The header is built via [`ShmRingHeader::new`], the same constructor
/// production `VmmState::init_shm_region` uses — a schema change lands
/// once and both sites pick it up.
#[cfg(test)]
pub fn shm_init(buf: &mut [u8], shm_offset: usize, shm_size: usize) {
    let header = ShmRingHeader::new(shm_size);
    let hdr_bytes = header.as_bytes();
    buf[shm_offset..shm_offset + HEADER_SIZE].copy_from_slice(hdr_bytes);
    // Zero the data area.
    let data_start = shm_offset + HEADER_SIZE;
    let data_end = shm_offset + shm_size;
    buf[data_start..data_end].fill(0);
}

/// Read the ring header from guest memory via zerocopy.
///
/// `ShmRingHeader` derives `FromBytes` + `Immutable` + `KnownLayout`,
/// so any byte slice sized exactly `HEADER_SIZE` is a valid header —
/// all fields are fixed-width scalars with no invalid bit patterns.
/// `read_from_bytes` returns a `Result<Self>` that only fails on
/// size mismatch; the slice is always exactly `HEADER_SIZE` bytes by
/// construction, so `.unwrap()` is justified.
fn read_header(buf: &[u8], shm_offset: usize) -> ShmRingHeader {
    let s = &buf[shm_offset..shm_offset + HEADER_SIZE];
    ShmRingHeader::read_from_bytes(s).expect("HEADER_SIZE matches ShmRingHeader layout")
}

/// Read `len` bytes from the ring's data area starting at monotonic offset
/// `ptr`, handling wraparound.
fn read_ring_bytes(
    buf: &[u8],
    data_start: usize,
    capacity: usize,
    ptr: u64,
    len: usize,
) -> Vec<u8> {
    let mut out = vec![0u8; len];
    read_ring_into(buf, data_start, capacity, ptr, &mut out);
    out
}

/// Read `len` bytes from the ring into an existing buffer, handling wraparound.
fn read_ring_into(buf: &[u8], data_start: usize, capacity: usize, ptr: u64, out: &mut [u8]) {
    let len = out.len();
    let mut remaining = len;
    let mut src_pos = (ptr % capacity as u64) as usize;
    let mut dst_pos = 0;
    while remaining > 0 {
        let chunk = remaining.min(capacity - src_pos);
        out[dst_pos..dst_pos + chunk]
            .copy_from_slice(&buf[data_start + src_pos..data_start + src_pos + chunk]);
        dst_pos += chunk;
        src_pos = 0; // wrap
        remaining -= chunk;
    }
}

/// Drain all complete messages from the ring buffer.
///
/// `buf` is the full guest memory (read-only). `shm_offset` is the byte
/// offset where the SHM region starts.
pub fn shm_drain(buf: &[u8], shm_offset: usize) -> ShmDrainResult {
    // A misconfigured caller (e.g., an `shm_size` smaller than
    // `HEADER_SIZE` propagating into `collect_results`) must not panic
    // the host: `read_header` indexes `buf[shm_offset..shm_offset +
    // HEADER_SIZE]` unconditionally and would slice-panic on a
    // too-small buffer. Bail with an empty drain instead.
    if shm_offset.saturating_add(HEADER_SIZE) > buf.len() {
        return ShmDrainResult::default();
    }
    let header = read_header(buf, shm_offset);
    if header.magic != SHM_RING_MAGIC {
        return ShmDrainResult::default();
    }

    // Guest-supplied `capacity` must be a plausible ring size. Zero
    // would divide-by-zero inside `read_ring_into`'s `ptr % capacity`;
    // values above `MAX_SHM_CAPACITY` indicate torn init or
    // corruption and would indexing-panic on `buf` once the walk
    // advances past the real region. Either case: treat as
    // uninitialized and bail out cleanly.
    if header.capacity == 0 || header.capacity > MAX_SHM_CAPACITY {
        return ShmDrainResult::default();
    }

    let capacity = header.capacity as usize;
    let data_start = shm_offset + HEADER_SIZE;
    let mut read_pos = header.read_ptr;
    let write_pos = header.write_ptr;
    let mut entries = Vec::new();
    // Ring invariant: at most `capacity` bytes are unread at any
    // moment. A torn/corrupt header where `read_ptr > write_ptr`
    // (or where the gap exceeds capacity for any reason) makes
    // `wrapping_sub` return a near-`u64::MAX` distance — without
    // this guard the loop would iterate ~2^60 times before
    // termination, OOMing the host long before that.
    //
    // Mirrors the writer-side invariant check in `shm_write` /
    // `shm_write_raw` (`if used > capacity { return; }`).
    if write_pos.wrapping_sub(read_pos) > capacity as u64 {
        return ShmDrainResult {
            entries,
            drops: header.drops,
        };
    }
    // Largest payload that can fit in a valid message: any
    // `msg.length` larger than this is necessarily torn/corrupt and
    // must not be trusted as an allocation size. Without this cap a
    // guest-controlled `u32` (up to 4 GiB) could trigger an OOM via
    // `vec![0u8; msg.length as usize]` in `read_ring_bytes`.
    let max_payload = (capacity - MSG_HEADER_SIZE.min(capacity)) as u64;

    // Modular distance: `write_pos.wrapping_sub(read_pos)` handles
    // both the normal case and the (extremely rare) u64 overflow of
    // `write_pos` ahead of `read_pos`. Raw addition would overflow
    // when `read_pos` is near `u64::MAX`.
    while write_pos.wrapping_sub(read_pos) >= MSG_HEADER_SIZE as u64 {
        let mut hdr_buf = [0u8; MSG_HEADER_SIZE];
        read_ring_into(buf, data_start, capacity, read_pos, &mut hdr_buf);
        let msg = ShmMessage::read_from_bytes(&hdr_buf)
            .expect("MSG_HEADER_SIZE matches ShmMessage layout");

        // Reject implausible message lengths before allocating: a
        // length larger than the entire ring's payload area cannot
        // come from a complete, non-torn write. Stop draining rather
        // than allocating multi-GiB or chasing further corrupt
        // entries.
        if msg.length as u64 > max_payload {
            break;
        }

        let total_msg_size = MSG_HEADER_SIZE as u64 + msg.length as u64;
        if write_pos.wrapping_sub(read_pos) < total_msg_size {
            // Incomplete message — stop.
            break;
        }

        let payload = read_ring_bytes(
            buf,
            data_start,
            capacity,
            read_pos.wrapping_add(MSG_HEADER_SIZE as u64),
            msg.length as usize,
        );

        let computed_crc = crc32fast::hash(&payload);
        entries.push(ShmEntry {
            msg_type: msg.msg_type,
            payload,
            crc_ok: computed_crc == msg.crc32,
        });

        read_pos = read_pos.wrapping_add(total_msg_size);
    }

    ShmDrainResult {
        entries,
        drops: header.drops,
    }
}

/// Drain messages from the SHM ring while the guest VM is running.
///
/// Unlike `shm_drain` (which operates on a post-mortem snapshot),
/// this reads from live guest memory via volatile pointers and writes
/// `read_ptr` back so the guest can reclaim ring space.
///
/// `mem` provides volatile access to guest DRAM.
/// `shm_base_pa` is the DRAM-relative offset of the SHM region.
///
/// Returns drained entries. Call periodically (~10ms) from the monitor
/// thread to prevent ring overflow during long scenarios.
pub fn shm_drain_live(mem: &crate::monitor::reader::GuestMem, shm_base_pa: u64) -> ShmDrainResult {
    let magic = mem.read_u32(shm_base_pa, 0);
    if magic != SHM_RING_MAGIC {
        return ShmDrainResult::default();
    }

    let capacity_raw = mem.read_u32(shm_base_pa, 8);
    // `capacity` is guest-supplied and not covered by the magic gate
    // above — a torn init (guest writes magic before capacity) or
    // corruption can leave capacity at zero while magic reads valid.
    // `read_ring_volatile` would then compute `ptr % capacity as u64`
    // and divide by zero, panicking the monitor thread and — under
    // `panic = "abort"` — aborting the entire host. A wild value like
    // `0xFFFF_FFFF` is just as damaging: `read_ring_volatile` would
    // then read_volatile past the end of the mapped SHM region. Reject
    // both ends: treat as uninitialized and let the next tick re-read.
    if capacity_raw == 0 || capacity_raw > MAX_SHM_CAPACITY {
        return ShmDrainResult::default();
    }
    let capacity = capacity_raw as usize;
    let write_ptr = mem.read_u64(shm_base_pa, 16);
    let read_ptr = mem.read_u64(shm_base_pa, 24);
    let drops = mem.read_u64(shm_base_pa, 32);

    let data_start_pa = shm_base_pa + HEADER_SIZE as u64;
    let mut read_pos = read_ptr;
    let mut entries = Vec::new();
    // Ring invariant: at most `capacity` bytes are unread. A
    // torn/corrupt header where `read_ptr > write_ptr` (or where
    // the gap exceeds capacity) makes `wrapping_sub` return a
    // near-`u64::MAX` distance — without this guard the loop
    // would iterate ~2^60 times before terminating and OOM the
    // host monitor thread.
    //
    // Bail without advancing `read_ptr` so the next tick re-reads
    // the (hopefully recovered) header. Mirrors the writer-side
    // invariant check in `shm_write` / `shm_write_raw`.
    if write_ptr.wrapping_sub(read_pos) > capacity as u64 {
        return ShmDrainResult { entries, drops };
    }
    // Same OOM cap as `shm_drain`: a guest-supplied `msg.length`
    // larger than the ring's payload area can never come from a
    // complete write, and trusting it would let an untrusted guest
    // request a multi-GiB host allocation.
    let max_payload = (capacity - MSG_HEADER_SIZE.min(capacity)) as u64;

    // Modular distance via `wrapping_sub` — `read_pos + ...` would
    // overflow when `read_pos` is near `u64::MAX` (post-wraparound
    // case validated by `shm_write_wrapping_sub_handles_u64_overflow_of_write_ptr`).
    while write_ptr.wrapping_sub(read_pos) >= MSG_HEADER_SIZE as u64 {
        // Read message header via volatile.
        let mut hdr_buf = [0u8; MSG_HEADER_SIZE];
        read_ring_volatile(mem, data_start_pa, capacity, read_pos, &mut hdr_buf);
        let msg = ShmMessage::read_from_bytes(&hdr_buf)
            .expect("MSG_HEADER_SIZE matches ShmMessage layout");

        // Reject implausible `msg.length` before allocating: prevents
        // a torn-write or guest-controlled u32 (up to ~4GiB) from
        // triggering an OOM via `vec![0u8; msg.length as usize]`.
        if msg.length as u64 > max_payload {
            break;
        }

        let total_msg_size = MSG_HEADER_SIZE as u64 + msg.length as u64;
        if write_ptr.wrapping_sub(read_pos) < total_msg_size {
            break;
        }

        let mut payload = vec![0u8; msg.length as usize];
        if !payload.is_empty() {
            read_ring_volatile(
                mem,
                data_start_pa,
                capacity,
                read_pos.wrapping_add(MSG_HEADER_SIZE as u64),
                &mut payload,
            );
        }

        let computed_crc = crc32fast::hash(&payload);
        entries.push(ShmEntry {
            msg_type: msg.msg_type,
            payload,
            crc_ok: computed_crc == msg.crc32,
        });

        read_pos = read_pos.wrapping_add(total_msg_size);
    }

    // Advance read_ptr so the guest can reuse the drained space.
    if read_pos != read_ptr {
        mem.write_u64(shm_base_pa, 24, read_pos);
    }

    ShmDrainResult { entries, drops }
}

/// Read `out.len()` bytes from the ring data area via volatile reads,
/// handling wraparound. Uses byte-by-byte volatile reads since the
/// data area is in guest memory that the guest may be writing to.
fn read_ring_volatile(
    mem: &crate::monitor::reader::GuestMem,
    data_start_pa: u64,
    capacity: usize,
    ptr: u64,
    out: &mut [u8],
) {
    let mut remaining = out.len();
    let mut src_pos = (ptr % capacity as u64) as usize;
    let mut dst_pos = 0;
    while remaining > 0 {
        let chunk = remaining.min(capacity - src_pos);
        for i in 0..chunk {
            let pa = data_start_pa + (src_pos + i) as u64;
            out[dst_pos + i] = mem.read_u8(pa, 0);
        }
        dst_pos += chunk;
        src_pos = 0; // wrap
        remaining -= chunk;
    }
}

// ---------------------------------------------------------------------------
// Helper: write a message into the ring (for testing / guest-side simulation)
// ---------------------------------------------------------------------------

/// Write a TLV message into the ring buffer. Returns the number of bytes
/// written (MSG_HEADER_SIZE + payload.len()), or 0 if the ring is full
/// (and increments the drops counter).
///
/// This is the guest-side write operation, used in tests to simulate a
/// producer.
#[allow(dead_code)]
pub fn shm_write(buf: &mut [u8], shm_offset: usize, msg_type: u32, payload: &[u8]) -> usize {
    let header = read_header(buf, shm_offset);
    let capacity = header.capacity as usize;
    let Some(total) = MSG_HEADER_SIZE.checked_add(payload.len()) else {
        // Pathological payload whose size overflows with the header
        // prefix: treat as ring-full so the drops counter reflects the
        // lost message, matching the capacity-overflow path below.
        let drops_offset = shm_offset + 32;
        let current = u64::from_ne_bytes(buf[drops_offset..drops_offset + 8].try_into().unwrap());
        buf[drops_offset..drops_offset + 8]
            .copy_from_slice(&current.saturating_add(1).to_ne_bytes());
        return 0;
    };

    // Available space: capacity - (write_ptr - read_ptr). Both ptrs
    // are monotonic u64 counters; `wrapping_sub` is the semantically
    // correct distance under modular arithmetic and handles the
    // (extremely rare) u64 overflow of write_ptr ahead of read_ptr.
    //
    // If the distance exceeds capacity, the ring invariant is
    // violated — torn memory, corruption, or a bug elsewhere. Log
    // and drop the message rather than returning a meaningless value.
    let used = header.write_ptr.wrapping_sub(header.read_ptr) as usize;
    if used > capacity {
        tracing::warn!(
            write_ptr = header.write_ptr,
            read_ptr = header.read_ptr,
            capacity = capacity,
            used = used,
            "shm_ring: used > capacity; ring invariant violated (torn memory?)"
        );
        return 0;
    }
    // `checked_add` guards against overflow on a pathological payload
    // (MSG_HEADER_SIZE + usize::MAX). Treat overflow as ring-full.
    let needed = used.checked_add(total);
    if needed.is_none_or(|n| n > capacity) {
        // Ring full — increment drops counter. `saturating_add` because
        // a pinned-at-u64::MAX counter is the right observable state
        // when drops overflow; a wraparound to 0 would masquerade as
        // "no drops" to the host telemetry reader.
        let drops_offset = shm_offset + 32; // offset of `drops` field
        let current = u64::from_ne_bytes(buf[drops_offset..drops_offset + 8].try_into().unwrap());
        buf[drops_offset..drops_offset + 8]
            .copy_from_slice(&current.saturating_add(1).to_ne_bytes());
        return 0;
    }

    let data_start = shm_offset + HEADER_SIZE;

    // Write message header. `ShmMessage.length` is `u32`; a payload
    // whose length exceeds u32::MAX cannot be faithfully represented
    // in the header, so drop it rather than silently truncating and
    // producing a header whose CRC+length mismatch would either
    // crash the reader or cause it to skip downstream messages.
    //
    // Defense-in-depth: in the current schema `capacity: u32` (see
    // `ShmHeader`) makes this branch unreachable — the `needed >
    // capacity` check above already rejects payloads larger than ~4GB
    // well before the u32 conversion here could fail. Kept so that a
    // future refactor widening `capacity` to `u64` cannot silently
    // produce a torn header with a truncated length field.
    let Ok(length_u32) = u32::try_from(payload.len()) else {
        let drops_offset = shm_offset + 32;
        let current = u64::from_ne_bytes(buf[drops_offset..drops_offset + 8].try_into().unwrap());
        buf[drops_offset..drops_offset + 8]
            .copy_from_slice(&current.saturating_add(1).to_ne_bytes());
        return 0;
    };
    let msg = ShmMessage {
        msg_type,
        length: length_u32,
        crc32: crc32fast::hash(payload),
        _pad: 0,
    };
    write_ring_bytes(buf, data_start, capacity, header.write_ptr, msg.as_bytes());

    // Write payload
    if !payload.is_empty() {
        write_ring_bytes(
            buf,
            data_start,
            capacity,
            header.write_ptr + MSG_HEADER_SIZE as u64,
            payload,
        );
    }

    // Update write_ptr
    let new_write = header.write_ptr + total as u64;
    let wp_offset = shm_offset + 16; // offset of `write_ptr` field
    buf[wp_offset..wp_offset + 8].copy_from_slice(&new_write.to_ne_bytes());

    total
}

/// Write bytes into the ring's data area at monotonic offset `ptr`,
/// handling wraparound.
#[allow(dead_code)]
fn write_ring_bytes(buf: &mut [u8], data_start: usize, capacity: usize, ptr: u64, data: &[u8]) {
    let mut remaining = data.len();
    let mut src_pos = 0;
    let mut dst_pos = (ptr % capacity as u64) as usize;
    while remaining > 0 {
        let chunk = remaining.min(capacity - dst_pos);
        buf[data_start + dst_pos..data_start + dst_pos + chunk]
            .copy_from_slice(&data[src_pos..src_pos + chunk]);
        src_pos += chunk;
        dst_pos = 0; // wrap
        remaining -= chunk;
    }
}

/// Raw-pointer mirror of [`shm_write`] used by the guest production
/// writers (`write_msg`, `write_msg_nonblocking`). Operates entirely
/// through `ptr::read_volatile` / `ptr::write_volatile` so the SHM
/// region is never materialized as `&mut [u8]` — the host monitor
/// thread reads the same memory concurrently via `shm_drain_live`,
/// so a `&mut` slice would alias the host's view and violate Rust's
/// reference rules even though `SHM_WRITE_LOCK` serializes guest-side
/// writers.
///
/// SAFETY: caller must ensure `base` points to a valid `size`-byte
/// mapping that outlives the call, and that no other code holds a
/// `&` or `&mut` reference to bytes in that mapping.
#[allow(dead_code)]
unsafe fn shm_write_raw(base: *mut u8, size: usize, msg_type: u32, payload: &[u8]) {
    if size < HEADER_SIZE {
        return;
    }
    // Read the current header field-by-field via volatile loads.
    // Order mirrors `ShmRingHeader`: magic (0), version (4),
    // capacity (8), control_bytes (12), write_ptr (16), read_ptr (24),
    // drops (32). Only `capacity`, `write_ptr`, `read_ptr`, and
    // `drops` are needed by the write path.
    let capacity = unsafe { ptr::read_volatile(base.add(8) as *const u32) } as usize;
    if capacity == 0 || capacity > size - HEADER_SIZE {
        return;
    }
    let write_ptr = unsafe { ptr::read_volatile(base.add(16) as *const u64) };
    let read_ptr = unsafe { ptr::read_volatile(base.add(24) as *const u64) };

    let bump_drops = || {
        // SAFETY: caller invariant; `drops` field at offset 32 lies
        // wholly within the `size`-byte mapping because
        // `size >= HEADER_SIZE` was checked above.
        let drops = unsafe { ptr::read_volatile(base.add(32) as *const u64) };
        unsafe { ptr::write_volatile(base.add(32) as *mut u64, drops.saturating_add(1)) };
    };

    let Some(total) = MSG_HEADER_SIZE.checked_add(payload.len()) else {
        bump_drops();
        return;
    };

    let used = write_ptr.wrapping_sub(read_ptr) as usize;
    if used > capacity {
        return;
    }
    let needed = used.checked_add(total);
    if needed.is_none_or(|n| n > capacity) {
        bump_drops();
        return;
    }

    let Ok(length_u32) = u32::try_from(payload.len()) else {
        bump_drops();
        return;
    };

    let msg = ShmMessage {
        msg_type,
        length: length_u32,
        crc32: crc32fast::hash(payload),
        _pad: 0,
    };
    let msg_bytes = msg.as_bytes();

    // Data area starts immediately after the header.
    let data_base = unsafe { base.add(HEADER_SIZE) };

    // Write the TLV header bytes into the ring data area.
    // SAFETY: caller invariant + all bounds derived from `capacity`
    // which we just verified is `<= size - HEADER_SIZE`.
    unsafe {
        write_ring_volatile(data_base, capacity, write_ptr, msg_bytes);
        if !payload.is_empty() {
            write_ring_volatile(
                data_base,
                capacity,
                write_ptr.wrapping_add(MSG_HEADER_SIZE as u64),
                payload,
            );
        }
    }

    // Publish: bump write_ptr last so a concurrent host reader never
    // observes a partially written message past the previous
    // `write_ptr`. `wrapping_add` matches the modular distance used
    // throughout the drain path (`shm_drain` / `shm_drain_live`),
    // which already tolerates `write_ptr` wrapping past `u64::MAX`.
    let new_write = write_ptr.wrapping_add(total as u64);
    unsafe { ptr::write_volatile(base.add(16) as *mut u64, new_write) };
}

/// Volatile byte-by-byte write of `data` into the ring's data area
/// starting at monotonic offset `ptr`, handling wraparound. Mirror
/// of [`read_ring_volatile`] for guest-side writers.
///
/// SAFETY: `data_base` must point to a `capacity`-byte ring data area
/// that lives for the duration of the call.
#[allow(dead_code)]
unsafe fn write_ring_volatile(data_base: *mut u8, capacity: usize, ptr: u64, data: &[u8]) {
    let mut remaining = data.len();
    let mut src_pos = 0usize;
    let mut dst_pos = (ptr % capacity as u64) as usize;
    while remaining > 0 {
        let chunk = remaining.min(capacity - dst_pos);
        for i in 0..chunk {
            // SAFETY: `dst_pos + i < capacity` because chunk is bounded
            // by `capacity - dst_pos`; caller guarantees `data_base`
            // points to a `capacity`-byte mapping.
            unsafe {
                ptr::write_volatile(data_base.add(dst_pos + i), data[src_pos + i]);
            }
        }
        src_pos += chunk;
        remaining -= chunk;
        dst_pos = 0; // wrap
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    // Compile-time size assertions (also present above, but explicit tests
    // for visibility in test output).
    #[test]
    fn header_size_is_40() {
        assert_eq!(std::mem::size_of::<ShmRingHeader>(), 40);
    }

    #[test]
    fn message_size_is_16() {
        assert_eq!(std::mem::size_of::<ShmMessage>(), 16);
    }

    // ---- ShmRingHeader::new saturating-sub boundaries ---------------
    //
    // `new` subtracts `HEADER_SIZE` from `shm_size` via `saturating_sub`
    // so a mis-sized region surfaces as a zero-capacity ring rather than
    // panicking the VMM on integer underflow before the layout error can
    // surface. The four fixtures pin the three "too small" shapes and
    // the "just enough for a non-empty data area" shape against the
    // HEADER_SIZE boundary.

    #[test]
    fn shm_ring_header_new_zero_input_clamps_to_zero_capacity() {
        assert_eq!(ShmRingHeader::new(0).capacity, 0);
    }

    #[test]
    fn shm_ring_header_new_below_header_size_clamps_to_zero_capacity() {
        // HEADER_SIZE - 1: saturating_sub returns 0, no underflow.
        assert_eq!(ShmRingHeader::new(HEADER_SIZE - 1).capacity, 0);
    }

    #[test]
    fn shm_ring_header_new_exactly_header_size_yields_zero_capacity() {
        // HEADER_SIZE exactly: data area is empty but the header is
        // well-formed. `shm_write` hits the ring-full branch on every
        // call — internally consistent, not a panic.
        assert_eq!(ShmRingHeader::new(HEADER_SIZE).capacity, 0);
    }

    #[test]
    fn shm_ring_header_new_above_header_size_carries_delta_as_capacity() {
        // HEADER_SIZE + 100: 100 bytes of data area is addressable.
        assert_eq!(ShmRingHeader::new(HEADER_SIZE + 100).capacity, 100);
    }

    /// Allocate a buffer and initialize a ring of the given size.
    fn make_ring(shm_size: usize) -> Vec<u8> {
        let mut buf = vec![0u8; shm_size];
        shm_init(&mut buf, 0, shm_size);
        buf
    }

    #[test]
    fn init_sets_magic_and_capacity() {
        let buf = make_ring(1024);
        let hdr = read_header(&buf, 0);
        assert_eq!(hdr.magic, SHM_RING_MAGIC);
        assert_eq!(hdr.version, SHM_RING_VERSION);
        assert_eq!(hdr.capacity, (1024 - HEADER_SIZE) as u32);
        assert_eq!(hdr.write_ptr, 0);
        assert_eq!(hdr.read_ptr, 0);
        assert_eq!(hdr.drops, 0);
    }

    #[test]
    fn shm_write_rejects_torn_header_read_ptr_past_write_ptr() {
        // If the shared header is torn (or corrupt) such that
        // read_ptr > write_ptr, `wrapping_sub` yields a huge "used"
        // value. The capacity check must detect the invariant
        // violation and return 0 rather than dropping via the
        // ordinary "ring full" path or silently corrupting state.
        let mut buf = make_ring(1024);
        // write_ptr is at shm_offset + 16; read_ptr is at shm_offset + 24.
        // Force read_ptr > write_ptr to simulate the torn state.
        let wp_offset = 16;
        let rp_offset = 24;
        buf[wp_offset..wp_offset + 8].copy_from_slice(&0u64.to_ne_bytes());
        buf[rp_offset..rp_offset + 8].copy_from_slice(&100u64.to_ne_bytes());

        let result = shm_write(&mut buf, 0, 1, b"probe");
        assert_eq!(
            result, 0,
            "torn header (read_ptr > write_ptr) must return 0, got {result}"
        );
    }

    #[test]
    fn shm_write_wrapping_sub_handles_u64_overflow_of_write_ptr() {
        // When the monotonic write_ptr overflows u64 (extremely rare
        // but theoretically possible over long runs), `wrapping_sub`
        // gives the correct modular distance while raw subtraction
        // would underflow. Set write_ptr just past wrap (= 10) and
        // read_ptr just before wrap (= u64::MAX - 5). Used distance
        // = 16 via wrapping_sub.
        //
        // Ring is sized generously so `used + total_msg <= capacity`
        // and the write should succeed.
        let mut buf = make_ring(4096);
        let wp_offset = 16;
        let rp_offset = 24;
        let new_write_ptr: u64 = 10;
        let new_read_ptr: u64 = u64::MAX - 5;
        buf[wp_offset..wp_offset + 8].copy_from_slice(&new_write_ptr.to_ne_bytes());
        buf[rp_offset..rp_offset + 8].copy_from_slice(&new_read_ptr.to_ne_bytes());

        // Sanity: wrapping_sub gives 16, well below capacity 4096-40.
        assert_eq!(new_write_ptr.wrapping_sub(new_read_ptr), 16);

        let result = shm_write(&mut buf, 0, 1, b"probe");
        assert!(
            result > 0,
            "post-wraparound write should succeed, got {result}"
        );
    }

    /// A torn-init or corrupt guest can leave `capacity = 0` with a
    /// valid magic. Drain must not divide-by-zero under `ptr %
    /// capacity` inside the ring walk; return an empty drain instead.
    #[test]
    fn drain_rejects_zero_capacity() {
        let mut buf = make_ring(1024);
        // Overwrite capacity (u32 at offset 8) with 0.
        buf[8..12].copy_from_slice(&0u32.to_ne_bytes());
        // Set a non-zero write_ptr so the walk would otherwise start.
        buf[16..24].copy_from_slice(&64u64.to_ne_bytes());

        let result = shm_drain(&buf, 0);
        assert!(
            result.entries.is_empty(),
            "capacity=0 must bail out, got {} entries",
            result.entries.len(),
        );
    }

    /// A wild `capacity` (e.g. `0xFFFF_FFFF` from torn init) must be
    /// rejected, not trusted. Trusting it would index past the end of
    /// `buf` in `read_ring_into` once the walk advances, panicking the
    /// host.
    #[test]
    fn drain_rejects_oversized_capacity() {
        let mut buf = make_ring(1024);
        buf[8..12].copy_from_slice(&u32::MAX.to_ne_bytes());
        buf[16..24].copy_from_slice(&64u64.to_ne_bytes());

        let result = shm_drain(&buf, 0);
        assert!(
            result.entries.is_empty(),
            "capacity>MAX_SHM_CAPACITY must bail out, got {} entries",
            result.entries.len(),
        );
    }

    /// Boundary: a capacity exactly equal to `MAX_SHM_CAPACITY + 1`
    /// must be rejected, while the allocation-realistic small cases
    /// already covered by the other tests continue to work.
    #[test]
    fn drain_rejects_capacity_one_past_max() {
        let mut buf = make_ring(1024);
        let over = (MAX_SHM_CAPACITY as u64) + 1;
        // `as u32` truncates but MAX_SHM_CAPACITY+1 < u32::MAX so it
        // is representable directly.
        let over_u32 = over as u32;
        buf[8..12].copy_from_slice(&over_u32.to_ne_bytes());
        buf[16..24].copy_from_slice(&64u64.to_ne_bytes());

        let result = shm_drain(&buf, 0);
        assert!(result.entries.is_empty());
    }

    #[test]
    fn drain_empty_ring() {
        let buf = make_ring(1024);
        let result = shm_drain(&buf, 0);
        assert!(result.entries.is_empty());
        assert_eq!(result.drops, 0);
    }

    #[test]
    fn drain_bad_magic() {
        let mut buf = vec![0u8; 1024];
        // Don't initialize — magic is 0.
        let result = shm_drain(&buf, 0);
        assert!(result.entries.is_empty());

        // Set wrong magic.
        buf[0..4].copy_from_slice(&0xDEADBEEFu32.to_ne_bytes());
        let result = shm_drain(&buf, 0);
        assert!(result.entries.is_empty());
    }

    #[test]
    fn write_and_drain_single_message() {
        let mut buf = make_ring(1024);
        let payload = b"hello world";
        let written = shm_write(&mut buf, 0, 1, payload);
        assert_eq!(written, MSG_HEADER_SIZE + payload.len());

        let result = shm_drain(&buf, 0);
        assert_eq!(result.entries.len(), 1);
        assert_eq!(result.entries[0].msg_type, 1);
        assert_eq!(result.entries[0].payload, payload);
        assert!(result.entries[0].crc_ok);
        assert_eq!(result.drops, 0);
    }

    #[test]
    fn write_and_drain_multiple_messages() {
        let mut buf = make_ring(1024);
        shm_write(&mut buf, 0, 1, b"first");
        shm_write(&mut buf, 0, 2, b"second");
        shm_write(&mut buf, 0, 3, b"third");

        let result = shm_drain(&buf, 0);
        assert_eq!(result.entries.len(), 3);
        assert_eq!(result.entries[0].msg_type, 1);
        assert_eq!(result.entries[0].payload, b"first");
        assert_eq!(result.entries[1].msg_type, 2);
        assert_eq!(result.entries[1].payload, b"second");
        assert_eq!(result.entries[2].msg_type, 3);
        assert_eq!(result.entries[2].payload, b"third");
        for e in &result.entries {
            assert!(e.crc_ok);
        }
    }

    #[test]
    fn write_empty_payload() {
        let mut buf = make_ring(1024);
        let written = shm_write(&mut buf, 0, 42, b"");
        assert_eq!(written, MSG_HEADER_SIZE);

        let result = shm_drain(&buf, 0);
        assert_eq!(result.entries.len(), 1);
        assert_eq!(result.entries[0].msg_type, 42);
        assert!(result.entries[0].payload.is_empty());
        assert!(result.entries[0].crc_ok);
    }

    #[test]
    fn ring_full_increments_drops() {
        // Small ring: header (40) + data (60) = 100 bytes.
        // MSG_HEADER_SIZE = 16, so a message with 44 bytes payload = 60 bytes total
        // fills the ring exactly.
        let shm_size = HEADER_SIZE + 60;
        let mut buf = make_ring(shm_size);
        let payload = vec![0xAA; 44]; // 16 + 44 = 60, fills ring
        let written = shm_write(&mut buf, 0, 1, &payload);
        assert_eq!(written, 60);

        // Second write should fail — ring full.
        let written = shm_write(&mut buf, 0, 2, b"x");
        assert_eq!(written, 0);

        let result = shm_drain(&buf, 0);
        assert_eq!(result.entries.len(), 1);
        assert_eq!(result.drops, 1);
    }

    #[test]
    fn ring_full_multiple_drops() {
        let shm_size = HEADER_SIZE + 32;
        let mut buf = make_ring(shm_size);
        let payload = vec![0xBB; 16]; // 16 + 16 = 32, fills ring
        shm_write(&mut buf, 0, 1, &payload);

        // Three failed writes.
        assert_eq!(shm_write(&mut buf, 0, 2, b"a"), 0);
        assert_eq!(shm_write(&mut buf, 0, 3, b"b"), 0);
        assert_eq!(shm_write(&mut buf, 0, 4, b"c"), 0);

        let result = shm_drain(&buf, 0);
        assert_eq!(result.drops, 3);
    }

    #[test]
    fn wraparound_single_message() {
        // Ring with capacity = 48. Write a 32-byte message (16 hdr + 16 payload)
        // to advance write_ptr to 32. Then simulate the host advancing read_ptr
        // to 32. Then write another 32-byte message that wraps around.
        let shm_size = HEADER_SIZE + 48;
        let mut buf = make_ring(shm_size);

        // First message: 16 + 16 = 32 bytes.
        let payload1 = vec![0x11; 16];
        shm_write(&mut buf, 0, 1, &payload1);

        // Simulate host draining: advance read_ptr to match write_ptr.
        let hdr = read_header(&buf, 0);
        buf[24..32].copy_from_slice(&hdr.write_ptr.to_ne_bytes());

        // Second message: 16 + 16 = 32 bytes. Starts at position 32 in a
        // 48-byte ring, so it wraps around.
        let payload2 = vec![0x22; 16];
        shm_write(&mut buf, 0, 2, &payload2);

        // Drain should see only the second message.
        let result = shm_drain(&buf, 0);
        assert_eq!(result.entries.len(), 1);
        assert_eq!(result.entries[0].msg_type, 2);
        assert_eq!(result.entries[0].payload, payload2);
        assert!(result.entries[0].crc_ok);
    }

    #[test]
    fn wraparound_message_header_splits() {
        // Ring with capacity = 40. Write 32 bytes to advance to position 32.
        // Then advance read_ptr. Write another message starting at 32 —
        // the 16-byte message header crosses the 40-byte boundary.
        let shm_size = HEADER_SIZE + 40;
        let mut buf = make_ring(shm_size);

        // First: 16 + 16 = 32 bytes.
        shm_write(&mut buf, 0, 1, &[0xAA; 16]);

        // Advance read_ptr.
        let hdr = read_header(&buf, 0);
        buf[24..32].copy_from_slice(&hdr.write_ptr.to_ne_bytes());

        // Second: 16 + 4 = 20 bytes, starting at position 32 in a 40-byte ring.
        // Header bytes: 32..40 (8 bytes) then 0..8 (8 bytes) — wraps mid-header.
        let payload2 = vec![0xBB; 4];
        shm_write(&mut buf, 0, 2, &payload2);

        let result = shm_drain(&buf, 0);
        assert_eq!(result.entries.len(), 1);
        assert_eq!(result.entries[0].msg_type, 2);
        assert_eq!(result.entries[0].payload, payload2);
        assert!(result.entries[0].crc_ok);
    }

    #[test]
    fn crc_detects_corruption() {
        let mut buf = make_ring(1024);
        shm_write(&mut buf, 0, 1, b"integrity check");

        // Corrupt one byte of the payload in the ring data area.
        let data_start = HEADER_SIZE;
        let payload_start = data_start + MSG_HEADER_SIZE;
        buf[payload_start] ^= 0xFF;

        let result = shm_drain(&buf, 0);
        assert_eq!(result.entries.len(), 1);
        assert!(!result.entries[0].crc_ok);
    }

    #[test]
    fn crc_empty_payload_is_zero_for_empty() {
        // CRC32 of empty input is 0x00000000.
        assert_eq!(crc32fast::hash(b""), 0x0000_0000);
    }

    #[test]
    fn crc32_known_vectors() {
        // Standard CRC32 test vectors.
        assert_eq!(crc32fast::hash(b"123456789"), 0xCBF4_3926);
        assert_eq!(crc32fast::hash(b""), 0x0000_0000);
        assert_eq!(crc32fast::hash(b"a"), 0xE8B7_BE43);
    }

    #[test]
    fn nonzero_shm_offset() {
        // SHM region at offset 4096 in a larger buffer (simulating guest memory).
        let offset = 4096;
        let shm_size = 512;
        let total = offset + shm_size;
        let mut buf = vec![0xFFu8; total];
        shm_init(&mut buf, offset, shm_size);

        shm_write(&mut buf, offset, 7, b"offset test");

        let result = shm_drain(&buf, offset);
        assert_eq!(result.entries.len(), 1);
        assert_eq!(result.entries[0].msg_type, 7);
        assert_eq!(result.entries[0].payload, b"offset test");
        assert!(result.entries[0].crc_ok);
    }

    #[test]
    fn large_payload() {
        let mut buf = make_ring(65536);
        let payload = vec![0x42; 60000];
        let written = shm_write(&mut buf, 0, 99, &payload);
        assert_eq!(written, MSG_HEADER_SIZE + 60000);

        let result = shm_drain(&buf, 0);
        assert_eq!(result.entries.len(), 1);
        assert_eq!(result.entries[0].payload.len(), 60000);
        assert!(result.entries[0].payload.iter().all(|&b| b == 0x42));
        assert!(result.entries[0].crc_ok);
    }

    #[test]
    fn incomplete_message_not_drained() {
        let mut buf = make_ring(1024);
        shm_write(&mut buf, 0, 1, b"complete");

        // Manually advance write_ptr by 20 bytes (pretend a message header
        // was written but payload is incomplete).
        let hdr = read_header(&buf, 0);
        let fake_write = hdr.write_ptr + 20;
        // Write a fake message header at the current write position claiming
        // 100 bytes of payload (which we don't actually write).
        let fake_msg = ShmMessage {
            msg_type: 99,
            length: 100,
            crc32: 0,
            _pad: 0,
        };
        let data_start = HEADER_SIZE;
        let capacity = hdr.capacity as usize;
        write_ring_bytes(
            &mut buf,
            data_start,
            capacity,
            hdr.write_ptr,
            fake_msg.as_bytes(),
        );
        // Advance write_ptr to only partially cover the fake message.
        buf[16..24].copy_from_slice(&fake_write.to_ne_bytes());

        let result = shm_drain(&buf, 0);
        // Only the first complete message should be drained.
        assert_eq!(result.entries.len(), 1);
        assert_eq!(result.entries[0].msg_type, 1);
        assert_eq!(result.entries[0].payload, b"complete");
    }

    #[test]
    fn stimulus_payload_size_is_24() {
        assert_eq!(std::mem::size_of::<StimulusPayload>(), 24);
    }

    #[test]
    fn msg_type_stimulus_ascii() {
        let bytes = MSG_TYPE_STIMULUS.to_be_bytes();
        assert_eq!(&bytes, b"STIM");
    }

    #[test]
    fn msg_type_scenario_start_ascii() {
        let bytes = MSG_TYPE_SCENARIO_START.to_be_bytes();
        assert_eq!(&bytes, b"SCST");
    }

    #[test]
    fn msg_type_scenario_end_ascii() {
        let bytes = MSG_TYPE_SCENARIO_END.to_be_bytes();
        assert_eq!(&bytes, b"SCEN");
    }

    #[test]
    fn msg_type_sched_exit_ascii() {
        let bytes = MSG_TYPE_SCHED_EXIT.to_be_bytes();
        assert_eq!(&bytes, b"SCDX");
    }

    #[test]
    fn msg_type_crash_ascii() {
        let bytes = MSG_TYPE_CRASH.to_be_bytes();
        assert_eq!(&bytes, b"CRSH");
    }

    #[test]
    fn stimulus_payload_roundtrip() {
        let payload = StimulusPayload {
            elapsed_ms: 1234,
            step_index: 3,
            op_count: 5,
            op_kinds: 0b1010_0101,
            cgroup_count: 4,
            worker_count: 16,
            total_iterations: 99999,
        };
        let bytes = payload.as_bytes();
        let event = StimulusEvent::from_payload(bytes).unwrap();
        assert_eq!(event.elapsed_ms, 1234);
        assert_eq!(event.step_index, 3);
        assert_eq!(event.op_count, 5);
        assert_eq!(event.op_kinds, 0b1010_0101);
        assert_eq!(event.cgroup_count, 4);
        assert_eq!(event.worker_count, 16);
        assert_eq!(event.total_iterations, 99999);
    }

    #[test]
    fn stimulus_event_from_short_payload() {
        assert!(StimulusEvent::from_payload(&[0u8; 19]).is_none());
        assert!(StimulusEvent::from_payload(&[0u8; 24]).is_some());
    }

    #[test]
    fn stimulus_write_and_drain() {
        let mut buf = make_ring(1024);
        let payload = StimulusPayload {
            elapsed_ms: 500,
            step_index: 1,
            op_count: 3,
            op_kinds: 7,
            cgroup_count: 2,
            worker_count: 8,
            total_iterations: 42000,
        };
        let written = shm_write(&mut buf, 0, MSG_TYPE_STIMULUS, payload.as_bytes());
        assert_eq!(written, MSG_HEADER_SIZE + 24);

        let result = shm_drain(&buf, 0);
        assert_eq!(result.entries.len(), 1);
        assert_eq!(result.entries[0].msg_type, MSG_TYPE_STIMULUS);
        assert!(result.entries[0].crc_ok);
        let event = StimulusEvent::from_payload(&result.entries[0].payload).unwrap();
        assert_eq!(event.elapsed_ms, 500);
        assert_eq!(event.step_index, 1);
        assert_eq!(event.op_count, 3);
    }

    #[test]
    fn header_fields_at_expected_offsets() {
        let mut buf = make_ring(256);
        // Write known values and verify byte-level layout.
        let hdr = ShmRingHeader {
            magic: SHM_RING_MAGIC,
            version: SHM_RING_VERSION,
            capacity: 216,
            control_bytes: 0,
            write_ptr: 0x1122_3344_5566_7788,
            read_ptr: 0xAABB_CCDD_EEFF_0011,
            drops: 42,
        };
        buf[..HEADER_SIZE].copy_from_slice(hdr.as_bytes());

        assert_eq!(
            u32::from_ne_bytes(buf[0..4].try_into().unwrap()),
            SHM_RING_MAGIC
        );
        assert_eq!(
            u32::from_ne_bytes(buf[4..8].try_into().unwrap()),
            SHM_RING_VERSION
        );
        assert_eq!(u32::from_ne_bytes(buf[8..12].try_into().unwrap()), 216);
        assert_eq!(
            u64::from_ne_bytes(buf[16..24].try_into().unwrap()),
            0x1122_3344_5566_7788
        );
        assert_eq!(
            u64::from_ne_bytes(buf[24..32].try_into().unwrap()),
            0xAABB_CCDD_EEFF_0011
        );
        assert_eq!(u64::from_ne_bytes(buf[32..40].try_into().unwrap()), 42);
    }

    #[test]
    fn dump_req_offset_in_control_bytes() {
        assert_eq!(DUMP_REQ_OFFSET, 12);
        assert_eq!(DUMP_REQ_SYSRQ_D, b'D');
    }

    #[test]
    fn stall_req_offset_in_control_bytes() {
        assert_eq!(STALL_REQ_OFFSET, 13);
        assert_eq!(STALL_REQ_ACTIVATE, b'S');
    }

    #[test]
    fn stimulus_event_from_exact_size_payload() {
        let payload = StimulusPayload {
            elapsed_ms: 42,
            step_index: 7,
            op_count: 3,
            op_kinds: 0xFF,
            cgroup_count: 2,
            worker_count: 10,
            total_iterations: 4,
        };
        let bytes = payload.as_bytes();
        assert_eq!(bytes.len(), 24);
        let event = StimulusEvent::from_payload(bytes).unwrap();
        assert_eq!(event.elapsed_ms, 42);
        assert_eq!(event.step_index, 7);
        assert_eq!(event.op_count, 3);
        assert_eq!(event.op_kinds, 0xFF);
        assert_eq!(event.cgroup_count, 2);
        assert_eq!(event.worker_count, 10);
        assert_eq!(event.total_iterations, 4);
    }

    #[test]
    fn stimulus_event_from_oversized_payload() {
        let mut bytes = vec![0u8; 32];
        // Set elapsed_ms to 123 at offset 0.
        bytes[0..4].copy_from_slice(&123u32.to_ne_bytes());
        let event = StimulusEvent::from_payload(&bytes).unwrap();
        assert_eq!(event.elapsed_ms, 123);
    }

    #[test]
    fn concurrent_producer_consumer_simulated() {
        // Simulate alternating writes and drains to exercise the read_ptr
        // advancement path.
        let shm_size = HEADER_SIZE + 128;
        let mut buf = make_ring(shm_size);

        // Write 3 messages, drain, advance read_ptr, write 3 more, drain.
        for round in 0..3 {
            let base_type = round * 10;
            shm_write(&mut buf, 0, base_type + 1, b"aa");
            shm_write(&mut buf, 0, base_type + 2, b"bb");
            shm_write(&mut buf, 0, base_type + 3, b"cc");

            let result = shm_drain(&buf, 0);
            assert_eq!(result.entries.len(), 3);
            for e in &result.entries {
                assert!(e.crc_ok);
            }

            // Advance read_ptr to write_ptr (simulate host consuming).
            let hdr = read_header(&buf, 0);
            buf[24..32].copy_from_slice(&hdr.write_ptr.to_ne_bytes());
        }
    }

    #[test]
    fn stimulus_event_from_empty_payload() {
        assert!(StimulusEvent::from_payload(&[]).is_none());
    }

    #[test]
    fn stimulus_event_clone_preserves_fields() {
        let event = StimulusEvent {
            elapsed_ms: 999,
            step_index: 7,
            op_count: 3,
            op_kinds: 0xF0,
            cgroup_count: 5,
            worker_count: 20,
            total_iterations: 16,
        };
        let c = event.clone();
        assert_eq!(c.elapsed_ms, 999);
        assert_eq!(c.step_index, 7);
        assert_eq!(c.op_count, 3);
        assert_eq!(c.op_kinds, 0xF0);
        assert_eq!(c.cgroup_count, 5);
        assert_eq!(c.worker_count, 20);
        assert_eq!(c.total_iterations, 16);
    }

    #[test]
    fn shm_drain_result_default_empty() {
        let r = ShmDrainResult::default();
        assert!(r.entries.is_empty());
        assert_eq!(r.drops, 0);
    }

    #[test]
    fn write_exact_capacity_then_empty() {
        // Exactly fill capacity with one message, drain, verify empty after.
        let data_size = 64;
        let shm_size = HEADER_SIZE + data_size;
        let mut buf = make_ring(shm_size);
        let payload_len = data_size - MSG_HEADER_SIZE;
        let payload = vec![0x55u8; payload_len];
        let written = shm_write(&mut buf, 0, 1, &payload);
        assert_eq!(written, data_size);

        let result = shm_drain(&buf, 0);
        assert_eq!(result.entries.len(), 1);
        assert!(result.entries[0].crc_ok);
        assert_eq!(result.entries[0].payload.len(), payload_len);
    }

    #[test]
    fn write_ring_bytes_wraparound_exact() {
        // Data area of 16 bytes, write 8 bytes starting at position 12 —
        // first 4 bytes fit, then wraps to start for remaining 4.
        let data_start = HEADER_SIZE;
        let capacity = 16;
        let shm_size = HEADER_SIZE + capacity;
        let mut buf = vec![0u8; shm_size];
        let data = [1u8, 2, 3, 4, 5, 6, 7, 8];
        write_ring_bytes(&mut buf, data_start, capacity, 12, &data);
        // Bytes at positions 12..16 then 0..4
        assert_eq!(&buf[data_start + 12..data_start + 16], &[1, 2, 3, 4]);
        assert_eq!(&buf[data_start..data_start + 4], &[5, 6, 7, 8]);
    }

    #[test]
    fn read_ring_bytes_wraparound_exact() {
        let data_start = HEADER_SIZE;
        let capacity = 16;
        let shm_size = HEADER_SIZE + capacity;
        let mut buf = vec![0u8; shm_size];
        // Plant data that wraps: positions 14..16 and 0..2
        buf[data_start + 14] = 0xAA;
        buf[data_start + 15] = 0xBB;
        buf[data_start] = 0xCC;
        buf[data_start + 1] = 0xDD;
        let out = read_ring_bytes(&buf, data_start, capacity, 14, 4);
        assert_eq!(out, vec![0xAA, 0xBB, 0xCC, 0xDD]);
    }

    #[test]
    fn stimulus_payload_as_bytes_roundtrip() {
        let p = StimulusPayload {
            elapsed_ms: u32::MAX,
            step_index: u16::MAX,
            op_count: u16::MAX,
            op_kinds: u32::MAX,
            cgroup_count: u16::MAX,
            worker_count: u16::MAX,
            total_iterations: u64::MAX,
        };
        let bytes = p.as_bytes();
        let e = StimulusEvent::from_payload(bytes).unwrap();
        assert_eq!(e.elapsed_ms, u32::MAX);
        assert_eq!(e.step_index, u16::MAX);
        assert_eq!(e.op_count, u16::MAX);
        assert_eq!(e.op_kinds, u32::MAX);
        assert_eq!(e.cgroup_count, u16::MAX);
        assert_eq!(e.worker_count, u16::MAX);
        assert_eq!(e.total_iterations, u64::MAX);
    }

    #[test]
    fn multiple_writes_fill_and_drop() {
        // Ring with 80 bytes of data. Each message = 16 + 8 = 24 bytes.
        // Can fit 3 messages (72 bytes). 4th should drop.
        let shm_size = HEADER_SIZE + 80;
        let mut buf = make_ring(shm_size);
        assert_eq!(shm_write(&mut buf, 0, 1, &[0xAA; 8]), 24);
        assert_eq!(shm_write(&mut buf, 0, 2, &[0xBB; 8]), 24);
        assert_eq!(shm_write(&mut buf, 0, 3, &[0xCC; 8]), 24);
        assert_eq!(shm_write(&mut buf, 0, 4, &[0xDD; 8]), 0); // dropped

        let result = shm_drain(&buf, 0);
        assert_eq!(result.entries.len(), 3);
        assert_eq!(result.drops, 1);
    }

    // ---- signal_guest_value edge-case offsets ---------------------
    //
    // `signal_guest_value` writes one byte at
    // `shm_base + SIGNAL_SLOT_BASE + slot` via `GuestMem::write_u8`,
    // which is the bounds-checked entry point. These tests pin the
    // host-side signal API against the boundary cases of the
    // GuestMem mapping it sits inside.

    #[test]
    fn signal_guest_value_writes_to_correct_slot() {
        // Lay out: [pre-shm bytes] [SHM region]. shm_base is the
        // offset of the SHM region within the GuestMem mapping.
        // Slot 0 lands at shm_base + SIGNAL_SLOT_BASE; slot 1 lands
        // at the next byte.
        let shm_base: u64 = 64;
        let total: u64 = shm_base + 32;
        let mut buf = vec![0u8; total as usize];
        // SAFETY: buf outlives the GuestMem use.
        let mem = unsafe { crate::monitor::reader::GuestMem::new(buf.as_mut_ptr(), total) };
        signal_guest_value(&mem, shm_base, 0, SIGNAL_SHUTDOWN_REQ);
        signal_guest_value(&mem, shm_base, 1, SIGNAL_PROBES_READY);
        assert_eq!(
            buf[shm_base as usize + SIGNAL_SLOT_BASE],
            SIGNAL_SHUTDOWN_REQ
        );
        assert_eq!(
            buf[shm_base as usize + SIGNAL_SLOT_BASE + 1],
            SIGNAL_PROBES_READY
        );
    }

    #[test]
    fn signal_guest_value_at_zero_shm_base() {
        // shm_base = 0: slot bytes land at SIGNAL_SLOT_BASE and
        // SIGNAL_SLOT_BASE + 1 from the start of the mapping.
        let mut buf = vec![0u8; 32];
        let len = buf.len() as u64;
        // SAFETY: buf outlives the GuestMem use.
        let mem = unsafe { crate::monitor::reader::GuestMem::new(buf.as_mut_ptr(), len) };
        signal_guest_value(&mem, 0, 0, 0xAB);
        signal_guest_value(&mem, 0, 1, 0xCD);
        assert_eq!(buf[SIGNAL_SLOT_BASE], 0xAB);
        assert_eq!(buf[SIGNAL_SLOT_BASE + 1], 0xCD);
    }

    #[test]
    fn signal_guest_value_at_exact_boundary_succeeds() {
        // GuestMem sized so the last slot byte is the very last
        // byte of the mapping. write_u8's bounds check
        // (`addr + 1 > size`) admits exactly this position.
        let shm_base: u64 = 0;
        // Last slot is at SIGNAL_SLOT_BASE + 1 (slot index 1).
        // Size the mapping so that byte sits at size - 1.
        let total: u64 = (SIGNAL_SLOT_BASE + 2) as u64;
        let mut buf = vec![0u8; total as usize];
        // SAFETY: buf outlives the GuestMem use.
        let mem = unsafe { crate::monitor::reader::GuestMem::new(buf.as_mut_ptr(), total) };
        signal_guest_value(&mem, shm_base, 1, 0xEE);
        assert_eq!(buf[SIGNAL_SLOT_BASE + 1], 0xEE);
    }

    #[test]
    fn signal_guest_value_past_boundary_is_noop() {
        // shm_base places the slot byte past the GuestMem size.
        // GuestMem.write_u8 silently noops; the backing buffer is
        // unmodified. This covers the defensive path: a misconfigured
        // shm_base must not corrupt host memory past the declared
        // GuestMem size.
        let total: u64 = 32;
        let mut buf = vec![0xAAu8; (total + 64) as usize]; // sentinel
        // Declare a mapping smaller than the underlying buffer so we
        // can detect any write that would have escaped the bounds.
        // SAFETY: buf is larger than `total`; the in-bounds writes
        // stay within `buf` and the out-of-bounds case must noop.
        let mem = unsafe { crate::monitor::reader::GuestMem::new(buf.as_mut_ptr(), total) };
        // shm_base 32 + SIGNAL_SLOT_BASE 14 + slot 0 = 46, which
        // is past the declared size of 32.
        signal_guest_value(&mem, 32, 0, 0xFF);
        // Buffer bytes past the declared size remain at their
        // sentinel value 0xAA — write_u8 dropped silently.
        assert!(buf.iter().all(|&b| b == 0xAA));
    }

    #[test]
    fn signal_guest_value_offset_only_partially_in_bounds_is_noop() {
        // shm_base is in-bounds, but shm_base + SIGNAL_SLOT_BASE +
        // slot crosses the boundary. write_u8 must combine pa and
        // offset before the bounds check, not check pa alone.
        let total: u64 = (SIGNAL_SLOT_BASE + 1) as u64; // 15
        let mut buf = vec![0u8; (total + 64) as usize];
        // SAFETY: buf is larger than `total`; in-bounds writes are
        // within buf and out-of-bounds writes are dropped.
        let mem = unsafe { crate::monitor::reader::GuestMem::new(buf.as_mut_ptr(), total) };
        // shm_base = 1, SIGNAL_SLOT_BASE = 14, slot = 0
        // -> addr 15, addr + 1 = 16 > 15 -> noop.
        signal_guest_value(&mem, 1, 0, 0x77);
        assert!(buf.iter().all(|&b| b == 0));
    }

    // ---- read_ring_volatile multi-region routing -------------------
    //
    // `read_ring_volatile` reads each byte through `mem.read_u8(pa, 0)`,
    // so on a multi-region GuestMem each byte must resolve to the
    // region that contains its DRAM offset. This test wires up a
    // 2-region GuestMem with the ring's data area placed inside
    // region 1 (well past region 0's end) and verifies that the
    // bytes returned reflect region 1's host buffer, not stale
    // memory past region 0.

    #[test]
    fn read_ring_volatile_routes_through_correct_region() {
        use crate::monitor::reader::{GuestMem, MemRegion};

        // Region 0: 4 KiB at DRAM offset 0, filled with 0xAA.
        // Region 1: 4 KiB at DRAM offset 1 MiB, filled with 0xBB
        //           except for a planted 32-byte "ring data area"
        //           starting at byte 0 of region 1.
        let mut buf0 = vec![0xAAu8; 4096];
        let mut buf1 = vec![0xBBu8; 4096];
        let pattern: [u8; 32] = [
            0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, //
            0x09, 0x0A, 0x0B, 0x0C, 0x0D, 0x0E, 0x0F, 0x10, //
            0x11, 0x12, 0x13, 0x14, 0x15, 0x16, 0x17, 0x18, //
            0x19, 0x1A, 0x1B, 0x1C, 0x1D, 0x1E, 0x1F, 0x20, //
        ];
        buf1[0..32].copy_from_slice(&pattern);

        let regions = vec![
            MemRegion {
                host_ptr: buf0.as_mut_ptr(),
                offset: 0,
                size: 4096,
            },
            MemRegion {
                host_ptr: buf1.as_mut_ptr(),
                offset: 1 << 20, // 1 MiB
                size: 4096,
            },
        ];
        // SAFETY: buf0 and buf1 outlive the GuestMem use.
        let mem = unsafe { GuestMem::from_regions_for_test(regions) };

        // Read 32 bytes starting at the data area in region 1.
        // capacity is set generously so no wraparound occurs in this
        // straight-line read.
        let data_start_pa: u64 = 1 << 20;
        let capacity: usize = 4096;
        let mut out = vec![0u8; 32];
        read_ring_volatile(&mem, data_start_pa, capacity, 0, &mut out);

        assert_eq!(out, pattern);
        // No byte from region 0 (0xAA) leaked into the result.
        assert!(!out.contains(&0xAA));
    }

    #[test]
    fn read_ring_volatile_wraparound_routes_through_correct_region() {
        // Wraparound case: capacity is small enough that a read of
        // length capacity starting near the end wraps back to byte 0
        // of the data area. With multi-region routing, BOTH halves
        // of the wrapped read must hit the right region.
        use crate::monitor::reader::{GuestMem, MemRegion};

        let mut buf0 = vec![0xAAu8; 4096];
        let mut buf1 = vec![0u8; 4096];
        // Plant a 64-byte pattern; data area starts at offset 0 of
        // region 1, capacity 64. A read of 64 bytes starting at
        // pos 60 wraps: bytes at positions 60..64 then 0..60.
        for (i, slot) in buf1[..64].iter_mut().enumerate() {
            *slot = i as u8;
        }
        let regions = vec![
            MemRegion {
                host_ptr: buf0.as_mut_ptr(),
                offset: 0,
                size: 4096,
            },
            MemRegion {
                host_ptr: buf1.as_mut_ptr(),
                offset: 1 << 20,
                size: 4096,
            },
        ];
        // SAFETY: backing buffers outlive the GuestMem use.
        let mem = unsafe { GuestMem::from_regions_for_test(regions) };

        let data_start_pa: u64 = 1 << 20;
        let capacity: usize = 64;
        let mut out = vec![0u8; 64];
        // ptr 60 -> src_pos = 60 % 64 = 60. First chunk covers 4
        // bytes (positions 60..63), then wraps to position 0 for
        // the remaining 60.
        read_ring_volatile(&mem, data_start_pa, capacity, 60, &mut out);

        assert_eq!(out[0..4], [60u8, 61, 62, 63]);
        for (i, &b) in out[4..].iter().enumerate() {
            assert_eq!(b, i as u8);
        }
    }

    // ---- shm_write_raw round-trip --------------------------------
    //
    // `shm_write_raw` is the production guest-side writer that
    // operates on `*mut u8` instead of `&mut [u8]` so the host's
    // concurrent volatile reader cannot violate Rust aliasing. The
    // round-trip tests pin its byte-level layout against the slice
    // reader (`shm_drain`) — they must agree on every field.

    #[test]
    fn shm_write_raw_round_trips_through_drain() {
        // Allocate a buffer, init the ring header via the slice
        // helper, then write through the raw-pointer path. Drain via
        // the slice reader and verify the message arrived intact.
        let shm_size = 1024usize;
        let mut buf = vec![0u8; shm_size];
        shm_init(&mut buf, 0, shm_size);
        let payload = b"raw-ptr round trip";
        // SAFETY: `buf` outlives the call; no concurrent reference
        // to its bytes exists during the write.
        unsafe {
            shm_write_raw(buf.as_mut_ptr(), shm_size, MSG_TYPE_STIMULUS, payload);
        }

        let result = shm_drain(&buf, 0);
        assert_eq!(result.entries.len(), 1);
        assert_eq!(result.entries[0].msg_type, MSG_TYPE_STIMULUS);
        assert_eq!(result.entries[0].payload, payload);
        assert!(result.entries[0].crc_ok);
        assert_eq!(result.drops, 0);
    }

    #[test]
    fn shm_write_raw_handles_wraparound() {
        // Force the ring to wrap mid-payload by pre-advancing
        // read_ptr/write_ptr near the boundary.
        let shm_size = HEADER_SIZE + 48;
        let mut buf = vec![0u8; shm_size];
        shm_init(&mut buf, 0, shm_size);
        let capacity_u64 = (shm_size - HEADER_SIZE) as u64;
        // Place write_ptr at capacity - 8 so a 16-byte header + 4-byte
        // payload (= 20 bytes) wraps.
        let pre = capacity_u64 - 8;
        buf[16..24].copy_from_slice(&pre.to_ne_bytes());
        buf[24..32].copy_from_slice(&pre.to_ne_bytes());

        let payload = [0xAAu8, 0xBB, 0xCC, 0xDD];
        // SAFETY: see `shm_write_raw_round_trips_through_drain`.
        unsafe {
            shm_write_raw(buf.as_mut_ptr(), shm_size, 0xDEAD_BEEF, &payload);
        }
        let result = shm_drain(&buf, 0);
        assert_eq!(result.entries.len(), 1);
        assert_eq!(result.entries[0].msg_type, 0xDEAD_BEEF);
        assert_eq!(result.entries[0].payload, payload);
        assert!(result.entries[0].crc_ok);
    }

    #[test]
    fn shm_write_raw_rejects_undersized_mapping() {
        // A `size` smaller than HEADER_SIZE must noop without
        // touching memory — the volatile reads at offsets 8/16/24/32
        // would read past the end of the mapping otherwise.
        let mut buf = vec![0xAAu8; HEADER_SIZE - 1];
        // SAFETY: buf is `HEADER_SIZE - 1` bytes; the call must
        // bail before touching offset 8.
        unsafe {
            shm_write_raw(buf.as_mut_ptr(), buf.len(), 1, b"x");
        }
        // No bytes mutated — sentinel intact.
        assert!(buf.iter().all(|&b| b == 0xAA));
    }

    #[test]
    fn shm_write_raw_rejects_torn_capacity() {
        // capacity = 0 (torn init): drop without writing.
        let shm_size = 1024usize;
        let mut buf = vec![0u8; shm_size];
        shm_init(&mut buf, 0, shm_size);
        // Overwrite capacity field with 0.
        buf[8..12].copy_from_slice(&0u32.to_ne_bytes());

        // SAFETY: see `shm_write_raw_round_trips_through_drain`.
        unsafe {
            shm_write_raw(buf.as_mut_ptr(), shm_size, 1, b"probe");
        }
        // write_ptr unchanged.
        assert_eq!(u64::from_ne_bytes(buf[16..24].try_into().unwrap()), 0);
    }

    #[test]
    fn shm_write_raw_increments_drops_when_full() {
        // Fill the ring, then write once more — drops must bump.
        let shm_size = HEADER_SIZE + 32;
        let mut buf = vec![0u8; shm_size];
        shm_init(&mut buf, 0, shm_size);
        let big = vec![0xCCu8; 16]; // 16 + 16 = 32, fills capacity.
        // SAFETY: see `shm_write_raw_round_trips_through_drain`.
        unsafe {
            shm_write_raw(buf.as_mut_ptr(), shm_size, 1, &big);
            shm_write_raw(buf.as_mut_ptr(), shm_size, 2, b"x");
        }
        // drops field at offset 32.
        let drops = u64::from_ne_bytes(buf[32..40].try_into().unwrap());
        assert_eq!(drops, 1);
    }

    // ---- shm_drain panic-guard / OOM-cap / overflow guards ------

    #[test]
    fn shm_drain_returns_empty_when_buf_smaller_than_header() {
        // A misconfigured caller (e.g. shm_size < HEADER_SIZE)
        // must produce an empty drain instead of slice-panicking
        // inside `read_header`.
        let buf = vec![0u8; HEADER_SIZE - 1];
        let result = shm_drain(&buf, 0);
        assert!(result.entries.is_empty());
        assert_eq!(result.drops, 0);
    }

    #[test]
    fn shm_drain_returns_empty_when_offset_pushes_past_end() {
        // shm_offset positioned so that `shm_offset + HEADER_SIZE`
        // exceeds buf.len(). Must not panic.
        let buf = vec![0u8; 1024];
        let result = shm_drain(&buf, 1024 - HEADER_SIZE + 1);
        assert!(result.entries.is_empty());
    }

    #[test]
    fn shm_drain_caps_torn_msg_length_against_capacity() {
        // Torn header advertises `msg.length = u32::MAX`. The OOM
        // guard must reject before allocation.
        let shm_size = 1024usize;
        let mut buf = vec![0u8; shm_size];
        shm_init(&mut buf, 0, shm_size);
        // Plant a TLV header at the start of the data area:
        // msg_type=1, length=u32::MAX, crc=0, _pad=0.
        let data_start = HEADER_SIZE;
        buf[data_start..data_start + 4].copy_from_slice(&1u32.to_ne_bytes());
        buf[data_start + 4..data_start + 8].copy_from_slice(&u32::MAX.to_ne_bytes());
        buf[data_start + 8..data_start + 12].copy_from_slice(&0u32.to_ne_bytes());
        buf[data_start + 12..data_start + 16].copy_from_slice(&0u32.to_ne_bytes());
        // Set write_ptr large enough that the loop enters and the
        // torn `length` is read.
        buf[16..24].copy_from_slice(&64u64.to_ne_bytes());

        let result = shm_drain(&buf, 0);
        // Must not have allocated 4GiB; result is empty (cap break).
        assert!(result.entries.is_empty());
    }

    #[test]
    fn shm_drain_handles_read_pos_near_u64_max() {
        // Set read_ptr near u64::MAX and write_ptr just past wrap;
        // the modular distance check (`wrapping_sub`) must drive the
        // loop without overflowing the additive form.
        let shm_size = HEADER_SIZE + 64;
        let mut buf = vec![0u8; shm_size];
        shm_init(&mut buf, 0, shm_size);
        // First write a normal message via shm_write so the layout
        // is valid, then force read_ptr/write_ptr to the post-wrap
        // window while keeping the pattern length consistent.
        let _ = shm_write(&mut buf, 0, 1, b"abcd"); // total 20.
        // Move write_ptr := 5, read_ptr := u64::MAX - 14
        // (so wrapping_sub(write_ptr, read_ptr) == 20).
        let new_write: u64 = 5;
        let new_read: u64 = u64::MAX - 14;
        buf[16..24].copy_from_slice(&new_write.to_ne_bytes());
        buf[24..32].copy_from_slice(&new_read.to_ne_bytes());
        // Sanity.
        assert_eq!(new_write.wrapping_sub(new_read), 20);

        // The drain enters the loop, computes total_msg_size=20,
        // attempts to read the header from `data_start + (read_pos %
        // capacity)`. Since the original write left valid bytes only
        // at position 0, the header read at position
        // `(u64::MAX - 14) % 64` resolves to a defined byte position
        // in the data area — the test's purpose is to confirm no
        // panic, not to verify the parsed payload. Drain returns
        // either empty (CRC mismatch / max_payload reject) or a
        // single entry, but never panics.
        let result = shm_drain(&buf, 0);
        // The pre-write established `drops = 0` and modular
        // arithmetic must not have aborted.
        assert!(result.entries.len() <= 1);
    }

    #[test]
    fn shm_drain_rejects_used_greater_than_capacity() {
        // Writer-side invariant: `used = write_ptr - read_ptr` is at
        // most `capacity`. A torn header can violate this (e.g.
        // `read_ptr > write_ptr` so `wrapping_sub` returns a
        // near-`u64::MAX` distance). The drain MUST bail before
        // entering the loop — without the guard it would iterate
        // ~2^60 times and OOM the host.
        let shm_size = 1024usize;
        let mut buf = vec![0u8; shm_size];
        shm_init(&mut buf, 0, shm_size);
        // Force read_ptr > write_ptr to make wrapping_sub huge.
        let bogus_write: u64 = 0;
        let bogus_read: u64 = 100;
        buf[16..24].copy_from_slice(&bogus_write.to_ne_bytes());
        buf[24..32].copy_from_slice(&bogus_read.to_ne_bytes());
        // Sanity: distance vastly exceeds capacity (984 bytes).
        assert!(bogus_write.wrapping_sub(bogus_read) > (shm_size - HEADER_SIZE) as u64);

        // Must return immediately — no allocation, no iteration.
        let result = shm_drain(&buf, 0);
        assert!(result.entries.is_empty());
    }

    #[test]
    fn shm_drain_rejects_used_one_past_capacity() {
        // Boundary case: distance is exactly `capacity + 1`. The
        // strict `>` comparison must reject; `==` would accept.
        let shm_size = HEADER_SIZE + 64;
        let mut buf = vec![0u8; shm_size];
        shm_init(&mut buf, 0, shm_size);
        // capacity = 64. Set distance to 65 via wrapping_sub.
        let new_write: u64 = 65;
        let new_read: u64 = 0;
        buf[16..24].copy_from_slice(&new_write.to_ne_bytes());
        buf[24..32].copy_from_slice(&new_read.to_ne_bytes());

        let result = shm_drain(&buf, 0);
        assert!(result.entries.is_empty());
    }

    #[test]
    fn shm_drain_accepts_used_exactly_capacity() {
        // The invariant permits `used == capacity` (full ring); the
        // guard must NOT trip on this case. Build a full ring via
        // the writer and verify drain succeeds.
        let shm_size = HEADER_SIZE + 32;
        let mut buf = vec![0u8; shm_size];
        shm_init(&mut buf, 0, shm_size);
        let payload = vec![0xCCu8; 16]; // 16 + 16 = 32 = capacity.
        let written = shm_write(&mut buf, 0, 1, &payload);
        assert_eq!(written, 32);

        let result = shm_drain(&buf, 0);
        assert_eq!(result.entries.len(), 1);
        assert_eq!(result.entries[0].payload, payload);
    }
}