ktstr 0.6.0

//! `#[ktstr_test]` entry registration and scheduler configuration.
//!
//! Every `#[ktstr_test]` proc-macro expansion writes a static
//! [`KtstrTestEntry`] into [`KTSTR_TESTS`] via `linkme`; programmatic
//! callers can do the same by pushing their own `KtstrTestEntry` values
//! into the slice. [`find_test`] is the name-to-entry lookup used by
//! host-side dispatch.
//!
//! Each entry points at a [`Scheduler`] definition — a `&'static` value
//! that captures the scheduler binary ([`SchedulerSpec`]), guest
//! [`Sysctl`]s, kernel args, [`CgroupPath`] parent, topology and
//! [`TopologyConstraints`], and monitor assertions. [`BpfMapWrite`]
//! describes a host-side map write the runtime performs mid-run.

use anyhow::Result;
use linkme::distributed_slice;
use std::time::Duration;

use crate::assert::AssertResult;
use crate::scenario::Ctx;

/// Re-exports of topology types for use in [`KtstrTestEntry`] statics
/// generated by the `#[ktstr_test]` macro.
pub use crate::vmm::topology::{MemSideCache, NumaDistance, NumaNode, Topology};

/// How to specify the scheduler for an `#[ktstr_test]`.
///
/// The four variants form a semantic taxonomy, not a syntactic one:
/// [`Eevdf`](Self::Eevdf) is the no-scx control ("kernel default,
/// don't launch anything"); [`Discover`](Self::Discover) and
/// [`Path`](Self::Path) both locate a userspace scheduler binary, by
/// name-lookup vs. explicit filesystem path respectively; and
/// [`KernelBuiltin`](Self::KernelBuiltin) activates an in-kernel
/// scheduling policy via shell commands rather than any binary.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum SchedulerSpec {
    /// No userspace scheduler — run under the kernel's default
    /// scheduler. On current kernels that's EEVDF; the variant
    /// name is fixed so the assertion isn't coupled to the kernel
    /// version's default.
    Eevdf,
    /// Auto-discover the scheduler binary by name (looks in
    /// `KTSTR_SCHEDULER` env, the ktstr binary's sibling dir,
    /// `target/debug/`, `target/release/`, and invokes
    /// `cargo build` if nothing's found).
    Discover(&'static str),
    /// Explicit filesystem path to a scheduler binary. The file must
    /// already exist; `resolve_scheduler` does not auto-build this
    /// variant.
    Path(&'static str),
    /// Kernel-built scheduler (e.g. BPF-less sched_ext or
    /// debugfs-tuned). Activated/deactivated via shell commands
    /// rather than a userspace binary.
    KernelBuiltin {
        /// Shell commands invoked before the scenario runs to switch
        /// the kernel into this scheduling policy (e.g. write to
        /// `/sys/kernel/debug/sched/...`).
        enable: &'static [&'static str],
        /// Shell commands invoked after the scenario finishes to
        /// restore the kernel's baseline scheduling policy.
        disable: &'static [&'static str],
    },
}

impl SchedulerSpec {
    /// Whether this spec represents an active scheduling policy
    /// (anything other than the kernel default EEVDF).
    pub const fn has_active_scheduling(&self) -> bool {
        !matches!(self, SchedulerSpec::Eevdf)
    }

    /// Short, human-readable name for logging and sidecar output.
    ///
    /// Maps `Eevdf` → `"eevdf"`, `Discover(n)` → `n`, `Path(p)` → `p`,
    /// `KernelBuiltin { .. }` → `"kernel"`. Single source of truth
    /// for "what do we call this scheduler when telling the user" —
    /// sidecar serialization and failure-header formatting both use
    /// this, and any future consumer gets identical naming for free.
    pub const fn display_name(&self) -> &'static str {
        match self {
            SchedulerSpec::Eevdf => "eevdf",
            SchedulerSpec::Discover(n) => n,
            SchedulerSpec::Path(p) => p,
            SchedulerSpec::KernelBuiltin { .. } => "kernel",
        }
    }

    /// Best-effort git commit of the scheduler binary used for this
    /// run, or `None` when the commit cannot be determined honestly.
    ///
    /// Currently ALWAYS returns `None`. The field is reserved on the
    /// sidecar schema so stats tooling can enrich it once a reliable
    /// source exists, but no variant today has one:
    ///
    /// - `Eevdf` — no userspace scheduler binary at all. Kernel
    ///   default; the running kernel's identity belongs in
    ///   `host.kernel_release`, not here.
    /// - `Discover(_)` — `resolve_scheduler` has a 5-path cascade
    ///   (`KTSTR_SCHEDULER` env override → sibling of the ktstr
    ///   binary → `target/debug/` → `target/release/` → cargo
    ///   rebuild fallback). Only the rebuild path guarantees the
    ///   resulting binary was built from the current tree; the
    ///   four pre-built discovery paths can point at a binary
    ///   whose commit is unknown to this process. Synthesizing a
    ///   commit would be a lie in 4 of 5 cases and would silently
    ///   attribute regressions to the wrong commit. A future
    ///   enhancement can probe the binary itself (e.g.
    ///   `--version` output, an ELF note) and return `Some(..)`
    ///   ONLY when the actual commit is introspected; until then,
    ///   `None` is the only honest answer.
    /// - `Path(p)` — arbitrary externally-built binary. No
    ///   reliable introspection path (no shared ABI, no required
    ///   `--version` format).
    /// - `KernelBuiltin` — in-kernel scheduler, no userspace
    ///   binary commit to record.
    ///
    /// Returning `None` rather than `Some("unknown")` keeps the
    /// sidecar schema's nullable semantics honest: `stats compare`
    /// distinguishes "unset" from "set to a sentinel" without a
    /// magic string, and a future enhancement that learns to
    /// introspect a scheduler binary can flip a single arm to
    /// `Some(..)` without retrofitting consumers to strip a
    /// sentinel.
    pub const fn scheduler_commit(&self) -> Option<&'static str> {
        match self {
            SchedulerSpec::Eevdf
            | SchedulerSpec::Discover(_)
            | SchedulerSpec::Path(_)
            | SchedulerSpec::KernelBuiltin { .. } => None,
        }
    }
}

/// A `key=value` sysctl applied to the guest before the scheduler
/// starts, injected into the guest kernel cmdline as
/// `sysctl.<key>=<value>`.
///
/// Use the **dot-separated** form for `key` (e.g. `"kernel.foo"`, not
/// `"kernel/foo"`). Duplicate keys in a scheduler's sysctls slice are
/// applied in order; the last write wins.
///
/// Construct with [`Sysctl::new`], which const-asserts the key format
/// at compile time. Direct struct-literal construction is rejected
/// (fields are crate-private) to ensure every constructed `Sysctl`
/// passed through the format gate.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Sysctl {
    key: &'static str,
    value: &'static str,
}

impl Sysctl {
    /// Const constructor for use in `static`/`const` context.
    ///
    /// # Panics
    ///
    /// Panics at compile time (or const-eval time) when:
    /// - `key` is empty
    /// - `key` contains `/` (common typo from sysctl conf-file paths —
    ///   sysctl-write uses the dotted form, not the slash form)
    /// - `key` contains whitespace, `=`, or any control byte
    ///   (would corrupt the `sysctl.<key>=<value>` cmdline form)
    /// - `key` does not contain at least one `.` (sysctls are
    ///   namespaced like `kernel.foo` / `net.core.bar`; a bare
    ///   single-segment name is almost certainly a typo)
    /// - `key` starts or ends with `.`
    /// - `key` contains an empty segment (`..` — kernel sysctl
    ///   parser rejects)
    /// - `value` is empty
    /// - `value` contains a newline / carriage return / `=`
    ///   (would corrupt the `sysctl.<key>=<value>` cmdline form)
    pub const fn new(key: &'static str, value: &'static str) -> Self {
        let key_bytes = key.as_bytes();
        assert!(!key_bytes.is_empty(), "Sysctl key must not be empty");
        assert!(
            key_bytes[0] != b'.' && key_bytes[key_bytes.len() - 1] != b'.',
            "Sysctl key must not start or end with `.`",
        );
        let mut i = 0;
        let mut has_dot = false;
        let mut prev = 0u8;
        while i < key_bytes.len() {
            let b = key_bytes[i];
            assert!(
                b != b'/',
                "Sysctl key must use the dotted form (e.g. `kernel.foo`), not the slash form (`kernel/foo`)",
            );
            assert!(
                b != b' ' && b != b'\t' && b != b'\n' && b != b'\r',
                "Sysctl key must not contain whitespace (would corrupt cmdline form)",
            );
            assert!(
                b != b'=',
                "Sysctl key must not contain `=` (would corrupt `sysctl.<key>=<value>` cmdline split)",
            );
            assert!(
                b >= 0x20 && b < 0x7f,
                "Sysctl key must be printable ASCII only (no control bytes / high-bit chars)",
            );
            if b == b'.' {
                has_dot = true;
                assert!(
                    i == 0 || prev != b'.',
                    "Sysctl key must not contain `..` (empty segment — kernel sysctl parser rejects)",
                );
            }
            prev = b;
            i += 1;
        }
        assert!(
            has_dot,
            "Sysctl key must be namespaced (contain at least one `.`, e.g. `kernel.foo`)",
        );
        let value_bytes = value.as_bytes();
        assert!(!value_bytes.is_empty(), "Sysctl value must not be empty");
        let mut j = 0;
        while j < value_bytes.len() {
            let b = value_bytes[j];
            assert!(
                b != b'\n',
                "Sysctl value must not contain a newline (would corrupt cmdline form)",
            );
            assert!(
                b != b'\r',
                "Sysctl value must not contain a carriage return (would corrupt cmdline form)",
            );
            assert!(
                b != b'=',
                "Sysctl value must not contain `=` (would corrupt `sysctl.<key>=<value>` cmdline form)",
            );
            j += 1;
        }
        Self { key, value }
    }

    /// The validated dotted sysctl key (e.g. `"kernel.sched_cfs_bandwidth_slice_us"`).
    pub const fn key(&self) -> &'static str {
        self.key
    }

    /// The validated sysctl value, written as a string.
    pub const fn value(&self) -> &'static str {
        self.value
    }
}

/// Validated cgroup parent path.
///
/// Wraps a `&'static str` that is guaranteed to start with `/` and not
/// be `"/"` alone. Construct via [`CgroupPath::new`] (which const-panics
/// on invalid input) or the [`Scheduler::cgroup_parent`] builder.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct CgroupPath(&'static str);

impl CgroupPath {
    /// Const constructor. Mirrors the runtime gate in
    /// `test_support::args::cell_parent_path_is_valid` byte-for-byte.
    /// `Path::components` and `Components` are not yet const-fn
    /// reachable, so the walk iterates bytes directly via a stable
    /// `while` loop.
    ///
    /// So `cgroup_parent = "/foo/.."` panics at const-eval just as
    /// `--cell-parent-cgroup=/foo/..` panics at test setup; the two
    /// share validation contract, and the asymmetry that would
    /// otherwise exist (const-fn rejecting `/foo/./bar` while the
    /// runtime accepts it) is eliminated by the auto-strip rule.
    ///
    /// # Panics
    ///
    /// Panics at compile time (or const-eval time) when `path` would
    /// normalize back to (or escape) the host cgroup root once
    /// concatenated with `/sys/fs/cgroup` and canonicalized by the
    /// kernel:
    ///
    ///   - must start with `/`
    ///   - must not contain any `..` (ParentDir) segments
    ///   - must contain at least one non-`.`, non-empty segment
    ///   - empty segments (consecutive `/`) and `.` (CurDir) segments
    ///     are auto-stripped, matching `Path::components` semantics
    ///     used by the runtime validator
    pub const fn new(path: &'static str) -> Self {
        assert!(
            !path.is_empty() && path.as_bytes()[0] == b'/',
            "CgroupPath must begin with '/' (e.g. \"/ktstr\")"
        );
        assert!(
            path.len() > 1,
            "CgroupPath must not be \"/\" alone (that is the cgroup root)"
        );
        let bytes = path.as_bytes();
        let mut seg_start: usize = 1; // skip leading `/`
        let mut i: usize = 1;
        let mut has_normal = false;
        while i <= bytes.len() {
            let at_end = i == bytes.len();
            if at_end || bytes[i] == b'/' {
                let seg_len = i - seg_start;
                let is_dotdot =
                    seg_len == 2 && bytes[seg_start] == b'.' && bytes[seg_start + 1] == b'.';
                assert!(
                    !is_dotdot,
                    "CgroupPath must not contain `..` segments \
                     (they escape `/sys/fs/cgroup` once the kernel \
                     canonicalizes the resolved path)"
                );
                let is_empty_or_dot = seg_len == 0 || (seg_len == 1 && bytes[seg_start] == b'.');
                if !is_empty_or_dot {
                    has_normal = true;
                }
                seg_start = i + 1;
            }
            i += 1;
        }
        assert!(
            has_normal,
            "CgroupPath must contain at least one non-`.`/non-empty segment \
             (paths like `/`, `///`, or `/.` normalize back to `/sys/fs/cgroup`)"
        );
        Self(path)
    }

    /// The raw path string (e.g. `"/ktstr"`).
    pub const fn as_str(&self) -> &'static str {
        self.0
    }

    /// The full sysfs cgroup directory (e.g. `"/sys/fs/cgroup/ktstr"`).
    pub fn sysfs_path(&self) -> String {
        format!("/sys/fs/cgroup{}", self.0)
    }
}

impl std::ops::Deref for CgroupPath {
    type Target = str;
    fn deref(&self) -> &str {
        self.0
    }
}

impl std::fmt::Display for CgroupPath {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(self.0)
    }
}

/// Host-side BPF map write performed during VM execution.
///
/// The write is event-driven: the host polls for BPF map discoverability
/// (scheduler loaded), then polls the SHM ring for scenario start, then
/// writes.
///
/// Construct with [`BpfMapWrite::new`], which const-asserts the
/// `map_name_suffix` format at compile time. Direct struct-literal
/// construction is rejected (fields are crate-private) so every
/// constructed `BpfMapWrite` passes through the format gate.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub struct BpfMapWrite {
    map_name_suffix: &'static str,
    offset: usize,
    value: u32,
}

impl BpfMapWrite {
    /// Const constructor for use in `static`/`const` context.
    ///
    /// # Panics
    ///
    /// Panics at compile time (or const-eval time) when:
    /// - `map_name_suffix` is empty
    /// - `map_name_suffix` does not start with `.` (BPF map names
    ///   are derived from ELF section names like `.bss`, `.data`,
    ///   `.rodata`; a suffix without the leading `.` would never
    ///   match any loaded map and is almost certainly a typo)
    /// - `map_name_suffix` is a bare `.` or starts with `..`
    ///   (no real BPF section name has that shape)
    /// - `map_name_suffix` contains whitespace, `/`, `\`, or any
    ///   non-printable / control byte (no real BPF map name carries
    ///   those characters; NULL would truncate libbpf C-string
    ///   comparison)
    pub const fn new(map_name_suffix: &'static str, offset: usize, value: u32) -> Self {
        let bytes = map_name_suffix.as_bytes();
        assert!(
            !bytes.is_empty(),
            "BpfMapWrite map_name_suffix must not be empty"
        );
        assert!(
            bytes[0] == b'.',
            "BpfMapWrite map_name_suffix must start with `.` (BPF map suffixes match ELF section names like `.bss`, `.data`, `.rodata`)",
        );
        assert!(
            bytes.len() >= 2,
            "BpfMapWrite map_name_suffix must be longer than a bare `.` (no real BPF section name is just `.`)",
        );
        assert!(
            bytes[1] != b'.',
            "BpfMapWrite map_name_suffix must not start with `..` (no real BPF section name has that shape)",
        );
        let mut i = 0;
        while i < bytes.len() {
            let b = bytes[i];
            assert!(
                b != b' ' && b != b'\t' && b != b'\n' && b != b'\r',
                "BpfMapWrite map_name_suffix must not contain whitespace",
            );
            assert!(
                b != b'/' && b != b'\\',
                "BpfMapWrite map_name_suffix must not contain path separators",
            );
            assert!(
                b >= 0x20 && b < 0x7f,
                "BpfMapWrite map_name_suffix must be printable ASCII only (no control bytes / NULL / high-bit chars)",
            );
            i += 1;
        }
        Self {
            map_name_suffix,
            offset,
            value,
        }
    }

    /// The validated map-name suffix to match against loaded BPF maps
    /// (e.g. `".bss"`).
    pub const fn map_name_suffix(&self) -> &'static str {
        self.map_name_suffix
    }

    /// Byte offset within the map's value region.
    pub const fn offset(&self) -> usize {
        self.offset
    }

    /// u32 value to write at `offset` inside the matched map.
    pub const fn value(&self) -> u32 {
        self.value
    }
}

/// Gauntlet topology filtering constraints.
///
/// Controls which gauntlet presets are eligible for a test entry.
/// Presets that don't meet all constraints are skipped.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct TopologyConstraints {
    /// Minimum number of NUMA nodes.
    pub min_numa_nodes: u32,
    /// Maximum number of NUMA nodes.
    pub max_numa_nodes: Option<u32>,
    /// Minimum number of LLCs.
    pub min_llcs: u32,
    /// Maximum number of LLCs.
    pub max_llcs: Option<u32>,
    /// Whether the test requires SMT (threads_per_core > 1).
    pub requires_smt: bool,
    /// Minimum total CPU count.
    pub min_cpus: u32,
    /// Maximum total CPU count.
    pub max_cpus: Option<u32>,
}

impl TopologyConstraints {
    /// Conservative default constraints: single NUMA node, 1-12 LLCs,
    /// no SMT requirement, 1-192 CPUs. Accepts most single-node
    /// gauntlet presets ktstr ships while rejecting multi-NUMA presets
    /// (numa2-*, numa4-*) and the scale-boundary single-node presets
    /// that exceed the CPU/LLC caps (near-max-llc, max-cpu, and their
    /// -nosmt variants). Test authors that want broader coverage must
    /// raise `max_numa_nodes`, `max_llcs`, or `max_cpus` explicitly.
    ///
    /// The canonical const handle — use directly when you need an
    /// explicit const binding (`pub const X: TopologyConstraints =
    /// TopologyConstraints::DEFAULT;`). For struct-literal spread in
    /// a `static` / `const` initializer, either form works (modern
    /// Rust promotes the trivially-Copy temporary returned by the
    /// const-fn constructor):
    ///
    /// ```ignore
    /// pub static A: TopologyConstraints = TopologyConstraints {
    ///     min_llcs: 4,
    ///     ..TopologyConstraints::DEFAULT  // const path
    /// };
    /// pub static B: TopologyConstraints = TopologyConstraints {
    ///     min_llcs: 4,
    ///     ..TopologyConstraints::new()    // const-fn constructor
    /// };
    /// ```
    ///
    /// [`Self::new()`] and `Default::default()` delegate to this const
    /// for the `const fn` / trait-impl entry points.
    pub const DEFAULT: Self = Self {
        min_numa_nodes: 1,
        max_numa_nodes: Some(1),
        min_llcs: 1,
        max_llcs: Some(12),
        requires_smt: false,
        min_cpus: 1,
        max_cpus: Some(192),
    };

    /// Build the default constraints. Equivalent to [`Self::DEFAULT`].
    /// Either form works for struct-literal spread in `static` /
    /// `const`: `..Self::DEFAULT` (const path) or `..Self::new()`
    /// (const-fn constructor — modern Rust promotes the trivially-Copy
    /// temporary). `Default::default()` is the trait-impl entry point
    /// for non-const contexts.
    pub const fn new() -> Self {
        Self::DEFAULT
    }
}

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

impl TopologyConstraints {
    /// Whether a topology preset is eligible under these constraints
    /// and the host's physical limits.
    pub fn accepts(
        &self,
        topo: &Topology,
        host_cpus: u32,
        host_llcs: u32,
        host_max_cpus_per_llc: u32,
    ) -> bool {
        topo.num_numa_nodes() >= self.min_numa_nodes
            && self
                .max_numa_nodes
                .is_none_or(|max| topo.num_numa_nodes() <= max)
            && topo.num_llcs() >= self.min_llcs
            && self.max_llcs.is_none_or(|max| topo.num_llcs() <= max)
            && (!self.requires_smt || topo.threads_per_core >= 2)
            && topo.total_cpus() >= self.min_cpus
            && self.max_cpus.is_none_or(|max| topo.total_cpus() <= max)
            && topo.total_cpus() <= host_cpus
            && topo.num_llcs() <= host_llcs
            && topo.cores_per_llc * topo.threads_per_core <= host_max_cpus_per_llc
    }

    /// No-perf-mode variant of [`Self::accepts`]. The VM topology is
    /// emulated via KVM (vCPUs, ACPI SRAT/SLIT, CPUID), not pinned to
    /// host hardware, so the host's NUMA-node count, LLC count, and
    /// per-LLC CPU width do not constrain it. Only the total-CPU
    /// inequality survives — the guest needs `topo.total_cpus()` host
    /// CPUs to schedule its vCPU threads, regardless of how those
    /// vCPUs are grouped into virtual LLCs and nodes.
    ///
    /// The entry's `min_numa_nodes` / `min_llcs` / `min_cpus` /
    /// `requires_smt` / `max_*` fields still gate which gauntlet
    /// presets the test author wants to exercise — those are
    /// expressing test scope, not host capability — so they keep
    /// firing here. The host-side checks (`<= host_cpus`,
    /// `<= host_llcs`, `<= host_max_cpus_per_llc`) collapse to the
    /// single CPU-budget check.
    pub fn accepts_no_perf_mode(&self, topo: &Topology, host_cpus: u32) -> bool {
        topo.num_numa_nodes() >= self.min_numa_nodes
            && self
                .max_numa_nodes
                .is_none_or(|max| topo.num_numa_nodes() <= max)
            && topo.num_llcs() >= self.min_llcs
            && self.max_llcs.is_none_or(|max| topo.num_llcs() <= max)
            && (!self.requires_smt || topo.threads_per_core >= 2)
            && topo.total_cpus() >= self.min_cpus
            && self.max_cpus.is_none_or(|max| topo.total_cpus() <= max)
            && topo.total_cpus() <= host_cpus
    }

    /// Reject inverted ranges (any `max_*` strictly less than the
    /// matching `min_*`). An inverted range cannot match ANY topology
    /// — [`Self::accepts`] / [`Self::accepts_no_perf_mode`] would
    /// silently return `false` for every preset and the test would
    /// be skipped without diagnostic.
    ///
    /// Wired into [`KtstrTestEntry::validate`], which fires at the
    /// start of `run_ktstr_test` (before any VM boot or preset
    /// enumeration). A struct-literal like `TopologyConstraints {
    /// min_numa_nodes: 5, max_numa_nodes: Some(2), ..DEFAULT }`
    /// surfaces a loud error within seconds of test dispatch instead
    /// of as a silently-empty gauntlet sweep at runtime. NOTE: this
    /// only covers the BASE test invocation path — nextest's
    /// `--list` output for gauntlet variant lines still elides
    /// presets that don't satisfy `accepts()`, so an inverted entry
    /// produces zero gauntlet variant lines in `--list` output
    /// without a per-variant diagnostic. The base-test invocation
    /// surfaces the error definitively; the listing-time silent
    /// elision of gauntlet variants is a separate concern.
    pub fn validate(&self) -> anyhow::Result<()> {
        if let Some(max) = self.max_numa_nodes
            && max < self.min_numa_nodes
        {
            anyhow::bail!(
                "TopologyConstraints inverted: max_numa_nodes={} < \
                 min_numa_nodes={}. No topology can satisfy both bounds, \
                 so every gauntlet preset would silently skip.",
                max,
                self.min_numa_nodes,
            );
        }
        if let Some(max) = self.max_llcs
            && max < self.min_llcs
        {
            anyhow::bail!(
                "TopologyConstraints inverted: max_llcs={} < min_llcs={}. \
                 No topology can satisfy both bounds.",
                max,
                self.min_llcs,
            );
        }
        if let Some(max) = self.max_cpus
            && max < self.min_cpus
        {
            anyhow::bail!(
                "TopologyConstraints inverted: max_cpus={} < min_cpus={}. \
                 No topology can satisfy both bounds.",
                max,
                self.min_cpus,
            );
        }
        Ok(())
    }
}

impl TopologyConstraints {
    /// Override `min_numa_nodes`.
    #[must_use = "builder methods consume self; bind the result"]
    pub const fn with_min_numa_nodes(mut self, min_numa_nodes: u32) -> Self {
        self.min_numa_nodes = min_numa_nodes;
        self
    }

    /// Override `max_numa_nodes`.
    #[must_use = "builder methods consume self; bind the result"]
    pub const fn with_max_numa_nodes(mut self, max_numa_nodes: u32) -> Self {
        self.max_numa_nodes = Some(max_numa_nodes);
        self
    }

    /// Clear `max_numa_nodes` (lift the upper bound).
    #[must_use = "builder methods consume self; bind the result"]
    pub const fn without_max_numa_nodes(mut self) -> Self {
        self.max_numa_nodes = None;
        self
    }

    /// Override `min_llcs`.
    #[must_use = "builder methods consume self; bind the result"]
    pub const fn with_min_llcs(mut self, min_llcs: u32) -> Self {
        self.min_llcs = min_llcs;
        self
    }

    /// Override `max_llcs`.
    #[must_use = "builder methods consume self; bind the result"]
    pub const fn with_max_llcs(mut self, max_llcs: u32) -> Self {
        self.max_llcs = Some(max_llcs);
        self
    }

