ktstr/

verifier.rs

//! BPF verifier log parsing, cycle detection, and output formatting.
//!
//! Provides:
//! - [`VerifierStats`] / [`ProgStats`] / [`DiffRow`] — data types
//! - [`collect_verifier_output`] — boot VM, collect stats via host introspection
//! - [`format_verifier_output`] / [`format_verifier_diff`] — text formatting
//! - [`extract_verifier_log`] — extract verifier trace from libbpf log blob
//! - [`parse_verifier_stats`] — extract insn/state counts from verifier log
//! - [`normalize_verifier_line`] — strip variable register state annotations
//! - [`detect_cycle`] / [`collapse_cycles`] — loop iteration compression
//! - [`build_b_map`] / [`build_diff_rows`] — A/B comparison helpers
//! - `SCHED_OUTPUT_START` / `SCHED_OUTPUT_END` — delimiters the
//!   guest's rust_init emits over the bulk port (as `MSG_TYPE_SCHED_LOG`
//!   frames) around the scheduler log region;
//!   `parse_sched_output` extracts the enclosed block

use std::collections::HashMap;

/// Delimiter the guest's rust_init emits over the bulk port (as a
/// `MSG_TYPE_SCHED_LOG` frame) immediately before the scheduler log
/// block. Paired with [`SCHED_OUTPUT_END`].
pub(crate) const SCHED_OUTPUT_START: &str = "===SCHED_OUTPUT_START===";
/// Delimiter the guest's rust_init emits over the bulk port (as a
/// `MSG_TYPE_SCHED_LOG` frame) immediately after the scheduler log
/// block. Paired with [`SCHED_OUTPUT_START`].
pub(crate) const SCHED_OUTPUT_END: &str = "===SCHED_OUTPUT_END===";

/// Extract the scheduler log from guest output between
/// [`SCHED_OUTPUT_START`] and [`SCHED_OUTPUT_END`]. Returns `None` if
/// the delimiters are absent or the enclosed content is empty after
/// trimming.
///
/// Uses `find` on the start marker and `rfind` on the end marker: if
/// the scheduler log itself contains the end sentinel string (e.g. a
/// stack trace that quotes the marker), `rfind` anchors on the last
/// occurrence, which is the real terminator emitted by the guest's
/// post-scenario shutdown path.
pub(crate) fn parse_sched_output(output: &str) -> Option<&str> {
    let start = output.find(SCHED_OUTPUT_START)?;
    let end = output.rfind(SCHED_OUTPUT_END)?;
    let after_marker = start + SCHED_OUTPUT_START.len();
    if after_marker >= end {
        return None;
    }
    let content = output[after_marker..end].trim();
    if content.is_empty() {
        return None;
    }
    Some(content)
}

/// Concatenate every CRC-valid `MSG_TYPE_SCHED_LOG` chunk in the
/// bulk-port drain into one `String`, in arrival order.
///
/// The guest's `dump_sched_output` emits the `SCHED_OUTPUT_START`
/// and `SCHED_OUTPUT_END` markers as their own
/// [`crate::vmm::wire::MsgType::SchedLog`] frames, with the file
/// content split across one or more intermediate frames. Replaying
/// the chunks back-to-back reproduces the byte-for-byte stream the
/// prior COM2 path appended to `output`, so [`parse_sched_output`]
/// runs unchanged on the result.
///
/// Empty / `None` drain yields an empty string.
pub(crate) fn concat_sched_log_chunks(
    drain: Option<&crate::vmm::host_comms::BulkDrainResult>,
) -> String {
    let Some(drain) = drain else {
        return String::new();
    };
    let mut acc = String::new();
    for e in &drain.entries {
        if e.msg_type != crate::vmm::wire::MSG_TYPE_SCHED_LOG || !e.crc_ok {
            continue;
        }
        acc.push_str(&String::from_utf8_lossy(&e.payload));
    }
    acc
}

/// Extract scheduler log content even when the closing delimiter is
/// absent. Tries [`parse_sched_output`] first (well-formed
/// open+close); on failure, returns the slice from
/// [`SCHED_OUTPUT_START`] to the end of `output` when only the start
/// marker is present. Returns `None` only when neither marker is
/// found or every candidate slice is empty after trimming.
///
/// Used by the auto-repro path: a scheduler that crashes mid-run
/// emits SCHED_OUTPUT_START but never reaches the post-scenario
/// shutdown that writes SCHED_OUTPUT_END. The partial content still
/// holds the stack frames the probe pipeline needs to seed kprobe
/// targets, so discarding it would lose the only crash signal.
pub(crate) fn parse_sched_output_partial(output: &str) -> Option<&str> {
    if let Some(content) = parse_sched_output(output) {
        return Some(content);
    }
    let start = output.find(SCHED_OUTPUT_START)?;
    let after_marker = start + SCHED_OUTPUT_START.len();
    let content = output[after_marker..].trim();
    if content.is_empty() {
        return None;
    }
    Some(content)
}

/// Parsed verifier stats from the kernel log line:
/// `processed N insns (limit M) max_states_per_insn X total_states Y peak_states Z mark_read W`
pub struct VerifierStats {
    /// Instructions processed during verification.
    pub processed_insns: u64,
    /// Total explored verifier states.
    pub total_states: u64,
    /// Peak concurrent explored states.
    pub peak_states: u64,
    /// Total verification wall time in microseconds, when
    /// BPF_LOG_STATS emitted a "verification time" line.
    pub time_usec: Option<u64>,
    /// Stack depth in the format `"<prog>+<subprog>+<main>"` (e.g.
    /// `"32+16+8"`) when BPF_LOG_STATS emitted a "stack depth" line.
    pub stack_depth: Option<String>,
}

/// Per-program verifier statistics collected from a VM run.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct ProgStats {
    /// Program name as registered with the kernel.
    pub name: String,
    /// Instructions processed by the verifier (from host-side
    /// `bpf_prog_aux->verified_insns`).
    pub verified_insns: u32,
}

/// A single row in the A/B diff output.
pub struct DiffRow {
    /// Program name present in both A and B runs.
    pub name: String,
    /// `verified_insns` from the A run.
    pub a: u64,
    /// `verified_insns` from the B run.
    pub b: u64,
    /// Signed delta (`b - a`); positive means B's verifier cost grew
    /// relative to A.
    pub delta: i64,
}

/// Parse `raw` as `u64`, warning with `field` context on failure.
/// Returns 0 on parse error. Centralizes the
/// warn-and-default-to-zero path for the three count words
/// (`processed_insns`, `total_states`, `peak_states`) so their
/// error handling stays in lock-step.
fn parse_or_warn(raw: &str, field: &str) -> u64 {
    match raw.parse() {
        Ok(n) => n,
        Err(e) => {
            tracing::warn!(
                field,
                word = raw,
                err = %e,
                "malformed BPF verifier count; leaving 0",
            );
            0
        }
    }
}

/// Parse verifier stats from the log output.
///
/// The kernel always emits a "processed N insns ..." line. When
/// BPF_LOG_STATS is set, it also emits "verification time" and
/// "stack depth" lines.
pub fn parse_verifier_stats(log: &str) -> VerifierStats {
    let mut stats = VerifierStats {
        processed_insns: 0,
        total_states: 0,
        peak_states: 0,
        time_usec: None,
        stack_depth: None,
    };

    let mut found_insns = false;
    let mut found_time = false;
    let mut found_stack = false;

    for line in log.lines().rev() {
        if !found_insns && line.starts_with("processed ") {
            found_insns = true;
            let words: Vec<&str> = line.split_whitespace().collect();
            if words.len() >= 2 {
                stats.processed_insns = parse_or_warn(words[1], "processed_insns");
            }
            for (i, &w) in words.iter().enumerate() {
                if w == "total_states"
                    && let Some(v) = words.get(i + 1)
                {
                    stats.total_states = parse_or_warn(v, "total_states");
                }
                if w == "peak_states"
                    && let Some(v) = words.get(i + 1)
                {
                    stats.peak_states = parse_or_warn(v, "peak_states");
                }
            }
        }
        if !found_time && line.contains("verification time") {
            found_time = true;
            for word in line.split_whitespace() {
                if let Ok(n) = word.parse::<u64>() {
                    stats.time_usec = Some(n);
                    break;
                }
            }
        }
        if !found_stack && line.contains("stack depth") {
            found_stack = true;
            if let Some(pos) = line.find("stack depth") {
                let after = &line[pos + "stack depth".len()..];
                let depth_str = after.trim();
                if !depth_str.is_empty() {
                    stats.stack_depth = Some(depth_str.to_string());
                }
            }
        }
        if found_insns && found_time && found_stack {
            break;
        }
    }

    stats
}

/// Normalize a BPF verifier log line by stripping variable register-state
/// annotations so that lines from different loop iterations compare equal.
///
/// Handles:
/// - Instruction with `; frame` annotation: `3006: (07) r9 += 1  ; frame1: R9_w=2`
/// - Instruction with `; R` + digit annotation: `9: (15) if r7 == 0x0 goto pc+1  ; R7=scalar(...)`
/// - Branch with inline target state: `3026: (b5) if r6 <= 0x11dc0 goto pc+2 3029: frame1: R0=1`
/// - Standalone register dump with frame: `3041: frame1: R0_w=scalar()`
/// - Standalone register dump without frame: `3029: R0=1 R6=scalar()`
///
/// Preserves source comments (`; for (int j = 0; ...)`) and non-annotation
/// semicolons (`; Return value`) -- these serve as cycle anchors.
pub fn normalize_verifier_line(line: &str) -> &str {
    let trimmed = line.trim();
    if trimmed.is_empty() || !trimmed.as_bytes()[0].is_ascii_digit() {
        return trimmed;
    }
    // "3041: frame1: ..." or "3041: R0_w=scalar()" — standalone register dump.
    // State-only lines; keep just the instruction index.
    if let Some(colon) = trimmed.find(": ") {
        let after = &trimmed[colon + 2..];
        if after.starts_with("frame")
            || (after.starts_with('R')
                && after.as_bytes().get(1).is_some_and(|b| b.is_ascii_digit()))
        {
            return &trimmed[..colon + 1];
        }
    }
    // "; frame" annotation on instruction line
    if let Some(pos) = trimmed.find("; frame") {
        return trimmed[..pos].trim_end();
    }
    // "; R" followed by digit — register annotation without frame prefix
    if let Some(pos) = trimmed.find("; R")
        && trimmed
            .as_bytes()
            .get(pos + 3)
            .is_some_and(|b| b.is_ascii_digit())
    {
        return trimmed[..pos].trim_end();
    }
    // Inline branch-target state: "goto pc+2 3029: frame1: ..."
    if let Some(goto_pos) = trimmed.find("goto pc") {
        let after_goto = &trimmed[goto_pos + 7..];
        let end = after_goto
            .find(|c: char| c != '+' && c != '-' && !c.is_ascii_digit())
            .unwrap_or(after_goto.len());
        let insn_end = goto_pos + 7 + end;
        if insn_end < trimmed.len() {
            return trimmed[..insn_end].trim_end();
        }
    }
    trimmed
}

/// Normalize for cycle detection: strip register annotations (via
/// `normalize_verifier_line`) then strip the leading instruction address
/// (`NNN: `). Unrolled loops place each copy at different addresses, so
/// the address must be removed for block comparison to find repeats.
fn normalize_for_cycle_detection(line: &str) -> &str {
    let n = normalize_verifier_line(line);
    // Strip leading digits + ": " prefix (e.g. "42: (07) r1 += 8" -> "(07) r1 += 8").
    if let Some(colon) = n.find(": ") {
        let before = &n[..colon];
        if !before.is_empty() && before.bytes().all(|b| b.is_ascii_digit()) {
            return &n[colon + 2..];
        }
    }
    n
}

/// Detect a single repeating cycle in a slice of lines.
///
/// Returns `Some((start, period, count))` where the cycle begins at
/// `start`, each iteration is `period` lines, and it repeats `count` times.
pub fn detect_cycle(lines: &[&str]) -> Option<(usize, usize, usize)> {
    const MIN_PERIOD: usize = 5;
    const MIN_REPS: usize = 3;

    if lines.len() < MIN_PERIOD * MIN_REPS {
        return None;
    }

    // Two normalization levels:
    // - anchor_norms: keeps addresses, strips register annotations. Used for
    //   anchor frequency counting — prevents within-period duplicates at
    //   different addresses from inflating frequency.
    // - block_norms: also strips addresses. Used for block equality comparison
    //   so unrolled loops (same instructions at different addresses) can match.
    let anchor_norms: Vec<&str> = lines.iter().map(|l| normalize_verifier_line(l)).collect();
    let block_norms: Vec<&str> = lines
        .iter()
        .map(|l| normalize_for_cycle_detection(l))
        .collect();

    // Find most frequent non-trivial anchor-normalized line.
    let mut sorted_norms: Vec<&str> = anchor_norms
        .iter()
        .filter(|l| l.len() >= 10)
        .copied()
        .collect();
    sorted_norms.sort_unstable();

    let mut best_anchor: Option<(&str, usize)> = None;
    let mut i = 0;
    while i < sorted_norms.len() {
        let mut j = i + 1;
        while j < sorted_norms.len() && sorted_norms[j] == sorted_norms[i] {
            j += 1;
        }
        let count = j - i;
        if count >= MIN_REPS && best_anchor.is_none_or(|(_, best)| count > best) {
            best_anchor = Some((sorted_norms[i], count));
        }
        i = j;
    }

    // If address-preserving anchor search found nothing (unrolled loops
    // where every address is unique), fall back to address-stripped norms.
    let (anchor, use_block_norms_for_positions) = match best_anchor {
        Some((a, _)) => (a, false),
        None => {
            let mut sorted_block: Vec<&str> = block_norms
                .iter()
                .filter(|l| l.len() >= 10)
                .copied()
                .collect();
            sorted_block.sort_unstable();
            let mut ba: Option<(&str, usize)> = None;
            let mut bi = 0;
            while bi < sorted_block.len() {
                let mut bj = bi + 1;
                while bj < sorted_block.len() && sorted_block[bj] == sorted_block[bi] {
                    bj += 1;
                }
                let c = bj - bi;
                if c >= MIN_REPS && ba.is_none_or(|(_, best)| c > best) {
                    ba = Some((sorted_block[bi], c));
                }
                bi = bj;
            }
            match ba {
                Some((a, _)) => (a, true),
                None => return None,
            }
        }
    };

    let norms_for_pos = if use_block_norms_for_positions {
        &block_norms
    } else {
        &anchor_norms
    };
    let positions: Vec<usize> = norms_for_pos
        .iter()
        .enumerate()
        .filter(|(_, l)| **l == anchor)
        .map(|(i, _)| i)
        .collect();

    // Try strides 1..=3 to handle anchors appearing K times per cycle.
    for stride in 1..=3usize {
        if positions.len() <= stride {
            continue;
        }

        let mut gaps: Vec<usize> = positions
            .windows(stride + 1)
            .map(|w| w[stride] - w[0])
            .filter(|g| *g >= MIN_PERIOD)
            .collect();
        gaps.sort_unstable();

        let mut best_period = 0;
        let mut best_gap_count = 0;
        let mut gi = 0;
        while gi < gaps.len() {
            let mut gj = gi + 1;
            while gj < gaps.len() && gaps[gj] == gaps[gi] {
                gj += 1;
            }
            let count = gj - gi;
            if count > best_gap_count {
                best_gap_count = count;
                best_period = gaps[gi];
            }
            gi = gj;
        }
        if best_period == 0 || best_gap_count < MIN_REPS - 1 {
            continue;
        }
        let period = best_period;

        for &pos in &positions {
            if pos + 2 * period > lines.len() {
                break;
            }
            if block_norms[pos..pos + period] == block_norms[pos + period..pos + 2 * period] {
                let first_block = &block_norms[pos..pos + period];
                let mut count = 1;
                while pos + (count + 1) * period <= lines.len() {
                    if block_norms[pos + count * period..pos + (count + 1) * period] != *first_block
                    {
                        break;
                    }
                    count += 1;
                }
                // Try earlier starts to find best alignment.
                let mut best_start = pos;
                let mut best_count = count;
                for offset in 1..period {
                    let Some(cand) = pos.checked_sub(offset) else {
                        break;
                    };
                    if cand + 2 * period > lines.len() {
                        continue;
                    }
                    if block_norms[cand..cand + period]
                        != block_norms[cand + period..cand + 2 * period]
                    {
                        continue;
                    }
                    let mut c = 2;
                    while cand + (c + 1) * period <= lines.len()
                        && block_norms[cand + c * period..cand + (c + 1) * period]
                            == block_norms[cand..cand + period]
                    {
                        c += 1;
                    }
                    if c > best_count {
                        best_start = cand;
                        best_count = c;
                    }
                }
                if best_count >= MIN_REPS {
                    return Some((best_start, period, best_count));
                }
            }
        }
    }

    None
}