    /// Clear `max_llcs` (lift the upper bound).
    #[must_use = "builder methods consume self; bind the result"]
    pub const fn without_max_llcs(mut self) -> Self {
        self.max_llcs = None;
        self
    }

    /// Override `requires_smt`.
    #[must_use = "builder methods consume self; bind the result"]
    pub const fn with_requires_smt(mut self, requires_smt: bool) -> Self {
        self.requires_smt = requires_smt;
        self
    }

    /// Override `min_cpus`.
    #[must_use = "builder methods consume self; bind the result"]
    pub const fn with_min_cpus(mut self, min_cpus: u32) -> Self {
        self.min_cpus = min_cpus;
        self
    }

    /// Override `max_cpus`.
    #[must_use = "builder methods consume self; bind the result"]
    pub const fn with_max_cpus(mut self, max_cpus: u32) -> Self {
        self.max_cpus = Some(max_cpus);
        self
    }

    /// Clear `max_cpus` (lift the upper bound).
    #[must_use = "builder methods consume self; bind the result"]
    pub const fn without_max_cpus(mut self) -> Self {
        self.max_cpus = None;
        self
    }
}

/// Definition of a scheduler for the test framework.
///
/// Captures everything the framework needs to know about a scheduler:
/// its name, binary spec, sysctls, kernel args, scheduler args,
/// cgroup parent, default topology, gauntlet topology constraints,
/// config-file plumbing, kernel sweep set, and assertion overrides.
///
/// Construct via the `declare_scheduler!` macro (the production
/// path) or the [`Scheduler::named`] const builder chain. Test bodies
/// reference declared schedulers via the `scheduler = MY_SCHED`
/// attribute on `#[ktstr_test]`.
#[derive(Debug)]
pub struct Scheduler {
    /// Short human name for the scheduler, used in logs and sidecar
    /// metadata.
    pub name: &'static str,
    /// Source of the scheduler: a built-in spec variant (`Eevdf`,
    /// `Discover`, `Path`, or `KernelBuiltin`).
    ///
    /// The `declare_scheduler!` macro exposes three pseudo-keys that
    /// each map to one [`SchedulerSpec`] variant — the macro never
    /// accepts an enum literal, so authors must pick the matching
    /// key:
    /// - `binary = "scx_name"` → [`SchedulerSpec::Discover("scx_name")`](SchedulerSpec::Discover) (PATH lookup in the guest).
    /// - `binary_path = "/abs/path"` → [`SchedulerSpec::Path("/abs/path")`](SchedulerSpec::Path) (explicit absolute path).
    /// - `kernel_builtin_enable = "…", kernel_builtin_disable = "…"` (paired) → [`SchedulerSpec::KernelBuiltin`].
    ///
    /// Code that constructs a [`Scheduler`] outside the macro
    /// (manual `..Scheduler::EEVDF` spread, programmatic builders)
    /// uses the typed [`SchedulerSpec`] enum directly — see the
    /// chainable [`Scheduler::binary_discover`] sugar for the
    /// `Discover` common case.
    pub binary: SchedulerSpec,
    /// Guest sysctls applied before the scheduler starts (injected
    /// into the guest kernel cmdline as `sysctl.<key>=<value>`).
    /// Applied in order; duplicate keys last-write-wins.
    pub sysctls: &'static [Sysctl],
    /// Extra guest kernel command-line arguments appended when booting
    /// the VM. This is the GUEST KERNEL cmdline, not the scheduler
    /// binary's CLI — use [`sched_args`](field@Self::sched_args) for that.
    ///
    /// Do not override the kargs ktstr injects itself (`console=`,
    /// `loglevel=`, `init=`): those break guest-side init
    /// and leave the VM unable to run tests.
    pub kargs: &'static [&'static str],
    /// Scheduler-wide assertion overrides merged on top of
    /// `Assert::default_checks()` and below each per-entry `assert`.
    ///
    /// Construct via [`crate::assert::Assert::NO_OVERRIDES`] (the
    /// zero-overrides baseline) chained through the `Assert` builder
    /// methods. The `Assert` builder surface (every overridable
    /// threshold + scheduler-tunable knob) is documented at the
    /// [Checking](https://likewhatevs.github.io/ktstr/guide/concepts/checking.html)
    /// guide chapter; see [`crate::assert::Assert`] for the full
    /// per-method threshold list.
    pub assert: crate::assert::Assert,
    /// Cgroup parent path. Must begin with `/` and must not be `"/"`
    /// alone (that is the cgroup root). Example: `"/ktstr"`.
    /// When set, the init creates the sysfs directory before starting
    /// the scheduler, and `--cell-parent-cgroup {path}` is injected
    /// into scheduler args.
    pub cgroup_parent: Option<CgroupPath>,
    /// Scheduler CLI args, prepended before per-test `extra_sched_args`.
    pub sched_args: &'static [&'static str],
    /// Default VM topology for tests using this scheduler. Tests inherit
    /// this topology unless they override `numa_nodes`, `llcs`, `cores`,
    /// or `threads` explicitly in `#[ktstr_test]`.
    pub topology: Topology,
    /// Gauntlet topology constraints. Tests inherit these unless they
    /// override specific fields in `#[ktstr_test]`.
    pub constraints: TopologyConstraints,
    /// Host-side path to an opaque config file passed to the scheduler
    /// binary inside the VM. The file is included in the initramfs at
    /// `/include-files/{filename}` and `--config /include-files/{filename}`
    /// is prepended to `sched_args`.
    pub config_file: Option<&'static str>,
    /// Declares how an inline config is passed to the scheduler.
    /// First element: CLI arg template with `{file}` placeholder
    /// (e.g. `"f:{file}"`, `"--config {file}"`). Second element:
    /// guest filesystem path where the JSON is written
    /// (e.g. `"/include-files/layered.json"`). The framework
    /// `mkdir -p`s the parent and writes the config content there.
    pub config_file_def: Option<(&'static str, &'static str)>,
    /// Per-scheduler filter on the verifier sweep matrix.
    ///
    /// Each entry is a string consumed by [`KernelId::parse`](crate::kernel_path::KernelId::parse)
    /// at verifier runtime — same parser as the
    /// `cargo ktstr verifier --kernel <SPEC>` CLI flag. Accepts exact
    /// versions (`"6.14"`), closed ranges spelled either `..` or `..=`
    /// (`"6.14..7.0"` or `"6.14..=7.0"` — both inclusive on both
    /// endpoints), git refs (`"git+URL#REF"`), paths, and cache keys.
    ///
    /// The verifier sweep matrix is driven by the operator's
    /// `cargo ktstr verifier --kernel <SPEC>` set (which the
    /// dispatcher always populates into `KTSTR_KERNEL_LIST` — even
    /// the no-`--kernel` case synthesizes a single auto-discovered
    /// entry). For each scheduler,
    /// `list_verifier_cells_all` in `test_support::dispatch`
    /// emits a cell per (kernel-list entry that passes this filter ×
    /// accepted gauntlet topology preset).
    ///
    /// Match semantics per spec variant:
    /// - `Version`: raw-label string equality OR sanitized-label
    ///   match against the kernel-list entry.
    /// - `Range`: range-membership check via
    ///   `decompose_version_for_compare` on the entry's raw version
    ///   string. Lets a scheduler declaring
    ///   `kernels = ["6.14..6.16"]` match any operator-supplied
    ///   kernel whose version falls in `[6.14, 6.16]` inclusive.
    /// - `Path` / `CacheKey` / `Git`: sanitized-label equality.
    ///
    /// Empty (`&[]`) means no filter — the scheduler verifies
    /// against every entry in `KTSTR_KERNEL_LIST`.
    pub kernels: &'static [&'static str],
}

impl Scheduler {
    /// Placeholder scheduler representing "no scx scheduler." Tests
    /// that use `Scheduler::EEVDF` run under the kernel's default
    /// scheduler (EEVDF on current kernels), with no scheduler binary
    /// launched. Useful as a baseline and for tests that exercise
    /// framework behavior independent of any scx scheduler.
    ///
    /// The `.name` is the compile-time-fixed string `"eevdf"` — NOT
    /// runtime-derived from the live kernel. A sidecar written on a
    /// kernel whose default is a successor scheduling class still
    /// records `"eevdf"` here as long as the test attributes this
    /// `Scheduler` to a run. The sidecar reads
    /// `self.name` directly via the `scheduler` field
    /// projection.
    ///
    /// The sidecar itself does not carry the mapping from kernel
    /// version to scheduling class: the `host.kernel_release`
    /// field records the live kernel's release string (`6.6.12`,
    /// `6.14.2`, …) and nothing else. Consumers that need to answer
    /// "did this run actually use EEVDF?" must combine
    /// `host.kernel_release` with version-to-class knowledge
    /// maintained OUTSIDE the sidecar — e.g. an external lookup
    /// table that records the default scheduling class per upstream
    /// kernel release. A sidecar alone cannot distinguish a true
    /// EEVDF run from a run on a successor-class kernel that
    /// reused the same `"eevdf"` label via
    /// [`Scheduler::EEVDF`]'s compile-time-fixed `.name`.
    pub const EEVDF: Scheduler = Scheduler {
        name: "eevdf",
        binary: SchedulerSpec::Eevdf,
        sysctls: &[],
        kargs: &[],
        assert: crate::assert::Assert::NO_OVERRIDES,
        cgroup_parent: None,
        sched_args: &[],
        topology: Topology {
            llcs: 1,
            cores_per_llc: 2,
            threads_per_core: 1,
            numa_nodes: 1,
            nodes: None,
            distances: None,
        },
        constraints: TopologyConstraints::DEFAULT,
        config_file: None,
        config_file_def: None,
        kernels: &[],
    };

    /// Const constructor for defining schedulers in static context.
    /// Caller chains [`Self::binary`] (or
    /// [`Self::binary_discover`]) plus any of the other const-fn
    /// builders to override the per-field defaults; the unset
    /// fields stay at the values below.
    ///
    /// **Defaults** (intentional — these compose with the
    /// per-test `#[ktstr_test]` attributes via merge, so every
    /// field has a no-op identity that lets per-test overrides
    /// take precedence):
    /// - `binary`: [`SchedulerSpec::Eevdf`] — kernel default
    ///   scheduling class, no scheduler binary launched.
    ///   Override via [`Self::binary`] / [`Self::binary_discover`].
    /// - `sysctls` / `kargs` / `sched_args`: empty slices — no
    ///   guest sysctls applied, no extra kernel cmdline args, no
    ///   extra scheduler CLI args.
    /// - `assert`: [`crate::assert::Assert::NO_OVERRIDES`] — no
    ///   threshold overrides on top of
    ///   [`crate::assert::Assert::default_checks`].
    /// - `cgroup_parent`: `None` — no `--cell-parent-cgroup`
    ///   injection.
    /// - `topology`: `1 numa × 1 llc × 2 cores × 1 thread` —
    ///   smallest meaningful topology for a quick smoke
    ///   scheduler. Override via [`Self::topology`].
    /// - `constraints`: [`TopologyConstraints::DEFAULT`] — no
    ///   gauntlet limits.
    /// - `config_file` / `config_file_def`: `None` — no config
    ///   file plumbing. Setting both at construction is rejected
    ///   at validation time.
    /// - `kernels`: empty slice — verifies against every kernel
    ///   in the operator's `--kernel` set (no per-scheduler
    ///   filter).
    pub const fn named(name: &'static str) -> Scheduler {
        Scheduler {
            name,
            binary: SchedulerSpec::Eevdf,
            sysctls: &[],
            kargs: &[],
            assert: crate::assert::Assert::NO_OVERRIDES,
            cgroup_parent: None,
            sched_args: &[],
            topology: Topology {
                llcs: 1,
                cores_per_llc: 2,
                threads_per_core: 1,
                numa_nodes: 1,
                nodes: None,
                distances: None,
            },
            constraints: TopologyConstraints::DEFAULT,
            config_file: None,
            config_file_def: None,
            kernels: &[],
        }
    }

    /// Set the binary spec. Returns self for const chaining.
    pub const fn binary(mut self, binary: SchedulerSpec) -> Self {
        self.binary = binary;
        self
    }

    /// Sugar for `.binary(SchedulerSpec::Discover(name))` — the
    /// dominant Scheduler construction path. Use when the scheduler
    /// binary is on `PATH` under the named filename (e.g.
    /// `.binary_discover("scx_rusty")`) and the framework should
    /// resolve it at guest-init time via `which`-style lookup.
    /// Equivalent to:
    ///
    /// ```ignore
    /// Scheduler::named("rusty").binary(SchedulerSpec::Discover("scx_rusty"))
    /// // is equivalent to
    /// Scheduler::named("rusty").binary_discover("scx_rusty")
    /// ```
    ///
    /// For an explicit absolute path (no `PATH` lookup), call
    /// `.binary(SchedulerSpec::Path("/absolute/path"))` directly —
    /// the path variant has no chainable sugar because absolute
    /// paths are the rare case (cross-compiled trees, ad-hoc
    /// installs) and a path-typed setter would obscure the intent.
    pub const fn binary_discover(self, name: &'static str) -> Self {
        self.binary(SchedulerSpec::Discover(name))
    }

    /// Set sysctls. Returns self for const chaining.
    pub const fn sysctls(mut self, sysctls: &'static [Sysctl]) -> Self {
        self.sysctls = sysctls;
        self
    }

    /// Set kernel args. Returns self for const chaining.
    pub const fn kargs(mut self, kargs: &'static [&'static str]) -> Self {
        self.kargs = kargs;
        self
    }

    /// Set assertion config. Returns self for const chaining.
    pub const fn assert(mut self, assert: crate::assert::Assert) -> Self {
        self.assert = assert;
        self
    }

    /// Set cgroup parent path. See the [`cgroup_parent`](field@Self::cgroup_parent)
    /// field for path format requirements (must begin with `/`, must
    /// not be `"/"` alone).
    pub const fn cgroup_parent(mut self, path: &'static str) -> Self {
        self.cgroup_parent = Some(CgroupPath::new(path));
        self
    }

    /// Set scheduler CLI args prepended before per-test
    /// `extra_sched_args`.
    pub const fn sched_args(mut self, args: &'static [&'static str]) -> Self {
        self.sched_args = args;
        self
    }

    /// Set the default VM topology for tests using this scheduler.
    /// Tests inherit this unless they override individual dimensions
    /// explicitly in `#[ktstr_test]`.
    pub const fn topology(mut self, numa_nodes: u32, llcs: u32, cores: u32, threads: u32) -> Self {
        self.topology = Topology {
            llcs,
            cores_per_llc: cores,
            threads_per_core: threads,
            numa_nodes,
            nodes: None,
            distances: None,
        };
        self
    }

    /// Set gauntlet topology constraints. Tests inherit these unless
    /// they override specific fields in `#[ktstr_test]`.
    pub const fn constraints(mut self, constraints: TopologyConstraints) -> Self {
        self.constraints = constraints;
        self
    }

    /// Set minimum number of NUMA nodes.
    pub const fn min_numa_nodes(mut self, n: u32) -> Self {
        self.constraints.min_numa_nodes = n;
        self
    }

    /// Set maximum number of NUMA nodes.
    pub const fn max_numa_nodes(mut self, n: u32) -> Self {
        self.constraints.max_numa_nodes = Some(n);
        self
    }

    /// Set minimum number of LLCs.
    pub const fn min_llcs(mut self, n: u32) -> Self {
        self.constraints.min_llcs = n;
        self
    }

    /// Set maximum number of LLCs.
    pub const fn max_llcs(mut self, n: u32) -> Self {
        self.constraints.max_llcs = Some(n);
        self
    }

    /// Set whether the scheduler requires SMT.
    pub const fn requires_smt(mut self, v: bool) -> Self {
        self.constraints.requires_smt = v;
        self
    }

    /// Set minimum total CPU count.
    pub const fn min_cpus(mut self, n: u32) -> Self {
        self.constraints.min_cpus = n;
        self
    }

    /// Set maximum total CPU count.
    pub const fn max_cpus(mut self, n: u32) -> Self {
        self.constraints.max_cpus = Some(n);
        self
    }

    /// Set a host-side config file path. The file is included in the
    /// guest initramfs and `--config` is injected into scheduler args.
    pub const fn config_file(mut self, path: &'static str) -> Self {
        self.config_file = Some(path);
        self
    }

    /// Declare how inline config content is passed to the scheduler.
    /// `arg_template`: CLI arg with `{file}` placeholder for the
    /// guest path (e.g. `"f:{file}"`, `"--config {file}"`).
    /// `guest_path`: where the JSON is written in the guest
    /// (e.g. `"/include-files/layered.json"`).
    pub const fn config_file_def(
        mut self,
        arg_template: &'static str,
        guest_path: &'static str,
    ) -> Self {
        self.config_file_def = Some((arg_template, guest_path));
        self
    }

    /// Set the kernel specs the verifier should exercise this scheduler
    /// against. See [`Self::kernels`](field@Self::kernels) for the
    /// accepted string shapes.
    pub const fn kernels(mut self, kernels: &'static [&'static str]) -> Self {
        self.kernels = kernels;
        self
    }

    /// Whether this scheduler runs an active scheduling policy
    /// (anything other than the kernel default EEVDF). Forwards to
    /// [`SchedulerSpec::has_active_scheduling`].
    ///
    /// Kept (unlike the trivial `name` / `binary` accessors that were
    /// dropped) because the 11+ call sites in `eval.rs` / `probe.rs`
    /// would otherwise spell `entry.scheduler.binary.has_active_scheduling()`
    /// at every dispatch decision — the `.binary.` indirection
    /// repeats noise without adding meaning. The forwarder is one
    /// line of glue for a recurring readability win.
    pub const fn has_active_scheduling(&self) -> bool {
        self.binary.has_active_scheduling()
    }
}

/// Registration entry for an `#[ktstr_test]`-annotated function.
///
/// Construct via the [`#[ktstr_test]`] macro (the production path)
/// or struct-literal with `..KtstrTestEntry::DEFAULT` (the
/// integration-test / gauntlet rewriter path). The macro emits the
/// `linkme` distributed-slice registration; programmatic callers
/// register via [`KTSTR_TESTS`].
#[derive(Debug)]
pub struct KtstrTestEntry {
    /// Fully qualified test name as it appears in nextest output.
    pub name: &'static str,
    /// Entry point invoked once per replica, inside the guest VM when
    /// `host_only` is false and on the host when it is true.
    pub func: fn(&Ctx) -> Result<AssertResult>,
    /// Base virtual topology; gauntlet expansion produces additional
    /// variants layered on top of this baseline.
    pub topology: Topology,
    /// Host-topology constraints (CPU and LLC bounds) that gate
    /// whether this entry is eligible on the current machine.
    pub constraints: TopologyConstraints,
    /// Guest memory in MiB (binary mebibytes; conversion at
    /// VM-launch is `value << 20` bytes, not `value * 1_000_000`).
    pub memory_mib: u32,
    /// Primary scheduler that drives the test. Defaults to
    /// [`Scheduler::EEVDF`] (the no-scx-scheduler placeholder; tests
    /// then run under the kernel's default scheduling class).
    ///
    /// Reference to a static [`Scheduler`] — typically the const
    /// emitted by [`declare_scheduler!`](crate::declare_scheduler).
    /// Pure-binary-workload tests use `Scheduler::EEVDF` here and
    /// supply their binary via the `payload` field below.
    pub scheduler: &'static crate::test_support::Scheduler,
    /// Additional schedulers staged into the guest initramfs alongside
    /// [`Self::scheduler`] so the scheduler-lifecycle ops
    /// ([`Op::ReplaceScheduler`](crate::scenario::ops::Op::ReplaceScheduler)
    /// and siblings) can swap to a different scheduler mid-experiment
    /// without a VM reboot. The boot-time scheduler is always
    /// [`Self::scheduler`] — entries here are the candidate set the
    /// test will swap TO via `Op::AttachScheduler` /
    /// `Op::ReplaceScheduler`.
    ///
    /// Each staged scheduler must have a `Scheduler::name` that is
    /// unique within the set AND distinct from a small reserved-name
    /// list (currently `scheduler`, `sched_args`, `init`, `args`,
    /// `exec_cmd`, `sched_enable`, `sched_disable`) that the
    /// framework uses for boot-time initramfs entries. Name
    /// collisions OR reserved-name collisions bail at
    /// [`Self::validate`] time with an actionable diagnostic naming
    /// the offending entries.
    ///
    /// The boot-time [`Self::scheduler`] is NOT auto-included in this
    /// set AND cannot be repeated here — [`Self::validate`] rejects
    /// any staged entry whose `name` matches the boot scheduler's
    /// `name`. Keep the boot scheduler in [`Self::scheduler`] and
    /// stage only the additional candidates the test wants to swap
    /// to. Tests that don't use scheduler-lifecycle ops leave this
    /// field at its `&[]` default and pay no initramfs cost for the
    /// staging machinery.
    ///
    /// Staged binary content is hashed into the initramfs cache key
    /// (see `BaseKey` in `crate::vmm::initramfs_cache`) — rebuilding a
    /// staged scheduler between test runs invalidates the cache
    /// automatically; no manual cache clean is needed.
    pub staged_schedulers: &'static [&'static crate::test_support::Scheduler],
    /// Optional binary payload to run as the primary workload. When
    /// `Some`, the test runs the referenced [`Payload`](crate::test_support::Payload)
    /// (which must be [`PayloadKind::Binary`](crate::test_support::PayloadKind::Binary))
    /// alongside the configured scheduler. When `None`, the test runs
    /// a scheduler-only scenario.
    ///
    /// Populated by `#[ktstr_test(payload = SOME_BIN)]`; direct
    /// programmatic callers may also set this.
    pub payload: Option<&'static crate::test_support::Payload>,
    /// Additional binary payloads composed with the primary. Each
    /// entry is launched via [`Ctx::payload`](crate::scenario::Ctx)
    /// in the test body.
    ///
    /// Populated by `#[ktstr_test(workloads = [A, B])]`.
    pub workloads: &'static [&'static crate::test_support::Payload],
    /// When true, a crash triggers an auto-repro run with BPF probes
    /// attached to the crash call chain.
    pub auto_repro: bool,
    /// Per-entry assertion overrides merged on top of
    /// `Assert::default_checks()` and the scheduler's `assert`.
    pub assert: crate::assert::Assert,
    /// Extra CLI arguments appended to the scheduler invocation.
    pub extra_sched_args: &'static [&'static str],
    /// `scx_sched.watchdog_timeout` override applied to the guest kernel.
    pub watchdog_timeout: Duration,
    /// Host-side BPF map writes to perform during VM execution.
    ///
    /// Empty slice (the default) means "no writes." Setup failures
    /// (accessor init, map resolution, probes-ready wait) abort before
    /// any writes are attempted. Within the write phase itself, every
    /// entry is attempted; individual write failures are logged but do
    /// not abort remaining writes — partial success suppresses the
    /// final signal to the guest so it times out rather than observing
    /// half-applied state.
    pub bpf_map_write: &'static [&'static BpfMapWrite],
    /// Pin vCPU threads to host cores matching the virtual topology's LLC
    /// structure, use 2MB hugepages for guest memory, NUMA mbind guest
    /// memory to pinned vCPU nodes, and promote vCPU threads to
    /// SCHED_FIFO. Validates that the host has enough CPUs and LLCs to
    /// satisfy the request without oversubscription.
    ///
    /// On x86_64, additionally: set KVM_HINTS_REALTIME CPUID hint
    /// (disables PV spinlocks, PV TLB flush, PV sched_yield; enables
    /// haltpoll cpuidle), disable PAUSE and HLT VM exits via
    /// KVM_CAP_X86_DISABLE_EXITS (HLT falls back to PAUSE-only when
    /// mitigate_smt_rsb is active), skip KVM_CAP_HALT_POLL (guest
    /// haltpoll cpuidle disables host halt polling via
    /// MSR_KVM_POLL_CONTROL), and check TSC stability.
    ///
    /// On aarch64, KVM exit suppression and CPUID hints are not
    /// available. The four host-side optimizations (vCPU pinning,
    /// hugepages, NUMA mbind, RT scheduling) apply.
    pub performance_mode: bool,
    /// Decouple virtual topology from host hardware. When set:
    ///
    /// - The VM's virtual topology (`numa_nodes`, `llcs`, `cores`,
    ///   `threads`) is built as declared — the guest sees the full
    ///   requested topology via KVM vCPU layout, ACPI SRAT/SLIT
    ///   tables, etc.
    /// - Host-side cpuset/LLC locking still applies (the no-perf
    ///   `LlcPlan` path), so concurrent perf-mode peers are still
    ///   serialised against this VM.
    /// - Host-side performance_mode optimisations are skipped:
    ///   no vCPU-to-host-core pinning, no 2 MB hugepages, no NUMA
    ///   mbind, no `SCHED_FIFO` promotion, no `KVM_HINTS_REALTIME`
    ///   CPUID hint, no `KVM_CAP_X86_DISABLE_EXITS`.
    /// - Host topology constraints are relaxed during gauntlet
    ///   preset filtering — the entry's `min_numa_nodes` /
    ///   `min_llcs` / `requires_smt` / per-LLC CPU limits are not
    ///   compared against host hardware. The only host check that
    ///   stays is "total host CPUs >= total vCPUs", so a test
    ///   declaring `numa_nodes = 3` runs on a 1-NUMA-node host.
    ///
    /// Equivalent to setting `KTSTR_NO_PERF_MODE=1` per-test —
    /// either source forces the no-perf path. Mutually exclusive
    /// with `performance_mode = true`; [`KtstrTestEntry::validate`]
    /// rejects the combination because "I want pinning" and "I
    /// explicitly don't want pinning" are contradictory.
    pub no_perf_mode: bool,
    /// Workload duration.
    pub duration: Duration,
    /// When true, the test expects run_ktstr_test to return Err.
    /// Disables auto_repro (no point probing a deliberately failing test).
    pub expect_err: bool,
    /// When true, a terminal Inconclusive verdict (e.g. zero-denominator
    /// ratio gate that couldn't evaluate) routes to EXIT_PASS instead
    /// of EXIT_INCONCLUSIVE at the dispatch layer. The test process
    /// exits 0 and CI gates keying off the per-test exit code see no
    /// failure. Use only when the test author has reason to accept an
    /// Inconclusive arm as not-a-failure for this specific test —
    /// e.g. an exploratory benchmark whose ratio gate may legitimately
    /// see no signal under certain host topologies. Inconclusive is
    /// still recorded in the sidecar so stats tooling can surface it,
    /// and the operator-facing failure dump still renders the
    /// Inconclusive diagnostic. This flag changes only the dispatch
    /// exit-code projection.
    ///
    /// Mutually orthogonal with [`Self::expect_err`]: when both are
    /// true and the result is Inconclusive, expect_err still wins
    /// (expect_err demands a real Fail; Inconclusive doesn't satisfy
    /// that and routes to EXIT_FAIL with the expect_err unsatisfied
    /// explainer).
    ///
    /// Populated by `#[ktstr_test(allow_inconclusive)]` /
    /// `#[ktstr_test(allow_inconclusive = true)]` or by direct entry
    /// construction.
    pub allow_inconclusive: bool,
    /// When true, the test runs directly on the host instead of
    /// booting a VM. Used for tests that need host tools (cargo,
    /// nested VMs) unavailable in the guest initramfs.
    pub host_only: bool,
    /// Extra host-side file specs beyond what the entry's
    /// [`scheduler`](Self::scheduler) / [`payload`](Self::payload) /
    /// [`workloads`](Self::workloads) declare. Unions with those
    /// per-payload specs at `run_ktstr_test` time; see
    /// [`all_include_files`](Self::all_include_files) for the
    /// aggregation contract. Use this slot for test-level
    /// dependencies that don't belong on a specific Payload —
    /// auxiliary data files, per-test helper scripts, fixtures.
    pub extra_include_files: &'static [&'static str],
    /// Maximum acceptable wall-clock duration of host-side VM teardown
    /// (BSP exit through SHM drain). Compared against
    /// [`VmResult::cleanup_duration`](crate::vmm::VmResult::cleanup_duration)
    /// in `evaluate_vm_result`; when the budget is exceeded the test's
    /// `AssertResult` is folded with a failing
    /// [`AssertDetail`](crate::assert::AssertDetail). Catches
    /// sub-watchdog cleanup regressions (e.g. a 30s teardown that the
    /// 60s host watchdog would silently absorb) at the test that
    /// declares the budget rather than at gross-timeout failure.
    /// `None` (the default) disables the check, leaving the watchdog
    /// as the only guard. Populated by
    /// `#[ktstr_test(cleanup_budget_ms = N)]` or by direct entry
    /// construction.
    pub cleanup_budget: Option<Duration>,
    /// Inline config content (JSON string) written to the guest path
    /// declared by the scheduler's `config_file_def`. The framework
    /// writes this string to a temp file, packs it into the initramfs,
    /// and passes the scheduler's arg template with `{file}` replaced.
    ///
    /// Populated by `#[ktstr_test(config = EXPR)]` (literal or path to
    /// a `const &'static str`) or by direct entry construction.
    ///
    /// Pairing gate: `config_content` and the scheduler's
    /// `config_file_def` must both be `Some(_)` or both be `None`.
    /// [`KtstrTestEntry::validate`] enforces this at runtime so direct
    /// programmatic-entry callers see the misconfiguration before VM
    /// boot, and the `#[ktstr_test]` macro emits a `const _: () = {
    /// assert!(...) };` block that catches the same mismatch at
    /// compile time for attribute-built entries. A `Some` here without
    /// a scheduler `config_file_def` would be silently dropped at
    /// dispatch (no `--config` flag derives from it); a `None` here
    /// against a scheduler that declares `config_file_def` would
    /// launch the scheduler binary without `--config`. Both are
    /// rejected.
    pub config_content: Option<&'static str>,
    /// Optional virtio-blk disk attached to the VM at `/dev/vda`.
    /// `None` (the default) boots without a disk; `Some(cfg)` calls
    /// `crate::vmm::KtstrVmBuilder::disk` in
    /// `crate::test_support::runtime::build_vm_builder_base` so the
    /// guest sees a raw block device sized per `cfg.capacity_mib`.
    /// The `#[ktstr_test]` macro does not currently surface this
    /// slot — direct construction via `..KtstrTestEntry::DEFAULT`
    /// is the only path. Mutually exclusive with `host_only`:
    /// `validate` rejects the combination because `host_only`
    /// skips the VM boot that owns the disk lifecycle.
    pub disk: Option<crate::vmm::disk_config::DiskConfig>,
    /// Host-side callback invoked after `vm.run()` returns, with
    /// access to the full `VmResult`. Runs on the HOST, not inside
    /// the guest. Use for assertions that need host-side state
    /// (e.g., `VmResult.snapshot_bridge` content after a snapshot
    /// capture pipeline fires inside the VM).
    ///
    /// `None` (the default) skips the callback. When `Some`, the
    /// closure receives `&VmResult` and returns `Result<()>` — an
    /// `Err` fails the test with the returned message.
    pub post_vm: Option<fn(&crate::vmm::VmResult) -> Result<()>>,
    /// Periodic snapshot count: when non-zero, the freeze
    /// coordinator divides the 10%–90% slice of the workload
    /// duration (anchored at the FIRST `MSG_TYPE_SCENARIO_START`
    /// the coordinator observes) into `num_snapshots + 1` equal
    /// intervals and fires a host-side `freeze_and_capture(false)`
    /// at each of the `num_snapshots` interior boundaries —
    /// e.g. `N = 1` lands a single capture at the workload's
    /// midpoint (`start + 0.5·d`); `N = 3` lands captures at
    /// `start + 0.3·d`, `start + 0.5·d`, `start + 0.7·d`. No
    /// boundary lands at exactly `start + 0.1·d` or
    /// `start + 0.9·d` — the buffers reserve those edges for
    /// workload ramp-up / ramp-down. Each boundary is stored
    /// under `"periodic_NNN"` (zero-padded 3-digit index) on the
    /// host's [`crate::scenario::snapshot::SnapshotBridge`].
    /// Anchoring at ScenarioStart means boot + verifier time do
    /// not eat the budget. Pauses observed via
    /// `MSG_TYPE_SCENARIO_PAUSE` / `MSG_TYPE_SCENARIO_RESUME` shift
    /// every un-fired boundary by the cumulative pause duration —
    /// the boundary clock is workload-time, not wall-clock, so a
    /// guest that pauses for `P` ns delays each remaining boundary
    /// by `P` ns. `0` (the default) disables periodic capture
    /// entirely; the coordinator's run-loop never even computes
    /// boundary timestamps.
    ///
    /// **Capture cost.** Each periodic boundary fires the same
    /// host-side `freeze_and_capture(false)` path that
    /// [`crate::scenario::ops::Op::CaptureSnapshot`] dispatches: every
    /// vCPU is parked under `FREEZE_RENDEZVOUS_TIMEOUT` (30 s
    /// hard ceiling), BPF maps are walked, the dump is serialised
    /// to JSON, and the report is stored on the
    /// [`crate::scenario::snapshot::SnapshotBridge`]. On a healthy
    /// guest with a typical scheduler-state map size the freeze
    /// is tens of milliseconds (10–100 ms is the steady-state
    /// observation; cold-cache and large guest-memory walks can
    /// push higher). The host-side watchdog deadline is extended
    /// by the freeze duration after each fire, so periodic
    /// captures do not eat into the workload's wall-clock budget.
    ///
    /// **Best-effort delivery.** Up to `N` captures fire; an early
    /// VM exit (kill flag, BSP done, rendezvous timeout, watchdog
    /// deadline) can cut the periodic sequence short, and the
    /// run-loop stops servicing periodic boundaries the moment the
    /// kill flag fires. Tests should assert `>= some_lower_bound`
    /// rather than `== num_snapshots`. `Op::CaptureSnapshot` captures
    /// composed by the test author land on the same bridge
    /// alongside the `periodic_NNN` tags; total bridge occupancy
    /// is `num_snapshots + user_captures` and the bridge
    /// FIFO-evicts past
    /// [`crate::scenario::snapshot::MAX_STORED_SNAPSHOTS`].
    /// Additionally, the run-loop abandons the remaining sequence
    /// after 2 consecutive rendezvous timeouts and emits a
    /// `tracing::warn` naming the consecutive-timeout count.
    ///
    /// **Minimum spacing.** Each capture freezes every vCPU for
    /// tens of milliseconds at minimum (see "Capture cost" above),
    /// so boundaries scheduled closer than ~100 ms apart would
    /// fire back-to-back without any workload progress in between.
    /// `validate()` rejects entries where
    /// `0.8 · duration / (N + 1) < 100 ms` — choose `N` and
    /// `duration` so the resulting interval clears that floor.
    ///
    /// **Ordering.** Periodic captures are stored in order
    /// (`periodic_000` first, `periodic_NNN` last). Tests that
    /// need to walk them in time order should call
    /// [`crate::scenario::snapshot::SnapshotBridge::drain_ordered`]
    /// rather than [`crate::scenario::snapshot::SnapshotBridge::drain`]
    /// — the latter returns a `HashMap` and loses ordering.
    ///
    /// `validate()` rejects `num_snapshots >
    /// crate::scenario::snapshot::MAX_STORED_SNAPSHOTS` (= 64
    /// today): the bridge enforces FIFO eviction at that cap, so a
    /// higher count would silently drop the earliest periodic
    /// samples once `store()` started evicting. Refusing the
    /// configuration is more honest than half-delivering it. The
    /// 64 cap also ensures the 3-digit `:03` width on
    /// `periodic_NNN` is always sufficient.
    pub num_snapshots: u32,
    /// Cgroup directory that the framework creates BEFORE the
    /// scheduler starts and uses as the parent for every workload
    /// cgroup the test author declares via [`Ctx::cgroup_def`]
    /// (`ctx.cgroup_def("cg_0")` etc.). When `Some(path)`, the
    /// guest mkdir's `/sys/fs/cgroup{path}` and the per-test
    /// CgroupManager places its children there
    /// (`/sys/fs/cgroup{path}/cg_0` etc.); when `None`, the
    /// framework falls back to the legacy resolution
    /// (`crate::test_support::args::resolve_cgroup_root` —
    /// `--cell-parent-cgroup` in `sched_args` overrides; default
    /// `/sys/fs/cgroup/ktstr`).
    ///
    /// Distinct from [`crate::test_support::Scheduler::cgroup_parent`]:
    /// `cgroup_parent` is a scheduler-only knob that controls the
    /// scheduler argv (`--cell-parent-cgroup` flag, only when the
    /// scheduler declaration explicitly carries it in `sched_args`);
    /// `workload_root_cgroup` is a framework knob for the workload
    /// side and never reaches the scheduler argv. A test can set
    /// `workload_root_cgroup` without affecting the scheduler's
    /// cgroup placement, and a scheduler can set `cgroup_parent`
    /// without affecting where workloads land.
    pub workload_root_cgroup: Option<CgroupPath>,
    /// Whether KASLR is enabled in the guest kernel for this test.
    /// `true` (the default) lets the guest randomize kernel virt + direct-map
    /// addresses (CONFIG_RANDOMIZE_BASE=y + CONFIG_RANDOMIZE_MEMORY=y in
    /// `ktstr.kconfig`); `false` appends `nokaslr` to the guest cmdline so
    /// the kernel-image slide and the `page_offset_base` direct-map randomization
    /// both stay at compile-time defaults. KASLR-off is the determinism escape
    /// for tests that depend on fixed kernel addresses or that need to reproduce
    /// bugs masked by randomization; the default-on case exercises ktstr's
    /// host-side derivation chain (MSR_LSTAR readback, KERN_ADDRS guest channel,
    /// /proc/kallsyms `page_offset_base` lookup) end-to-end. The
    /// `kaslr_*_e2e` regression tests guard the derivation; the
    /// `kaslr_disabled_via_macro_attribute` regression guards this opt-out.
    /// Operator-level alternative: `kargs = ["nokaslr"]` on the scheduler decl
    /// — same effect, declared once for every test that uses the scheduler.
    /// Combining `kaslr = false` with `kargs = ["nokaslr"]` is redundant
    /// but harmless — `nokaslr` appearing twice on the cmdline is a no-op
    /// (kernel parses it as a bool flag, not a value).
    pub kaslr: bool,
}

/// Placeholder function for [`KtstrTestEntry::DEFAULT`].
///
/// Returns `Err` — NOT a panic — so a programmatic caller that
/// accidentally uses `KtstrTestEntry::DEFAULT` without overriding
/// `func` gets an immediate actionable failure inside the test-run
/// loop rather than taking down the whole dispatch process. The
/// `..KtstrTestEntry::DEFAULT` struct-update spread only populates
/// unfilled fields; if the caller spread the default without
/// setting `func`, this stub runs and bails with a message pointing
/// at the mistake.
fn default_test_func(_ctx: &Ctx) -> Result<AssertResult> {
    anyhow::bail!("KtstrTestEntry::DEFAULT func called — override func before use")
}

impl KtstrTestEntry {
    /// Sensible defaults for all fields. Override `name`, `func`, and
    /// `scheduler` (at minimum) via struct update syntax. Manual
    /// consumers should also set `auto_repro` explicitly: the default
    /// `true` boots a second VM with BPF probes attached on failure,
    /// which roughly doubles a failing test's wall-clock time.
    ///
    /// [`Self::DEFAULT`] is the source of truth (struct-literal
    /// const); [`Self::new()`] is a delegating alias for method-style
    /// use and `Default::default()` is the trait-shim — both
    /// equivalent in non-const contexts. For `static` / `const`
    /// initializer spread sites (e.g. `#[distributed_slice(KTSTR_TESTS)]`
    /// macro expansions), `..Self::DEFAULT` is the canonical shape —
    /// it spreads the struct-literal const directly without taking a
    /// detour through a const-fn return.
    ///
    /// ```
    /// use ktstr::prelude::*;
    ///
    /// fn my_test_fn(_ctx: &Ctx) -> Result<AssertResult> {
    ///     Ok(AssertResult::pass())
    /// }
    ///
    /// #[distributed_slice(KTSTR_TESTS)]
    /// #[linkme(crate = ktstr::linkme)]
    /// static ENTRY: KtstrTestEntry = KtstrTestEntry {
    ///     name: "my_test",
    ///     func: my_test_fn,
    ///     scheduler: &Scheduler::EEVDF,
    ///     ..KtstrTestEntry::DEFAULT
    /// };
    /// ```
    ///
    /// The `#[linkme(crate = ktstr::linkme)]` annotation is required
    /// when the downstream crate does not depend on `linkme` directly
    /// — see [`crate::distributed_slice`] for the full rationale.
    pub const DEFAULT: Self = Self {
        name: "",
        func: default_test_func,
        topology: Topology {
            llcs: 1,
            cores_per_llc: 2,
            threads_per_core: 1,
            numa_nodes: 1,
            nodes: None,
            distances: None,
        },
        constraints: TopologyConstraints::DEFAULT,
        memory_mib: 2048,
        scheduler: &crate::test_support::Scheduler::EEVDF,
        staged_schedulers: &[],
        payload: None,
        workloads: &[],
        auto_repro: true,
        assert: crate::assert::Assert::NO_OVERRIDES,
        extra_sched_args: &[],
        watchdog_timeout: Duration::from_secs(5),
        bpf_map_write: &[],
        performance_mode: false,
        no_perf_mode: false,
        duration: Duration::from_secs(12),
        expect_err: false,
        allow_inconclusive: false,
        host_only: false,
        extra_include_files: &[],
        cleanup_budget: None,
        config_content: None,
        disk: None,
        post_vm: None,
        num_snapshots: 0,
        workload_root_cgroup: None,
        kaslr: true,
    };

    /// Build the default entry. Equivalent to [`Self::DEFAULT`].
    /// Either `..Self::DEFAULT` or `..Self::new()` works in
    /// `static` / `const` initializer spread sites since `new()`
    /// is `const fn` and KtstrTestEntry has no Drop-bearing
    /// fields. `Default::default()` is the trait-shim equivalent
    /// for non-const contexts.
    pub const fn new() -> Self {
        Self::DEFAULT
    }

    /// Reject values that would boot a broken VM or leave assertions
    /// vacuously passing. The `#[ktstr_test]` proc macro enforces the
    /// same constraints at compile time for attribute-built entries;
    /// this method covers directly-constructed entries (library
    /// callers building `KtstrTestEntry` values to push into
    /// [`KTSTR_TESTS`] programmatically).
    ///
    /// Rules:
    /// - `name` must be non-empty (empty names collapse into each
    ///   other in nextest output and in sidecar lookups).
    /// - `name` must not contain `/` or `\` (path separators embed in
    ///   sidecar filenames and nextest test IDs; a separator would
    ///   create a synthetic subdirectory in sidecar output and
    ///   mangle `cargo nextest run -E 'test(name)'` filtering).
    /// - `memory_mib` must be `> 0` (a VM with zero memory cannot boot).
    /// - `duration` must be `> 0` (a zero-duration run never exercises
    ///   the scheduler and produces no telemetry).
    pub fn validate(&self) -> anyhow::Result<()> {
        if self.name.is_empty() {
            anyhow::bail!(
                "KtstrTestEntry.name must be non-empty (empty names \
                 collide in nextest output and sidecar lookups)"
            );
        }
        if self.name.contains('/') || self.name.contains('\\') {
            anyhow::bail!(
                "KtstrTestEntry '{}' name must not contain path \
                 separators ('/' or '\\') — they embed in sidecar \
                 filenames and nextest test IDs, creating synthetic \
                 subdirectories in sidecar output and mangling \
                 nextest -E 'test(name)' filtering",
                self.name,
            );
        }
        if self.memory_mib == 0 {
            anyhow::bail!(
                "KtstrTestEntry '{}'.memory_mib must be > 0 (a VM with \
                 zero memory cannot boot)",
                self.name,
            );
        }
        if self.duration.is_zero() {
            anyhow::bail!(
                "KtstrTestEntry '{}'.duration must be > 0 (a zero-duration \
                 run never exercises the scheduler and produces no data \
                 for assertions)",
                self.name,
            );
        }
        if let Some(p) = self.payload
            && p.is_scheduler()
        {
            anyhow::bail!(
                "KtstrTestEntry '{}'.payload must be PayloadKind::Binary, \
                 not Scheduler-kind (schedulers belong in the `scheduler` \
                 slot; the `payload` slot is for userspace binaries \
                 composed under the scheduler)",
                self.name,
            );
        }
        if self.host_only && self.disk.is_some() {
            anyhow::bail!(
                "KtstrTestEntry '{}'.host_only=true with disk=Some(..) — \
                 host_only skips the VM boot that owns the virtio-blk \
                 device lifecycle, so the disk would never be attached. \
                 Drop one of host_only or disk.",
                self.name,
            );
        }
        // staged_schedulers names must (a) pass the per-name shape
        // checks (non-empty, no path separators, no NUL bytes, no
        // leading dot, not a reserved framework slot — see
        // [`crate::test_support::staged::validate_staged_scheduler_name`])
        // and (b) be unique within the set AND disjoint from the
        // boot scheduler's `name`. A collision on either axis would
        // land two distinct schedulers at the same guest path —
        // silent overwrite, the second-staged binary clobbering the
        // first OR shadowing a boot-time framework slot. The
        // boot-name seed catches the "stage all the schedulers I
        // might use" misuse (author includes the boot scheduler in
        // the staged set thinking it's required there too). Bails
        // here at validate time so the error surfaces ahead of any
        // VM boot or initramfs construction.
        let mut seen_names: std::collections::BTreeSet<&'static str> =
            std::collections::BTreeSet::new();
        seen_names.insert(self.scheduler.name);
        let staged_who = format!("KtstrTestEntry '{}'.staged_schedulers", self.name);
        for staged in self.staged_schedulers {
            crate::test_support::staged::validate_staged_scheduler_name(&staged_who, staged.name)?;
            if !seen_names.insert(staged.name) {
                if staged.name == self.scheduler.name {
                    anyhow::bail!(
                        "KtstrTestEntry '{}'.staged_schedulers cannot include \
                         the boot scheduler '{}' — the boot slot already \
                         stages it. Staged entries are the ADDITIONAL \
                         candidates the test will swap TO via \
                         Op::AttachScheduler / Op::ReplaceScheduler.",
                        self.name,
                        staged.name,
                    );
                }
                anyhow::bail!(
                    "KtstrTestEntry '{}'.staged_schedulers has duplicate \
                     Scheduler.name '{}'; each staged scheduler must have \
                     a unique name (the name maps 1:1 to the guest-side \
                     staging path)",
                    self.name,
                    staged.name,
                );
            }
        }
        // Defense-in-depth for the programmatic-construction path
        // (struct-literal `KtstrTestEntry { .. }` in integration tests,
        // gauntlet-rewritten entries). The macro at
        // ktstr-macros/src/lib.rs rejects `host_only = true` paired with
        // any `scheduler = ...` attribute at compile time, but
        // programmatic construction bypasses that gate. Match against
        // `SchedulerSpec::Eevdf` (the value-level marker for the
        // no-scx-scheduler placeholder) so a struct literal that sets
        // `scheduler: &SOME_REAL_SCHED` under host_only is caught while
        // the default `scheduler: &Scheduler::EEVDF` (whose binary is
        // `SchedulerSpec::Eevdf`) is accepted. The variant-based check
        // is spec-safe — unlike a pointer-identity check against
        // `&Scheduler::EEVDF`, which depends on rustc/LLVM's const-
        // deduplication of `&CONST_EXPR` materializations.
        if self.host_only
            && !matches!(
                self.scheduler.binary,
                crate::test_support::SchedulerSpec::Eevdf
            )
        {
            anyhow::bail!(
                "KtstrTestEntry '{}'.host_only=true with scheduler=&{:?} — \
                 host_only skips the VM boot that owns the scheduler \
                 lifecycle, so the declared scheduler would never attach. \
                 Drop one of host_only or scheduler; the host's \
                 currently-active scheduler (default EEVDF when none is \
                 loaded) runs the test under host_only.",
                self.name,
                self.scheduler.name,
            );
        }
        if self.performance_mode && self.no_perf_mode {
            anyhow::bail!(
                "KtstrTestEntry '{}'.performance_mode=true with \
                 no_perf_mode=true — the two flags are contradictory \
                 (\"I want pinning\" vs. \"I explicitly don't want \
                 pinning\"). Drop one of them.",
                self.name,
            );
        }
        if (self.assert.expect_scx_bpf_error_contains.is_some()
            || self.assert.expect_scx_bpf_error_matches.is_some())
            && !self.expect_err
        {
            anyhow::bail!(
                "KtstrTestEntry '{}' sets an scx_bpf_error matcher \
                 (expect_scx_bpf_error_contains or expect_scx_bpf_error_matches) \
                 without expect_err = true — a reproducer matcher narrows \
                 which failure counts as the expected bug and only \
                 applies to expected-error tests. Set expect_err = true \
                 or drop the matcher.",
                self.name,
            );
        }
        // Periodic snapshots route through SnapshotBridge::store, which
        // FIFO-evicts at MAX_STORED_SNAPSHOTS. Allowing num_snapshots
        // past the cap would silently lose the earliest samples — a
        // periodic run with N=128 today would only retain
        // periodic_064..periodic_127 in the bridge.
        let max = crate::scenario::snapshot::MAX_STORED_SNAPSHOTS as u32;
        if self.num_snapshots > max {
            anyhow::bail!(
                "KtstrTestEntry '{}'.num_snapshots={} exceeds \
                 MAX_STORED_SNAPSHOTS={} — the bridge would FIFO-evict \
                 the earliest periodic samples. Lower the count or split \
                 into multiple test entries.",
                self.name,
                self.num_snapshots,
                max,
            );
        }
        if self.num_snapshots > 0 {
            // host_only skips the VM boot that owns the freeze
            // coordinator's run-loop. Without that loop there is no
            // thread to stamp `scenario_start_ns`, no thread to fire
            // `freeze_and_capture(false)` at each boundary, and no
            // `SnapshotBridge` plumbed onto a `VmResult` for the
            // test author to drain post-run. The combination is
            // unsatisfiable; reject at validate time so a
            // misconfigured entry surfaces during nextest discovery
            // rather than as silently-empty bridge results.
            if self.host_only {
                anyhow::bail!(
                    "KtstrTestEntry '{}'.host_only=true with \
                     num_snapshots={} > 0 — host_only skips the VM \
                     boot that owns the freeze coordinator's \
                     periodic-capture loop, so no snapshot would \
                     ever fire. Drop one of host_only or \
                     num_snapshots.",
                    self.name,
                    self.num_snapshots,
                );
            }
            // Refuse interval shorter than the minimum useful capture
            // cadence. Each boundary fire freezes every vCPU, walks
            // BPF maps, serialises the dump, and writes to the
            // bridge — under the FREEZE_RENDEZVOUS_TIMEOUT (30 s)
            // hard ceiling but commonly tens of milliseconds on a
            // healthy guest. An interval shorter than ~100 ms would
            // back-to-back the captures with no actual workload
            // progress between them, defeating the periodic-sampling
            // purpose. Compute the interval in nanoseconds in u128
            // to avoid overflow on long durations: the formula
            // mirrors the run-loop's
            // `compute_periodic_boundaries_ns` (10 % pre-buffer,
            // 80 % usable span, divided into N+1 equal intervals).
            let usable_span_ns = self
                .duration
                .as_nanos()
                .saturating_sub(2u128.saturating_mul(self.duration.as_nanos() / 10));
            let interval_ns = usable_span_ns / (self.num_snapshots as u128 + 1);
            const MIN_INTERVAL_NS: u128 = 100 * 1_000_000; // 100 ms
            if interval_ns < MIN_INTERVAL_NS {
                anyhow::bail!(
                    "KtstrTestEntry '{}'.num_snapshots={} with \
                     duration={:?} produces a periodic interval of \
                     {} ns ({} ms) — below the 100 ms minimum the \
                     freeze-and-capture path can sustain without \
                     back-to-back firing. Either reduce num_snapshots \
                     or extend duration so 0.8·duration / (N+1) >= 100 ms.",
                    self.name,
                    self.num_snapshots,
                    self.duration,
                    interval_ns,
                    interval_ns / 1_000_000,
                );
            }
        }
        // Pair `scheduler.config_file_def` with `config_content`. The
        // `#[ktstr_test]` macro emits a `const _: () = assert!(...)`
        // block that catches the same mismatch at compile time for
        // attribute-built entries; this branch covers programmatic
        // construction (callers building `KtstrTestEntry` values
        // directly) and surfaces the misconfiguration before VM boot
        // rather than as a silent missing-`--config` flag.
        let scheduler_has_def = self.scheduler.config_file_def.is_some();
        let entry_has_content = self.config_content.is_some();
        if scheduler_has_def && !entry_has_content {
            anyhow::bail!(
                "KtstrTestEntry '{}'.scheduler '{}' declares \
                 `config_file_def` but the entry does not supply \
                 `config_content`; the scheduler binary expects an \
                 inline config and would launch without `--config`. \
                 Set `config = ...` on `#[ktstr_test]` or assign \
                 `config_content` directly.",
                self.name,
                self.scheduler.name,
            );
        }
        if !scheduler_has_def && entry_has_content {
            anyhow::bail!(
                "KtstrTestEntry '{}'.config_content is set but the \
                 scheduler '{}' does not declare `config_file_def`; \
                 the content would be silently dropped at dispatch. \
                 Remove `config = ...` or add \
                 `config_file_def(arg_template, guest_path)` to the \
                 scheduler.",
                self.name,
                self.scheduler.name,
            );
        }
        // Mirror the payload-slot gate for every workload entry. The
        // `workloads` slot is for userspace binaries composed with
        // the primary payload under the scheduler; a scheduler-kind
        // Payload here would be silently ignored at spawn time. The
        // narrow typo path post-`declare_scheduler!` rollout is
        // pasting [`Payload::KERNEL_DEFAULT`] (the only Scheduler-kind
        // Payload still in the prelude) into a `workloads = [...]`
        // attribute instead of the `scheduler = ...` slot.
        for (idx, w) in self.workloads.iter().enumerate() {
            if w.is_scheduler() {
                anyhow::bail!(
                    "KtstrTestEntry '{}'.workloads[{idx}] (name='{}') must be \
                     PayloadKind::Binary, not Scheduler-kind (schedulers belong \
                     in the `scheduler` slot; the `workloads` slot is for \
                     userspace binaries composed under the scheduler)",
                    self.name,
                    w.name,
                );
            }
        }
        // Reject inverted topology ranges before they silently filter
        // every gauntlet preset to zero matches. The per-entry
        // constraints gate which gauntlet presets the test author wants
        // to exercise; an inverted bound (e.g. min_numa_nodes=5 with
        // max_numa_nodes=Some(2)) would yield false on every preset.
        self.constraints
            .validate()
            .map_err(|e| anyhow::anyhow!("KtstrTestEntry '{}'.constraints: {e}", self.name))?;
        // Same for the scheduler-level constraints, which apply on top
        // of the per-entry ones. A scheduler whose declared topology
        // requirements are themselves inverted has the same silent-
        // filter pathology regardless of what test entries declare.
        self.scheduler.constraints.validate().map_err(|e| {
            anyhow::anyhow!(
                "KtstrTestEntry '{}'.scheduler '{}'.constraints: {e}",
                self.name,
                self.scheduler.name
            )
        })?;
        Ok(())
    }