/// Collapse repeating cycles in a verifier log.
///
/// Runs cycle detection iteratively (up to 5 passes for nested loops).
/// Each cycle is replaced with:
/// - `--- Nx of the following M lines ---` (count header, no closing marker)
/// - first iteration (with original register annotations)
/// - `--- K identical iterations omitted ---` (omission marker)
/// - last iteration (with original register annotations)
/// - `--- end repeat ---` (closes the omission)
pub fn collapse_cycles(log: &str) -> String {
    const MAX_PASSES: usize = 5;
    let mut text = log.to_string();

    for _ in 0..MAX_PASSES {
        let lines: Vec<&str> = text.lines().collect();
        let (start, period, count) = match detect_cycle(&lines) {
            Some(c) => c,
            None => break,
        };

        let mut out = String::new();
        for line in &lines[..start] {
            out.push_str(line);
            out.push('\n');
        }
        out.push_str(&format!(
            "--- {}x of the following {} lines ---\n",
            count, period
        ));
        for line in &lines[start..start + period] {
            out.push_str(line);
            out.push('\n');
        }
        out.push_str(&format!(
            "--- {} identical iterations omitted ---\n",
            count - 2
        ));
        let last_start = start + (count - 1) * period;
        for line in &lines[last_start..last_start + period] {
            out.push_str(line);
            out.push('\n');
        }
        out.push_str("--- end repeat ---\n");
        let suffix_start = start + count * period;
        for line in &lines[suffix_start..] {
            out.push_str(line);
            out.push('\n');
        }
        text = out;
    }

    text
}

/// Build diff rows from A stats and B lookup map.
pub fn build_diff_rows(stats_a: &[ProgStats], b_map: &HashMap<String, u64>) -> Vec<DiffRow> {
    let mut rows = Vec::new();
    for ps in stats_a {
        let a = ps.verified_insns as u64;
        let b = b_map.get(&ps.name).copied().unwrap_or(0);
        rows.push(DiffRow {
            name: ps.name.clone(),
            a,
            b,
            delta: a as i64 - b as i64,
        });
    }
    rows
}

/// Build the B-side lookup map from collected stats.
pub fn build_b_map(stats_b: &[ProgStats]) -> HashMap<String, u64> {
    stats_b
        .iter()
        .map(|ps| (ps.name.clone(), ps.verified_insns as u64))
        .collect()
}

// ---------------------------------------------------------------------------
// VM-based verifier collection
// ---------------------------------------------------------------------------

/// Whether the scheduler positively confirmed it turned on during a
/// verifier VM run.
///
/// The guest init's attach gate (`poll_startup` + `poll_scx_attached`
/// in `crate::vmm::rust_init::scheduler`) already runs on every verifier
/// VM boot: it confirms the scheduler process survived BPF load AND that
/// `/sys/kernel/sched_ext/state` reached `enabled`. The kernel sets
/// `enabled` only after `ops.init`, per-task init, and switching
/// eligible tasks to the sched_ext class (`kernel/sched/ext.c`
/// `scx_root_enable_workfn`), so `enabled` proves the scheduler turned
/// on and is scheduling — not merely that its BPF loaded.
///
/// The verdict is POSITIVE-confirmation, not absence-of-failure: a
/// verifier cell PASSes only when the guest reached its post-attach
/// dispatch phase — a `PayloadStarting` lifecycle frame, emitted at
/// `ktstr_guest_init` Phase 5, which is reached ONLY if `start_scheduler`
/// succeeded in Phase 3. On attach failure the guest emits
/// `SchedulerDied` / `SchedulerNotAttached` and force-reboots in Phase 3
/// BEFORE Phase 5, so no `PayloadStarting` arrives. A guest that vanishes
/// before Phase 5 with NO frame at all — e.g. a kernel panic, which
/// reboots via the guest's `panic=-1` (an i8042 reset →
/// `ExitAction::Shutdown`, NOT a host watchdog timeout) — is
/// [`AttachOutcome::Unconfirmed`], also a FAIL. Absence-of-failure alone
/// would false-PASS that vanish case. [`collect_verifier_output`]
/// consumes this verdict instead of discarding it. Scheduler-agnostic
/// (kernel sysfs state), so it holds for every declared scheduler.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum AttachOutcome {
    /// The guest reached its post-attach dispatch phase (a
    /// `PayloadStarting` frame) with no failure frame: the scheduler
    /// loaded, stayed alive, and reached sched_ext `enabled`.
    Attached,
    /// Scheduler process exited during BPF load / startup
    /// (`LifecyclePhase::SchedulerDied`).
    Died,
    /// Scheduler stayed alive but never reached `enabled`
    /// (`LifecyclePhase::SchedulerNotAttached`); carries the guest's
    /// reason suffix when present.
    NotAttached(String),
    /// No failure frame AND no `PayloadStarting` frame: the guest never
    /// reached the post-attach dispatch phase, so attach was never
    /// positively confirmed (e.g. an early guest kernel panic that
    /// reboots via `panic=-1` before Phase 3 — emitting no lifecycle
    /// frame and NOT tripping the host watchdog).
    Unconfirmed,
}

impl AttachOutcome {
    /// Human-readable failure reason, or `None` when attached.
    pub fn failure_reason(&self) -> Option<String> {
        match self {
            AttachOutcome::Attached => None,
            AttachOutcome::Died => {
                Some("scheduler process exited during BPF load/startup".to_string())
            }
            AttachOutcome::NotAttached(reason) if reason.is_empty() => {
                Some("scheduler never reached sched_ext 'enabled'".to_string())
            }
            AttachOutcome::NotAttached(reason) => Some(format!(
                "scheduler never reached sched_ext 'enabled': {reason}"
            )),
            AttachOutcome::Unconfirmed => Some(
                "scheduler attach unconfirmed — guest never reached the dispatch phase \
                 (no PayloadStarting frame; possible early guest kernel panic)"
                    .to_string(),
            ),
        }
    }
}

/// Derive the [`AttachOutcome`] from a VM run's bulk-port lifecycle
/// frames using a POSITIVE-confirmation rule:
/// - a `SchedulerDied` frame ⇒ [`AttachOutcome::Died`] (wins outright —
///   a process that exited cannot have attached);
/// - else a `SchedulerNotAttached` frame ⇒ [`AttachOutcome::NotAttached`]
///   (with its reason suffix);
/// - else a `PayloadStarting` frame ⇒ [`AttachOutcome::Attached`] (the
///   guest reached its post-attach dispatch phase, so the scheduler
///   turned on);
/// - else [`AttachOutcome::Unconfirmed`] — no failure AND no positive
///   frame, so attach was never confirmed (e.g. an early guest kernel
///   panic that reboots before Phase 5 without emitting any frame).
///
/// Corrupt frames (`crc_ok == false`) and empty payloads are skipped. A
/// `None` `guest_messages` (no frames at all) is
/// [`AttachOutcome::Unconfirmed`].
pub(crate) fn attach_outcome_from_messages(
    guest_messages: Option<&crate::vmm::host_comms::BulkDrainResult>,
) -> AttachOutcome {
    let Some(drain) = guest_messages else {
        return AttachOutcome::Unconfirmed;
    };
    let mut not_attached: Option<String> = None;
    let mut payload_starting = false;
    for e in &drain.entries {
        if e.msg_type != crate::vmm::wire::MSG_TYPE_LIFECYCLE || !e.crc_ok || e.payload.is_empty() {
            continue;
        }
        match crate::vmm::wire::LifecyclePhase::from_wire(e.payload[0]) {
            Some(crate::vmm::wire::LifecyclePhase::SchedulerDied) => return AttachOutcome::Died,
            Some(crate::vmm::wire::LifecyclePhase::SchedulerNotAttached) => {
                not_attached = Some(String::from_utf8_lossy(&e.payload[1..]).into_owned());
            }
            Some(crate::vmm::wire::LifecyclePhase::PayloadStarting) => {
                payload_starting = true;
            }
            _ => {}
        }
    }
    if let Some(reason) = not_attached {
        AttachOutcome::NotAttached(reason)
    } else if payload_starting {
        AttachOutcome::Attached
    } else {
        AttachOutcome::Unconfirmed
    }
}

/// Whether the guest confirmed workload dispatch: at least one
/// `WorkloadDispatched` lifecycle frame (crc-ok, non-empty payload) in
/// the run's bulk-port frames. Emitted by `ktstr_guest_init` Phase 5 only
/// when the injected SpinWait workload recorded a worker with non-zero
/// `iterations` under a confirmed SCHED_EXT policy (`sched_policy_error`
/// is None) after the scheduler attached — so a fair-class fallback
/// cannot false-confirm — a positive, scheduler-agnostic proof the
/// scheduler dispatched a task onto a CPU.
/// Corrupt frames (`crc_ok == false`) and empty payloads are skipped. A
/// `None` `guest_messages` (no frames at all) is `false`.
fn dispatch_confirmed_from_messages(
    guest_messages: Option<&crate::vmm::host_comms::BulkDrainResult>,
) -> bool {
    let Some(drain) = guest_messages else {
        return false;
    };
    drain.entries.iter().any(|e| {
        e.msg_type == crate::vmm::wire::MSG_TYPE_LIFECYCLE
            && e.crc_ok
            && !e.payload.is_empty()
            && crate::vmm::wire::LifecyclePhase::from_wire(e.payload[0])
                == Some(crate::vmm::wire::LifecyclePhase::WorkloadDispatched)
    })
}

/// Result of collecting verifier output from a VM run.
pub struct VerifierVmResult {
    /// Per-program verifier statistics from host-side memory
    /// introspection (`bpf_prog_aux->verified_insns`).
    pub stats: Vec<ProgStats>,
    /// Scheduler log (stdout+stderr) from the VM. Contains libbpf's
    /// verifier instruction traces when BPF load fails.
    pub scheduler_log: String,
    /// Whether the scheduler positively confirmed attach. Derived from
    /// the guest's lifecycle frames ([`AttachOutcome`]). Attach is
    /// necessary but NOT sufficient for a cell PASS: this must be
    /// [`AttachOutcome::Attached`] (the guest reached its post-attach
    /// dispatch phase) AND [`Self::dispatched`] must be true. Verification
    /// alone (non-empty `stats`) is not enough — a scheduler whose BPF
    /// loads but never reaches sched_ext `enabled`, or a guest that
    /// vanishes before the dispatch phase, is a real failure.
    pub attach: AttachOutcome,
    /// Whether the guest confirmed the injected verifier workload
    /// dispatched — a `WorkloadDispatched` lifecycle frame, emitted by
    /// `ktstr_guest_init` Phase 5 when the SpinWait probe recorded a
    /// worker with non-zero `iterations` under a confirmed SCHED_EXT
    /// policy after attach (so a fair-class fallback cannot false-confirm).
    /// A cell PASSes only
    /// when this is true AND [`Self::attach`] is
    /// [`AttachOutcome::Attached`]: a scheduler that turns on (sched_ext
    /// `enabled`) but never dispatches a runnable task is a real, distinct
    /// failure — worse than never attaching — that the attach verdict
    /// alone cannot catch. Derived from the run's lifecycle frames;
    /// scheduler-agnostic — the probe runs as SCHED_EXT, so the BPF
    /// scheduler dispatches it under any switch mode (full or
    /// `SCX_OPS_SWITCH_PARTIAL`) and non-zero worker progress proves
    /// dispatch, unlike an scx-specific `nr_dispatched` counter.
    pub dispatched: bool,
    /// The host watchdog fired (hard-deadline hang) before the guest
    /// exited. Orthogonal to [`Self::attach`]: the attach verdict already
    /// fails a guest that vanished BEFORE the dispatch phase — an early
    /// kernel panic reboots via `panic=-1` (an i8042 reset →
    /// `ExitAction::Shutdown`, `timed_out == false`) and is
    /// [`AttachOutcome::Unconfirmed`]. This flag catches the remaining
    /// case: a guest that wedges AFTER attaching (during teardown), which
    /// leaves `attach == Attached` but never exits. A verifier cell FAILs
    /// on it too — but NOT on the guest exit code, which is `1` even on
    /// the verifier success path (no `#[ktstr_test]` body to dispatch).
    pub timed_out: bool,
}