    /// Aggregate every declared include-file spec: the entry's
    /// primary [`payload`](Self::payload) (if present) contributes
    /// its [`Payload::include_files`](crate::test_support::Payload::include_files),
    /// each entry in [`workloads`](Self::workloads) contributes its
    /// own, and [`extra_include_files`](Self::extra_include_files)
    /// contributes test-level extras. Pre-dedupe aggregation order:
    /// payload → workloads (in declaration order) → extras. The
    /// scheduler tier does not contribute — `Scheduler` has no
    /// `include_files` field, and the scheduler binary path is
    /// resolved separately at run time. Duplicate spec strings at
    /// this layer are NOT deduped — the framework's include-file
    /// pipeline at `run_ktstr_test` resolves each entry to a
    /// `(archive_path, host_path)` pair and dedupes on identical
    /// pairs while erroring on archive_path collisions with
    /// conflicting host_paths. This aggregation order does NOT
    /// survive downstream: the final resolved list is sorted
    /// alphabetically by archive_path after deduplication.
    /// Alphabetical ordering ensures deterministic initramfs layout
    /// regardless of declaration order.
    pub fn all_include_files(&self) -> Vec<&'static str> {
        let mut out: Vec<&'static str> = Vec::new();
        if let Some(p) = self.payload {
            out.extend(p.include_files.iter().copied());
        }
        for w in self.workloads {
            out.extend(w.include_files.iter().copied());
        }
        out.extend(self.extra_include_files.iter().copied());
        out
    }
}

/// Programmatic builder methods. Use at runtime (let bindings, fn
/// returns). For `static` / `const` initializers and
/// `#[distributed_slice(KTSTR_TESTS)]` registration, prefer the
/// struct-literal `..KtstrTestEntry::DEFAULT` spread — chained
/// `with_X` calls fail in `const` context with E0015 ("cannot call
/// non-const fn in constants") because these setters are declared
/// `pub fn`, not `pub const fn`. See [`Self::DEFAULT`] for the
/// worked struct-literal example.
impl KtstrTestEntry {
    /// Override `name`.
    #[must_use = "builder methods consume self; bind the result"]
    pub fn with_name(mut self, name: &'static str) -> Self {
        self.name = name;
        self
    }

    /// Override `func`.
    #[must_use = "builder methods consume self; bind the result"]
    pub fn with_func(mut self, func: fn(&Ctx) -> Result<AssertResult>) -> Self {
        self.func = func;
        self
    }

    /// Override `topology`.
    #[must_use = "builder methods consume self; bind the result"]
    pub fn with_topology(mut self, topology: Topology) -> Self {
        self.topology = topology;
        self
    }

    /// Override `constraints`.
    #[must_use = "builder methods consume self; bind the result"]
    pub fn with_constraints(mut self, constraints: TopologyConstraints) -> Self {
        self.constraints = constraints;
        self
    }

    /// Override `memory_mib`.
    #[must_use = "builder methods consume self; bind the result"]
    pub fn with_memory_mib(mut self, memory_mib: u32) -> Self {
        self.memory_mib = memory_mib;
        self
    }

    /// Override `scheduler`.
    #[must_use = "builder methods consume self; bind the result"]
    pub fn with_scheduler(mut self, scheduler: &'static crate::test_support::Scheduler) -> Self {
        self.scheduler = scheduler;
        self
    }

    /// Override `staged_schedulers` — the candidate set the test can
    /// swap to mid-experiment via the scheduler-lifecycle ops
    /// ([`Op::AttachScheduler`](crate::scenario::ops::Op::AttachScheduler) /
    /// [`Op::ReplaceScheduler`](crate::scenario::ops::Op::ReplaceScheduler)).
    /// See the field doc on [`Self::staged_schedulers`] for the
    /// per-scheduler-name uniqueness + reserved-name validation
    /// contract.
    #[must_use = "builder methods consume self; bind the result"]
    pub fn with_staged_schedulers(
        mut self,
        staged: &'static [&'static crate::test_support::Scheduler],
    ) -> Self {
        self.staged_schedulers = staged;
        self
    }

    /// Override `payload`.
    #[must_use = "builder methods consume self; bind the result"]
    pub fn with_payload(mut self, payload: &'static crate::test_support::Payload) -> Self {
        self.payload = Some(payload);
        self
    }

    /// Clear `payload` (run a scheduler-only scenario with no primary
    /// binary).
    #[must_use = "builder methods consume self; bind the result"]
    pub fn without_payload(mut self) -> Self {
        self.payload = None;
        self
    }

    /// Override `workloads`.
    #[must_use = "builder methods consume self; bind the result"]
    pub fn with_workloads(
        mut self,
        workloads: &'static [&'static crate::test_support::Payload],
    ) -> Self {
        self.workloads = workloads;
        self
    }

    /// Override `auto_repro`.
    #[must_use = "builder methods consume self; bind the result"]
    pub fn with_auto_repro(mut self, auto_repro: bool) -> Self {
        self.auto_repro = auto_repro;
        self
    }

    /// Override `assert`.
    ///
    /// Replaces the entry's per-test overrides wholesale; the assertion
    /// resolution at run time still layers `Assert::default_checks()`
    /// and the scheduler-level `assert` underneath.
    #[must_use = "builder methods consume self; bind the result"]
    pub fn with_assert(mut self, assert: crate::assert::Assert) -> Self {
        self.assert = assert;
        self
    }

    /// Override `extra_sched_args`.
    #[must_use = "builder methods consume self; bind the result"]
    pub fn with_extra_sched_args(mut self, extra_sched_args: &'static [&'static str]) -> Self {
        self.extra_sched_args = extra_sched_args;
        self
    }

    /// Override `watchdog_timeout`.
    #[must_use = "builder methods consume self; bind the result"]
    pub fn with_watchdog_timeout(mut self, watchdog_timeout: Duration) -> Self {
        self.watchdog_timeout = watchdog_timeout;
        self
    }

    /// Override `bpf_map_write`.
    #[must_use = "builder methods consume self; bind the result"]
    pub fn with_bpf_map_write(mut self, bpf_map_write: &'static [&'static BpfMapWrite]) -> Self {
        self.bpf_map_write = bpf_map_write;
        self
    }

    /// Override `performance_mode`.
    #[must_use = "builder methods consume self; bind the result"]
    pub fn with_performance_mode(mut self, performance_mode: bool) -> Self {
        self.performance_mode = performance_mode;
        self
    }

    /// Override `no_perf_mode`.
    #[must_use = "builder methods consume self; bind the result"]
    pub fn with_no_perf_mode(mut self, no_perf_mode: bool) -> Self {
        self.no_perf_mode = no_perf_mode;
        self
    }

    /// Override `duration`.
    #[must_use = "builder methods consume self; bind the result"]
    pub fn with_duration(mut self, duration: Duration) -> Self {
        self.duration = duration;
        self
    }

    /// Override `expect_err`.
    #[must_use = "builder methods consume self; bind the result"]
    pub fn with_expect_err(mut self, expect_err: bool) -> Self {
        self.expect_err = expect_err;
        self
    }

    /// Override [`Self::allow_inconclusive`]. When true, an
    /// Inconclusive terminal verdict routes to EXIT_PASS instead
    /// of EXIT_INCONCLUSIVE at the dispatch layer.
    #[must_use = "builder methods consume self; bind the result"]
    pub fn with_allow_inconclusive(mut self, allow_inconclusive: bool) -> Self {
        self.allow_inconclusive = allow_inconclusive;
        self
    }

    /// Override `host_only`.
    #[must_use = "builder methods consume self; bind the result"]
    pub fn with_host_only(mut self, host_only: bool) -> Self {
        self.host_only = host_only;
        self
    }

    /// Override `extra_include_files`.
    #[must_use = "builder methods consume self; bind the result"]
    pub fn with_extra_include_files(
        mut self,
        extra_include_files: &'static [&'static str],
    ) -> Self {
        self.extra_include_files = extra_include_files;
        self
    }

    /// Override `cleanup_budget`.
    #[must_use = "builder methods consume self; bind the result"]
    pub fn with_cleanup_budget(mut self, cleanup_budget: Duration) -> Self {
        self.cleanup_budget = Some(cleanup_budget);
        self
    }

    /// Clear `cleanup_budget` (leave the host watchdog as the only
    /// guard).
    #[must_use = "builder methods consume self; bind the result"]
    pub fn without_cleanup_budget(mut self) -> Self {
        self.cleanup_budget = None;
        self
    }

    /// Override `config_content`.
    #[must_use = "builder methods consume self; bind the result"]
    pub fn with_config_content(mut self, config_content: &'static str) -> Self {
        self.config_content = Some(config_content);
        self
    }

    /// Clear `config_content`.
    #[must_use = "builder methods consume self; bind the result"]
    pub fn without_config_content(mut self) -> Self {
        self.config_content = None;
        self
    }

    /// Override `disk`.
    ///
    /// Pairs with the `host_only = false` requirement enforced by
    /// [`Self::validate`] — `host_only = true` with a `Some(..)` disk
    /// is rejected because host-only skips the VM boot that owns the
    /// virtio-blk lifecycle.
    #[must_use = "builder methods consume self; bind the result"]
    pub fn with_disk(mut self, disk: crate::vmm::disk_config::DiskConfig) -> Self {
        self.disk = Some(disk);
        self
    }

    /// Clear `disk` (boot without a virtio-blk device).
    #[must_use = "builder methods consume self; bind the result"]
    pub fn without_disk(mut self) -> Self {
        self.disk = None;
        self
    }

    /// Override `post_vm`.
    ///
    /// The closure runs on the host after `vm.run()` returns with
    /// access to the full `VmResult`; an `Err` from the closure fails
    /// the test with the returned message.
    #[must_use = "builder methods consume self; bind the result"]
    pub fn with_post_vm(mut self, post_vm: fn(&crate::vmm::VmResult) -> Result<()>) -> Self {
        self.post_vm = Some(post_vm);
        self
    }

    /// Clear `post_vm` (skip the host-side callback).
    #[must_use = "builder methods consume self; bind the result"]
    pub fn without_post_vm(mut self) -> Self {
        self.post_vm = None;
        self
    }

    /// Override `num_snapshots`.
    #[must_use = "builder methods consume self; bind the result"]
    pub fn with_num_snapshots(mut self, num_snapshots: u32) -> Self {
        self.num_snapshots = num_snapshots;
        self
    }

    /// Override `workload_root_cgroup` with a validated path.
    /// `path` must satisfy [`CgroupPath::new`]'s requirements
    /// (starts with `/`, not bare `/`, no `..` components); the
    /// const-eval gate panics on invalid input so programmatic
    /// callers see the same validation as the macro path.
    #[must_use = "builder methods consume self; bind the result"]
    pub const fn with_workload_root_cgroup(mut self, path: &'static str) -> Self {
        self.workload_root_cgroup = Some(CgroupPath::new(path));
        self
    }
}

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

/// Distributed slice collecting all `#[ktstr_test]` entries via linkme.
#[distributed_slice]
pub static KTSTR_TESTS: [KtstrTestEntry];

/// Distributed slice collecting all `declare_scheduler!` registrations
/// via linkme. Each entry is a `&'static Scheduler` pointing at a
/// const emitted by the macro. The verifier discovers schedulers by
/// spawning the test binary with `--ktstr-list-schedulers`; a per-binary
/// ctor walks this slice and serializes each entry to JSON.
#[distributed_slice]
pub static KTSTR_SCHEDULERS: [&'static Scheduler];

/// Look up a registered test function by name.
pub fn find_test(name: &str) -> Option<&'static KtstrTestEntry> {
    KTSTR_TESTS.iter().find(|e| e.name == name)
}

/// Look up a registered scheduler by its [`Scheduler::name`] field
/// (the `name = "..."` value supplied to
/// [`declare_scheduler!`](crate::declare_scheduler), not the
/// SCREAMING_SNAKE_CASE const identifier the macro emits). Returns
/// `None` if no registered scheduler matches.
///
/// Two `declare_scheduler!` invocations that share a `name = "..."`
/// value (under distinct const idents) both register in
/// [`KTSTR_SCHEDULERS`]; a linear scan returns the first match and
/// the second is unreachable. This function panics on the first call
/// when any duplicate exists so the misconfiguration surfaces
/// loudly instead of silently dropping a registration. The scan and
/// map construction happen once per process via a `LazyLock`.
pub fn find_scheduler(name: &str) -> Option<&'static Scheduler> {
    scheduler_index().get(name).copied()
}

/// Process-wide index from [`Scheduler::name`] to the registered
/// `&'static Scheduler`. Built once on first lookup. Detects
/// duplicate names and panics with both colliding consts' addresses
/// so the test author can identify the two declarations.
fn scheduler_index() -> &'static std::collections::HashMap<&'static str, &'static Scheduler> {
    static INDEX: std::sync::LazyLock<std::collections::HashMap<&'static str, &'static Scheduler>> =
        std::sync::LazyLock::new(|| {
            build_scheduler_index_or_panic(KTSTR_SCHEDULERS.iter().copied())
        });
    &INDEX
}

/// Build a name → scheduler map from an iterator of registered
/// schedulers, panicking on the first duplicate name. Factored out
/// so `tests` can exercise the duplicate-detection branch against a
/// mock slice — the real [`KTSTR_SCHEDULERS`] is the union of every
/// `declare_scheduler!` invocation in the linked binary and so
/// cannot host an intentional duplicate without poisoning every
/// other test.
fn build_scheduler_index_or_panic<I>(
    schedulers: I,
) -> std::collections::HashMap<&'static str, &'static Scheduler>
where
    I: IntoIterator<Item = &'static Scheduler>,
{
    let mut map: std::collections::HashMap<&'static str, &'static Scheduler> =
        std::collections::HashMap::new();
    for sched in schedulers {
        if let Some(prev) = map.insert(sched.name, sched)
            && !std::ptr::eq(prev, sched)
        {
            // `{:p}` prints the pointer in the standard `0x…` form
            // so a user diff'ing two declarations can tell at a
            // glance whether they're literally the same static
            // (re-export of the same const, harmless) or two
            // distinct consts with the same `name = "..."` (the
            // collision this guard exists for).
            panic!(
                "ktstr: duplicate scheduler name `{name}` registered \
                 in KTSTR_SCHEDULERS:\n  \
                 first: {prev:p}\n  \
                 second: {sched:p}\n\
                 Two `declare_scheduler!` invocations declared the \
                 same `name = \"{name}\"`. The first registration \
                 wins under linear scan and the second is unreachable. \
                 Rename one of the declarations or remove the \
                 duplicate.",
                name = sched.name,
            );
        }
    }
    map
}

/// JSON shape projected from a registered [`Scheduler`]. Each entry
/// carries scheduler name, a [`BinaryKindJson`]-tagged binary
/// specification (Discover / Path / Eevdf / KernelBuiltin),
/// per-scheduler default [`TopologyJson`], always-on scheduler
/// args, declared kernel set, and gauntlet constraints. Internal
/// fields (assertion overrides, sysctls, kargs, cgroup parent,
/// config-file plumbing) are intentionally omitted.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct SchedulerJson {
    /// Scheduler name — the `name = "..."` value supplied to
    /// [`declare_scheduler!`](crate::declare_scheduler) or
    /// [`Scheduler::named`].
    pub name: String,
    /// Binary specification: distinguishes Discover (build via cargo
    /// `[[bin]]` name), Path (use absolute path verbatim), Eevdf
    /// (kernel default scheduler, no binary), and KernelBuiltin
    /// (built into the kernel, enable/disable via guest commands).
    /// The variant tag lets the verifier dispatch exhaustively `match`
    /// on the binary type without parsing the string.
    pub binary_kind: BinaryKindJson,
    /// Default VM topology for tests using this scheduler. The
    /// verifier sweep's per-cell topology comes from gauntlet presets
    /// filtered through `constraints`; this field carries the
    /// per-scheduler baseline that test-entry plumbing inherits when
    /// a test does not override `numa_nodes`/`llcs`/`cores`/`threads`
    /// in its `#[ktstr_test]` attributes. Mirror of
    /// [`Scheduler::topology`].
    pub topology: TopologyJson,
    /// Always-on scheduler CLI args.
    pub sched_args: Vec<String>,
    /// Kernel specs (consumed by `cargo_ktstr::kernel::resolve_kernel_set`).
    pub kernels: Vec<String>,
    /// Gauntlet preset constraints (filter the verifier's topology sweep).
    pub constraints: TopologyConstraintsJson,
}

/// JSON-friendly form of [`SchedulerSpec`] tagged so the verifier
/// dispatch can exhaustively `match` on the variant. `Discover` and
/// `Path` both carry a string identifier; `Eevdf` and
/// `KernelBuiltin` both signal "no BPF to verify".
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "snake_case", tag = "kind", content = "value")]
pub enum BinaryKindJson {
    /// Cargo `[[bin]]` name. Verifier resolves via `build_and_find_binary`.
    Discover(String),
    /// Absolute filesystem path. Verifier checks `path.exists()` and uses verbatim.
    Path(String),
    /// Kernel default scheduler (EEVDF on current kernels). No BPF, no binary.
    Eevdf,
    /// Built into the kernel (e.g. `scx_simple` enabled via sysfs). No userspace binary.
    KernelBuiltin,
}

/// JSON-friendly mirror of `Topology` for the verifier wire format.
/// Captures the four-tuple shape (numa nodes × LLCs × cores × threads)
/// the per-scheduler baseline topology was declared with. The verifier
/// uses this when computing the sweep matrix — the baseline anchors
/// the default cell when no gauntlet preset matches the
/// scheduler's constraints.
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct TopologyJson {
    /// NUMA node count. Maps to [`Topology::num_numa_nodes`].
    pub num_numa_nodes: u32,
    /// Last-level cache (LLC) count, equivalent to "sockets" on
    /// pre-CCX/CCD x86. Maps to [`Topology::num_llcs`].
    pub num_llcs: u32,
    /// Physical cores per LLC. Mirrors `Topology::cores_per_llc`.
    pub cores_per_llc: u32,
    /// Hardware threads per physical core. `1` for non-SMT; `2`
    /// for x86 hyperthreading. Mirrors `Topology::threads_per_core`.
    pub threads_per_core: u32,
}

impl TopologyJson {
    /// Single-CPU baseline: 1 NUMA node × 1 LLC × 1 core × 1 thread.
    /// Used by `cargo ktstr verifier` and verifier-pipeline tests as
    /// the no-scheduling-workload default.
    pub const SINGLE_CPU: Self = Self {
        num_numa_nodes: 1,
        num_llcs: 1,
        cores_per_llc: 1,
        threads_per_core: 1,
    };
}

/// Result-based validation for wire-format topology values. Lets the
/// verifier dispatch surface a per-cell "topology rejected" diagnostic
/// instead of taking the [`Topology::new`] panic surface in the builder.
/// Mirrors [`Topology::validate`] — any field == 0, overflow in total
/// CPU count, or `llcs` not divisible by `numa_nodes` returns `Err`.
/// The result is a uniform-distribution [`Topology`] (`nodes = None`,
/// `distances = None`); explicit per-node config and distance matrices
/// require constructing [`Topology`] directly.
impl TryFrom<TopologyJson> for Topology {
    type Error = String;

    fn try_from(value: TopologyJson) -> Result<Self, Self::Error> {
        let topo = Self {
            llcs: value.num_llcs,
            cores_per_llc: value.cores_per_llc,
            threads_per_core: value.threads_per_core,
            numa_nodes: value.num_numa_nodes,
            nodes: None,
            distances: None,
        };
        topo.validate()?;
        Ok(topo)
    }
}

/// Project a [`Topology`] into its wire-format mirror. Drops the
/// `nodes` and `distances` fields (uniform-distribution shape only);
/// callers that need to preserve explicit per-node config or distance
/// matrices must not use this conversion. Takes [`Topology`] by value
/// (it derives [`Copy`]) to match the by-value shape of
/// [`From<TopologyConstraintsJson> for TopologyConstraints`].
impl From<Topology> for TopologyJson {
    fn from(t: Topology) -> Self {
        Self {
            num_numa_nodes: t.numa_nodes,
            num_llcs: t.llcs,
            cores_per_llc: t.cores_per_llc,
            threads_per_core: t.threads_per_core,
        }
    }
}

/// JSON-friendly mirror of [`TopologyConstraints`] — the host-side
/// `Option<u32>` fields serialize as `null` (default serde behavior;
/// no `skip_serializing_if`) rather than the `Some(N)`/`None`-tagged
/// shapes serde uses for `Option` inside larger struct graphs.
/// Field semantics match [`TopologyConstraints`] verbatim; see that
/// type for per-field documentation.
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct TopologyConstraintsJson {
    pub min_numa_nodes: u32,
    pub max_numa_nodes: Option<u32>,
    pub min_llcs: u32,
    pub max_llcs: Option<u32>,
    pub requires_smt: bool,
    pub min_cpus: u32,
    pub max_cpus: Option<u32>,
}

/// Infallible shape conversion — every field maps 1:1 to
/// [`TopologyConstraints`], so the verifier sweep dispatch can reuse
/// the same `accepts` / `accepts_no_perf_mode` filters that gauntlet
/// dispatch uses.
impl From<TopologyConstraintsJson> for TopologyConstraints {
    fn from(j: TopologyConstraintsJson) -> Self {
        Self {
            min_numa_nodes: j.min_numa_nodes,
            max_numa_nodes: j.max_numa_nodes,
            min_llcs: j.min_llcs,
            max_llcs: j.max_llcs,
            requires_smt: j.requires_smt,
            min_cpus: j.min_cpus,
            max_cpus: j.max_cpus,
        }
    }
}

impl SchedulerJson {
    /// Project a `Scheduler` static into its JSON shape.
    pub fn from_scheduler(s: &Scheduler) -> Self {
        let binary_kind = match s.binary {
            SchedulerSpec::Discover(n) => BinaryKindJson::Discover(n.to_string()),
            SchedulerSpec::Path(p) => BinaryKindJson::Path(p.to_string()),
            SchedulerSpec::Eevdf => BinaryKindJson::Eevdf,
            SchedulerSpec::KernelBuiltin { .. } => BinaryKindJson::KernelBuiltin,
        };
        Self {
            name: s.name.to_string(),
            binary_kind,
            topology: TopologyJson {
                num_numa_nodes: s.topology.num_numa_nodes(),
                num_llcs: s.topology.num_llcs(),
                cores_per_llc: s.topology.cores_per_llc,
                threads_per_core: s.topology.threads_per_core,
            },
            sched_args: s.sched_args.iter().map(|a| a.to_string()).collect(),
            kernels: s.kernels.iter().map(|k| k.to_string()).collect(),
            constraints: TopologyConstraintsJson {
                min_numa_nodes: s.constraints.min_numa_nodes,
                max_numa_nodes: s.constraints.max_numa_nodes,
                min_llcs: s.constraints.min_llcs,
                max_llcs: s.constraints.max_llcs,
                requires_smt: s.constraints.requires_smt,
                min_cpus: s.constraints.min_cpus,
                max_cpus: s.constraints.max_cpus,
            },
        }
    }
}

::ctor::declarative::ctor! {
/// Ctor that intercepts `--ktstr-list-schedulers` before `main()` runs.
/// Walks [`KTSTR_SCHEDULERS`], serializes each entry to JSON via
/// [`SchedulerJson::from_scheduler`], prints the resulting array on
/// stdout, and exits with status 0.
///
/// One ctor per binary, regardless of how many schedulers the binary
/// registers — walks the slice once and emits a single JSON array.
///
/// Uses ctor's declarative `ctor::declarative::ctor! { ... }` form;
/// the proc-macro `#[ctor::ctor(...)]` form is re-exported at
/// `crate::__private::ctor::ctor` for downstream consumers.
#[ctor(unsafe)]
fn __ktstr_list_schedulers() {
    if !std::env::args().any(|a| a == "--ktstr-list-schedulers") {
        return;
    }
    let entries: Vec<SchedulerJson> = KTSTR_SCHEDULERS
        .iter()
        .map(|s| SchedulerJson::from_scheduler(s))
        .collect();
    let json = ::serde_json::to_string(&entries).expect("serialize schedulers");
    println!("{json}");
    std::process::exit(0);
}
}