impl VerifierVmResult {
    /// The verifier cell PASS/FAIL verdict: `Ok(())` when the scheduler
    /// verified its BPF, attached (sched_ext `enabled`), AND dispatched
    /// the injected workload; `Err(reason)` naming the first failing gate
    /// otherwise. Gate order — `timed_out` (a post-attach teardown hang),
    /// then attach (did it turn on?), then dispatch (did it schedule a
    /// task?) — so the root-cause failure is reported first: an attach
    /// failure is named before the dispatch gate it necessarily also
    /// trips. Does NOT key on the guest exit code, which is `1` even on
    /// the verifier success path (no `#[ktstr_test]` body to dispatch).
    pub fn cell_verdict(&self) -> Result<(), String> {
        // A hard-deadline hang. The attach verdict (Unconfirmed) already
        // fails a guest that vanished BEFORE the dispatch phase (an early
        // panic reboots via panic=-1 with timed_out=false), so this
        // catches the orthogonal case: a guest that wedges AFTER attaching,
        // during teardown.
        if self.timed_out {
            return Err("VM timed out (hung after attach, before exit)".to_string());
        }
        // PASS requires the scheduler to have turned ON, not just to have
        // loaded + verified its BPF.
        if let Some(reason) = self.attach.failure_reason() {
            return Err(format!("scheduler did not turn on — {reason}"));
        }
        // PASS also requires DISPATCH: the guest injects a SpinWait
        // workload after attach and only emits WorkloadDispatched when a
        // SCHED_EXT worker makes forward progress. A scheduler that turns
        // on but never dispatches a runnable task is a distinct, worse
        // failure the attach gate can't catch.
        if !self.dispatched {
            return Err(
                "scheduler attached but did not dispatch the injected workload (0 iterations)"
                    .to_string(),
            );
        }
        Ok(())
    }
}

/// Boot a VM and collect verifier statistics via host-side memory
/// introspection. Per-program `verified_insns` comes from
/// `bpf_prog_aux->verified_insns` read through the guest's physical
/// memory. On load failure, libbpf prints the verifier log to stderr;
/// the returned `scheduler_log` field contains the scheduler's captured
/// output from the VM.
///
/// `topology` selects the verifier VM's emulated topology via
/// [`crate::test_support::TopologyJson`] — the same named-field
/// shape carried in the per-scheduler `--ktstr-list-schedulers`
/// JSON. The named fields force callers to spell every dimension
/// at the call site, preventing position-swap misorders. The sole
/// caller is the sweep cell handler (`crate::test_support::dispatch`'s
/// `run_verifier_cell`), which passes the topology of the gauntlet
/// preset named in the cell — the verifier sweeps each scheduler across
/// every preset its constraints accept under no_perf_mode.
pub fn collect_verifier_output(
    sched_bin: &std::path::Path,
    ktstr_bin: &std::path::Path,
    kernel: &std::path::Path,
    extra_sched_args: &[String],
    topology: crate::test_support::TopologyJson,
) -> anyhow::Result<VerifierVmResult> {
    use anyhow::Context;

    // Pre-validate via TryFrom so a clean "topology rejected" error
    // surfaces here instead of the builder's Topology::new panic on
    // bad input.
    let validated: crate::vmm::topology::Topology = topology
        .try_into()
        .map_err(|e: String| anyhow::anyhow!("invalid topology {topology:?}: {e}"))?;

    let sched_args: Vec<String> = extra_sched_args.to_vec();

    // The verifier only loads the scheduler's BPF and reads the kernel
    // verifier's load-time `verified_insns` counts via host-side
    // introspection — a value fixed at BPF load, wholly INDEPENDENT of
    // perf-mode tuning (CPU pinning, RT priority, hugepages, NUMA mbind,
    // KVM exit suppression). So the verifier VM ALWAYS runs with
    // performance mode disabled: it needs none of that tuning. Disabling
    // perf mode also moves the run OFF the default run-lock path — whose
    // per-offset `LOCK_SH` search hard-fails "all N LLC slots busy
    // (LOCK_SH)" (`acquire_default_run_locks`, src/vmm/mod.rs) when no
    // candidate offset is free, the failure the verifier was hitting —
    // ONTO the no-perf-mode plan, which reserves a shared (`LOCK_SH`)
    // SUBSET of LLCs via `acquire_llc_plan`. `LOCK_SH` holders are
    // mutually compatible, so parallel verifier / no-perf cells no longer
    // starve each other on the LLC lock; a `performance_mode` peer holding
    // `LOCK_EX` on those LLCs can still defer a cell (nextest retries it),
    // which is correct — the verifier must not perturb an isolated peer's
    // pinned CPUs. Note `no_perf_mode(true)` does NOT skip the reservation
    // ENTIRELY — only `KTSTR_BYPASS_LLC_LOCKS=1` does; see
    // [`crate::vmm::KtstrVmBuilder::no_perf_mode`].
    // Pass the validated Topology directly so misorder cannot occur
    // at the builder boundary (the TryFrom above already enforces the
    // type-level invariants).
    let vm = crate::vmm::KtstrVm::builder()
        .kernel(kernel)
        .init_binary(ktstr_bin)
        .scheduler_binary(sched_bin)
        .sched_args(&sched_args)
        // Boot the guest into the verifier dispatch probe: the sweep VM
        // has no `#[ktstr_test]` body, so Phase 5 spawns a SpinWait
        // workload and emits `WorkloadDispatched` on confirmed progress.
        // The PASS verdict requires that frame in addition to attach.
        .run_args(&[crate::test_support::VERIFIER_WORKLOAD_FLAG.to_string()])
        .topology(validated)
        .memory_mib(2048)
        .timeout(std::time::Duration::from_secs(120))
        .no_perf_mode(true)
        .build()
        .context("build verifier VM")?;

    let result = vm.run().context("run verifier VM")?;

    // Concatenate bulk-port `MSG_TYPE_SCHED_LOG` chunks, then run
    // the marker-pair extractor on the merged stream — the
    // SCHED_OUTPUT_START/END markers travel verbatim inside chunk
    // bytes so the existing parser slices the same content the
    // prior COM2 dump produced. Falls back to `result.output` when
    // the bulk-port drain has no SchedLog frames (verifier VM
    // running on a kernel without the bulk port, for instance).
    let merged = concat_sched_log_chunks(result.guest_messages.as_ref());
    let scheduler_log = if !merged.is_empty() {
        parse_sched_output(&merged).unwrap_or("").to_string()
    } else {
        parse_sched_output(&result.output).unwrap_or("").to_string()
    };

    // Build ProgStats from host-side ProgVerifierStats. Each program
    // that loaded successfully is visible in prog_idr with its
    // verified_insns count.
    let stats: Vec<ProgStats> = result
        .verifier_stats
        .iter()
        .map(|pvs| ProgStats {
            name: pvs.name.clone(),
            verified_insns: pvs.verified_insns,
        })
        .collect();

    let attach = attach_outcome_from_messages(result.guest_messages.as_ref());
    let dispatched = dispatch_confirmed_from_messages(result.guest_messages.as_ref());

    Ok(VerifierVmResult {
        stats,
        scheduler_log,
        attach,
        dispatched,
        timed_out: result.timed_out,
    })
}

/// Extract the verifier instruction trace from a scheduler log blob.
///
/// libbpf wraps the kernel verifier log between marker lines:
///   `-- BEGIN PROG LOAD LOG --`
///   `-- END PROG LOAD LOG --`
///
/// Returns the content between the first pair of markers, or `None` if
/// no markers are found (backward compat with logs that contain only
/// raw verifier output).
pub fn extract_verifier_log(scheduler_log: &str) -> Option<&str> {
    const BEGIN: &str = "-- BEGIN PROG LOAD LOG --";
    const END: &str = "-- END PROG LOAD LOG --";

    let begin_pos = scheduler_log.find(BEGIN)?;
    let content_start = begin_pos + BEGIN.len();
    // Skip the newline after the BEGIN marker if present.
    let content_start = if scheduler_log.as_bytes().get(content_start) == Some(&b'\n') {
        content_start + 1
    } else {
        content_start
    };
    let end_pos = scheduler_log[content_start..].find(END)?;
    let content = &scheduler_log[content_start..content_start + end_pos];
    // The END marker may appear mid-line (e.g. "libbpf: -- END ...").
    // Trim back to the last newline to drop the partial prefix.
    let content = content
        .rfind('\n')
        .map(|p| &content[..p])
        .unwrap_or(content);
    Some(content.trim_end_matches('\n'))
}

/// Format verifier results as text: brief lines per program and collapsed
/// logs.
pub fn format_verifier_output(label: &str, result: &VerifierVmResult, raw: bool) -> String {
    let mut out = String::new();
    out.push_str(&format!("\n{label}\n"));
    if result.timed_out {
        out.push_str("  scheduler: UNKNOWN — VM timed out before exit\n");
    } else {
        match result.attach.failure_reason() {
            None => {
                out.push_str("  scheduler: attached (sched_ext enabled)\n");
                if result.dispatched {
                    out.push_str("  dispatch: confirmed (injected workload ran)\n");
                } else {
                    out.push_str(
                        "  dispatch: NOT CONFIRMED — attached but injected workload made no progress\n",
                    );
                }
            }
            Some(reason) => out.push_str(&format!("  scheduler: NOT ATTACHED — {reason}\n")),
        }
    }
    for ps in &result.stats {
        out.push_str(&format!(
            "  {:<40} verified_insns={}\n",
            ps.name, ps.verified_insns
        ));
    }

    if !result.scheduler_log.is_empty() {
        // Extract the verifier log from between libbpf's markers.
        // Falls back to the full scheduler_log when no markers exist.
        let verifier_log =
            extract_verifier_log(&result.scheduler_log).unwrap_or(&result.scheduler_log);

        let vs = parse_verifier_stats(verifier_log);
        if vs.processed_insns > 0 {
            out.push_str(&format!("\n{label} --- verifier stats ---\n"));
            out.push_str(&format!(
                "  processed={}  states={}/{}",
                vs.processed_insns, vs.peak_states, vs.total_states
            ));
            if let Some(t) = vs.time_usec {
                out.push_str(&format!("  time={t}us"));
            }
            if let Some(ref s) = vs.stack_depth {
                out.push_str(&format!("  stack={s}"));
            }
            out.push('\n');
        }

        out.push_str(&format!("\n{label} --- scheduler log ---\n"));
        if raw {
            out.push_str(&result.scheduler_log);
        } else {
            out.push_str(&collapse_cycles(verifier_log));
        }
    }

    out
}

/// Format an A/B diff table comparing two sets of verifier stats.
pub fn format_verifier_diff(
    label_a: &str,
    stats_a: &[ProgStats],
    label_b: &str,
    stats_b: &[ProgStats],
) -> String {
    let b_map = build_b_map(stats_b);
    let diff_rows = build_diff_rows(stats_a, &b_map);

    let mut out = String::new();
    out.push_str(&format!("\ndelta A/B diff: {label_a} vs {label_b}\n"));
    let mut table = crate::cli::new_table();
    table.set_header(vec!["program", "A", "B", "delta"]);
    for row in &diff_rows {
        table.add_row(vec![
            row.name.clone(),
            row.a.to_string(),
            row.b.to_string(),
            format!("{:+}", row.delta),
        ]);
    }
    out.push_str(&table.to_string());
    out.push('\n');
    out
}

// ---------------------------------------------------------------------------
// Per-cell PASS/FAIL result capture (for the run-summary table)
// ---------------------------------------------------------------------------

/// One `cargo ktstr verifier` cell's outcome. The cell process writes it
/// (via `write_cell_record`) into the directory named by
/// [`crate::KTSTR_VERIFIER_RESULT_DIR_ENV`]; after nextest returns the
/// dispatcher reads them back (via [`read_cell_records`]) and renders the
/// per-(topology × scheduler) summary table. A cell is one
/// (scheduler, kernel, topology): the verifier sweeps each declared
/// scheduler across topologies, so topology IS a result axis — a
/// scheduler can pass on one topology and fail on another.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct VerifierCellRecord {
    /// Declared scheduler name (the `<sched>` cell-name segment).
    pub scheduler: String,
    /// Sanitized kernel label (the `<kernel>` cell-name segment).
    pub kernel: String,
    /// Gauntlet topology preset (the `<preset>` cell-name segment).
    pub topology: String,
    /// Whether the cell passed (exit 0 from the cell handler).
    pub passed: bool,
    /// Per-program stats (program name + its `verified_insns` count)
    /// captured for this cell, copied from the VM run's
    /// [`VerifierVmResult::stats`]. Empty when the cell failed before
    /// producing stats. Drives the per-scheduler `verified_insns` tables
    /// ([`render_instruction_count_tables`], rows = kernel, cols = BPF
    /// program, cell = the count summarized across topologies) that the
    /// dispatcher prints before the PASS/FAIL grid.
    pub stats: Vec<ProgStats>,
}

/// Map a cell's full name to a filesystem-safe record filename: every
/// non-alphanumeric byte becomes `_`. The input is the cell's full name
/// (`verifier/<sched>/<kernel>/<preset>`, unique per cell), so the
/// mapping is unique per cell — a nextest RETRY of the same cell overwrites its
/// own prior record, so the FINAL attempt's outcome wins (a cell that
/// failed then passed on retry records PASS).
fn cell_record_filename(full_name: &str) -> String {
    let mut s: String = full_name
        .chars()
        .map(|c| if c.is_ascii_alphanumeric() { c } else { '_' })
        .collect();
    s.push_str(".json");
    s
}

/// Write a cell's PASS/FAIL record into `dir`. Parses `full_name`
/// (`verifier/<sched>/<kernel>/<preset>`); a name that does not fit that
/// shape is skipped (the cell already errored on the malformed name).
/// Best-effort: a write failure is logged and swallowed — the summary
/// table is a convenience over the per-cell nextest output, so a lost
/// record must never turn an otherwise-passing cell into a failure.
///
/// `pub(crate)`: the only writer is the cell handler in
/// `test_support::dispatch` (same crate); the reader side
/// ([`read_cell_records`] / [`render_result_table`]) is `pub` for the
/// `cargo-ktstr` binary crate.
pub(crate) fn write_cell_record(
    dir: &std::path::Path,
    full_name: &str,
    passed: bool,
    stats: &[ProgStats],
) {
    let Some(rest) = full_name.strip_prefix("verifier/") else {
        return;
    };
    let parts: Vec<&str> = rest.splitn(3, '/').collect();
    if parts.len() != 3 {
        return;
    }
    let record = VerifierCellRecord {
        scheduler: parts[0].to_string(),
        kernel: parts[1].to_string(),
        topology: parts[2].to_string(),
        passed,
        stats: stats.to_vec(),
    };
    let path = dir.join(cell_record_filename(full_name));
    match serde_json::to_vec(&record) {
        Ok(bytes) => {
            if let Err(e) = std::fs::write(&path, bytes) {
                eprintln!(
                    "ktstr verifier: warning: could not write result record {}: {e}",
                    path.display(),
                );
            }
        }
        Err(e) => eprintln!("ktstr verifier: warning: serialize result record: {e}"),
    }
}