/// Default `post_vm` callback emitted by `#[ktstr_test]` when the
/// attribute omits `post_vm = ...`. Asserts that at least one
/// periodic snapshot produced REAL BPF state during the workload
/// window WHEN periodic was configured (`num_snapshots > 0`); when
/// periodic was disabled (`num_snapshots == 0`), the helper is a
/// no-op and returns `Ok`.
///
/// This is the smoke-floor for any periodic-configured test:
/// "the test produced meaningful data." Importantly, the floor
/// reads
/// [`SnapshotBridge::periodic_real_count`](crate::scenario::snapshot::SnapshotBridge::periodic_real_count) —
/// NOT [`VmResult::periodic_fired`](crate::vmm::VmResult::periodic_fired).
/// The latter counts every periodic boundary the coordinator
/// attempted, INCLUDING placeholder rows from rendezvous timeouts;
/// a scheduler that attaches but produces nothing but placeholders
/// would pass the `periodic_fired >= 1` floor and obscure the
/// broken-scheduler diagnosis. The real-count floor catches that
/// case (placeholder-only fills surface as zero) while tolerating
/// the realistic single-snapshot timeout flake (one real capture
/// among N is enough to pass the floor).
///
/// Tests that need stronger assertions (per-snapshot field reads,
/// per-phase ratios, etc.) supply their own `post_vm = my_checker`
/// instead.
///
/// The macro routes this fn pointer into [`KtstrTestEntry::post_vm`]
/// so the runner's existing dispatch path applies it identically
/// to an author-supplied callback.
pub fn default_post_vm_periodic_fired(result: &crate::vmm::VmResult) -> anyhow::Result<()> {
    // Short-circuit when the VM run already failed: the underlying
    // crash/timeout already drives the test failure with its own
    // diagnostic. Emitting a periodic-floor Err on top obscures the
    // real cause (e.g., a scheduler that exited before any periodic
    // boundary fired would surface as "0 real captures" here
    // instead of the scheduler-log message that explains the
    // exit). The runner's own success path renders the crash; let
    // it.
    if !result.success {
        return Ok(());
    }
    if result.periodic_target == 0 {
        return Ok(());
    }
    let real = result.snapshot_bridge.periodic_real_count();
    anyhow::ensure!(
        real >= 1,
        "no periodic snapshot produced real BPF state \
         (periodic_real_count=0, periodic_fired={}, target={}) — \
         scheduler attached but every snapshot was a placeholder \
         (typical cause: scheduler stalled during the workload and \
         the freeze rendezvous timed out fetching state; less \
         commonly: gate suppression rejected every capture)",
        result.periodic_fired,
        result.periodic_target,
    );
    Ok(())
}

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

    /// Minimal Ctx for invoking `default_test_func` without booting a
    /// real workload. The func signature only requires `&Ctx`; the
    /// stub returns Err unconditionally so no field on Ctx is read.
    fn dummy_ctx() -> (crate::cgroup::CgroupManager, crate::topology::TestTopology) {
        let cgroups = crate::cgroup::CgroupManager::new("/sys/fs/cgroup/ktstr-dummy");
        let topo = crate::topology::TestTopology::from_vm_topology(&Topology {
            llcs: 1,
            cores_per_llc: 1,
            threads_per_core: 1,
            numa_nodes: 1,
            nodes: None,
            distances: None,
        });
        (cgroups, topo)
    }

    #[test]
    fn ktstr_test_entry_default_fields() {
        let d = KtstrTestEntry::DEFAULT;
        assert_eq!(d.name, "");
        // func is the stub — verified separately in
        // default_test_func_returns_err.
        assert_eq!(d.topology.llcs, 1);
        assert_eq!(d.topology.cores_per_llc, 2);
        assert_eq!(d.topology.threads_per_core, 1);
        assert_eq!(d.topology.numa_nodes, 1);
        assert!(d.topology.nodes.is_none());
        assert!(d.topology.distances.is_none());
        assert_eq!(d.constraints, TopologyConstraints::DEFAULT);
        assert_eq!(d.memory_mib, 2048);
        // scheduler defaults to `&Scheduler::EEVDF`, whose
        // compile-time-fixed `.name = "eevdf"`. Read directly via
        // field access — no kind dispatch.
        assert_eq!(d.scheduler.name, "eevdf");
        assert!(!d.scheduler.has_active_scheduling());
        assert!(d.auto_repro);
        assert!(d.extra_sched_args.is_empty());
        assert_eq!(d.watchdog_timeout, Duration::from_secs(5));
        assert!(d.bpf_map_write.is_empty());
        assert!(!d.performance_mode);
        assert!(!d.no_perf_mode);
        assert_eq!(d.duration, Duration::from_secs(12));
        assert!(!d.expect_err);
        assert!(!d.host_only);
        // Payload slot defaults to None (scheduler-only entry); workloads
        // slice defaults to empty. Macro emits these as explicit None/&[]
        // so struct-update spreaders also get the right values.
        assert!(d.payload.is_none());
        assert!(d.workloads.is_empty());
    }

    /// Empirical pin: `..Self::new()` works in a `static` spread.
    /// KtstrTestEntry has no Drop-bearing transitive fields (every
    /// type in the struct is Copy or holds only `&'static`/`Option`/
    /// primitives), so `Self::new()` returning a const Self literal
    /// is const-evaluable AND the temporary it produces needs no
    /// destructor. Both DEFAULT and new() are valid in const-spread.
    #[allow(dead_code)]
    static ENTRY_VIA_NEW: KtstrTestEntry = KtstrTestEntry {
        name: "via_new",
        ..KtstrTestEntry::new()
    };

    #[allow(dead_code)]
    static ENTRY_VIA_DEFAULT: KtstrTestEntry = KtstrTestEntry {
        name: "via_default",
        ..KtstrTestEntry::DEFAULT
    };

    #[test]
    fn ktstr_test_entry_const_spread_works_via_both_new_and_default() {
        assert_eq!(ENTRY_VIA_NEW.name, "via_new");
        assert_eq!(ENTRY_VIA_DEFAULT.name, "via_default");
        // Both spread forms produce identical non-name fields.
        assert_eq!(ENTRY_VIA_NEW.memory_mib, ENTRY_VIA_DEFAULT.memory_mib);
        assert_eq!(ENTRY_VIA_NEW.duration, ENTRY_VIA_DEFAULT.duration);
    }

    #[test]
    fn ktstr_test_entry_with_chain_overrides_target_fields() {
        let entry = KtstrTestEntry::DEFAULT
            .with_name("chain_test")
            .with_memory_mib(4096)
            .with_duration(Duration::from_secs(30))
            .with_auto_repro(false)
            .with_performance_mode(true)
            .with_num_snapshots(2);
        assert_eq!(entry.name, "chain_test");
        assert_eq!(entry.memory_mib, 4096);
        assert_eq!(entry.duration, Duration::from_secs(30));
        assert!(!entry.auto_repro);
        assert!(entry.performance_mode);
        assert_eq!(entry.num_snapshots, 2);
        // Untouched defaults survive.
        assert_eq!(entry.scheduler.name, "eevdf");
        assert!(!entry.host_only);
        // Validate succeeds — the chain produced a usable entry.
        entry.validate().expect("chained entry must validate");
    }

    /// `without_<field>` returns the original Option<T> field to
    /// `None`. The chain symmetry pin: `with_X(v).without_X() ==
    /// DEFAULT-state for that field`.
    #[test]
    fn ktstr_test_entry_without_chain_clears_option_fields() {
        use crate::test_support::{OutputFormat, Payload, PayloadKind};
        const FIO: Payload = Payload {
            name: "fio",
            kind: PayloadKind::Binary("fio"),
            output: OutputFormat::Json,
            default_args: &[],
            default_checks: &[],
            metrics: &[],
            include_files: &[],
            uses_parent_pgrp: false,
            known_flags: None,
            metric_bounds: None,
        };
        let entry = KtstrTestEntry::DEFAULT
            .with_name("clear_test")
            .with_payload(&FIO)
            .with_cleanup_budget(Duration::from_secs(10))
            .without_payload()
            .without_cleanup_budget();
        assert!(entry.payload.is_none());
        assert!(entry.cleanup_budget.is_none());
    }

    #[test]
    fn ktstr_test_entry_payload_slot_can_be_populated() {
        use crate::test_support::{OutputFormat, Payload, PayloadKind};
        const FIO: Payload = Payload {
            name: "fio",
            kind: PayloadKind::Binary("fio"),
            output: OutputFormat::Json,
            default_args: &[],
            default_checks: &[],
            metrics: &[],
            include_files: &[],
            uses_parent_pgrp: false,
            known_flags: None,
            metric_bounds: None,
        };
        let entry = KtstrTestEntry {
            name: "payload_entry",
            payload: Some(&FIO),
            ..KtstrTestEntry::DEFAULT
        };
        let p = entry.payload.expect("payload set");
        assert_eq!(p.name, "fio");
        assert!(!p.is_scheduler());
    }

    #[test]
    fn ktstr_test_entry_workloads_slot_accepts_multiple_payloads() {
        use crate::test_support::{OutputFormat, Payload, PayloadKind};
        const FIO: Payload = Payload {
            name: "fio",
            kind: PayloadKind::Binary("fio"),
            output: OutputFormat::Json,
            default_args: &[],
            default_checks: &[],
            metrics: &[],
            include_files: &[],
            uses_parent_pgrp: false,
            known_flags: None,
            metric_bounds: None,
        };
        // stress-ng emits progress / metrics / summaries to stderr; stdout
        // is blank. `OutputFormat::Json` yields zero metrics — stdout has
        // nothing JSON-shaped to parse, and the stderr fallback sees prose
        // rather than JSON so the extraction pipeline returns empty.
        // `OutputFormat::LlmExtract` MAY extract numbers from the stderr
        // fallback, but results depend on the local model's tolerance for
        // stress-ng's prose format — unstable without a stderr→stdout
        // redirect wired into `default_args`. Keep `ExitCode` unless you
        // are prepared for that tradeoff.
        const STRESS_NG: Payload = Payload {
            name: "stress-ng",
            kind: PayloadKind::Binary("stress-ng"),
            output: OutputFormat::ExitCode,
            default_args: &[],
            default_checks: &[],
            metrics: &[],
            include_files: &[],
            uses_parent_pgrp: false,
            known_flags: None,
            metric_bounds: None,
        };
        let entry = KtstrTestEntry {
            name: "multi_workload",
            workloads: &[&FIO, &STRESS_NG],
            ..KtstrTestEntry::DEFAULT
        };
        assert_eq!(entry.workloads.len(), 2);
        assert_eq!(entry.workloads[0].name, "fio");
        assert_eq!(entry.workloads[1].name, "stress-ng");
    }

    /// `validate()` rejects any `workloads[i]` that is a
    /// Scheduler-kind Payload — symmetric with the existing
    /// `payload`-slot rejection. Catches the typo where a test
    /// author writes `workloads = [Payload::KERNEL_DEFAULT]` instead of a
    /// binary payload.
    #[test]
    fn validate_rejects_scheduler_kind_in_workloads() {
        use crate::test_support::{OutputFormat, Payload, PayloadKind};
        const GOOD: Payload = Payload {
            name: "fio",
            kind: PayloadKind::Binary("fio"),
            output: OutputFormat::Json,
            default_args: &[],
            default_checks: &[],
            metrics: &[],
            include_files: &[],
            uses_parent_pgrp: false,
            known_flags: None,
            metric_bounds: None,
        };
        fn good_test_func(_: &Ctx) -> Result<AssertResult> {
            Ok(AssertResult::pass())
        }
        let entry = KtstrTestEntry {
            name: "mixed_kinds",
            func: good_test_func,
            // Second workload is scheduler-kind → must bail.
            workloads: &[&GOOD, &Payload::KERNEL_DEFAULT],
            ..KtstrTestEntry::DEFAULT
        };
        let err = entry.validate().unwrap_err();
        let msg = format!("{err}");
        assert!(
            msg.contains("workloads[1]") && msg.contains("Scheduler-kind"),
            "expected workloads[1] Scheduler-kind bail, got: {msg}"
        );
        assert!(
            msg.contains("kernel_default"),
            "error must name the offending workload entry, got: {msg}"
        );
    }

    /// Binary-only workloads slip past `validate()` cleanly — the
    /// Scheduler-kind check does not over-reject. Pins the happy
    /// path against the rejection path so future edits to the
    /// workloads loop don't flip polarity.
    #[test]
    fn validate_accepts_binary_only_workloads() {
        use crate::test_support::{OutputFormat, Payload, PayloadKind};
        const FIO: Payload = Payload {
            name: "fio",
            kind: PayloadKind::Binary("fio"),
            output: OutputFormat::Json,
            default_args: &[],
            default_checks: &[],
            metrics: &[],
            include_files: &[],
            uses_parent_pgrp: false,
            known_flags: None,
            metric_bounds: None,
        };
        // stress-ng emits progress / metrics / summaries to stderr; stdout
        // is blank. `OutputFormat::Json` yields zero metrics — stdout has
        // nothing JSON-shaped to parse, and the stderr fallback sees prose
        // rather than JSON so the extraction pipeline returns empty.
        // `OutputFormat::LlmExtract` MAY extract numbers from the stderr
        // fallback, but results depend on the local model's tolerance for
        // stress-ng's prose format — unstable without a stderr→stdout
        // redirect wired into `default_args`. Keep `ExitCode` unless you
        // are prepared for that tradeoff.
        const STRESS_NG: Payload = Payload {
            name: "stress-ng",
            kind: PayloadKind::Binary("stress-ng"),
            output: OutputFormat::ExitCode,
            default_args: &[],
            default_checks: &[],
            metrics: &[],
            include_files: &[],
            uses_parent_pgrp: false,
            known_flags: None,
            metric_bounds: None,
        };
        fn good_test_func(_: &Ctx) -> Result<AssertResult> {
            Ok(AssertResult::pass())
        }
        let entry = KtstrTestEntry {
            name: "all_binary",
            func: good_test_func,
            workloads: &[&FIO, &STRESS_NG],
            ..KtstrTestEntry::DEFAULT
        };
        entry.validate().expect("binary-only workloads must pass");
    }

    // -- staged_schedulers validation tests --
    //
    // KtstrTestEntry::validate at L1556 walks staged_schedulers and
    // enforces (a) per-name shape rules via the delegated
    // validate_staged_scheduler_name + (b) within-set uniqueness
    // via the BTreeSet insert. Both are silent-data-loss classes
    // if they regress — invalid names
    // would land at corrupt guest staging paths; duplicate names
    // would silently overwrite at the same guest path. The
    // delegated shape rules are independently tested at
    // src/test_support/staged.rs; the smoke test below proves
    // the delegation is wired (catches a future refactor that
    // inlines dup-check and forgets the shape-check).

    /// Empty `staged_schedulers` slice — the default for every
    /// test that doesn't use scheduler-lifecycle ops — must validate.
    #[test]
    fn validate_accepts_empty_staged_schedulers() {
        let entry = KtstrTestEntry {
            name: "no_staged",
            staged_schedulers: &[],
            ..KtstrTestEntry::DEFAULT
        };
        entry.validate().expect("empty staged_schedulers must pass");
    }

    /// Two distinct well-formed staged schedulers — both pass shape
    /// check, names differ. Pins the happy path for the
    /// mid-experiment swap use case (mitosis args A → mitosis args
    /// B declared as two distinct schedulers).
    #[test]
    fn validate_accepts_well_formed_unique_staged_schedulers() {
        static SCX_MITOSIS_A: Scheduler =
            Scheduler::named("scx_mitosis_a").binary_discover("scx_mitosis");
        static SCX_MITOSIS_B: Scheduler =
            Scheduler::named("scx_mitosis_b").binary_discover("scx_mitosis");
        // Slice itself must be `static` so the &'static [...] field
        // can borrow it; binding the slice literal to a static at the
        // call site moves the const-promotion responsibility off the
        // struct-literal expression where lifetime inference can't
        // see the longer-lived destination.
        static SCHEDS: &[&Scheduler] = &[&SCX_MITOSIS_A, &SCX_MITOSIS_B];
        let entry = KtstrTestEntry {
            name: "staged_two",
            staged_schedulers: SCHEDS,
            ..KtstrTestEntry::DEFAULT
        };
        entry
            .validate()
            .expect("two distinct well-formed staged schedulers must pass");
    }

    /// Two staged schedulers with the SAME name collide at the
    /// guest-side staging path (`/staging/schedulers/<name>/`). The
    /// validate-time dedup catches this before any VM boot; without
    /// the pin, a regression that drops the BTreeSet insert check
    /// would silently produce a guest layout where the second
    /// scheduler overwrites the first.
    #[test]
    fn validate_rejects_duplicate_staged_scheduler_names() {
        static DUPE_A: Scheduler = Scheduler::named("scx_dupe").binary_discover("scx_first");
        static DUPE_B: Scheduler = Scheduler::named("scx_dupe").binary_discover("scx_second");
        static SCHEDS: &[&Scheduler] = &[&DUPE_A, &DUPE_B];
        let entry = KtstrTestEntry {
            name: "staged_dupe",
            staged_schedulers: SCHEDS,
            ..KtstrTestEntry::DEFAULT
        };
        let err = entry
            .validate()
            .expect_err("duplicate staged Scheduler.name must reject");
        let msg = err.to_string();
        assert!(
            msg.contains("duplicate"),
            "error must name the duplicate-name violation, got: {msg}"
        );
        assert!(
            msg.contains("scx_dupe"),
            "error must name the colliding scheduler name, got: {msg}"
        );
        assert!(
            msg.contains("staged_schedulers"),
            "error must name the field, got: {msg}"
        );
    }

    /// A staged scheduler with a shape-violating name (path
    /// separator) must reject — proves the validate gate delegates
    /// to validate_staged_scheduler_name. The exhaustive shape
    /// rejections (empty, NUL, leading dot, reserved names) are
    /// covered at src/test_support/staged.rs:138+; this smoke test
    /// pins the delegation site against a future refactor that
    /// inlines the dup-check and forgets the shape-check.
    #[test]
    fn validate_rejects_shape_violating_staged_scheduler_name() {
        static BAD_SLASH: Scheduler = Scheduler::named("scx/path").binary_discover("scx_x");
        static SCHEDS: &[&Scheduler] = &[&BAD_SLASH];
        let entry = KtstrTestEntry {
            name: "staged_bad_shape",
            staged_schedulers: SCHEDS,
            ..KtstrTestEntry::DEFAULT
        };
        let err = entry
            .validate()
            .expect_err("path-separator Scheduler.name must reject");
        let msg = err.to_string();
        assert!(
            msg.contains("staged_schedulers"),
            "error must name the field — proves the `who` context propagated \
             through the delegate call, got: {msg}"
        );
        // The exact shape-message ("path separators") is the
        // validate_staged_scheduler_name contract pinned at
        // src/test_support/staged.rs:151-156, not duplicated here.
    }

    /// A staged scheduler whose `name` matches the boot
    /// [`KtstrTestEntry::scheduler`]'s name must reject — the boot
    /// slot already provides that scheduler, and adding it to the
    /// staged set produces a guest layout where the boot scheduler
    /// is shadowed by the same binary under
    /// `/staging/schedulers/<name>/scheduler`. Catches the
    /// dev-advocate's "stage all the schedulers I might use"
    /// misuse pattern at validate time.
    #[test]
    fn validate_rejects_staged_scheduler_duplicating_boot_scheduler() {
        static BOOT: Scheduler = Scheduler::named("scx_mitosis").binary_discover("scx_mitosis");
        static STAGED_COPY: Scheduler =
            Scheduler::named("scx_mitosis").binary_discover("scx_mitosis");
        static SCHEDS: &[&Scheduler] = &[&STAGED_COPY];
        let entry = KtstrTestEntry {
            name: "staged_dup_of_boot",
            scheduler: &BOOT,
            staged_schedulers: SCHEDS,
            ..KtstrTestEntry::DEFAULT
        };
        let err = entry
            .validate()
            .expect_err("staged scheduler duplicating boot scheduler must reject");
        let msg = err.to_string();
        assert!(
            msg.contains("boot scheduler"),
            "error must name the boot-scheduler violation, got: {msg}"
        );
        assert!(
            msg.contains("scx_mitosis"),
            "error must name the colliding scheduler, got: {msg}"
        );
        assert!(
            msg.contains("Op::AttachScheduler") || msg.contains("Op::ReplaceScheduler"),
            "error must point to the lifecycle Ops that USE the staged set, got: {msg}"
        );
    }

    /// `validate()` rejects `host_only=true` paired with a
    /// `Some(DiskConfig)`. The combination is unsatisfiable today:
    /// `host_only` skips the VM boot that owns the virtio-blk device
    /// lifecycle, so the disk never attaches. Catching it at validate
    /// time surfaces the misconfiguration during nextest discovery
    /// instead of after a confusing host-only run that silently
    /// ignored the disk request.
    #[test]
    fn validate_rejects_host_only_with_disk() {
        fn good_test_func(_: &Ctx) -> Result<AssertResult> {
            Ok(AssertResult::pass())
        }
        let entry = KtstrTestEntry {
            name: "host_only_with_disk",
            func: good_test_func,
            host_only: true,
            disk: Some(crate::vmm::disk_config::DiskConfig::default()),
            ..KtstrTestEntry::DEFAULT
        };
        let err = entry
            .validate()
            .expect_err("host_only=true + disk=Some must be rejected");
        let msg = format!("{err}");
        assert!(
            msg.contains("host_only=true") && msg.contains("disk"),
            "expected host_only+disk diagnostic, got: {msg}",
        );
        assert!(
            msg.contains("host_only_with_disk"),
            "error must name the offending entry, got: {msg}",
        );
    }

    /// `host_only=true` with `disk=None` is the legitimate host-side
    /// shape: a host-only test running without any VM device. Pins
    /// the happy path against the rejection path so future edits
    /// to the host_only/disk gate don't flip polarity (rejecting a
    /// legitimate combination would silently break every host-only
    /// test author).
    #[test]
    fn validate_accepts_host_only_without_disk() {
        fn good_test_func(_: &Ctx) -> Result<AssertResult> {
            Ok(AssertResult::pass())
        }
        let entry = KtstrTestEntry {
            name: "host_only_no_disk",
            func: good_test_func,
            host_only: true,
            disk: None,
            ..KtstrTestEntry::DEFAULT
        };
        entry
            .validate()
            .expect("host_only=true + disk=None must validate");
    }

    /// `host_only=false` (the default) with `disk=Some(..)` is the
    /// canonical disk-attached VM test. Pins that the gate fires
    /// only on the actual conflict (host_only=true) and not on any
    /// `disk=Some(..)` entry — a future tightening that rejected
    /// every Some(DiskConfig) would break the entire disk
    /// integration test surface.
    #[test]
    fn validate_accepts_vm_with_disk() {
        fn good_test_func(_: &Ctx) -> Result<AssertResult> {
            Ok(AssertResult::pass())
        }
        let entry = KtstrTestEntry {
            name: "vm_with_disk",
            func: good_test_func,
            host_only: false,
            disk: Some(crate::vmm::disk_config::DiskConfig::default()),
            ..KtstrTestEntry::DEFAULT
        };
        entry
            .validate()
            .expect("host_only=false + disk=Some must validate");
    }

    /// `validate()` rejects `num_snapshots` greater than the bridge
    /// cap (`MAX_STORED_SNAPSHOTS == 64`). Without the gate the
    /// periodic loop would publish all `N` boundary captures but the
    /// bridge's FIFO eviction at `store()` would silently drop the
    /// earliest samples — a periodic run with `N == 128` would only
    /// retain `periodic_064..periodic_127`. Refusing the entry is
    /// more honest than half-delivering it.
    #[test]
    fn validate_rejects_num_snapshots_above_max_stored() {
        fn good_test_func(_: &Ctx) -> Result<AssertResult> {
            Ok(AssertResult::pass())
        }
        let cap = crate::scenario::snapshot::MAX_STORED_SNAPSHOTS as u32;
        let entry = KtstrTestEntry {
            name: "too_many_snapshots",
            func: good_test_func,
            num_snapshots: cap + 1,
            ..KtstrTestEntry::DEFAULT
        };
        let err = entry
            .validate()
            .expect_err("num_snapshots > MAX_STORED_SNAPSHOTS must reject");
        let msg = format!("{err}");
        assert!(
            msg.contains("num_snapshots") && msg.contains("MAX_STORED_SNAPSHOTS"),
            "expected num_snapshots/MAX_STORED_SNAPSHOTS diagnostic, got: {msg}",
        );
        assert!(
            msg.contains("too_many_snapshots"),
            "error must name the offending entry, got: {msg}",
        );
    }

    /// `num_snapshots == MAX_STORED_SNAPSHOTS` (the boundary case)
    /// must validate when `duration` is large enough to keep every
    /// inter-boundary interval at or above the 100 ms minimum spacing
    /// (see `validate_rejects_tight_periodic_spacing` below). 64
    /// captures over a long duration is the documented happy path —
    /// pinning it here prevents an off-by-one in the cap gate from
    /// silently rejecting the canonical maximum.
    #[test]
    fn validate_accepts_num_snapshots_at_max_stored() {
        fn good_test_func(_: &Ctx) -> Result<AssertResult> {
            Ok(AssertResult::pass())
        }
        let cap = crate::scenario::snapshot::MAX_STORED_SNAPSHOTS as u32;
        // 100 s · 0.8 / 65 ≈ 1.23 s per inter-boundary interval —
        // comfortably above the 100 ms minimum spacing gate.
        let entry = KtstrTestEntry {
            name: "max_snapshots_ok",
            func: good_test_func,
            num_snapshots: cap,
            duration: Duration::from_secs(100),
            ..KtstrTestEntry::DEFAULT
        };
        entry
            .validate()
            .expect("num_snapshots == MAX_STORED_SNAPSHOTS at long duration must validate");
    }

    /// `validate()` rejects `num_snapshots > 0` paired with
    /// `host_only == true`. Periodic capture freezes guest vCPUs via
    /// the freeze coordinator's rendezvous; a host-only entry never
    /// boots a VM, so there are no vCPUs to freeze and the periodic
    /// boundaries would fire against an empty bridge. Catching the
    /// combination at validate time surfaces the misconfiguration
    /// before dispatch instead of after a confusing host-only run
    /// that silently produced zero captures.
    #[test]
    fn validate_rejects_num_snapshots_with_host_only() {
        fn good_test_func(_: &Ctx) -> Result<AssertResult> {
            Ok(AssertResult::pass())
        }
        let entry = KtstrTestEntry {
            name: "host_only_periodic",
            func: good_test_func,
            host_only: true,
            num_snapshots: 1,
            ..KtstrTestEntry::DEFAULT
        };
        let err = entry
            .validate()
            .expect_err("host_only=true + num_snapshots>0 must be rejected");
        let msg = format!("{err}");
        assert!(
            msg.contains("host_only") && msg.contains("num_snapshots"),
            "expected host_only/num_snapshots diagnostic, got: {msg}",
        );
        assert!(
            msg.contains("host_only_periodic"),
            "error must name the offending entry, got: {msg}",
        );
    }

    /// `host_only == true` with `num_snapshots == 0` (the default)
    /// is the legitimate host-side shape and must continue to
    /// validate. Pins the gate to fire only on the actual conflict
    /// (host_only=true AND num_snapshots>0) and not on every
    /// host-only entry — a future tightening that rejected every
    /// host-only test author would silently break the host-only
    /// surface.
    #[test]
    fn validate_accepts_host_only_with_zero_snapshots() {
        fn good_test_func(_: &Ctx) -> Result<AssertResult> {
            Ok(AssertResult::pass())
        }
        let entry = KtstrTestEntry {
            name: "host_only_no_periodic",
            func: good_test_func,
            host_only: true,
            num_snapshots: 0,
            ..KtstrTestEntry::DEFAULT
        };
        entry
            .validate()
            .expect("host_only=true + num_snapshots=0 must validate");
    }

    /// `validate()` rejects an entry whose periodic boundaries would
    /// land closer than 100 ms apart. The freeze coordinator's
    /// periodic loop divides the 80 % usable span (10 % pre-buffer +
    /// 10 % post-buffer = 20 % buffer; the remainder is the usable
    /// span) into `num_snapshots + 1` equal intervals; if any
    /// interval falls under 100 ms the captures crowd into
    /// rendezvous serialisation slack and one or more boundaries
    /// will defer-and-drop. Refusing the entry is more honest than
    /// emitting a partial timeline.
    ///
    /// Bound math: a 1 s duration with `num_snapshots == 8` yields
    /// `(0.8 · 1e9 ns) / 9 ≈ 88.9 ms` per interval — under the 100 ms
    /// floor, so the gate must reject. Below the cap (8 < 64) so the
    /// rejection MUST come from the spacing check, not the
    /// MAX_STORED_SNAPSHOTS gate.
    #[test]
    fn validate_rejects_tight_periodic_spacing() {
        fn good_test_func(_: &Ctx) -> Result<AssertResult> {
            Ok(AssertResult::pass())
        }
        let entry = KtstrTestEntry {
            name: "tight_periodic_spacing",
            func: good_test_func,
            duration: Duration::from_secs(1),
            num_snapshots: 8,
            ..KtstrTestEntry::DEFAULT
        };
        let err = entry
            .validate()
            .expect_err("tight periodic spacing must be rejected");
        let msg = format!("{err}");
        assert!(
            msg.contains("100"),
            "error must mention the 100 ms floor, got: {msg}",
        );
        assert!(
            msg.contains("num_snapshots") || msg.contains("periodic"),
            "error must mention num_snapshots or periodic, got: {msg}",
        );
        assert!(
            msg.contains("tight_periodic_spacing"),
            "error must name the offending entry, got: {msg}",
        );
    }

    /// Periodic spacing gate must NOT fire when boundaries are at
    /// or above the 100 ms floor. 12 s default duration · 0.8 / 2
    /// = 4.8 s per interval (`N == 1`) is comfortably above the
    /// floor. Pins the polarity of the spacing gate so a future
    /// tightening that rejected every reasonable cadence would
    /// surface here as a regression instead of silently breaking
    /// every periodic test.
    #[test]
    fn validate_accepts_loose_periodic_spacing() {
        fn good_test_func(_: &Ctx) -> Result<AssertResult> {
            Ok(AssertResult::pass())
        }
        let entry = KtstrTestEntry {
            name: "loose_periodic_spacing",
            func: good_test_func,
            // 12 s default duration · 0.8 / (1 + 1) = 4.8 s per
            // interval — well above 100 ms.
            num_snapshots: 1,
            ..KtstrTestEntry::DEFAULT
        };
        entry
            .validate()
            .expect("loose periodic spacing (4.8 s per boundary) must validate");
    }

    /// `validate()` rejects an entry whose scheduler declares
    /// `config_file_def` but provides no `config_content`. The macro
    /// emits a `const _: () = assert!(...)` for attribute-built
    /// entries; this branch covers programmatic construction so
    /// dispatch never runs the scheduler binary without `--config`.
    #[test]
    fn validate_rejects_config_file_def_without_content() {
        static SCHED_WITH_CFG: Scheduler = Scheduler {
            name: "sched_with_cfg",
            binary: SchedulerSpec::Discover("sched_with_cfg_bin"),
            sysctls: &[],
            kargs: &[],
            assert: crate::assert::Assert::NO_OVERRIDES,
            cgroup_parent: None,
            sched_args: &[],
            topology: Topology {
                llcs: 1,
                cores_per_llc: 2,
                threads_per_core: 1,
                numa_nodes: 1,
                nodes: None,
                distances: None,
            },
            constraints: TopologyConstraints::DEFAULT,
            config_file: None,
            config_file_def: Some(("--config {file}", "/include-files/cfg.json")),
            kernels: &[],
        };
        fn good_test_func(_: &Ctx) -> Result<AssertResult> {
            Ok(AssertResult::pass())
        }
        let entry = KtstrTestEntry {
            name: "missing_config",
            func: good_test_func,
            scheduler: &SCHED_WITH_CFG,
            config_content: None,
            ..KtstrTestEntry::DEFAULT
        };
        let err = entry
            .validate()
            .expect_err("config_file_def without config_content must reject");
        let msg = format!("{err}");
        assert!(
            msg.contains("config_file_def") && msg.contains("config_content"),
            "expected config_file_def/config_content bail, got: {msg}"
        );
        assert!(
            msg.contains("missing_config"),
            "error must name the offending entry, got: {msg}"
        );
    }

    /// `validate()` rejects an entry that supplies `config_content`
    /// while pairing with a scheduler that declares no
    /// `config_file_def`. Symmetric with the missing-content gate:
    /// the content would be silently dropped at dispatch.
    #[test]
    fn validate_rejects_content_without_config_file_def() {
        fn good_test_func(_: &Ctx) -> Result<AssertResult> {
            Ok(AssertResult::pass())
        }
        let entry = KtstrTestEntry {
            name: "stray_config",
            func: good_test_func,
            scheduler: &crate::test_support::Scheduler::EEVDF,
            config_content: Some("{}"),
            ..KtstrTestEntry::DEFAULT
        };
        let err = entry
            .validate()
            .expect_err("config_content without config_file_def must reject");
        let msg = format!("{err}");
        assert!(
            msg.contains("config_content") && msg.contains("config_file_def"),
            "expected config_content/config_file_def bail, got: {msg}"
        );
        assert!(
            msg.contains("stray_config"),
            "error must name the offending entry, got: {msg}"
        );
    }

    /// `validate()` accepts both legitimate pairings: a scheduler
    /// with `config_file_def` paired with `config_content`, and a
    /// scheduler without `config_file_def` paired with no
    /// `config_content`. Pins the happy paths so a future tightening
    /// of the pairing gate can't silently break valid entries.
    #[test]
    fn validate_accepts_config_pairing() {
        static SCHED_WITH_CFG: Scheduler = Scheduler {
            name: "sched_paired",
            binary: SchedulerSpec::Discover("sched_paired_bin"),
            sysctls: &[],
            kargs: &[],
            assert: crate::assert::Assert::NO_OVERRIDES,
            cgroup_parent: None,
            sched_args: &[],
            topology: Topology {
                llcs: 1,
                cores_per_llc: 2,
                threads_per_core: 1,
                numa_nodes: 1,
                nodes: None,
                distances: None,
            },
            constraints: TopologyConstraints::DEFAULT,
            config_file: None,
            config_file_def: Some(("f:{file}", "/include-files/p.json")),
            kernels: &[],
        };
        fn good_test_func(_: &Ctx) -> Result<AssertResult> {
            Ok(AssertResult::pass())
        }
        // Scheduler with def + content set: accepted.
        let entry_paired = KtstrTestEntry {
            name: "paired_present",
            func: good_test_func,
            scheduler: &SCHED_WITH_CFG,
            config_content: Some("{\"layers\":[]}"),
            ..KtstrTestEntry::DEFAULT
        };
        entry_paired
            .validate()
            .expect("scheduler with config_file_def + content must validate");
        // Scheduler without def + content unset: also accepted.
        let entry_none = KtstrTestEntry {
            name: "neither_present",
            func: good_test_func,
            scheduler: &crate::test_support::Scheduler::EEVDF,
            config_content: None,
            ..KtstrTestEntry::DEFAULT
        };
        entry_none
            .validate()
            .expect("no config_file_def + no content must validate");
    }

    /// `validate()` rejects an entry that sets
    /// `expect_scx_bpf_error_contains` without also setting
    /// `expect_err = true`. The matcher narrows which failure counts
    /// as the expected bug, so it only makes sense on expected-error
    /// tests. Without the gate a matcher on a pass-expected entry
    /// would be silently inert.
    #[test]
    fn validate_rejects_expect_scx_bpf_error_contains_without_expect_err() {
        fn good_test_func(_: &Ctx) -> Result<AssertResult> {
            Ok(AssertResult::pass())
        }
        let entry = KtstrTestEntry {
            name: "bad_contains",
            func: good_test_func,
            assert: crate::assert::Assert::NO_OVERRIDES
                .expect_scx_bpf_error_contains("apply_cell_config"),
            expect_err: false,
            ..KtstrTestEntry::DEFAULT
        };
        let err = entry
            .validate()
            .expect_err("matcher without expect_err must be rejected");
        let msg = format!("{err}");
        assert!(
            msg.contains("expect_scx_bpf_error_contains") && msg.contains("expect_err"),
            "diagnostic must name BOTH the matcher field AND expect_err: {msg}",
        );
        assert!(
            msg.contains("bad_contains"),
            "error must name the offending entry: {msg}",
        );
    }

    /// Symmetric with the `_contains` rejection: setting
    /// `expect_scx_bpf_error_matches` without `expect_err = true`
    /// must also reject. Pins that the gate triggers on either
    /// matcher field — a future addition that only checked one
    /// would leave the other silently inert.
    #[test]
    fn validate_rejects_expect_scx_bpf_error_matches_without_expect_err() {
        fn good_test_func(_: &Ctx) -> Result<AssertResult> {
            Ok(AssertResult::pass())
        }
        let entry = KtstrTestEntry {
            name: "bad_matches",
            func: good_test_func,
            assert: crate::assert::Assert::NO_OVERRIDES
                .expect_scx_bpf_error_matches("apply_cell_config:[0-9]+"),
            expect_err: false,
            ..KtstrTestEntry::DEFAULT
        };
        let err = entry
            .validate()
            .expect_err("matcher without expect_err must be rejected");
        let msg = format!("{err}");
        assert!(
            msg.contains("expect_scx_bpf_error_matches") && msg.contains("expect_err"),
            "diagnostic must name BOTH the matcher field AND expect_err: {msg}",
        );
        assert!(
            msg.contains("bad_matches"),
            "error must name the offending entry: {msg}",
        );
    }

    /// Happy path: scx_bpf_error matchers paired with
    /// `expect_err = true` must validate cleanly. Pins the polarity
    /// of the gate so a future tightening that rejected every entry
    /// with a matcher would surface here instead of silently
    /// breaking every reproducer test.
    #[test]
    fn validate_accepts_expect_scx_bpf_error_matchers_with_expect_err() {
        fn good_test_func(_: &Ctx) -> Result<AssertResult> {
            Ok(AssertResult::pass())
        }
        let entry_contains = KtstrTestEntry {
            name: "good_contains",
            func: good_test_func,
            assert: crate::assert::Assert::NO_OVERRIDES
                .expect_scx_bpf_error_contains("apply_cell_config"),
            expect_err: true,
            ..KtstrTestEntry::DEFAULT
        };
        entry_contains
            .validate()
            .expect("contains matcher + expect_err=true must validate");
        let entry_matches = KtstrTestEntry {
            name: "good_matches",
            func: good_test_func,
            assert: crate::assert::Assert::NO_OVERRIDES
                .expect_scx_bpf_error_matches("apply_cell_config:[0-9]+"),
            expect_err: true,
            ..KtstrTestEntry::DEFAULT
        };
        entry_matches
            .validate()
            .expect("matches matcher + expect_err=true must validate");
        let entry_both = KtstrTestEntry {
            name: "good_both",
            func: good_test_func,
            assert: crate::assert::Assert::NO_OVERRIDES
                .expect_scx_bpf_error_contains("apply_cell_config")
                .expect_scx_bpf_error_matches("apply_cell_config:[0-9]+"),
            expect_err: true,
            ..KtstrTestEntry::DEFAULT
        };
        entry_both
            .validate()
            .expect("both matchers + expect_err=true must validate");
    }

    #[test]
    fn ktstr_test_entry_default_rejected_by_empty_name() {
        // DEFAULT has name = "" which validate() rejects — so the
        // stub entry cannot accidentally dispatch. This pins that
        // invariant.
        let err = KtstrTestEntry::DEFAULT.validate().unwrap_err();
        let msg = format!("{err}");
        assert!(
            msg.contains("name") && msg.contains("non-empty"),
            "expected name-non-empty bail, got: {msg}"
        );
    }

    #[test]
    fn default_test_func_returns_err() {
        // The stub bails with Err — NOT a panic. Callers that
        // accidentally leave `func: default_test_func` in their
        // entry must see a clean Err to surface the mistake.
        let (cgroups, topo) = dummy_ctx();
        let ctx = Ctx::builder(&cgroups, &topo)
            .duration(Duration::from_millis(1))
            .assert(crate::assert::Assert::NO_OVERRIDES)
            .build();
        let result = default_test_func(&ctx);
        let err = result.expect_err("default_test_func must return Err, not Ok");
        let msg = format!("{err}");
        assert!(
            msg.contains("KtstrTestEntry::DEFAULT func called"),
            "expected DEFAULT-called bail message, got: {msg}"
        );
        assert!(
            msg.contains("override func before use"),
            "expected actionable hint, got: {msg}"
        );
    }

    // -- Scheduler method tests --

    use super::super::test_helpers::validate_entry;

    #[test]
    fn scheduler_eevdf_defaults() {
        let s = &Scheduler::EEVDF;
        assert_eq!(s.name, "eevdf");
        assert!(s.sysctls.is_empty());
        assert!(s.kargs.is_empty());
        assert!(s.assert.not_starved.is_none());
        assert!(s.assert.max_imbalance_ratio.is_none());
    }

    #[test]
    fn scheduler_named_builder() {
        static TEST_SYSCTLS: &[Sysctl] =
            &[Sysctl::new("kernel.sched_cfs_bandwidth_slice_us", "1000")];
        let s = Scheduler::named("test_sched")
            .binary(SchedulerSpec::Discover("test_bin"))
            .sysctls(TEST_SYSCTLS)
            .kargs(&["nosmt"]);
        assert_eq!(s.name, "test_sched");
        assert_eq!(s.sysctls.len(), 1);
        assert_eq!(s.kargs.len(), 1);
    }

    #[test]
    fn scheduler_with_check() {
        let v = crate::assert::Assert::NO_OVERRIDES
            .check_not_starved()
            .max_imbalance_ratio(3.0);
        let s = Scheduler::named("sched").assert(v);
        assert_eq!(s.assert.not_starved, Some(true));
        assert_eq!(s.assert.max_imbalance_ratio, Some(3.0));
    }

    #[test]
    fn scheduler_named_default_topology_matches_macro_hardcode() {
        // The `declare_scheduler!` macro at
        // `ktstr-macros/src/lib.rs` hardcodes
        // `(numa=1, llcs=1, cores_per_llc=2, threads_per_core=1)`
        // as the fallback when the user omits `topology = (...)`.
        // That hardcode mirrors `Scheduler::named`'s default Topology.
        // The two sites are mechanically coupled by convention but
        // not by code: a change to `Scheduler::named` here would
        // silently let the macro check stale values, producing
        // misleading "effective topology llcs (N)" errors and/or
        // false-positives. Pin the defaults here so a drift fails
        // this test loudly and points at both sites.
        let s = Scheduler::named("__macro_default_topology_pin__");
        assert_eq!(s.topology.numa_nodes, 1);
        assert_eq!(s.topology.llcs, 1);
        assert_eq!(s.topology.cores_per_llc, 2);
        assert_eq!(s.topology.threads_per_core, 1);
    }

    // -- KtstrTestEntry::validate coverage --

    #[test]
    fn ktstr_test_entry_validate_accepts_defaults() {
        let e = validate_entry("ok", 512, Duration::from_secs(2));
        e.validate().unwrap();
    }

    #[test]
    fn ktstr_test_entry_validate_rejects_empty_name() {
        let e = validate_entry("", 512, Duration::from_secs(2));
        let err = e.validate().unwrap_err();
        let msg = format!("{err}");
        assert!(
            msg.contains("name") && msg.contains("non-empty"),
            "got: {msg}"
        );
    }

    #[test]
    fn ktstr_test_entry_validate_rejects_zero_memory() {
        let e = validate_entry("t", 0, Duration::from_secs(2));
        let err = e.validate().unwrap_err();
        let msg = format!("{err}");
        assert!(
            msg.contains("memory_mib") && msg.contains("> 0") && msg.contains("'t'"),
            "got: {msg}"
        );
    }

    #[test]
    fn ktstr_test_entry_validate_rejects_zero_duration() {
        let e = validate_entry("t", 512, Duration::ZERO);
        let err = e.validate().unwrap_err();
        let msg = format!("{err}");
        assert!(
            msg.contains("duration") && msg.contains("> 0"),
            "got: {msg}"
        );
    }

    // -- TopologyConstraints tests --

    #[test]
    fn topology_constraints_default_has_max_values() {
        let c = TopologyConstraints::DEFAULT;
        assert_eq!(c.max_llcs, Some(12));
        assert_eq!(c.max_numa_nodes, Some(1));
        assert_eq!(c.max_cpus, Some(192));
    }

    #[test]
    fn topology_constraints_max_fields_set() {
        let c = TopologyConstraints {
            max_llcs: Some(16),
            max_numa_nodes: Some(4),
            max_cpus: Some(128),
            ..TopologyConstraints::DEFAULT
        };
        assert_eq!(c.max_llcs, Some(16));
        assert_eq!(c.max_numa_nodes, Some(4));
        assert_eq!(c.max_cpus, Some(128));
        assert_eq!(c.min_numa_nodes, 1);
        assert_eq!(c.min_llcs, 1);
        assert_eq!(c.min_cpus, 1);
    }

    #[test]
    fn topology_constraints_with_chain_overrides_target_fields_only() {
        let c = TopologyConstraints::DEFAULT
            .with_min_numa_nodes(2)
            .with_max_numa_nodes(8)
            .with_min_llcs(3)
            .with_max_llcs(16)
            .with_requires_smt(true)
            .with_min_cpus(4)
            .with_max_cpus(64);
        assert_eq!(c.min_numa_nodes, 2);
        assert_eq!(c.max_numa_nodes, Some(8));
        assert_eq!(c.min_llcs, 3);
        assert_eq!(c.max_llcs, Some(16));
        assert!(c.requires_smt);
        assert_eq!(c.min_cpus, 4);
        assert_eq!(c.max_cpus, Some(64));
    }

    #[test]
    fn topology_constraints_without_chain_clears_option_fields() {
        let c = TopologyConstraints::DEFAULT
            .without_max_numa_nodes()
            .without_max_llcs()
            .without_max_cpus();
        assert!(c.max_numa_nodes.is_none());
        assert!(c.max_llcs.is_none());
        assert!(c.max_cpus.is_none());
        // min fields untouched.
        assert_eq!(c.min_numa_nodes, 1);
        assert_eq!(c.min_llcs, 1);
        assert_eq!(c.min_cpus, 1);
    }

    #[test]
    fn topology_constraints_with_chain_const_evaluable() {
        const C: TopologyConstraints = TopologyConstraints::DEFAULT
            .with_min_llcs(2)
            .with_max_llcs(4);
        assert_eq!(C.min_llcs, 2);
        assert_eq!(C.max_llcs, Some(4));
    }

    #[test]
    fn topology_constraints_equality() {
        let a = TopologyConstraints::DEFAULT;
        let b = TopologyConstraints::DEFAULT;
        assert_eq!(a, b);

        let c = TopologyConstraints {
            max_llcs: Some(8),
            ..TopologyConstraints::DEFAULT
        };
        assert_ne!(a, c);
    }

    #[test]
    fn accepts_default_allows_within_limits() {
        let c = TopologyConstraints::DEFAULT;
        // 1 NUMA, 8 LLCs, 4 cores, 2 threads = 64 CPUs
        let t = Topology::new(1, 8, 4, 2);
        assert!(c.accepts(&t, 128, 16, 32));
    }

    #[test]
    fn accepts_default_rejects_multi_numa() {
        let c = TopologyConstraints::DEFAULT;
        // 2 NUMA, 8 LLCs, 4 cores, 2 threads = 64 CPUs
        let t = Topology::new(2, 8, 4, 2);
        assert!(!c.accepts(&t, 128, 16, 32));
    }

    #[test]
    fn accepts_default_rejects_too_many_llcs() {
        let c = TopologyConstraints::DEFAULT;
        // 16 LLCs exceeds max_llcs=12
        let t = Topology::new(1, 16, 2, 1);
        assert!(!c.accepts(&t, 128, 32, 32));
    }

    #[test]
    fn accepts_none_means_no_limit() {
        let c = TopologyConstraints {
            max_llcs: None,
            max_numa_nodes: None,
            max_cpus: None,
            ..TopologyConstraints::DEFAULT
        };
        // 4 NUMA, 16 LLCs, 8 cores, 2 threads = 256 CPUs
        let t = Topology::new(4, 16, 8, 2);
        assert!(c.accepts(&t, 512, 32, 32));
    }

    #[test]
    fn accepts_rejects_too_many_llcs() {
        let c = TopologyConstraints {
            max_llcs: Some(4),
            ..TopologyConstraints::DEFAULT
        };
        let t = Topology::new(1, 8, 2, 1);
        assert!(!c.accepts(&t, 128, 16, 32));
    }

    #[test]
    fn accepts_allows_llcs_at_max() {
        let c = TopologyConstraints {
            max_llcs: Some(4),
            ..TopologyConstraints::DEFAULT
        };
        let t = Topology::new(1, 4, 2, 1);
        assert!(c.accepts(&t, 128, 16, 32));
    }

    #[test]
    fn accepts_rejects_too_many_numa_nodes() {
        let c = TopologyConstraints {
            max_numa_nodes: Some(2),
            ..TopologyConstraints::DEFAULT
        };
        let t = Topology::new(4, 4, 2, 1);
        assert!(!c.accepts(&t, 128, 16, 32));
    }

    #[test]
    fn accepts_allows_numa_at_max() {
        let c = TopologyConstraints {
            max_numa_nodes: Some(2),
            ..TopologyConstraints::DEFAULT
        };
        let t = Topology::new(2, 4, 2, 1);
        assert!(c.accepts(&t, 128, 16, 32));
    }

    #[test]
    fn accepts_rejects_too_many_cpus() {
        let c = TopologyConstraints {
            max_cpus: Some(16),
            ..TopologyConstraints::DEFAULT
        };
        // 4 LLCs * 4 cores * 2 threads = 32 CPUs
        let t = Topology::new(1, 4, 4, 2);
        assert!(!c.accepts(&t, 128, 16, 32));
    }

    #[test]
    fn accepts_allows_cpus_at_max() {
        let c = TopologyConstraints {
            max_cpus: Some(16),
            ..TopologyConstraints::DEFAULT
        };
        // 2 LLCs * 4 cores * 2 threads = 16 CPUs
        let t = Topology::new(1, 2, 4, 2);
        assert!(c.accepts(&t, 128, 16, 32));
    }

    #[test]
    fn accepts_rejects_too_few_llcs() {
        let c = TopologyConstraints {
            min_llcs: 4,
            ..TopologyConstraints::DEFAULT
        };
        let t = Topology::new(1, 2, 4, 1);
        assert!(!c.accepts(&t, 128, 16, 32));
    }

    #[test]
    fn accepts_rejects_exceeding_host_cpus() {
        let c = TopologyConstraints::DEFAULT;
        let t = Topology::new(1, 4, 4, 2); // 32 CPUs
        assert!(!c.accepts(&t, 16, 16, 32)); // host has only 16
    }

    #[test]
    fn accepts_rejects_exceeding_host_llcs() {
        let c = TopologyConstraints::DEFAULT;
        let t = Topology::new(1, 8, 2, 1);
        assert!(!c.accepts(&t, 128, 4, 32)); // host has only 4 LLCs
    }

    #[test]
    fn accepts_combined_min_and_max() {
        let c = TopologyConstraints {
            min_llcs: 2,
            max_llcs: Some(8),
            min_cpus: 4,
            max_cpus: Some(32),
            ..TopologyConstraints::DEFAULT
        };
        // 1 LLC, 4 CPUs -- rejected (min_llcs=2)
        assert!(!c.accepts(&Topology::new(1, 1, 4, 1), 128, 16, 32));
        // 2 LLCs, 4 CPUs -- accepted
        assert!(c.accepts(&Topology::new(1, 2, 2, 1), 128, 16, 32));
        // 16 LLCs, 32 CPUs -- rejected (max_llcs=8)
        assert!(!c.accepts(&Topology::new(1, 16, 2, 1), 128, 16, 32));
        // 8 LLCs, 16 CPUs -- accepted
        assert!(c.accepts(&Topology::new(1, 8, 2, 1), 128, 16, 32));
    }

    #[test]
    fn accepts_requires_smt() {
        let c = TopologyConstraints {
            requires_smt: true,
            ..TopologyConstraints::DEFAULT
        };
        let no_smt = Topology::new(1, 2, 4, 1);
        let with_smt = Topology::new(1, 2, 4, 2);
        assert!(!c.accepts(&no_smt, 128, 16, 32));
        assert!(c.accepts(&with_smt, 128, 16, 32));
    }

    #[test]
    fn accepts_rejects_too_few_numa_nodes() {
        let c = TopologyConstraints {
            min_numa_nodes: 2,
            max_numa_nodes: None,
            ..TopologyConstraints::DEFAULT
        };
        let t = Topology::new(1, 4, 4, 1);
        assert!(!c.accepts(&t, 128, 16, 32));
    }

    #[test]
    fn accepts_rejects_too_few_cpus() {
        let c = TopologyConstraints {
            min_cpus: 32,
            ..TopologyConstraints::DEFAULT
        };
        // 2 LLCs * 4 cores * 2 threads = 16 CPUs
        let t = Topology::new(1, 2, 4, 2);
        assert!(!c.accepts(&t, 128, 16, 32));
    }

    #[test]
    fn accepts_rejects_exceeding_host_cpus_per_llc() {
        let c = TopologyConstraints::DEFAULT;
        // cores_per_llc=8, threads_per_core=2 → 16 CPUs/LLC
        let t = Topology::new(1, 2, 8, 2);
        assert!(!c.accepts(&t, 128, 16, 8));
    }

    // -- TopologyConstraints::validate --

    #[test]
    fn validate_accepts_default_constraints() {
        // DEFAULT has min_numa_nodes=1, max_numa_nodes=Some(1);
        // min_llcs=1, max_llcs=Some(12); min_cpus=1, max_cpus=Some(192).
        // All min<=max — must pass.
        TopologyConstraints::DEFAULT
            .validate()
            .expect("DEFAULT constraints must be self-consistent");
    }

    #[test]
    fn validate_rejects_inverted_numa_nodes() {
        let c = TopologyConstraints {
            min_numa_nodes: 5,
            max_numa_nodes: Some(2),
            ..TopologyConstraints::DEFAULT
        };
        let err = c
            .validate()
            .expect_err("inverted min/max_numa_nodes must be rejected");
        let msg = err.to_string();
        assert!(
            msg.contains("max_numa_nodes=2") && msg.contains("min_numa_nodes=5"),
            "diagnostic must name both bounds: got {msg}",
        );
    }

    #[test]
    fn validate_rejects_inverted_llcs() {
        let c = TopologyConstraints {
            min_llcs: 8,
            max_llcs: Some(2),
            ..TopologyConstraints::DEFAULT
        };
        let err = c
            .validate()
            .expect_err("inverted min/max_llcs must be rejected");
        let msg = err.to_string();
        assert!(
            msg.contains("max_llcs=2") && msg.contains("min_llcs=8"),
            "diagnostic must name both bounds: got {msg}",
        );
    }

    #[test]
    fn validate_rejects_inverted_cpus() {
        let c = TopologyConstraints {
            min_cpus: 64,
            max_cpus: Some(16),
            ..TopologyConstraints::DEFAULT
        };
        let err = c
            .validate()
            .expect_err("inverted min/max_cpus must be rejected");
        let msg = err.to_string();
        assert!(
            msg.contains("max_cpus=16") && msg.contains("min_cpus=64"),
            "diagnostic must name both bounds: got {msg}",
        );
    }

    #[test]
    fn validate_accepts_max_equal_min() {
        // max == min is satisfiable (any topology with that exact
        // value matches), so the validator must allow it.
        let c = TopologyConstraints {
            min_numa_nodes: 3,
            max_numa_nodes: Some(3),
            min_llcs: 4,
            max_llcs: Some(4),
            min_cpus: 16,
            max_cpus: Some(16),
            ..TopologyConstraints::DEFAULT
        };
        c.validate()
            .expect("max==min is satisfiable and must be accepted");
    }

    #[test]
    fn validate_accepts_open_upper_bound() {
        // None for max_* means "no upper bound" — never inverted.
        let c = TopologyConstraints {
            min_numa_nodes: 16,
            max_numa_nodes: None,
            min_llcs: 32,
            max_llcs: None,
            min_cpus: 1024,
            max_cpus: None,
            ..TopologyConstraints::DEFAULT
        };
        c.validate()
            .expect("None upper bounds must always validate");
    }

    // -- KtstrTestEntry::validate wire-in for TopologyConstraints --

    /// Pin that an inverted per-entry `constraints` field surfaces
    /// through `KtstrTestEntry::validate` with the documented
    /// `KtstrTestEntry '<name>'.constraints:` prefix wrap. Catches a
    /// regression that drops the per-entry `.validate()?` call at the
    /// wire-in site in `KtstrTestEntry::validate`.
    #[test]
    fn entry_validate_propagates_per_entry_constraints_error() {
        let entry = KtstrTestEntry {
            name: "test_inverted_entry",
            constraints: TopologyConstraints {
                min_numa_nodes: 5,
                max_numa_nodes: Some(2),
                ..TopologyConstraints::DEFAULT
            },
            ..KtstrTestEntry::DEFAULT
        };
        let err = entry
            .validate()
            .expect_err("inverted per-entry constraints must surface");
        let msg = err.to_string();
        assert!(
            msg.contains("KtstrTestEntry 'test_inverted_entry'.constraints:"),
            "wrap prefix must name the entry + constraints field: got {msg}",
        );
        assert!(
            msg.contains("max_numa_nodes=2") && msg.contains("min_numa_nodes=5"),
            "underlying validator diagnostic must be preserved through map_err: got {msg}",
        );
    }

    /// Pin that an inverted scheduler-level `constraints` field
    /// surfaces through `KtstrTestEntry::validate` with the
    /// `KtstrTestEntry '<name>'.scheduler '<sched>'.constraints:`
    /// prefix wrap. Catches a regression that drops the
    /// `.scheduler.constraints.validate()?` call at the wire-in site.
    #[test]
    fn entry_validate_propagates_scheduler_constraints_error() {
        static BAD_SCHED: Scheduler = Scheduler {
            name: "bad_sched",
            binary: SchedulerSpec::Eevdf,
            sysctls: &[],
            kargs: &[],
            assert: crate::assert::Assert::NO_OVERRIDES,
            cgroup_parent: None,
            sched_args: &[],
            topology: Topology {
                llcs: 1,
                cores_per_llc: 1,
                threads_per_core: 1,
                numa_nodes: 1,
                nodes: None,
                distances: None,
            },
            constraints: TopologyConstraints {
                min_llcs: 8,
                max_llcs: Some(2),
                ..TopologyConstraints::DEFAULT
            },
            config_file: None,
            config_file_def: None,
            kernels: &[],
        };
        let entry = KtstrTestEntry {
            name: "test_inverted_scheduler",
            scheduler: &BAD_SCHED,
            ..KtstrTestEntry::DEFAULT
        };
        let err = entry
            .validate()
            .expect_err("inverted scheduler-level constraints must surface");
        let msg = err.to_string();
        assert!(
            msg.contains(
                "KtstrTestEntry 'test_inverted_scheduler'.scheduler 'bad_sched'.constraints:"
            ),
            "wrap prefix must name the entry + scheduler + constraints: got {msg}",
        );
        assert!(
            msg.contains("max_llcs=2") && msg.contains("min_llcs=8"),
            "underlying validator diagnostic must be preserved: got {msg}",
        );
    }

    // -- SchedulerSpec::display_name --

    #[test]
    fn display_name_eevdf() {
        assert_eq!(SchedulerSpec::Eevdf.display_name(), "eevdf");
    }

    #[test]
    fn display_name_discover_returns_binary_name() {
        assert_eq!(
            SchedulerSpec::Discover("scx_mitosis").display_name(),
            "scx_mitosis"
        );
    }

    #[test]
    fn display_name_path_returns_path_string() {
        assert_eq!(
            SchedulerSpec::Path("/usr/bin/scx_my_sched").display_name(),
            "/usr/bin/scx_my_sched"
        );
    }

    #[test]
    fn display_name_kernel_builtin_returns_kernel() {
        assert_eq!(
            SchedulerSpec::KernelBuiltin {
                enable: &[],
                disable: &[],
            }
            .display_name(),
            "kernel"
        );
    }

    // -- SchedulerSpec::scheduler_commit --
    //
    // Conservative by design: EVERY variant currently returns
    // None, including `Discover(_)`. `resolve_scheduler`'s 5-path
    // cascade can pick up a binary whose commit is unknown to this
    // process in four of the five paths, so `Discover` returns
    // None to avoid lying. The sidecar's nullable
    // semantics distinguish "unset" from a sentinel so consumers
    // can tell "no userspace binary" (Eevdf, KernelBuiltin) from
    // "external binary, commit unknown" (Path) and "discovered
    // binary, provenance unverified" (Discover). A future
    // introspection path (`binary --version`, ELF note, BuildId
    // lookup) can flip a variant to Some(..) when an authoritative
    // commit is available; until then, None keeps consumers from
    // attributing regressions to the wrong commit. See the method
    // doc for the per-variant rationale.

    #[test]
    fn scheduler_commit_eevdf_returns_none() {
        assert!(
            SchedulerSpec::Eevdf.scheduler_commit().is_none(),
            "Eevdf has no userspace binary — scheduler_commit must \
             be None so the sidecar field distinguishes this case \
             from `Path(_)` (external, unknown commit). Got: {:?}",
            SchedulerSpec::Eevdf.scheduler_commit(),
        );
    }

    #[test]
    fn scheduler_commit_discover_returns_none() {
        // `Discover` is resolved by `resolve_scheduler`'s 5-path
        // cascade. Only the rebuild fallback guarantees the binary
        // matches the current tree; the four pre-built discovery
        // paths (KTSTR_SCHEDULER env, ktstr-binary sibling dir,
        // target/debug/, target/release/) can pick up a binary
        // whose commit is unknown to this process. Synthesizing a
        // commit would be a lie in 4 of 5 cases — so the honest
        // answer today is `None`. A future enhancement that probes
        // the binary (e.g. `--version`, ELF note) can flip this to
        // `Some(..)` when an authoritative commit is available;
        // until then, `None` keeps consumers from attributing
        // regressions to the wrong commit.
        assert!(
            SchedulerSpec::Discover("scx_mitosis")
                .scheduler_commit()
                .is_none(),
            "Discover(_) must return None — resolve_scheduler's \
             cascade can pick up a binary whose commit doesn't \
             match the workspace. Got: {:?}",
            SchedulerSpec::Discover("scx_mitosis").scheduler_commit(),
        );
    }

    #[test]
    fn scheduler_commit_path_returns_none() {
        // External binaries have no reliable introspection path;
        // synthesizing a commit here would be a lie when the
        // binary was built from a different tree.
        assert!(
            SchedulerSpec::Path("/usr/bin/scx_external")
                .scheduler_commit()
                .is_none(),
            "Path(_) points at an externally-built binary — \
             scheduler_commit must be None so consumers don't treat \
             a fabricated commit as authoritative. Got: {:?}",
            SchedulerSpec::Path("/usr/bin/scx_external").scheduler_commit(),
        );
    }

    #[test]
    fn scheduler_commit_kernel_builtin_returns_none() {
        // In-kernel schedulers have no userspace binary. The
        // running kernel's identity belongs in
        // `host.kernel_release`, not here.
        let spec = SchedulerSpec::KernelBuiltin {
            enable: &[],
            disable: &[],
        };
        assert!(
            spec.scheduler_commit().is_none(),
            "KernelBuiltin has no userspace binary — \
             scheduler_commit must be None. Got: {:?}",
            spec.scheduler_commit(),
        );
    }

    // -- all_include_files aggregation tests --
    //
    // Pins the scheduler → payload → workloads → extras order. The
    // dedupe + archive-collision policy lives downstream in
    // `eval::dedupe_include_files`; this aggregator just gathers
    // raw spec strings.

    /// No Payload and no extras declare include_files → empty result.
    /// Regression guard for `all_include_files` returning an implicit
    /// non-empty list (e.g. leaking a default).
    #[test]
    fn all_include_files_empty_when_nothing_declared() {
        let entry = KtstrTestEntry {
            name: "t",
            ..KtstrTestEntry::DEFAULT
        };
        assert!(entry.all_include_files().is_empty());
    }

    /// Payload + workloads + extras merge in declaration order. Pins:
    /// - order is payload → workloads (preserving the slice order) →
    ///   extra_include_files
    /// - duplicates are NOT deduped at this layer (that's eval's job)
    /// - the scheduler field is `&Scheduler` (no `include_files`
    ///   surface), so the scheduler tier contributes nothing here
    #[test]
    fn all_include_files_merges_sources_in_order() {
        static PRIMARY: crate::test_support::Payload = crate::test_support::Payload {
            name: "primary",
            kind: crate::test_support::PayloadKind::Binary("fio"),
            output: crate::test_support::OutputFormat::ExitCode,
            default_args: &[],
            default_checks: &[],
            metrics: &[],
            include_files: &["fio"],
            uses_parent_pgrp: false,
            known_flags: None,
            metric_bounds: None,
        };
        static WL_A: crate::test_support::Payload = crate::test_support::Payload {
            name: "wl_a",
            kind: crate::test_support::PayloadKind::Binary("stress-ng"),
            output: crate::test_support::OutputFormat::ExitCode,
            default_args: &[],
            default_checks: &[],
            metrics: &[],
            include_files: &["stress-ng"],
            uses_parent_pgrp: false,
            known_flags: None,
            metric_bounds: None,
        };
        static WL_B: crate::test_support::Payload = crate::test_support::Payload {
            name: "wl_b",
            kind: crate::test_support::PayloadKind::Binary("schbench"),
            output: crate::test_support::OutputFormat::ExitCode,
            default_args: &[],
            default_checks: &[],
            metrics: &[],
            include_files: &["schbench"],
            uses_parent_pgrp: false,
            known_flags: None,
            metric_bounds: None,
        };
        static WORKLOADS: &[&crate::test_support::Payload] = &[&WL_A, &WL_B];
        let entry = KtstrTestEntry {
            name: "t",
            payload: Some(&PRIMARY),
            workloads: WORKLOADS,
            extra_include_files: &["test-fixture.json"],
            ..KtstrTestEntry::DEFAULT
        };
        let got = entry.all_include_files();
        assert_eq!(
            got,
            vec!["fio", "stress-ng", "schbench", "test-fixture.json"],
            "aggregation order must be payload → workloads → extras",
        );
    }

    /// Absent optional `payload` slot contributes nothing — the
    /// aggregator skips it without the `None → empty-push` misbehavior
    /// a future refactor might introduce. Scheduler tier no longer
    /// contributes after the `&Payload`→`&Scheduler` field-type change.
    #[test]
    fn all_include_files_skips_absent_payload() {
        let entry = KtstrTestEntry {
            name: "t",
            payload: None,
            workloads: &[],
            extra_include_files: &[],
            ..KtstrTestEntry::DEFAULT
        };
        assert!(entry.all_include_files().is_empty());
    }

    /// Dedup-detection: two distinct `&'static Scheduler` consts with
    /// the same `name` field must panic when `find_scheduler` builds
    /// its name → scheduler map. Exercises the
    /// `build_scheduler_index_or_panic` branch against a mock slice
    /// since the real `KTSTR_SCHEDULERS` is the union of every
    /// `declare_scheduler!` registration in the linked test binary
    /// and so cannot host an intentional duplicate without
    /// poisoning every other test that calls `find_scheduler`.
    #[test]
    #[should_panic(expected = "duplicate scheduler name `dup_name_test`")]
    fn build_scheduler_index_or_panic_rejects_duplicate_names() {
        // Two consts with the same name. Address-distinct so
        // `std::ptr::eq` returns false and the dedup branch fires.
        static A: Scheduler = Scheduler::named("dup_name_test");
        static B: Scheduler = Scheduler::named("dup_name_test");
        let _ = build_scheduler_index_or_panic([&A, &B]);
    }

    /// Distinct names build a populated index without panicking, and
    /// the returned map points back at the same `&'static Scheduler`
    /// references for lookup parity with `find_scheduler`.
    #[test]
    fn build_scheduler_index_or_panic_accepts_distinct_names() {
        static A: Scheduler = Scheduler::named("dup_name_a");
        static B: Scheduler = Scheduler::named("dup_name_b");
        let map = build_scheduler_index_or_panic([&A, &B]);
        assert!(std::ptr::eq(*map.get("dup_name_a").unwrap(), &A));
        assert!(std::ptr::eq(*map.get("dup_name_b").unwrap(), &B));
    }

    /// Same `&'static Scheduler` passed twice (a benign re-export of
    /// the same const, not a true duplicate-name collision across
    /// distinct consts) does not panic — `std::ptr::eq` short-circuits
    /// the dedup branch. Linkme's distributed slice is allowed to
    /// emit the same registration through multiple paths; only a
    /// pointer-distinct duplicate is a misconfiguration.
    #[test]
    fn build_scheduler_index_or_panic_tolerates_pointer_identity_aliases() {
        static A: Scheduler = Scheduler::named("alias_test");
        let map = build_scheduler_index_or_panic([&A, &A]);
        assert!(std::ptr::eq(*map.get("alias_test").unwrap(), &A));
    }

    #[test]
    fn topology_json_try_into_topology_accepts_single_cpu() {
        let topo: Topology = TopologyJson::SINGLE_CPU
            .try_into()
            .expect("SINGLE_CPU valid");
        assert_eq!(topo.numa_nodes, 1);
        assert_eq!(topo.llcs, 1);
        assert_eq!(topo.cores_per_llc, 1);
        assert_eq!(topo.threads_per_core, 1);
        assert!(topo.nodes.is_none());
        assert!(topo.distances.is_none());
    }

    #[test]
    fn topology_json_try_into_topology_accepts_multi_cpu() {
        let json = TopologyJson {
            num_numa_nodes: 2,
            num_llcs: 4,
            cores_per_llc: 8,
            threads_per_core: 2,
        };
        let topo: Topology = json.try_into().expect("2x4x8x2 valid");
        assert_eq!(topo.numa_nodes, 2);
        assert_eq!(topo.llcs, 4);
        assert_eq!(topo.cores_per_llc, 8);
        assert_eq!(topo.threads_per_core, 2);
    }

    #[test]
    fn topology_json_try_into_topology_rejects_zero_numa_nodes() {
        let json = TopologyJson {
            num_numa_nodes: 0,
            num_llcs: 1,
            cores_per_llc: 1,
            threads_per_core: 1,
        };
        let err = Topology::try_from(json).expect_err("zero numa_nodes must reject");
        assert!(
            err.contains("numa_nodes"),
            "error should mention numa_nodes: {err}"
        );
    }

    #[test]
    fn topology_json_try_into_topology_rejects_zero_llcs() {
        let json = TopologyJson {
            num_numa_nodes: 1,
            num_llcs: 0,
            cores_per_llc: 1,
            threads_per_core: 1,
        };
        let err = Topology::try_from(json).expect_err("zero llcs must reject");
        assert!(err.contains("llcs"), "error should mention llcs: {err}");
    }

    #[test]
    fn topology_json_try_into_topology_rejects_zero_cores() {
        let json = TopologyJson {
            num_numa_nodes: 1,
            num_llcs: 1,
            cores_per_llc: 0,
            threads_per_core: 1,
        };
        let err = Topology::try_from(json).expect_err("zero cores_per_llc must reject");
        assert!(
            err.contains("cores_per_llc"),
            "error should mention cores_per_llc: {err}"
        );
    }

    #[test]
    fn topology_json_try_into_topology_rejects_zero_threads() {
        let json = TopologyJson {
            num_numa_nodes: 1,
            num_llcs: 1,
            cores_per_llc: 1,
            threads_per_core: 0,
        };
        let err = Topology::try_from(json).expect_err("zero threads_per_core must reject");
        assert!(
            err.contains("threads_per_core"),
            "error should mention threads_per_core: {err}"
        );
    }

    #[test]
    fn topology_json_try_into_topology_rejects_indivisible_llcs() {
        // 3 LLCs across 2 NUMA nodes — not divisible.
        let json = TopologyJson {
            num_numa_nodes: 2,
            num_llcs: 3,
            cores_per_llc: 1,
            threads_per_core: 1,
        };
        let err = Topology::try_from(json).expect_err("indivisible llcs must reject");
        assert!(
            err.contains("divisible"),
            "error should mention divisibility: {err}"
        );
    }

    #[test]
    fn topology_json_try_into_topology_rejects_overflow_total_cpus() {
        // 2 LLCs × (u32::MAX / 4) cores × 4 threads overflows u32.
        let json = TopologyJson {
            num_numa_nodes: 1,
            num_llcs: 2,
            cores_per_llc: u32::MAX / 4,
            threads_per_core: 4,
        };
        let err = Topology::try_from(json).expect_err("u32 overflow on total cpus must reject");
        assert!(
            err.contains("overflow"),
            "error should mention overflow: {err}"
        );
    }

    #[test]
    fn topology_into_topology_json_drops_explicit_nodes_distances() {
        let topo = Topology {
            llcs: 4,
            cores_per_llc: 8,
            threads_per_core: 2,
            numa_nodes: 2,
            nodes: None,
            distances: None,
        };
        let json: TopologyJson = topo.into();
        assert_eq!(json.num_numa_nodes, 2);
        assert_eq!(json.num_llcs, 4);
        assert_eq!(json.cores_per_llc, 8);
        assert_eq!(json.threads_per_core, 2);
    }

    #[test]
    fn topology_constraints_json_into_topology_constraints_preserves_fields() {
        let json = TopologyConstraintsJson {
            min_numa_nodes: 1,
            max_numa_nodes: Some(4),
            min_llcs: 2,
            max_llcs: Some(8),
            requires_smt: true,
            min_cpus: 4,
            max_cpus: Some(64),
        };
        let c: TopologyConstraints = json.into();
        assert_eq!(c.min_numa_nodes, 1);
        assert_eq!(c.max_numa_nodes, Some(4));
        assert_eq!(c.min_llcs, 2);
        assert_eq!(c.max_llcs, Some(8));
        assert!(c.requires_smt);
        assert_eq!(c.min_cpus, 4);
        assert_eq!(c.max_cpus, Some(64));
    }

    #[test]
    fn topology_constraints_json_into_topology_constraints_handles_none_options() {
        let json = TopologyConstraintsJson {
            min_numa_nodes: 1,
            max_numa_nodes: None,
            min_llcs: 1,
            max_llcs: None,
            requires_smt: false,
            min_cpus: 1,
            max_cpus: None,
        };
        let c: TopologyConstraints = json.into();
        assert!(c.max_numa_nodes.is_none());
        assert!(c.max_llcs.is_none());
        assert!(c.max_cpus.is_none());
    }

    /// Const-fn parity with the runtime `--cell-parent-cgroup` gate:
    /// a `cgroup_parent` declaration containing `..` must compile-fail
    /// (or panic at runtime when the const fn is evaluated
    /// dynamically). Mirrors `runtime::append_base_sched_args`
    /// rejecting `--cell-parent-cgroup=/foo/..` so declarative and
    /// CLI sources share the same validation contract.
    #[test]
    #[should_panic(expected = "must not contain `..` segments")]
    fn cgroup_path_new_panics_on_parent_dir_segment() {
        let _ = CgroupPath::new("/foo/..");
    }

    /// Bare `/..` (ParentDir immediately after RootDir) — same
    /// rejection class as `/foo/..`, distinct shape.
    #[test]
    #[should_panic(expected = "must not contain `..` segments")]
    fn cgroup_path_new_panics_on_bare_parent_dir() {
        let _ = CgroupPath::new("/..");
    }

    /// `/.` — only a CurDir segment after root, no Normal anywhere.
    /// Hits the `has_normal=false` final assert (CurDir segments are
    /// stripped per the auto-strip rule, matching Path::components).
    #[test]
    #[should_panic(expected = "at least one non-`.`/non-empty segment")]
    fn cgroup_path_new_panics_on_only_dot_segment() {
        let _ = CgroupPath::new("/.");
    }

    /// Multiple consecutive slashes only — `///` produces 3 empty
    /// segments after the leading `/`. has_normal stays false →
    /// rejected by the final assert. Mirrors Path::components which
    /// collapses repeated separators and yields just `[RootDir]`.
    #[test]
    #[should_panic(expected = "at least one non-`.`/non-empty segment")]
    fn cgroup_path_new_panics_on_only_slashes() {
        let _ = CgroupPath::new("///");
    }

    /// `/foo/./bar` is ACCEPTED — the const-fn auto-strips the
    /// CurDir segment, matching `Path::components` semantics that
    /// the runtime validator uses. The two validators stay in
    /// lockstep on this shape (the runtime test
    /// `append_base_sched_args_accepts_embedded_dot_segment` pins
    /// the same accept behavior on the runtime side).
    #[test]
    fn cgroup_path_new_accepts_embedded_dot_segment() {
        let _ = CgroupPath::new("/foo/./bar");
    }

    /// Normal paths still pass — pin the accept path so a future
    /// over-tightening of the segment walker (e.g. rejecting
    /// trailing slash, or rejecting any segment of length 1) is
    /// caught.
    #[test]
    fn cgroup_path_new_accepts_normal_paths() {
        let _ = CgroupPath::new("/ktstr");
        let _ = CgroupPath::new("/sys/fs/cgroup/ktstr");
        let _ = CgroupPath::new("/a/b/c");
        let _ = CgroupPath::new("/foo/"); // trailing slash OK
    }

    /// GAP 14: pin that `SchedulerSpec` values survive a
    /// `HashSet` insert/lookup roundtrip — i.e. the Hash + Eq
    /// derives are present and agree on every variant. A
    /// regression that dropped Hash (e.g. swapping to a String
    /// payload without re-deriving) would silently break dedup
    /// in gauntlet expansion (which uses HashSet to collapse
    /// duplicate scheduler variants across topology combinations).
    #[test]
    fn scheduler_spec_hashset_roundtrip() {
        use std::collections::HashSet;
        let mut set: HashSet<SchedulerSpec> = HashSet::new();
        set.insert(SchedulerSpec::Eevdf);
        set.insert(SchedulerSpec::Discover("scx_lavd"));
        set.insert(SchedulerSpec::Path("./scx_rusty"));
        set.insert(SchedulerSpec::KernelBuiltin {
            enable: &["CONFIG_SCHED_DEBUG"],
            disable: &["CONFIG_SCHED_AUTOGROUP"],
        });
        assert_eq!(set.len(), 4);
        assert!(set.contains(&SchedulerSpec::Eevdf));
        assert!(set.contains(&SchedulerSpec::Discover("scx_lavd")));
        let dup_added = set.insert(SchedulerSpec::Eevdf);
        assert!(
            !dup_added,
            "duplicate SchedulerSpec insert must collapse via Hash+Eq"
        );
        assert_eq!(set.len(), 4);
    }

    /// GAP 14 sibling: pin that `BpfMapWrite` values survive a
    /// `HashSet` insert/lookup roundtrip — same rationale as
    /// `scheduler_spec_hashset_roundtrip` (the field uses an
    /// `&'static [&'static BpfMapWrite]` slice on `KtstrTestEntry`
    /// and dedup logic relies on the derived Hash+Eq).
    #[test]
    fn bpf_map_write_hashset_roundtrip() {
        use std::collections::HashSet;
        let w1 = BpfMapWrite::new(".data", 0, 1);
        let w2 = BpfMapWrite::new(".data", 4, 2);
        let w3 = BpfMapWrite::new(".other", 0, 1);
        let mut set: HashSet<BpfMapWrite> = HashSet::new();
        set.insert(w1);
        set.insert(w2);
        set.insert(w3);
        assert_eq!(set.len(), 3);
        assert!(set.contains(&w1));
        let dup_added = set.insert(w1);
        assert!(
            !dup_added,
            "duplicate BpfMapWrite insert must collapse via Hash+Eq"
        );
        assert_eq!(set.len(), 3);
    }

    // -- BpfMapWrite::new + Sysctl::new format-validation pins --
    //
    // Each `#[should_panic]` pins one branch of the const-assert
    // chain. A future relaxation of the format rules (e.g. dropping
    // the leading-`.` requirement on map_name_suffix) would have to
    // explicitly delete the corresponding test, surfacing the
    // semantic change rather than letting it pass silently.

    /// Empty `map_name_suffix` is intent-only invalid (matches no
    /// loaded BPF map). const-assert catches it at construction.
    #[test]
    #[should_panic(expected = "map_name_suffix must not be empty")]
    fn bpf_map_write_new_rejects_empty_suffix() {
        let _ = BpfMapWrite::new("", 0, 0);
    }

    /// BPF section names start with `.` (`.bss`, `.data`, `.rodata`);
    /// a suffix without the leading `.` matches no real map.
    #[test]
    #[should_panic(expected = "must start with `.`")]
    fn bpf_map_write_new_rejects_missing_dot_prefix() {
        let _ = BpfMapWrite::new("bss", 0, 0);
    }

    #[test]
    #[should_panic(expected = "must not contain whitespace")]
    fn bpf_map_write_new_rejects_whitespace_in_suffix() {
        let _ = BpfMapWrite::new(".b s", 0, 0);
    }

    #[test]
    #[should_panic(expected = "must not contain path separators")]
    fn bpf_map_write_new_rejects_path_separator_in_suffix() {
        let _ = BpfMapWrite::new(".bss/data", 0, 0);
    }

    /// Valid construction round-trips through the getters. Pinned to
    /// catch a getter-vs-field divergence (e.g. if a future refactor
    /// caches a transformed value but forgets to update the getter).
    #[test]
    fn bpf_map_write_new_valid_round_trips() {
        let w = BpfMapWrite::new(".bss", 16, 42);
        assert_eq!(w.map_name_suffix(), ".bss");
        assert_eq!(w.offset(), 16);
        assert_eq!(w.value(), 42);
    }

    #[test]
    #[should_panic(expected = "Sysctl key must not be empty")]
    fn sysctl_new_rejects_empty_key() {
        let _ = Sysctl::new("", "1");
    }

    /// Common operator typo: writing the sysctl path in slash-form
    /// (`kernel/foo`) instead of dotted form (`kernel.foo`).
    #[test]
    #[should_panic(expected = "must use the dotted form")]
    fn sysctl_new_rejects_slash_in_key() {
        let _ = Sysctl::new("kernel/foo", "1");
    }

    /// A bare single-segment key (`foo` instead of `kernel.foo`) is
    /// almost certainly a typo — the sysctl tree is always at least
    /// 2 segments deep.
    #[test]
    #[should_panic(expected = "must be namespaced")]
    fn sysctl_new_rejects_undotted_key() {
        let _ = Sysctl::new("foo", "1");
    }

    #[test]
    #[should_panic(expected = "must not start or end with `.`")]
    fn sysctl_new_rejects_leading_dot_key() {
        let _ = Sysctl::new(".foo", "1");
    }

    #[test]
    #[should_panic(expected = "must not start or end with `.`")]
    fn sysctl_new_rejects_trailing_dot_key() {
        let _ = Sysctl::new("foo.", "1");
    }

    #[test]
    #[should_panic(expected = "value must not be empty")]
    fn sysctl_new_rejects_empty_value() {
        let _ = Sysctl::new("kernel.foo", "");
    }

    #[test]
    #[should_panic(expected = "must not contain a newline")]
    fn sysctl_new_rejects_newline_in_value() {
        let _ = Sysctl::new("kernel.foo", "1\n2");
    }

    /// `=` in the value would corrupt the `sysctl.<key>=<value>`
    /// cmdline form (parser would split on the wrong `=`).
    #[test]
    #[should_panic(expected = "must not contain `=`")]
    fn sysctl_new_rejects_equals_in_value() {
        let _ = Sysctl::new("kernel.foo", "1=2");
    }

    #[test]
    fn sysctl_new_valid_round_trips() {
        let s = Sysctl::new("kernel.sched_cfs_bandwidth_slice_us", "1000");
        assert_eq!(s.key(), "kernel.sched_cfs_bandwidth_slice_us");
        assert_eq!(s.value(), "1000");
    }

    // -- additional Sysctl key/value rejection cases --

    #[test]
    #[should_panic(expected = "must not contain whitespace")]
    fn sysctl_new_rejects_space_in_key() {
        let _ = Sysctl::new(" kernel.foo", "1");
    }

    #[test]
    #[should_panic(expected = "must not contain whitespace")]
    fn sysctl_new_rejects_tab_in_key() {
        let _ = Sysctl::new("kernel\t.foo", "1");
    }

    #[test]
    #[should_panic(expected = "must not contain whitespace")]
    fn sysctl_new_rejects_newline_in_key() {
        let _ = Sysctl::new("kernel\n.foo", "1");
    }

    #[test]
    #[should_panic(expected = "must not contain `=`")]
    fn sysctl_new_rejects_equals_in_key() {
        let _ = Sysctl::new("kernel=foo.bar", "1");
    }

    #[test]
    #[should_panic(expected = "must be printable ASCII")]
    fn sysctl_new_rejects_control_byte_in_key() {
        let _ = Sysctl::new("kernel.\x01foo", "1");
    }

    #[test]
    #[should_panic(expected = "must not contain `..`")]
    fn sysctl_new_rejects_double_dot_in_key() {
        let _ = Sysctl::new("kernel..foo", "1");
    }

    #[test]
    #[should_panic(expected = "must not contain a carriage return")]
    fn sysctl_new_rejects_carriage_return_in_value() {
        let _ = Sysctl::new("kernel.foo", "1\r2");
    }

    // -- additional BpfMapWrite suffix rejection cases --

    #[test]
    #[should_panic(expected = "must be longer than a bare `.`")]
    fn bpf_map_write_new_rejects_bare_dot() {
        let _ = BpfMapWrite::new(".", 0, 0);
    }

    #[test]
    #[should_panic(expected = "must not start with `..`")]
    fn bpf_map_write_new_rejects_leading_double_dot() {
        let _ = BpfMapWrite::new("..bss", 0, 0);
    }

    #[test]
    #[should_panic(expected = "must be printable ASCII")]
    fn bpf_map_write_new_rejects_null_byte_in_suffix() {
        let _ = BpfMapWrite::new(".bss\0", 0, 0);
    }

    #[test]
    #[should_panic(expected = "must be printable ASCII")]
    fn bpf_map_write_new_rejects_control_byte_in_suffix() {
        let _ = BpfMapWrite::new(".b\x01ss", 0, 0);
    }

    /// Lock-step pin: `KtstrTestEntry::default()` must agree with
    /// `KtstrTestEntry::DEFAULT` field-by-field. The trait impl
    /// today delegates to the const, but a future divergence (e.g.
    /// someone "improves" `Default` to seed a different memory
    /// budget) would silently produce two construction paths
    /// disagreeing on what a baseline entry looks like —
    /// `..Default::default()` callers and `..KtstrTestEntry::DEFAULT`
    /// callers would yield different runs from "identical" code.
    #[test]
    fn ktstr_test_entry_default_matches_const() {
        let from_const = KtstrTestEntry::DEFAULT;
        let from_trait: KtstrTestEntry = Default::default();
        assert_eq!(from_trait.name, from_const.name);
        assert!(std::ptr::fn_addr_eq(from_trait.func, from_const.func));
        assert_eq!(from_trait.topology.llcs, from_const.topology.llcs);
        assert_eq!(
            from_trait.topology.cores_per_llc,
            from_const.topology.cores_per_llc
        );
        assert_eq!(
            from_trait.topology.threads_per_core,
            from_const.topology.threads_per_core
        );
        assert_eq!(
            from_trait.topology.numa_nodes,
            from_const.topology.numa_nodes
        );
        assert_eq!(from_trait.memory_mib, from_const.memory_mib);
        // Both default paths delegate to `Self::DEFAULT`, which sets
        // `scheduler: &Scheduler::EEVDF`. Rust may or may not dedupe
        // the `&CONST` materializations to the same address — pointer
        // equality is an implementation detail, not a contract.
        // Verify the scheduler-identity contract by NAME (the EEVDF
        // baseline has a unique stable name).
        assert_eq!(from_trait.scheduler.name, from_const.scheduler.name);
        assert_eq!(from_trait.scheduler.name, "eevdf");
        assert!(from_trait.payload.is_none() && from_const.payload.is_none());
        // Element-wise slice equality catches same-length content
        // drift (e.g. two different `&["--baseline=X"]` slices both
        // length 1 but with different X). Length-only compare would
        // miss that.
        assert_eq!(
            from_trait.workloads.len(),
            from_const.workloads.len(),
            "workloads count drift"
        );
        for (i, (a, b)) in from_trait
            .workloads
            .iter()
            .zip(from_const.workloads.iter())
            .enumerate()
        {
            assert!(
                std::ptr::eq(*a, *b),
                "workloads[{i}] pointer identity drift"
            );
        }
        assert_eq!(from_trait.auto_repro, from_const.auto_repro);
        assert_eq!(
            from_trait.extra_sched_args, from_const.extra_sched_args,
            "extra_sched_args content drift"
        );
        assert_eq!(from_trait.watchdog_timeout, from_const.watchdog_timeout);
        assert_eq!(
            from_trait.bpf_map_write.len(),
            from_const.bpf_map_write.len(),
            "bpf_map_write count drift"
        );
        for (i, (a, b)) in from_trait
            .bpf_map_write
            .iter()
            .zip(from_const.bpf_map_write.iter())
            .enumerate()
        {
            assert!(
                std::ptr::eq(*a, *b),
                "bpf_map_write[{i}] pointer identity drift"
            );
        }
        assert_eq!(from_trait.performance_mode, from_const.performance_mode);
        assert_eq!(from_trait.no_perf_mode, from_const.no_perf_mode);
        assert_eq!(from_trait.duration, from_const.duration);
        assert_eq!(from_trait.expect_err, from_const.expect_err);
        assert_eq!(from_trait.host_only, from_const.host_only);
        assert_eq!(
            from_trait.extra_include_files, from_const.extra_include_files,
            "extra_include_files content drift"
        );
        assert_eq!(from_trait.cleanup_budget, from_const.cleanup_budget);
        assert_eq!(from_trait.config_content, from_const.config_content);
        assert!(from_trait.disk.is_none() && from_const.disk.is_none());
        assert!(from_trait.post_vm.is_none() && from_const.post_vm.is_none());
        assert_eq!(from_trait.num_snapshots, from_const.num_snapshots);
        // `assert` field lock-step: Assert does not derive PartialEq,
        // so compare via `format_human()` (renders every
        // threshold field) plus the explicit bool field that
        // format_human omits.
        assert_eq!(
            from_trait.assert.format_human(),
            from_const.assert.format_human(),
            "Assert threshold-field drift between Default::default() and DEFAULT"
        );
        assert_eq!(
            from_trait.assert.enforce_monitor_thresholds,
            from_const.assert.enforce_monitor_thresholds,
            "Assert.enforce_monitor_thresholds drift between Default::default() and DEFAULT"
        );
    }

    /// `default_post_vm_periodic_fired` MUST return Ok when periodic
    /// was NOT configured (`periodic_target == 0`) — the helper is
    /// a no-op for non-periodic tests so they pass through the
    /// macro-default path without spurious assertions.
    #[test]
    fn default_post_vm_periodic_fired_skips_when_periodic_disabled() {
        let r = crate::vmm::VmResult {
            periodic_target: 0,
            periodic_fired: 0,
            ..crate::vmm::VmResult::test_fixture()
        };
        super::default_post_vm_periodic_fired(&r)
            .expect("periodic_target == 0 must short-circuit to Ok");
    }

    /// Build a synthetic non-placeholder `FailureDumpReport` —
    /// minimal but with `is_placeholder = false`. Used in the
    /// helpers below to drive the bridge's `periodic_real_count`
    /// without booting a real VM.
    fn real_report() -> crate::monitor::dump::FailureDumpReport {
        // Default puts `is_placeholder = false`; spell it explicitly
        // so a future Default-shape change doesn't silently flip the
        // test fixture.
        crate::monitor::dump::FailureDumpReport {
            schema: crate::monitor::dump::SCHEMA_SINGLE.to_string(),
            is_placeholder: false,
            ..Default::default()
        }
    }

    fn placeholder_report(reason: &str) -> crate::monitor::dump::FailureDumpReport {
        crate::monitor::dump::FailureDumpReport::placeholder(reason)
    }

    /// When periodic WAS configured AND at least one REAL (non-
    /// placeholder) snapshot landed on the bridge, the helper
    /// returns Ok. Pins the smoke-floor contract.
    #[test]
    fn default_post_vm_periodic_fired_ok_when_at_least_one_real_landed() {
        let r = crate::vmm::VmResult {
            periodic_target: 5,
            periodic_fired: 1,
            ..crate::vmm::VmResult::test_fixture()
        };
        r.snapshot_bridge.store("periodic_000", real_report());
        super::default_post_vm_periodic_fired(&r)
            .expect("at least one real periodic capture must surface as Ok");
    }

    /// When periodic WAS configured AND every fired snapshot was
    /// only a placeholder (rendezvous-timeout or gate-suppressed),
    /// the helper MUST fail — the scheduler attached but produced
    /// zero useful BPF state. This is the case the new semantic
    /// catches that the old `periodic_fired >= 1` floor missed.
    #[test]
    fn default_post_vm_periodic_fired_fails_when_only_placeholders_landed() {
        let r = crate::vmm::VmResult {
            periodic_target: 3,
            periodic_fired: 3,
            ..crate::vmm::VmResult::test_fixture()
        };
        // All three fires produced placeholders.
        r.snapshot_bridge
            .store("periodic_000", placeholder_report("rendezvous timed out"));
        r.snapshot_bridge
            .store("periodic_001", placeholder_report("rendezvous timed out"));
        r.snapshot_bridge
            .store("periodic_002", placeholder_report("rendezvous timed out"));
        let err = super::default_post_vm_periodic_fired(&r)
            .expect_err("placeholder-only fills must surface as Err");
        let msg = err.to_string();
        assert!(
            msg.contains("no periodic snapshot produced real BPF state")
                && msg.contains("periodic_real_count=0")
                && msg.contains("periodic_fired=3")
                && msg.contains("target=3"),
            "diagnostic must name the real-count floor + carry both counters; got {msg}",
        );
    }

    /// Pins the load-bearing tag-prefix discrimination: a bridge
    /// that holds ONLY non-periodic tagged reports (user
    /// `Op::CaptureSnapshot` captures, watchpoint fires, etc.)
    /// MUST surface `periodic_real_count == 0` even when those
    /// non-periodic reports are real (non-placeholder) and present
    /// in number. A regression that conflated "any real report" with
    /// "periodic real report" would pass this state spuriously.
    #[test]
    fn default_post_vm_periodic_fired_fails_when_only_non_periodic_tagged_reports_present() {
        let r = crate::vmm::VmResult {
            periodic_target: 3,
            periodic_fired: 3,
            ..crate::vmm::VmResult::test_fixture()
        };
        // Populate ONLY non-periodic-tagged reports — these come
        // from user `Op::CaptureSnapshot` calls or watchpoint
        // fires, not from the periodic dispatch loop. They MUST
        // NOT pollute the periodic floor.
        r.snapshot_bridge.store("user_capture", real_report());
        r.snapshot_bridge.store("snapshot_baseline", real_report());
        r.snapshot_bridge.store("watch_my_var", real_report());
        let err = super::default_post_vm_periodic_fired(&r).expect_err(
            "non-periodic tags MUST NOT count toward periodic_real_count; \
             bridge has 3 real reports but none periodic → must Err",
        );
        let msg = err.to_string();
        assert!(
            msg.contains("periodic_real_count=0"),
            "diagnostic must name the zero real-count; got {msg}",
        );
    }

    /// When periodic WAS configured AND ONE real capture landed
    /// alongside N placeholder timeouts (the realistic flake case),
    /// the helper MUST return Ok — single-snapshot timeouts under
    /// load shouldn't punish a working scheduler. Pins the
    /// tolerance gap that the strict `== periodic_target` floor
    /// would have failed.
    #[test]
    fn default_post_vm_periodic_fired_ok_when_one_real_among_placeholders() {
        let r = crate::vmm::VmResult {
            periodic_target: 5,
            periodic_fired: 5,
            ..crate::vmm::VmResult::test_fixture()
        };
        r.snapshot_bridge
            .store("periodic_000", placeholder_report("rendezvous timed out"));
        r.snapshot_bridge.store("periodic_001", real_report());
        r.snapshot_bridge
            .store("periodic_002", placeholder_report("rendezvous timed out"));
        r.snapshot_bridge
            .store("periodic_003", placeholder_report("rendezvous timed out"));
        r.snapshot_bridge
            .store("periodic_004", placeholder_report("rendezvous timed out"));
        super::default_post_vm_periodic_fired(&r).expect(
            "one real capture among placeholders must Ok — tolerance for single-snapshot flakes",
        );
    }

    /// Pins the early-return semantic at entry.rs:2470: the helper
    /// short-circuits to Ok on `periodic_target == 0` BEFORE
    /// checking `periodic_fired`. A regression that swapped these
    /// checks (e.g., "require fired matches target unconditionally")
    /// would break tests with `target=0, fired=5` (synthesized
    /// firings without a configured target). Currently impossible
    /// in production but cheap to pin.
    #[test]
    fn default_post_vm_periodic_fired_target_zero_fired_nonzero_returns_ok() {
        let r = crate::vmm::VmResult {
            periodic_target: 0,
            periodic_fired: 5,
            ..crate::vmm::VmResult::test_fixture()
        };
        super::default_post_vm_periodic_fired(&r)
            .expect("target == 0 takes priority over fired count; must Ok");
    }

    /// When `result.success == false` the underlying VM run
    /// failed (crash / timeout / signal). The default helper MUST
    /// short-circuit to Ok so the runner's own failure-rendering
    /// path drives the diagnostic; emitting `periodic_fired=0` on
    /// top would mask the real cause. Even if periodic was
    /// configured AND no snapshot fired, the success=false branch
    /// dominates.
    #[test]
    fn default_post_vm_periodic_fired_skips_when_vm_run_failed() {
        let r = crate::vmm::VmResult {
            success: false,
            periodic_target: 5,
            periodic_fired: 0,
            ..crate::vmm::VmResult::test_fixture()
        };
        super::default_post_vm_periodic_fired(&r)
            .expect("vm-run-failed must short-circuit to Ok so the runner's diagnostic dominates");
    }

    /// When periodic WAS configured BUT NO snapshot fired, the
    /// helper MUST return Err carrying both observed counters in
    /// the diagnostic. Pins that a regression to "always Ok" would
    /// surface immediately.
    #[test]
    fn default_post_vm_periodic_fired_fails_when_none_fired() {
        let r = crate::vmm::VmResult {
            periodic_target: 5,
            periodic_fired: 0,
            ..crate::vmm::VmResult::test_fixture()
        };
        let err = super::default_post_vm_periodic_fired(&r)
            .expect_err("periodic_fired == 0 with target > 0 must surface as Err");
        let msg = err.to_string();
        assert!(
            msg.contains("periodic_fired=0") && msg.contains("target=5"),
            "diagnostic must carry both counters; got {msg}",
        );
    }
}