/// Read every `*.json` cell record under `dir` (non-recursive). A missing
/// dir or an unparseable record is skipped (best-effort). Returns records
/// in filesystem-iteration order; [`render_result_table`] sorts for a
/// deterministic render.
pub fn read_cell_records(dir: &std::path::Path) -> Vec<VerifierCellRecord> {
    let Ok(entries) = std::fs::read_dir(dir) else {
        return Vec::new();
    };
    entries
        .flatten()
        .filter(|e| {
            e.path()
                .extension()
                .is_some_and(|x| x.eq_ignore_ascii_case("json"))
        })
        .filter_map(|e| std::fs::read(e.path()).ok())
        .filter_map(|bytes| serde_json::from_slice::<VerifierCellRecord>(&bytes).ok())
        .collect()
}

/// Decide the `cargo ktstr verifier` run outcome from nextest's exit
/// success, whether any per-cell records were produced, and the optional
/// `--scheduler` filter.
///
/// The dispatcher runs nextest with `--no-tests=pass`, so a run that
/// selects zero verifier cells exits 0 (success) with empty records
/// rather than nextest's generic "no tests to run" (exit 4). That lets
/// the dispatcher diagnose the empty case itself instead of surfacing a
/// cryptic nextest exit:
/// - `--scheduler <NAME>` set + no records: the name is not a declared
///   scheduler, or no topology preset fits this host for it.
/// - no `--scheduler` + no records: no `declare_scheduler!` is linked into
///   a test binary the sweep sees, or every declared scheduler's
///   constraints rejected all topology presets on this host.
///
/// A genuine build/exec failure still fails nextest (exit non-zero, which
/// `--no-tests=pass` does not mask), surfaced via the `exit_code` arm.
pub fn classify_run_outcome(
    success: bool,
    records_empty: bool,
    scheduler: Option<&str>,
    exit_code: Option<i32>,
) -> Result<(), String> {
    if !success {
        let code = exit_code.map_or_else(|| "signal".to_string(), |c| c.to_string());
        return Err(format!("cargo nextest run exited with {code}"));
    }
    if records_empty {
        return Err(match scheduler {
            Some(name) => format!(
                "--scheduler {name:?}: matched no verifier cell — no declared BPF \
                 scheduler by that name, or no topology preset fits this host for \
                 it. Run `cargo ktstr verifier` with no --scheduler to see the \
                 swept set."
            ),
            None => "no verifier cells ran — no scheduler is declared via \
                 declare_scheduler! in a linked test binary, or every declared \
                 scheduler's constraints rejected all topology presets on this \
                 host."
                .to_string(),
        });
    }
    Ok(())
}

/// Build the `cargo nextest run` argument vector for the verifier sweep.
///
/// Load-bearing tokens:
/// - `--run-ignored all`: verifier cells are emitted ignore-gated, so
///   nextest skips them unless opted in.
/// - `--no-tests pass`: a zero-cell selection exits 0 (not nextest's
///   default exit-4 "no tests to run"), so [`classify_run_outcome`] can
///   emit a targeted diagnostic instead of a cryptic nextest exit.
/// - `-E 'test(/^verifier/) & !test(/^verifier::/)'`: the `verifier/...`
///   cells, excluding the verifier module's own `verifier::tests::*`.
///
/// `nextest_profile`, if set, becomes nextest's `--profile <NAME>`,
/// emitted before `forward` so a forwarded token cannot shadow it.
/// `forward` is the user's trailing cargo/nextest args, appended verbatim.
pub fn build_nextest_args(nextest_profile: Option<&str>, forward: &[String]) -> Vec<String> {
    let mut args = vec![
        "nextest".to_string(),
        "run".to_string(),
        "--run-ignored".to_string(),
        "all".to_string(),
        "--no-tests".to_string(),
        "pass".to_string(),
        "-E".to_string(),
        "test(/^verifier/) & !test(/^verifier::/)".to_string(),
    ];
    if let Some(np) = nextest_profile {
        args.push("--profile".to_string());
        args.push(np.to_string());
    }
    args.extend(forward.iter().cloned());
    args
}

/// Render the per-cell records into a summary grid: one row per topology
/// preset, one column per scheduler, aggregating across kernels. Each
/// cell is:
/// - ✅ every kernel that ran this (topology, scheduler) passed,
/// - ❌ every kernel that ran it failed,
/// - 🇽 mixed — at least one kernel passed AND at least one failed,
/// - `-` no kernel ran this (topology, scheduler) (e.g. the scheduler's
///   constraints rejected the preset).
///
/// After the grid, the specific failing `(scheduler, kernel, topology)`
/// combinations are listed so a ❌ or 🇽 cell can be drilled to the exact
/// kernel(s) that failed. Returns `None` for an empty record set (the
/// caller prints nothing).
///
/// Rows, columns, and the failing list are BTreeSet-sorted so the same
/// run renders the same output (shell-pipeline stable). The header line
/// carries a ✅/❌/🇽 tally counting grid cells.
pub fn render_result_table(records: &[VerifierCellRecord]) -> Option<String> {
    if records.is_empty() {
        return None;
    }
    use std::collections::{BTreeMap, BTreeSet};
    let mut schedulers: BTreeSet<String> = BTreeSet::new(); // columns
    let mut rows: BTreeSet<String> = BTreeSet::new(); // topology presets
    // (topology, scheduler) -> (passes, fails) aggregated across kernels.
    let mut agg: BTreeMap<(String, String), (u32, u32)> = BTreeMap::new();
    // Distinct failing (scheduler, kernel, topology) combinations.
    let mut failing: BTreeSet<(String, String, String)> = BTreeSet::new();
    for r in records {
        schedulers.insert(r.scheduler.clone());
        rows.insert(r.topology.clone());
        let counts = agg
            .entry((r.topology.clone(), r.scheduler.clone()))
            .or_insert((0, 0));
        if r.passed {
            counts.0 += 1;
        } else {
            counts.1 += 1;
            failing.insert((r.scheduler.clone(), r.kernel.clone(), r.topology.clone()));
        }
    }

    let (mut n_pass, mut n_fail, mut n_mixed) = (0usize, 0usize, 0usize);
    let mut table = crate::cli::new_table();
    let mut header: Vec<String> = vec!["topology".to_string()];
    for sched in &schedulers {
        header.push(sched.clone());
    }
    table.set_header(header);
    for topo in &rows {
        let mut line: Vec<String> = vec![topo.clone()];
        for sched in &schedulers {
            // An entry only exists with >= 1 record, so (_, 0) means all
            // passed, (0, _) means all failed, and both-nonzero is mixed.
            let text = match agg.get(&(topo.clone(), sched.clone())) {
                None => "-",
                Some((_, 0)) => {
                    n_pass += 1;
                    "✅"
                }
                Some((0, _)) => {
                    n_fail += 1;
                    "❌"
                }
                Some(_) => {
                    n_mixed += 1;
                    "🇽"
                }
            };
            line.push(text.to_string());
        }
        table.add_row(line);
    }

    let mut out = format!("\nverifier summary: {n_pass} ✅  {n_fail} ❌  {n_mixed} 🇽\n{table}\n");
    if !failing.is_empty() {
        out.push_str("\nfailing combinations (scheduler / kernel / topology):\n");
        for (sched, kernel, topo) in &failing {
            out.push_str(&format!("  {sched} / {kernel} / {topo}\n"));
        }
    }
    Some(out)
}

/// Render one `verified_insns` table per declared scheduler. Within each
/// scheduler's section: rows = kernel version, columns = BPF program, and
/// each cell is that program's `verified_insns` for the (scheduler,
/// kernel) summarized ACROSS the topologies that ran it — a single number
/// when topology-invariant, `lo..hi` when it varies (`-` when that program
/// reported no stats on that kernel).
///
/// `verified_insns` is the verifier's PROCESSED-instruction count
/// (`env->insn_processed`) — fixed per load, but NOT topology-invariant
/// (a scheduler whose verification path depends on topology-derived
/// `.rodata`, e.g. `nr_cpus`, processes a different count per topology).
/// So topology is folded into the cell as a range rather than shown as its
/// own (usually all-identical) axis; the axes it genuinely varies on — BPF
/// program (x) and kernel version (y) — are the table axes, sectioned per
/// declared scheduler. Identical-binary declarations are sectioned
/// separately on purpose (they are run separately). Returns `None` when no
/// record carries any per-program stats (the caller prints nothing).
///
/// Schedulers, kernels, and programs are BTree-sorted so the same run
/// renders the same output (shell-pipeline stable). The range drops which
/// topology produced which count; a per-topology breakdown is a separate
/// detailed view, not this summary.
pub fn render_instruction_count_tables(records: &[VerifierCellRecord]) -> Option<String> {
    use std::collections::{BTreeMap, BTreeSet};
    // scheduler -> kernel -> program -> (min, max) verified_insns across
    // the topologies that ran it. Topology is folded into the (min, max)
    // range: a flat scheduler has min == max (one number), a
    // topology-sensitive one has min < max (`lo..hi`).
    type VerifiedInsnSpans = BTreeMap<String, BTreeMap<String, BTreeMap<String, (u32, u32)>>>;
    let mut by_sched: VerifiedInsnSpans = BTreeMap::new();
    // scheduler -> the union of program names it reported (the columns).
    let mut sched_progs: BTreeMap<String, BTreeSet<String>> = BTreeMap::new();
    for r in records {
        for s in &r.stats {
            let span = by_sched
                .entry(r.scheduler.clone())
                .or_default()
                .entry(r.kernel.clone())
                .or_default()
                .entry(s.name.clone())
                .or_insert((s.verified_insns, s.verified_insns));
            span.0 = span.0.min(s.verified_insns);
            span.1 = span.1.max(s.verified_insns);
            sched_progs
                .entry(r.scheduler.clone())
                .or_default()
                .insert(s.name.clone());
        }
    }
    if by_sched.is_empty() {
        return None;
    }

    let mut out = String::from(
        "\nverifier verified_insns (per scheduler; rows: kernel, cols: BPF program, \
         cell: range across topologies):\n",
    );
    for (sched, kernels) in &by_sched {
        let progs = &sched_progs[sched];
        let mut table = crate::cli::new_table();
        let mut header: Vec<String> = vec!["kernel".to_string()];
        for p in progs {
            header.push(p.clone());
        }
        table.set_header(header);
        for (kernel, prog_map) in kernels {
            let mut line: Vec<String> = vec![kernel.clone()];
            for p in progs {
                let text = match prog_map.get(p) {
                    Some((lo, hi)) if lo == hi => lo.to_string(),
                    Some((lo, hi)) => format!("{lo}..{hi}"),
                    None => "-".to_string(),
                };
                line.push(text);
            }
            table.add_row(line);
        }
        out.push_str(&format!("\n{sched}:\n{table}\n"));
    }
    Some(out)
}

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

    // -----------------------------------------------------------------------
    // per-cell result capture + summary table
    // -----------------------------------------------------------------------

    /// A malformed cell name is skipped (no record); a well-formed
    /// 3-segment cell records (scheduler/kernel/topology); and a nextest
    /// RETRY of the same cell overwrites its own prior record so the
    /// FINAL outcome wins (fail-then-pass -> PASS).
    #[test]
    fn cell_record_write_read_roundtrip_and_retry_overwrites() {
        let dir = std::env::temp_dir().join(format!("ktstr-verif-rec-{}", std::process::id()));
        std::fs::create_dir_all(&dir).expect("mk temp dir");
        // Malformed: no verifier/ prefix, and a 2-segment name (no
        // <preset> after the kernel) — both skipped.
        write_cell_record(&dir, "not_a_cell", true, &[]);
        write_cell_record(&dir, "verifier/only/two", true, &[]);
        // Well-formed cell: fail, then a retry passes -> overwrites.
        let name = "verifier/scx_a/kernel_6_14/tiny-1llc";
        write_cell_record(&dir, name, false, &[]);
        // The retry passes and carries per-program verified_insns, so the
        // final record has both the PASS outcome and the stats.
        let stats = [
            ProgStats {
                name: "ktstr_dispatch".into(),
                verified_insns: 321,
            },
            ProgStats {
                name: "ktstr_enqueue".into(),
                verified_insns: 123,
            },
        ];
        write_cell_record(&dir, name, true, &stats);
        let recs = read_cell_records(&dir);
        assert_eq!(
            recs.len(),
            1,
            "malformed names skipped; the retry overwrote its own record (one file): {recs:?}",
        );
        assert_eq!(recs[0].scheduler, "scx_a");
        assert_eq!(recs[0].kernel, "kernel_6_14");
        assert_eq!(recs[0].topology, "tiny-1llc");
        assert!(
            recs[0].passed,
            "final retry outcome (PASS) wins over the earlier FAIL"
        );
        // Per-program verified_insns survive the JSON roundtrip and reflect
        // the final (retry) write, not the earlier stat-less fail.
        assert_eq!(recs[0].stats, stats, "stats roundtrip via serde");
        std::fs::remove_dir_all(&dir).ok();
    }

    /// The summary grid aggregates across kernels per (topology,
    /// scheduler): all-pass -> ✅, all-fail -> ❌, with a ✅/❌/🇽 tally.
    /// Failing (scheduler, kernel, topology) combinations are listed after
    /// the grid; an empty record set renders nothing.
    #[test]
    fn render_result_table_matrix_tally_and_empty() {
        let recs = vec![
            VerifierCellRecord {
                scheduler: "scx_a".into(),
                kernel: "kernel_6_14".into(),
                topology: "tiny-1llc".into(),
                passed: true,
                stats: vec![],
            },
            VerifierCellRecord {
                scheduler: "scx_a".into(),
                kernel: "kernel_6_14".into(),
                topology: "large-4llc".into(),
                passed: false,
                stats: vec![],
            },
        ];
        let out = render_result_table(&recs).expect("non-empty records -> Some");
        assert!(
            out.contains("verifier summary: 1 ✅  1 ❌  0 🇽"),
            "tally: {out}"
        );
        // Columns are scheduler-only (kernels fold into the cell), so no
        // `scheduler @ kernel` labeling appears.
        assert!(
            out.contains("scx_a") && !out.contains(" @ "),
            "columns: {out}"
        );
        // The emoji must render in the GRID CELLS, not merely the tally
        // line: locate each topology's row and assert its cell glyph
        // (neither topology name appears on the `verifier summary:` line).
        let pass_row = out
            .lines()
            .find(|l| l.contains("tiny-1llc"))
            .expect("tiny-1llc row present");
        assert!(
            pass_row.contains('✅'),
            "all-pass cell renders ✅ in the grid row: {pass_row}"
        );
        let fail_row = out
            .lines()
            .find(|l| l.contains("large-4llc"))
            .expect("large-4llc row present");
        assert!(
            fail_row.contains('❌'),
            "all-fail cell renders ❌ in the grid row: {fail_row}"
        );
        // The single failure is listed after the grid.
        assert!(
            out.contains("failing combinations (scheduler / kernel / topology):")
                && out.contains("scx_a / kernel_6_14 / large-4llc"),
            "failing combinations listed: {out}"
        );
        assert!(render_result_table(&[]).is_none(), "empty -> None");
    }

    /// A (topology, scheduler) where one kernel passes and another fails
    /// renders 🇽 (mixed); an all-pass (topology, scheduler) across kernels
    /// renders ✅. Only the failing kernel appears in the failing list.
    #[test]
    fn render_result_table_mixed_kernels_blue_x() {
        let recs = vec![
            // tiny-1llc / scx_a: 6_14 passes, 6_15 fails -> mixed -> 🇽.
            VerifierCellRecord {
                scheduler: "scx_a".into(),
                kernel: "kernel_6_14".into(),
                topology: "tiny-1llc".into(),
                passed: true,
                stats: vec![],
            },
            VerifierCellRecord {
                scheduler: "scx_a".into(),
                kernel: "kernel_6_15".into(),
                topology: "tiny-1llc".into(),
                passed: false,
                stats: vec![],
            },
            // smt-2llc / scx_a: both kernels pass -> ✅.
            VerifierCellRecord {
                scheduler: "scx_a".into(),
                kernel: "kernel_6_14".into(),
                topology: "smt-2llc".into(),
                passed: true,
                stats: vec![],
            },
            VerifierCellRecord {
                scheduler: "scx_a".into(),
                kernel: "kernel_6_15".into(),
                topology: "smt-2llc".into(),
                passed: true,
                stats: vec![],
            },
        ];
        let out = render_result_table(&recs).expect("Some");
        assert!(
            out.contains("verifier summary: 1 ✅  0 ❌  1 🇽"),
            "tally counts one all-pass + one mixed cell: {out}"
        );
        let mixed_row = out
            .lines()
            .find(|l| l.contains("tiny-1llc"))
            .expect("tiny-1llc row present");
        assert!(
            mixed_row.contains('🇽'),
            "mixed (some pass, some fail) cell renders 🇽: {mixed_row}"
        );
        let pass_row = out
            .lines()
            .find(|l| l.contains("smt-2llc"))
            .expect("smt-2llc row present");
        assert!(
            pass_row.contains('✅'),
            "all-kernels-pass cell renders ✅: {pass_row}"
        );
        // Only the failing kernel is listed after the grid.
        assert!(
            out.contains("scx_a / kernel_6_15 / tiny-1llc"),
            "the failing kernel is listed: {out}"
        );
        assert!(
            !out.contains("kernel_6_14 / tiny-1llc"),
            "the passing kernel on the mixed topology is not listed: {out}"
        );
        assert!(
            !out.contains("/ smt-2llc"),
            "the all-pass topology contributes no failing combination: {out}"
        );
    }

    /// Per-scheduler verified_insns tables: one section per declared
    /// scheduler; within it rows = kernel version, columns = BPF program,
    /// each cell that program's verified_insns across the topologies that
    /// ran it — a single number when topology-invariant, `lo..hi` when it
    /// varies. A (kernel, program) that reported no stats shows `-`; an
    /// empty record set renders nothing.
    #[test]
    fn instruction_count_tables_per_scheduler_kernel_program_range() {
        let recs = vec![
            // scx_a / kernel_6_14 on two topologies with IDENTICAL counts
            // -> the cell collapses to a single number (topology-flat).
            VerifierCellRecord {
                scheduler: "scx_a".into(),
                kernel: "kernel_6_14".into(),
                topology: "tiny".into(),
                passed: true,
                stats: vec![
                    ProgStats {
                        name: "ktstr_dispatch".into(),
                        verified_insns: 128,
                    },
                    ProgStats {
                        name: "ktstr_enqueue".into(),
                        verified_insns: 64,
                    },
                ],
            },
            VerifierCellRecord {
                scheduler: "scx_a".into(),
                kernel: "kernel_6_14".into(),
                topology: "large".into(),
                passed: true,
                stats: vec![
                    ProgStats {
                        name: "ktstr_dispatch".into(),
                        verified_insns: 128,
                    },
                    ProgStats {
                        name: "ktstr_enqueue".into(),
                        verified_insns: 64,
                    },
                ],
            },
            // scx_a / kernel_6_15: ktstr_dispatch DIFFERS across topologies
            // -> `lo..hi` range; ktstr_enqueue is absent on this kernel
            // -> `-` in that column's kernel_6_15 row.
            VerifierCellRecord {
                scheduler: "scx_a".into(),
                kernel: "kernel_6_15".into(),
                topology: "tiny".into(),
                passed: true,
                stats: vec![ProgStats {
                    name: "ktstr_dispatch".into(),
                    verified_insns: 130,
                }],
            },
            VerifierCellRecord {
                scheduler: "scx_a".into(),
                kernel: "kernel_6_15".into(),
                topology: "large".into(),
                passed: true,
                stats: vec![ProgStats {
                    name: "ktstr_dispatch".into(),
                    verified_insns: 150,
                }],
            },
            // scx_b: a separate section (its own declaration).
            VerifierCellRecord {
                scheduler: "scx_b".into(),
                kernel: "kernel_6_14".into(),
                topology: "tiny".into(),
                passed: true,
                stats: vec![ProgStats {
                    name: "ktstr_dispatch".into(),
                    verified_insns: 200,
                }],
            },
        ];
        let out = render_instruction_count_tables(&recs).expect("stats present -> Some");
        // One section per declared scheduler.
        assert!(
            out.contains("scx_a:") && out.contains("scx_b:"),
            "one section per declared scheduler: {out}"
        );
        // Columns = BPF programs; rows = kernel version.
        assert!(
            out.contains("ktstr_dispatch") && out.contains("ktstr_enqueue"),
            "BPF-program columns: {out}"
        );
        assert!(
            out.contains("kernel_6_14") && out.contains("kernel_6_15"),
            "kernel-version rows: {out}"
        );
        // Topology folded into the cell as a range: flat -> "128",
        // varies across topologies -> "130..150".
        assert!(
            out.contains("128"),
            "topology-flat cell is a single number: {out}"
        );
        assert!(
            out.contains("130..150"),
            "topology-varying cell is a lo..hi range: {out}"
        );
        assert!(
            out.contains("64") && out.contains("200"),
            "other counts render: {out}"
        );
        // ktstr_enqueue reported no stats on kernel_6_15 -> `-`.
        assert!(
            out.contains('-'),
            "a (kernel, program) with no stats renders '-': {out}"
        );
        // Topology is NOT a table axis (folded into the range), so no
        // topology label appears in the output.
        assert!(
            !out.contains("tiny") && !out.contains("large"),
            "topology is not a table axis: {out}"
        );

        // No record carries stats -> nothing to render.
        let bare = vec![VerifierCellRecord {
            scheduler: "scx_a".into(),
            kernel: "kernel_6_14".into(),
            topology: "tiny".into(),
            passed: false,
            stats: vec![],
        }];
        assert!(
            render_instruction_count_tables(&bare).is_none(),
            "no stats -> None"
        );
    }

    /// classify_run_outcome: a build/exec failure surfaces nextest's exit;
    /// a successful-but-empty run is diagnosed by the dispatcher (friendly
    /// no-such-scheduler message when --scheduler was set, no-cells-ran
    /// message otherwise); a successful run with records is Ok.
    #[test]
    fn classify_run_outcome_cases() {
        // Records present + success -> Ok regardless of --scheduler.
        assert!(classify_run_outcome(true, false, None, Some(0)).is_ok());
        assert!(classify_run_outcome(true, false, Some("ktstr_sched"), Some(0)).is_ok());

        // Success + empty + --scheduler -> friendly "no such scheduler".
        // Reachable ONLY because --no-tests=pass turns a 0-cell match into
        // exit 0; under the old `auto` default a 0-match exited 4 -> the
        // failure arm, leaving this message dead.
        let e = classify_run_outcome(true, true, Some("nope"), Some(0)).unwrap_err();
        assert!(
            e.contains("--scheduler \"nope\"") && e.contains("matched no verifier cell"),
            "scheduler-empty diagnostic: {e}"
        );

        // Success + empty + no --scheduler -> "no cells ran" diagnosis
        // (must NOT silently succeed under --no-tests=pass).
        let e = classify_run_outcome(true, true, None, Some(0)).unwrap_err();
        assert!(
            e.contains("no verifier cells ran") && e.contains("declare_scheduler!"),
            "no-cells diagnostic: {e}"
        );

        // Failure surfaces nextest's exit code; a signal (no code) renders
        // as "signal".
        assert_eq!(
            classify_run_outcome(false, true, None, Some(4)).unwrap_err(),
            "cargo nextest run exited with 4"
        );
        assert_eq!(
            classify_run_outcome(false, false, Some("x"), None).unwrap_err(),
            "cargo nextest run exited with signal"
        );
    }

    /// build_nextest_args carries the flags that make the friendly
    /// diagnostic reachable: `--run-ignored all` (cells are ignore-gated)
    /// and `--no-tests pass` (a 0-cell selection exits 0 so
    /// classify_run_outcome runs instead of nextest's exit-4). Guards
    /// against a future edit silently dropping either, plus the
    /// profile-before-forwarded-args ordering.
    #[test]
    fn build_nextest_args_carries_load_bearing_flags() {
        let args = build_nextest_args(None, &[]);
        let ri = args
            .iter()
            .position(|a| a == "--run-ignored")
            .expect("--run-ignored present");
        assert_eq!(args[ri + 1], "all", "--run-ignored all");
        let nt = args
            .iter()
            .position(|a| a == "--no-tests")
            .expect("--no-tests present");
        assert_eq!(args[nt + 1], "pass", "--no-tests pass");
        assert!(
            args.iter()
                .any(|a| a == "test(/^verifier/) & !test(/^verifier::/)"),
            "verifier-cell filter present: {args:?}"
        );

        // --profile <NAME> is emitted before forwarded args so a forwarded
        // token cannot shadow it.
        let args = build_nextest_args(Some("ci"), &["--features".to_string(), "wprof".to_string()]);
        let p = args
            .iter()
            .position(|a| a == "--profile")
            .expect("--profile present");
        assert_eq!(args[p + 1], "ci");
        let f = args
            .iter()
            .position(|a| a == "--features")
            .expect("forwarded --features present");
        assert!(p < f, "profile emitted before forwarded args: {args:?}");
    }

    // -----------------------------------------------------------------------
    // scheduler attach verdict
    // -----------------------------------------------------------------------

    /// attach_outcome_from_messages positive-confirmation rule:
    /// Died > NotAttached > PayloadStarting(=Attached) > Unconfirmed.
    /// Absence of a positive PayloadStarting frame is Unconfirmed (FAIL),
    /// NOT a blind pass — this is what catches an early guest panic that
    /// reboots emitting no frame. Corrupt / empty / non-LIFECYCLE /
    /// unknown frames are skipped.
    #[test]
    fn attach_outcome_from_lifecycle_frames() {
        use crate::vmm::host_comms::BulkDrainResult;
        use crate::vmm::wire::{LifecyclePhase, MSG_TYPE_LIFECYCLE, ShmEntry};

        let frame = |phase: LifecyclePhase, reason: &str| -> ShmEntry {
            let mut payload = vec![phase.wire_value()];
            payload.extend_from_slice(reason.as_bytes());
            ShmEntry {
                msg_type: MSG_TYPE_LIFECYCLE,
                payload,
                crc_ok: true,
            }
        };
        let drain = |entries: Vec<ShmEntry>| BulkDrainResult { entries };

        // No frames at all -> Unconfirmed (guest vanished before any
        // phase; e.g. an early kernel panic that reboots via panic=-1).
        assert_eq!(
            attach_outcome_from_messages(None),
            AttachOutcome::Unconfirmed,
        );

        // Reached init but NOT the dispatch phase -> Unconfirmed.
        let init_only = drain(vec![frame(LifecyclePhase::InitStarted, "")]);
        assert_eq!(
            attach_outcome_from_messages(Some(&init_only)),
            AttachOutcome::Unconfirmed,
        );

        // Reached the dispatch phase (PayloadStarting), no failure -> Attached.
        let progress = drain(vec![
            frame(LifecyclePhase::InitStarted, ""),
            frame(LifecyclePhase::PayloadStarting, ""),
        ]);
        assert_eq!(
            attach_outcome_from_messages(Some(&progress)),
            AttachOutcome::Attached,
        );

        // SchedulerNotAttached carries its reason suffix verbatim.
        let not_attached = drain(vec![frame(LifecyclePhase::SchedulerNotAttached, "timeout")]);
        assert_eq!(
            attach_outcome_from_messages(Some(&not_attached)),
            AttachOutcome::NotAttached("timeout".to_string()),
        );

        // A failure frame wins over a (defensively) co-present
        // PayloadStarting.
        let fail_beats_positive = drain(vec![
            frame(LifecyclePhase::PayloadStarting, ""),
            frame(LifecyclePhase::SchedulerNotAttached, "sysfs absent"),
        ]);
        assert_eq!(
            attach_outcome_from_messages(Some(&fail_beats_positive)),
            AttachOutcome::NotAttached("sysfs absent".to_string()),
        );

        // SchedulerDied wins over NotAttached, in BOTH orders.
        for entries in [
            vec![
                frame(LifecyclePhase::SchedulerNotAttached, "timeout"),
                frame(LifecyclePhase::SchedulerDied, ""),
            ],
            vec![
                frame(LifecyclePhase::SchedulerDied, ""),
                frame(LifecyclePhase::SchedulerNotAttached, "timeout"),
            ],
        ] {
            let d = drain(entries);
            assert_eq!(attach_outcome_from_messages(Some(&d)), AttachOutcome::Died);
        }

        // Died wins even over a PayloadStarting.
        let died_beats_positive = drain(vec![
            frame(LifecyclePhase::PayloadStarting, ""),
            frame(LifecyclePhase::SchedulerDied, ""),
        ]);
        assert_eq!(
            attach_outcome_from_messages(Some(&died_beats_positive)),
            AttachOutcome::Died,
        );

        // Skipped frames (corrupt crc / empty payload / non-LIFECYCLE /
        // unknown discriminant) must NOT suppress a real PayloadStarting:
        // pairing each with a valid PayloadStarting must still resolve
        // Attached — proving the frame was skipped, not acted on (a
        // corrupt/non-LIFECYCLE Died byte would otherwise force Died).
        let corrupt_died = ShmEntry {
            msg_type: MSG_TYPE_LIFECYCLE,
            payload: vec![LifecyclePhase::SchedulerDied.wire_value()],
            crc_ok: false,
        };
        let empty = ShmEntry {
            msg_type: MSG_TYPE_LIFECYCLE,
            payload: Vec::new(),
            crc_ok: true,
        };
        let non_lifecycle_died = ShmEntry {
            msg_type: MSG_TYPE_LIFECYCLE + 1,
            payload: vec![LifecyclePhase::SchedulerDied.wire_value()],
            crc_ok: true,
        };
        let unknown_phase = ShmEntry {
            msg_type: MSG_TYPE_LIFECYCLE,
            payload: vec![250],
            crc_ok: true,
        };
        for skipped in [corrupt_died, empty, non_lifecycle_died, unknown_phase] {
            let d = drain(vec![skipped, frame(LifecyclePhase::PayloadStarting, "")]);
            assert_eq!(
                attach_outcome_from_messages(Some(&d)),
                AttachOutcome::Attached,
                "a skipped frame must not suppress a valid PayloadStarting",
            );
        }
    }

    /// dispatch_confirmed_from_messages: true only when a crc-ok,
    /// non-empty WorkloadDispatched frame is present; false for None / no
    /// such frame / corrupt / empty / non-LIFECYCLE.
    #[test]
    fn dispatch_confirmed_from_lifecycle_frames() {
        use crate::vmm::host_comms::BulkDrainResult;
        use crate::vmm::wire::{LifecyclePhase, MSG_TYPE_LIFECYCLE, ShmEntry};

        let frame = |phase: LifecyclePhase| -> ShmEntry {
            ShmEntry {
                msg_type: MSG_TYPE_LIFECYCLE,
                payload: vec![phase.wire_value()],
                crc_ok: true,
            }
        };
        let drain = |entries: Vec<ShmEntry>| BulkDrainResult { entries };

        // No frames at all -> false.
        assert!(!dispatch_confirmed_from_messages(None));

        // PayloadStarting but no WorkloadDispatched -> false (attached,
        // never dispatched).
        let attached_only = drain(vec![frame(LifecyclePhase::PayloadStarting)]);
        assert!(!dispatch_confirmed_from_messages(Some(&attached_only)));

        // WorkloadDispatched present -> true.
        let dispatched = drain(vec![
            frame(LifecyclePhase::PayloadStarting),
            frame(LifecyclePhase::WorkloadDispatched),
        ]);
        assert!(dispatch_confirmed_from_messages(Some(&dispatched)));

        // Corrupt crc / empty payload / non-LIFECYCLE WorkloadDispatched
        // frames are skipped and must not confirm dispatch.
        let corrupt = ShmEntry {
            msg_type: MSG_TYPE_LIFECYCLE,
            payload: vec![LifecyclePhase::WorkloadDispatched.wire_value()],
            crc_ok: false,
        };
        let empty = ShmEntry {
            msg_type: MSG_TYPE_LIFECYCLE,
            payload: Vec::new(),
            crc_ok: true,
        };
        let non_lifecycle = ShmEntry {
            msg_type: MSG_TYPE_LIFECYCLE + 1,
            payload: vec![LifecyclePhase::WorkloadDispatched.wire_value()],
            crc_ok: true,
        };
        for skipped in [corrupt, empty, non_lifecycle] {
            let d = drain(vec![skipped]);
            assert!(
                !dispatch_confirmed_from_messages(Some(&d)),
                "a corrupt/empty/non-LIFECYCLE frame must not confirm dispatch",
            );
        }
    }

    /// VerifierVmResult::cell_verdict gate order + messages: timed_out >
    /// attach failure > dispatch failure > PASS.
    #[test]
    fn cell_verdict_gate_order_and_messages() {
        let base = |attach: AttachOutcome, dispatched: bool, timed_out: bool| VerifierVmResult {
            stats: Vec::new(),
            scheduler_log: String::new(),
            attach,
            dispatched,
            timed_out,
        };

        // Verified + attached + dispatched, no hang -> PASS.
        assert_eq!(
            base(AttachOutcome::Attached, true, false).cell_verdict(),
            Ok(()),
        );

        // Attached but 0-dispatch -> FAIL naming the dispatch gate.
        let no_dispatch = base(AttachOutcome::Attached, false, false).cell_verdict();
        assert!(
            no_dispatch
                .as_ref()
                .unwrap_err()
                .contains("did not dispatch"),
            "dispatch gate must name the failure: {no_dispatch:?}",
        );

        // Attach failure -> FAIL naming attach, even if dispatched is
        // (defensively) true.
        let attach_fail = base(AttachOutcome::Died, true, false).cell_verdict();
        assert!(
            attach_fail
                .as_ref()
                .unwrap_err()
                .contains("did not turn on"),
            "attach gate must win over dispatch: {attach_fail:?}",
        );

        // timed_out wins over everything, even a clean attach + dispatch.
        let hung = base(AttachOutcome::Attached, true, true).cell_verdict();
        assert!(
            hung.as_ref().unwrap_err().contains("timed out"),
            "timed_out must win: {hung:?}",
        );

        // Attach failure outranks a co-present dispatch failure (root
        // cause reported first).
        let both = base(AttachOutcome::Died, false, false).cell_verdict();
        assert!(
            both.as_ref().unwrap_err().contains("did not turn on"),
            "attach failure reported before dispatch failure: {both:?}",
        );
    }

    /// AttachOutcome::failure_reason surfaces (None when attached, the
    /// distinct Died / NotAttached reasons otherwise).
    #[test]
    fn attach_outcome_failure_reason() {
        assert_eq!(AttachOutcome::Attached.failure_reason(), None);
        assert!(
            AttachOutcome::Died
                .failure_reason()
                .unwrap()
                .contains("exited during BPF load"),
        );
        assert!(
            AttachOutcome::NotAttached(String::new())
                .failure_reason()
                .unwrap()
                .contains("never reached sched_ext 'enabled'"),
        );
        assert_eq!(
            AttachOutcome::NotAttached("sysfs absent".to_string()).failure_reason(),
            Some("scheduler never reached sched_ext 'enabled': sysfs absent".to_string()),
        );
        assert!(
            AttachOutcome::Unconfirmed
                .failure_reason()
                .unwrap()
                .contains("attach unconfirmed"),
        );
    }

    /// A timed-out run shows UNKNOWN (not attached), overriding the
    /// attach line — catches a post-attach teardown hang (the frame scan
    /// + Unconfirmed already handle a guest that vanishes before attach).
    #[test]
    fn format_verifier_output_timed_out_shows_unknown() {
        let result = VerifierVmResult {
            stats: Vec::new(),
            scheduler_log: String::new(),
            attach: AttachOutcome::Attached,
            dispatched: false,
            timed_out: true,
        };
        let out = format_verifier_output("verifier", &result, false);
        assert!(
            out.contains("scheduler: UNKNOWN — VM timed out"),
            "timed-out run must show UNKNOWN: {out}",
        );
        assert!(
            !out.contains("scheduler: attached"),
            "timed-out run must not claim attached: {out}",
        );
    }

    /// An attached-but-not-dispatched run shows the attach line AND a
    /// "dispatch: NOT CONFIRMED" line — the signal that the scheduler
    /// turned on but never dispatched the injected workload.
    /// Guards the `format_verifier_output` dispatched==false render branch,
    /// which the snapshot tests (dispatched==true / Died) do not reach.
    #[test]
    fn format_verifier_output_attached_not_dispatched_shows_not_confirmed() {
        let result = VerifierVmResult {
            stats: Vec::new(),
            scheduler_log: String::new(),
            attach: AttachOutcome::Attached,
            dispatched: false,
            timed_out: false,
        };
        let out = format_verifier_output("verifier", &result, false);
        assert!(
            out.contains("scheduler: attached"),
            "attached run must show the attach line: {out}",
        );
        assert!(
            out.contains("dispatch: NOT CONFIRMED"),
            "attached-but-not-dispatched must render the NOT CONFIRMED signal: {out}",
        );
    }

    // -----------------------------------------------------------------------
    // parse_verifier_stats
    // -----------------------------------------------------------------------

    #[test]
    fn parse_verifier_stats_full_line() {
        let log = "processed 1234 insns (limit 1000000) max_states_per_insn 5 total_states 200 peak_states 50 mark_read 10\nverification time 42 usec\nstack depth 32+0\n";
        let vs = parse_verifier_stats(log);
        assert_eq!(vs.processed_insns, 1234);
        assert_eq!(vs.total_states, 200);
        assert_eq!(vs.peak_states, 50);
        assert_eq!(vs.time_usec, Some(42));
        assert_eq!(vs.stack_depth.as_deref(), Some("32+0"));
    }

    #[test]
    fn parse_verifier_stats_insns_only() {
        let log = "processed 500 insns (limit 1000000) max_states_per_insn 1 total_states 10 peak_states 3 mark_read 0\n";
        let vs = parse_verifier_stats(log);
        assert_eq!(vs.processed_insns, 500);
        assert_eq!(vs.total_states, 10);
        assert_eq!(vs.peak_states, 3);
        assert!(vs.time_usec.is_none());
        assert!(vs.stack_depth.is_none());
    }

    #[test]
    fn parse_verifier_stats_empty() {
        let vs = parse_verifier_stats("");
        assert_eq!(vs.processed_insns, 0);
        assert_eq!(vs.total_states, 0);
        assert_eq!(vs.peak_states, 0);
        assert!(vs.time_usec.is_none());
        assert!(vs.stack_depth.is_none());
    }

    #[test]
    fn parse_verifier_stats_garbage_lines() {
        let log = "some random output\nnot a stats line\n";
        let vs = parse_verifier_stats(log);
        assert_eq!(vs.processed_insns, 0);
        assert_eq!(vs.total_states, 0);
        assert!(vs.time_usec.is_none());
    }

    #[test]
    fn parse_verifier_stats_time_without_insns() {
        let log = "verification time 100 usec\nstack depth 64\n";
        let vs = parse_verifier_stats(log);
        assert_eq!(vs.processed_insns, 0);
        assert_eq!(vs.time_usec, Some(100));
        assert_eq!(vs.stack_depth.as_deref(), Some("64"));
    }

    #[test]
    fn parse_verifier_stats_multi_subprogram_stack() {
        let log = "processed 42 insns (limit 1000000) max_states_per_insn 1 total_states 5 peak_states 2 mark_read 0\nstack depth 32+16+8\n";
        let vs = parse_verifier_stats(log);
        assert_eq!(vs.processed_insns, 42);
        assert_eq!(vs.stack_depth.as_deref(), Some("32+16+8"));
    }

    #[test]
    fn parse_verifier_stats_noise_between_lines() {
        let log = "\
libbpf: loading something
processed 999 insns (limit 1000000) max_states_per_insn 3 total_states 77 peak_states 20 mark_read 5
libbpf: prog 'dispatch': attached
verification time 7 usec
stack depth 48+0
";
        let vs = parse_verifier_stats(log);
        assert_eq!(vs.processed_insns, 999);
        assert_eq!(vs.total_states, 77);
        assert_eq!(vs.peak_states, 20);
        assert_eq!(vs.time_usec, Some(7));
        assert_eq!(vs.stack_depth.as_deref(), Some("48+0"));
    }

    #[test]
    fn parse_verifier_stats_partial_insns_line() {
        let log = "processed 123\n";
        let vs = parse_verifier_stats(log);
        assert_eq!(vs.processed_insns, 123);
        assert_eq!(vs.total_states, 0);
        assert_eq!(vs.peak_states, 0);
    }

    #[test]
    fn parse_verifier_stats_only_stack_depth() {
        let log = "stack depth 128\n";
        let vs = parse_verifier_stats(log);
        assert_eq!(vs.stack_depth.as_deref(), Some("128"));
        assert_eq!(vs.processed_insns, 0);
    }

    #[test]
    fn parse_verifier_stats_zero_insns() {
        let log = "processed 0 insns (limit 1000000) max_states_per_insn 0 total_states 0 peak_states 0 mark_read 0\n";
        let vs = parse_verifier_stats(log);
        assert_eq!(vs.processed_insns, 0);
        assert_eq!(vs.total_states, 0);
        assert_eq!(vs.peak_states, 0);
    }

    #[test]
    fn parse_verifier_stats_large_values() {
        let log = "processed 999999 insns (limit 1000000) max_states_per_insn 100 total_states 50000 peak_states 12345 mark_read 9999\nverification time 123456 usec\n";
        let vs = parse_verifier_stats(log);
        assert_eq!(vs.processed_insns, 999999);
        assert_eq!(vs.total_states, 50000);
        assert_eq!(vs.peak_states, 12345);
        assert_eq!(vs.time_usec, Some(123456));
    }

    #[test]
    fn parse_verifier_stats_stack_depth_single() {
        let log = "stack depth 64\n";
        let vs = parse_verifier_stats(log);
        assert_eq!(vs.stack_depth.as_deref(), Some("64"));
    }

    #[test]
    fn parse_verifier_stats_stack_depth_many_subprograms() {
        let log = "stack depth 32+16+8+0+0\n";
        let vs = parse_verifier_stats(log);
        assert_eq!(vs.stack_depth.as_deref(), Some("32+16+8+0+0"));
    }

    #[test]
    fn parse_verifier_stats_multiple_processed_lines_takes_last() {
        let log = "processed 100 insns (limit 1000000) max_states_per_insn 1 total_states 5 peak_states 2 mark_read 0\nprocessed 200 insns (limit 1000000) max_states_per_insn 2 total_states 10 peak_states 4 mark_read 0\n";
        let vs = parse_verifier_stats(log);
        assert_eq!(vs.processed_insns, 200);
        assert_eq!(vs.total_states, 10);
    }

    #[test]
    fn parse_verifier_stats_complexity_error_with_stats() {
        let log = "\
func#0 @0
0: R1=ctx() R10=fp0
1: (bf) r6 = r1                       ; R1=ctx() R6_w=ctx()
back-edge from insn 42 to 10
BPF program is too complex
processed 131071 insns (limit 131072) max_states_per_insn 12 total_states 9999 peak_states 5000 mark_read 800
verification time 250000 usec
stack depth 96+32
";
        let vs = parse_verifier_stats(log);
        assert_eq!(vs.processed_insns, 131071);
        assert_eq!(vs.total_states, 9999);
        assert_eq!(vs.peak_states, 5000);
        assert_eq!(vs.time_usec, Some(250000));
        assert_eq!(vs.stack_depth.as_deref(), Some("96+32"));
    }

    #[test]
    fn parse_verifier_stats_complexity_error_no_stats() {
        let log = "\
func#0 @0
0: R1=ctx() R10=fp0
R1 type=ctx expected=fp
";
        let vs = parse_verifier_stats(log);
        assert_eq!(vs.processed_insns, 0);
        assert_eq!(vs.total_states, 0);
        assert!(vs.time_usec.is_none());
        assert!(vs.stack_depth.is_none());
    }

    #[test]
    fn parse_verifier_stats_loop_warning_with_stats() {
        let log = "\
infinite loop detected at insn 15
back-edge from insn 30 to 15
processed 500 insns (limit 1000000) max_states_per_insn 3 total_states 40 peak_states 15 mark_read 5
verification time 100 usec
";
        let vs = parse_verifier_stats(log);
        assert_eq!(vs.processed_insns, 500);
        assert_eq!(vs.total_states, 40);
        assert_eq!(vs.peak_states, 15);
        assert_eq!(vs.time_usec, Some(100));
    }

    #[test]
    fn parse_verifier_stats_processed_no_number() {
        let log = "processed\n";
        let vs = parse_verifier_stats(log);
        assert_eq!(vs.processed_insns, 0);
    }

    #[test]
    fn parse_verifier_stats_keyword_at_end_no_value() {
        let log = "processed 100 insns (limit 1000000) max_states_per_insn 1 total_states\n";
        let vs = parse_verifier_stats(log);
        assert_eq!(vs.processed_insns, 100);
        assert_eq!(vs.total_states, 0);
    }

    #[test]
    fn parse_verifier_stats_non_numeric_values() {
        let log = "processed 100 insns (limit 1000000) max_states_per_insn 1 total_states abc peak_states xyz mark_read 0\n";
        let vs = parse_verifier_stats(log);
        assert_eq!(vs.processed_insns, 100);
        assert_eq!(vs.total_states, 0);
        assert_eq!(vs.peak_states, 0);
    }

    #[test]
    fn parse_verifier_stats_verification_time_no_number() {
        let log = "verification time unknown usec\n";
        let vs = parse_verifier_stats(log);
        assert!(vs.time_usec.is_none());
    }

    #[test]
    fn parse_verifier_stats_stack_depth_empty() {
        let log = "stack depth   \n";
        let vs = parse_verifier_stats(log);
        assert!(vs.stack_depth.is_none());
    }

    #[test]
    fn parse_verifier_stats_peak_states_at_end() {
        let log = "processed 50 insns (limit 1000000) max_states_per_insn 1 total_states 10 peak_states\n";
        let vs = parse_verifier_stats(log);
        assert_eq!(vs.processed_insns, 50);
        assert_eq!(vs.total_states, 10);
        assert_eq!(vs.peak_states, 0);
    }

    #[test]
    fn parse_verifier_stats_windows_line_endings() {
        let log = "processed 42 insns (limit 1000000) max_states_per_insn 1 total_states 5 peak_states 2 mark_read 0\r\nverification time 10 usec\r\nstack depth 16\r\n";
        let vs = parse_verifier_stats(log);
        assert_eq!(vs.processed_insns, 42);
        assert_eq!(vs.time_usec, Some(10));
        assert!(vs.stack_depth.is_some());
    }

    // -----------------------------------------------------------------------
    // normalize_verifier_line
    // -----------------------------------------------------------------------

    #[test]
    fn normalize_plain_instruction() {
        assert_eq!(
            normalize_verifier_line("100: (07) r1 += 8"),
            "100: (07) r1 += 8"
        );
    }

    #[test]
    fn normalize_strips_frame_annotation() {
        assert_eq!(
            normalize_verifier_line("3006: (07) r9 += 1  ; frame1: R9_w=2"),
            "3006: (07) r9 += 1"
        );
    }

    #[test]
    fn normalize_strips_register_annotation() {
        assert_eq!(
            normalize_verifier_line("42: (bf) r6 = r1 ; R1=ctx() R6_w=ctx()"),
            "42: (bf) r6 = r1"
        );
    }

    #[test]
    fn normalize_standalone_register_dump() {
        assert_eq!(
            normalize_verifier_line("3041: frame1: R0_w=scalar()"),
            "3041:"
        );
    }

    #[test]
    fn normalize_goto_inline_state() {
        assert_eq!(
            normalize_verifier_line(
                "3026: (b5) if r6 <= 0x11dc0 goto pc+2 3029: frame1: R0=1 R6=scalar()"
            ),
            "3026: (b5) if r6 <= 0x11dc0 goto pc+2"
        );
    }

    #[test]
    fn normalize_goto_no_inline_state() {
        assert_eq!(
            normalize_verifier_line("50: (05) goto pc+10"),
            "50: (05) goto pc+10"
        );
    }

    #[test]
    fn normalize_non_instruction_line() {
        assert_eq!(normalize_verifier_line("func#0 @0"), "func#0 @0");
    }

    #[test]
    fn normalize_empty() {
        assert_eq!(normalize_verifier_line(""), "");
    }

    #[test]
    fn normalize_goto_negative_offset() {
        assert_eq!(
            normalize_verifier_line("50: (05) goto pc-10 60: frame1: R0=1"),
            "50: (05) goto pc-10"
        );
    }

    #[test]
    fn normalize_semicolon_source_comment() {
        let line = "100: (07) r1 += 8 ; for (int j = 0; j < n; j++)";
        assert_eq!(normalize_verifier_line(line), line);
    }

    #[test]
    fn normalize_semicolon_return_value_comment() {
        let line = "200: (b7) r0 = 0 ; Return value";
        assert_eq!(normalize_verifier_line(line), line);
    }

    #[test]
    fn normalize_standalone_bare_register_dump() {
        assert_eq!(
            normalize_verifier_line("3029: R0=1 R6=scalar(id=1)"),
            "3029:"
        );
    }

    #[test]
    fn normalize_standalone_r10_dump() {
        assert_eq!(normalize_verifier_line("42: R10=fp0"), "42:");
    }

    // -----------------------------------------------------------------------
    // detect_cycle / collapse_cycles
    // -----------------------------------------------------------------------

    fn repeating_log(prefix: usize, period: usize, reps: usize, suffix: usize) -> String {
        let mut lines = Vec::new();
        for i in 0..prefix {
            lines.push(format!("{}: (07) r1 += {i}", 1000 + i));
        }
        for rep in 0..reps {
            for j in 0..period {
                let insn = 100 + j;
                lines.push(format!(
                    "{insn}: (bf) r{} = r{} ; frame1: R{}_w={}",
                    j % 10,
                    (j + 1) % 10,
                    j % 10,
                    rep * 100 + j
                ));
            }
        }
        for i in 0..suffix {
            lines.push(format!("{}: (95) exit_{i}", 2000 + i));
        }
        lines.join("\n")
    }

    #[test]
    fn detect_cycle_basic() {
        let log = repeating_log(0, 10, 8, 0);
        let lines: Vec<&str> = log.lines().collect();
        let result = detect_cycle(&lines);
        assert!(result.is_some(), "should detect cycle");
        let (start, period, count) = result.unwrap();
        assert_eq!(period, 10);
        assert!(count >= 6, "count={count}");
        assert_eq!(start, 0);
    }

    #[test]
    fn detect_cycle_with_prefix_suffix() {
        let log = repeating_log(5, 10, 8, 5);
        let lines: Vec<&str> = log.lines().collect();
        let result = detect_cycle(&lines);
        assert!(result.is_some(), "should detect cycle with prefix/suffix");
        let (_start, period, count) = result.unwrap();
        assert_eq!(period, 10);
        assert!(count >= 6);
    }

    #[test]
    fn detect_cycle_too_few_reps() {
        let log = repeating_log(0, 10, 2, 0);
        let lines: Vec<&str> = log.lines().collect();
        assert!(detect_cycle(&lines).is_none());
    }

    #[test]
    fn detect_cycle_too_few_lines() {
        let lines: Vec<String> = (0..20)
            .map(|i| format!("{}: (07) r1 += {i}", 100 + i % 3))
            .collect();
        let refs: Vec<&str> = lines.iter().map(|s| s.as_str()).collect();
        assert!(detect_cycle(&refs).is_none());
    }

    #[test]
    fn detect_cycle_no_cycle() {
        let lines: Vec<String> = (0..100).map(|i| format!("{i}: unique_insn_{i}")).collect();
        let refs: Vec<&str> = lines.iter().map(|s| s.as_str()).collect();
        assert!(detect_cycle(&refs).is_none());
    }

    #[test]
    fn detect_cycle_empty() {
        let empty: Vec<&str> = vec![];
        assert!(detect_cycle(&empty).is_none());
    }

    #[test]
    fn detect_cycle_exact_boundary() {
        let log = repeating_log(0, 5, 6, 0);
        let lines: Vec<&str> = log.lines().collect();
        assert_eq!(lines.len(), 30);
        let result = detect_cycle(&lines);
        assert!(result.is_some(), "boundary case should detect cycle");
        let (_start, period, count) = result.unwrap();
        assert_eq!(period, 5);
        assert_eq!(count, 6);
    }

    #[test]
    fn collapse_cycles_empty_string() {
        assert_eq!(collapse_cycles(""), "");
    }

    #[test]
    fn collapse_cycles_basic() {
        let log = repeating_log(2, 10, 8, 2);
        let collapsed = collapse_cycles(&log);
        assert!(collapsed.contains("identical iterations omitted"));
        assert!(collapsed.contains("8x of the following 10 lines"));
        assert!(collapsed.contains("end repeat"));
        assert!(collapsed.lines().count() < log.lines().count());
    }

    #[test]
    fn collapse_cycles_no_cycle() {
        let log = "line 1\nline 2\nline 3\n";
        let collapsed = collapse_cycles(log);
        assert_eq!(collapsed, log);
    }

    #[test]
    fn collapse_cycles_preserves_stats() {
        let mut log = repeating_log(0, 10, 8, 0);
        log.push_str("\nprocessed 1000 insns (limit 1000000) max_states_per_insn 5 total_states 100 peak_states 30 mark_read 10\n");
        let collapsed = collapse_cycles(&log);
        assert!(collapsed.contains("processed 1000 insns"));
    }

    #[test]
    fn collapse_cycles_with_register_annotations() {
        let mut lines = Vec::new();
        lines.push("0: (07) r1 += 1".to_string());
        for rep in 0..8 {
            for j in 0..6 {
                let insn = 100 + j;
                lines.push(format!(
                    "{insn}: (bf) r{} = r{} ; frame1: R{}_w={}",
                    j % 10,
                    (j + 1) % 10,
                    j % 10,
                    rep * 100 + j
                ));
            }
        }
        lines.push("200: (95) exit".to_string());
        let log = lines.join("\n");
        let collapsed = collapse_cycles(&log);
        assert!(collapsed.contains("identical iterations omitted"));
    }

    // -----------------------------------------------------------------------
    // build_b_map / build_diff_rows
    // -----------------------------------------------------------------------

    fn prog(name: &str, verified_insns: u32) -> ProgStats {
        ProgStats {
            name: name.to_string(),
            verified_insns,
        }
    }

    #[test]
    fn build_b_map_basic() {
        let stats_b = vec![prog("dispatch", 500)];
        let map = build_b_map(&stats_b);
        assert_eq!(map.get("dispatch"), Some(&500));
    }

    #[test]
    fn build_b_map_empty() {
        let map = build_b_map(&[]);
        assert!(map.is_empty());
    }

    #[test]
    fn build_diff_rows_matching_programs() {
        let stats_a = vec![prog("dispatch", 500)];
        let mut b_map = HashMap::new();
        b_map.insert("dispatch".to_string(), 300u64);
        let rows = build_diff_rows(&stats_a, &b_map);
        assert_eq!(rows.len(), 1);
        assert_eq!(rows[0].name, "dispatch");
        assert_eq!(rows[0].a, 500);
        assert_eq!(rows[0].b, 300);
        assert_eq!(rows[0].delta, 200);
    }

    #[test]
    fn build_diff_rows_program_missing_from_b() {
        let stats_a = vec![prog("new_prog", 100)];
        let b_map = HashMap::new();
        let rows = build_diff_rows(&stats_a, &b_map);
        assert_eq!(rows.len(), 1);
        assert_eq!(rows[0].a, 100);
        assert_eq!(rows[0].b, 0);
        assert_eq!(rows[0].delta, 100);
    }

    #[test]
    fn build_diff_rows_negative_delta() {
        let stats_a = vec![prog("dispatch", 200)];
        let mut b_map = HashMap::new();
        b_map.insert("dispatch".to_string(), 500u64);
        let rows = build_diff_rows(&stats_a, &b_map);
        assert_eq!(rows[0].delta, -300);
    }

    #[test]
    fn build_diff_rows_empty_a() {
        let b_map = HashMap::new();
        let rows = build_diff_rows(&[], &b_map);
        assert!(rows.is_empty());
    }

    /// Simulates the verifier trace produced by #pragma unroll loops.
    /// Each copy is at a different base address but has the same
    /// instruction sequence. After normalize_for_cycle_detection strips
    /// addresses and register annotations, all copies look identical.
    fn unrolled_verifier_log(copies: usize, body_len: usize) -> String {
        let ops = [
            "(85) call bpf_ktime_get_ns#5",
            "(bf) r2 = r0",
            "(77) r0 >>= 16",
            "(af) r1 ^= r0",
            "(77) r2 >>= 32",
            "(0f) r1 += r2",
            "(24) w1 *= 7",
            "(04) w1 += 1",
        ];
        let mut lines = Vec::new();
        lines.push("func#0 @0".to_string());
        lines.push("0: R1=ctx() R10=fp0".to_string());
        let mut addr = 10;
        for copy in 0..copies {
            for (j, op) in ops.iter().enumerate().take(body_len) {
                lines.push(format!(
                    "{}: {op} ; R0_w=scalar(id={})",
                    addr,
                    copy * 100 + j
                ));
                addr += 1;
            }
        }
        lines.push(format!("{addr}: (05) goto pc-1"));
        lines.push(
            "processed 1000 insns (limit 1000000) max_states_per_insn 3 \
             total_states 50 peak_states 20 mark_read 5"
                .to_string(),
        );
        lines.join("\n")
    }

    #[test]
    fn detect_cycle_unrolled_loop() {
        let log = unrolled_verifier_log(8, 6);
        let lines: Vec<&str> = log.lines().collect();
        let result = detect_cycle(&lines);
        assert!(result.is_some(), "should detect cycle in unrolled loop");
        let (_start, period, count) = result.unwrap();
        assert_eq!(period, 6);
        assert!(count >= 6, "count={count}");
    }

    #[test]
    fn collapse_cycles_unrolled_loop() {
        let log = unrolled_verifier_log(8, 6);
        let collapsed = collapse_cycles(&log);
        assert!(
            collapsed.contains("identical iterations omitted"),
            "should collapse unrolled loop"
        );
        assert!(collapsed.lines().count() < log.lines().count());
    }

    // -----------------------------------------------------------------------
    // extract_verifier_log
    // -----------------------------------------------------------------------

    #[test]
    fn extract_verifier_log_basic() {
        let log = "\
libbpf: prog 'dispatch': BPF program load failed: -22
-- BEGIN PROG LOAD LOG --
func#0 @0
0: R1=ctx() R10=fp0
processed 100 insns (limit 1000000) max_states_per_insn 1 total_states 5 peak_states 2 mark_read 0
-- END PROG LOAD LOG --
libbpf: failed to load object 'ktstr_ops'
";
        let extracted = extract_verifier_log(log);
        assert!(extracted.is_some());
        let v = extracted.unwrap();
        assert!(v.starts_with("func#0 @0"));
        assert!(v.contains("processed 100 insns"));
        assert!(!v.contains("BEGIN PROG LOAD LOG"));
        assert!(!v.contains("END PROG LOAD LOG"));
        assert!(!v.contains("libbpf:"));
    }

    #[test]
    fn extract_verifier_log_none_without_markers() {
        let log = "func#0 @0\n0: R1=ctx()\nprocessed 50 insns\n";
        assert!(extract_verifier_log(log).is_none());
    }

    #[test]
    fn extract_verifier_log_empty() {
        assert!(extract_verifier_log("").is_none());
    }

    /// Attack 1: libbpf wraps verifier output with "libbpf: " prefix lines.
    /// `parse_verifier_stats` looks for `starts_with("processed ")` which
    /// won't match `libbpf: processed ...`. Without extraction, stats
    /// parsing fails on blobs where the `processed` line is only inside
    /// the markers.
    #[test]
    fn extract_verifier_log_attack1_stats_parse() {
        let blob = "\
libbpf: prog 'ktstr_ops_dispatch': BPF program load failed: -22
libbpf: -- BEGIN PROG LOAD LOG --
func#0 @0
0: R1=ctx() R10=fp0
1: (bf) r6 = r1 ; R1=ctx() R6_w=ctx()
back-edge from insn 42 to 10
BPF program is too complex
processed 131071 insns (limit 131072) max_states_per_insn 12 total_states 9999 peak_states 5000 mark_read 800
verification time 250000 usec
stack depth 96+32
libbpf: -- END PROG LOAD LOG --
libbpf: failed to load BPF skeleton 'ktstr_ops': -22
";
        let extracted = extract_verifier_log(blob);
        assert!(extracted.is_some(), "should find markers");
        let v = extracted.unwrap();
        let vs = parse_verifier_stats(v);
        assert_eq!(vs.processed_insns, 131071);
        assert_eq!(vs.total_states, 9999);
        assert_eq!(vs.peak_states, 5000);
        assert_eq!(vs.time_usec, Some(250000));
        assert_eq!(vs.stack_depth.as_deref(), Some("96+32"));

        // Without extraction, parsing the full blob must also work
        // because the "processed" line doesn't have a "libbpf: " prefix
        // inside the markers. But verify extraction gives cleaner input.
        let vs_raw = parse_verifier_stats(blob);
        assert_eq!(vs_raw.processed_insns, 131071);
    }

    /// Attack 3: three distinct program load logs in a single blob.
    /// Each has different instructions. `collapse_cycles` must NOT treat
    /// them as a repeating cycle.
    #[test]
    fn extract_verifier_log_attack3_no_false_collapse() {
        let blob = "\
libbpf: prog 'init': BPF program load failed: -22
libbpf: -- BEGIN PROG LOAD LOG --
func#0 @0
0: R1=ctx() R10=fp0
1: (bf) r6 = r1
2: (07) r6 += 8
3: (61) r0 = *(u32 *)(r6 + 0)
4: (95) exit
processed 5 insns (limit 1000000) max_states_per_insn 1 total_states 3 peak_states 1 mark_read 0
libbpf: -- END PROG LOAD LOG --
libbpf: prog 'dispatch': BPF program load failed: -22
libbpf: -- BEGIN PROG LOAD LOG --
func#1 @10
10: R1=ctx() R10=fp0
11: (bf) r7 = r1
12: (85) call bpf_ktime_get_ns#5
13: (77) r0 >>= 32
14: (95) exit
processed 5 insns (limit 1000000) max_states_per_insn 1 total_states 3 peak_states 1 mark_read 0
libbpf: -- END PROG LOAD LOG --
libbpf: prog 'enqueue': BPF program load failed: -22
libbpf: -- BEGIN PROG LOAD LOG --
func#2 @20
20: R1=ctx() R10=fp0
21: (b7) r0 = 0
22: (63) *(u32 *)(r10 - 4) = r0
23: (61) r1 = *(u32 *)(r10 - 4)
24: (95) exit
processed 5 insns (limit 1000000) max_states_per_insn 1 total_states 3 peak_states 1 mark_read 0
libbpf: -- END PROG LOAD LOG --
libbpf: failed to load BPF skeleton 'ktstr_ops': -22
";
        // extract_verifier_log returns the FIRST log section.
        let extracted = extract_verifier_log(blob);
        assert!(extracted.is_some());
        let v = extracted.unwrap();
        assert!(v.contains("func#0 @0"), "should get first program's log");
        assert!(!v.contains("func#1"), "should not include second program");

        // collapse_cycles on the extracted first section must not
        // collapse — it's only 7 lines total.
        let collapsed = collapse_cycles(v);
        assert!(
            !collapsed.contains("identical iterations omitted"),
            "must not false-collapse distinct program logs"
        );
    }

    // -- insta snapshot tests --

    #[test]
    fn snapshot_format_verifier_output_no_log() {
        let result = VerifierVmResult {
            stats: vec![
                ProgStats {
                    name: "enqueue".into(),
                    verified_insns: 500,
                },
                ProgStats {
                    name: "dispatch".into(),
                    verified_insns: 1200,
                },
                ProgStats {
                    name: "init".into(),
                    verified_insns: 300,
                },
            ],
            scheduler_log: String::new(),
            attach: AttachOutcome::Attached,
            dispatched: true,
            timed_out: false,
        };
        insta::assert_snapshot!(format_verifier_output("default", &result, false));
    }

    #[test]
    fn snapshot_format_verifier_output_with_log() {
        let log = "\
-- BEGIN PROG LOAD LOG --\n\
func#0 @0\n\
0: R1=ctx() R10=fp0\n\
processed 42 insns (limit 1000000) max_states_per_insn 1 total_states 10 peak_states 8 mark_read 5\n\
-- END PROG LOAD LOG --";
        let result = VerifierVmResult {
            stats: vec![ProgStats {
                name: "enqueue".into(),
                verified_insns: 42,
            }],
            scheduler_log: log.into(),
            // A load log present means the scheduler printed a verifier
            // trace then exited — the SchedulerDied failure path.
            attach: AttachOutcome::Died,
            dispatched: false,
            timed_out: false,
        };
        insta::assert_snapshot!(format_verifier_output("llc+steal", &result, false));
    }

    #[test]
    fn snapshot_format_verifier_diff() {
        let stats_a = vec![
            ProgStats {
                name: "enqueue".into(),
                verified_insns: 500,
            },
            ProgStats {
                name: "dispatch".into(),
                verified_insns: 1200,
            },
            ProgStats {
                name: "init".into(),
                verified_insns: 300,
            },
        ];
        let stats_b = vec![
            ProgStats {
                name: "enqueue".into(),
                verified_insns: 480,
            },
            ProgStats {
                name: "dispatch".into(),
                verified_insns: 1350,
            },
            ProgStats {
                name: "init".into(),
                verified_insns: 300,
            },
        ];
        insta::assert_snapshot!(format_verifier_diff("default", &stats_a, "llc", &stats_b));
    }

    #[test]
    fn snapshot_format_verifier_diff_missing_program() {
        let stats_a = vec![
            ProgStats {
                name: "enqueue".into(),
                verified_insns: 500,
            },
            ProgStats {
                name: "new_prog".into(),
                verified_insns: 100,
            },
        ];
        let stats_b = vec![ProgStats {
            name: "enqueue".into(),
            verified_insns: 500,
        }];
        insta::assert_snapshot!(format_verifier_diff("A", &stats_a, "B", &stats_b));
    }

    // -----------------------------------------------------------------------
    // extract_verifier_log — log extraction + cross-check against
    // parse_sched_output so the two slicers stay consistent on shared input.
    // -----------------------------------------------------------------------

    #[test]
    fn extract_verifier_log_between_begin_end_markers() {
        // libbpf wraps the verifier log between explicit marker lines;
        // the extractor returns the content between them, trimmed of
        // the BEGIN newline and the trailing libbpf END prefix.
        let blob = "\
            unrelated preamble\n\
            libbpf: -- BEGIN PROG LOAD LOG --\n\
            processed 1234 insns (limit 1000000) max_states_per_insn 5 total_states 200 peak_states 50 mark_read 10\n\
            libbpf: -- END PROG LOAD LOG --\n\
            trailing diagnostics\n";
        let log = extract_verifier_log(blob).expect("markers present");
        assert!(log.contains("processed 1234 insns"));
        assert!(!log.contains("BEGIN PROG LOAD LOG"));
        assert!(!log.contains("END PROG LOAD LOG"));
    }

    #[test]
    fn extract_verifier_log_returns_none_when_markers_absent() {
        // Backward compat: logs without the libbpf markers are treated
        // as "no markers" — the caller falls back to using the raw blob.
        assert!(extract_verifier_log("no markers in here").is_none());
        assert!(extract_verifier_log("only BEGIN marker -- BEGIN PROG LOAD LOG --").is_none());
    }

    #[test]
    fn extract_verifier_log_consistent_with_parse_sched_output() {
        // `collect_verifier_output` chains parse_sched_output →
        // extract_verifier_log on the VM stdout blob. Both slicers
        // operate on the same input without duplicating work, so a
        // single SCHED_OUTPUT block that wraps a libbpf-marked verifier
        // log must produce the same verifier text when extracted in
        // that order.
        let sched_inner = "\
            libbpf: -- BEGIN PROG LOAD LOG --\n\
            processed 7 insns (limit 1000000) max_states_per_insn 1 total_states 1 peak_states 1 mark_read 0\n\
            libbpf: -- END PROG LOAD LOG --\n";
        let vm_output = format!(
            "kernel boot junk\n{SCHED_OUTPUT_START}\n{sched_inner}{SCHED_OUTPUT_END}\nafterward\n",
        );
        let sched = parse_sched_output(&vm_output).expect("SCHED_OUTPUT block");
        let verifier_log = extract_verifier_log(sched).expect("verifier markers");
        assert!(verifier_log.contains("processed 7 insns"));
        assert!(!verifier_log.contains("SCHED_OUTPUT"));
        assert!(!verifier_log.contains("BEGIN PROG LOAD LOG"));
    }

    #[test]
    fn parse_sched_output_valid() {
        let output = format!(
            "noise\n{SCHED_OUTPUT_START}\nscheduler log line 1\nline 2\n{SCHED_OUTPUT_END}\nmore"
        );
        let parsed = parse_sched_output(&output);
        assert!(parsed.is_some());
        let content = parsed.unwrap();
        assert!(content.contains("scheduler log line 1"));
        assert!(content.contains("line 2"));
    }

    #[test]
    fn parse_sched_output_missing_start() {
        let output = format!("no start\n{SCHED_OUTPUT_END}\n");
        assert!(parse_sched_output(&output).is_none());
    }

    #[test]
    fn parse_sched_output_missing_end() {
        let output = format!("{SCHED_OUTPUT_START}\nsome content");
        assert!(parse_sched_output(&output).is_none());
    }

    #[test]
    fn parse_sched_output_empty_content() {
        let output = format!("{SCHED_OUTPUT_START}\n\n{SCHED_OUTPUT_END}");
        assert!(parse_sched_output(&output).is_none());
    }

    #[test]
    fn parse_sched_output_with_stack_traces() {
        let stack = "do_enqueue_task+0x1a0/0x380\nbalance_one+0x50/0x100\n";
        let output = format!("{SCHED_OUTPUT_START}\n{stack}\n{SCHED_OUTPUT_END}");
        let parsed = parse_sched_output(&output).unwrap();
        assert!(parsed.contains("do_enqueue_task"));
        assert!(parsed.contains("balance_one"));
    }

    #[test]
    fn parse_sched_output_rfind_survives_end_marker_in_content() {
        // Regression: if the scheduler log echoes the END marker
        // inside its own content (e.g. a shell heredoc, a diagnostic
        // that quotes the sentinel), `find` truncated the section at
        // the first occurrence — which was inside the content, not
        // at the terminator. `rfind` anchors on the last occurrence,
        // which is the real terminator.
        let content = format!("line1\nfake {SCHED_OUTPUT_END} inside\nline3");
        let output = format!("{SCHED_OUTPUT_START}\n{content}\n{SCHED_OUTPUT_END}\n");
        let parsed = parse_sched_output(&output).unwrap();
        assert!(
            parsed.contains("line3"),
            "rfind must keep content after an embedded END marker: {parsed:?}"
        );
        assert!(
            parsed.contains("fake"),
            "content before the embedded marker must also survive: {parsed:?}"
        );
    }

    // -- parse_sched_output_partial --

    #[test]
    fn parse_sched_output_partial_well_formed_matches_strict() {
        // When both delimiters are present, the partial parser
        // returns the same content as the strict parser.
        let output = format!(
            "noise\n{SCHED_OUTPUT_START}\nscheduler log line 1\nline 2\n{SCHED_OUTPUT_END}\nmore"
        );
        assert_eq!(
            parse_sched_output_partial(&output),
            parse_sched_output(&output),
        );
    }

    #[test]
    fn parse_sched_output_partial_missing_end_returns_partial() {
        // When SCHED_OUTPUT_END is absent (scheduler crashed mid-run
        // before writing the closing delimiter), the partial parser
        // returns content from after SCHED_OUTPUT_START to end of
        // buffer. The strict parser returns None for the same input.
        let output = format!("{SCHED_OUTPUT_START}\nstack frame 1\nstack frame 2");
        assert!(parse_sched_output(&output).is_none());
        let partial = parse_sched_output_partial(&output).unwrap();
        assert!(partial.contains("stack frame 1"));
        assert!(partial.contains("stack frame 2"));
    }

    #[test]
    fn parse_sched_output_partial_missing_start_returns_none() {
        // No start marker → no content recoverable. The end-marker-
        // only case is unrecoverable: we cannot infer where the log
        // begins.
        let output = format!("garbage\n{SCHED_OUTPUT_END}\n");
        assert!(parse_sched_output_partial(&output).is_none());
    }

    #[test]
    fn parse_sched_output_partial_empty_content_returns_none() {
        // Start marker present but no payload after it.
        let output = format!("{SCHED_OUTPUT_START}\n");
        assert!(parse_sched_output_partial(&output).is_none());
    }
}