ktstr 0.17.0

// Global allocator (jemalloc) is provided centrally by the ktstr
// library crate (src/lib.rs) and inherited by this bin.
use std::path::{Path, PathBuf};

use anyhow::Result;
use clap::{ArgAction, CommandFactory, Parser, Subcommand};

use ktstr::cli;
use ktstr::cli::KernelCommand;
use ktstr::ctprof;
use ktstr::ctprof_compare;
use ktstr::topology::TestTopology;

/// Per-section render helpers for `write_show`, factored into
/// `src/bin/ktstr/show_render.rs` to keep this file under the size
/// guard. `write_show` stays here and orchestrates; the per-section
/// writers live in the submodule and reach back to the PSI helpers
/// and the `ctprof_compare` alias via `use super::*`.
///
/// `#[path]` is required: a `[[bin]]` crate root resolves `mod foo;`
/// to a sibling `foo.rs`, not a `ktstr/`-stem subdir, so the explicit
/// path points at the submodule file (mirrors the cargo-ktstr bin's
/// `#[path = "cargo_ktstr/..."]` declarations).
#[path = "ktstr/show_render.rs"]
mod show_render;

#[derive(Parser)]
#[command(
    name = "ktstr",
    about = "Run ktstr scheduler test scenarios on the host",
    after_help = "See also: `cargo ktstr` for cargo-integrated workflows \
                  (test, coverage, llvm-cov, verifier, stats)."
)]
struct Cli {
    #[command(subcommand)]
    command: Command,
}

#[derive(Subcommand)]
enum Command {
    /// Show host CPU topology.
    Topo,
    /// Manage cached kernel images.
    Kernel {
        #[command(subcommand)]
        command: KernelCommand,
    },
    /// Boot an interactive shell in a KVM virtual machine.
    ///
    /// Launches a VM with busybox and drops into a shell. Files and
    /// directories passed via -i are available at /include-files/<name>
    /// inside the guest. Directories are walked recursively, preserving
    /// structure. Dynamically-linked ELF binaries get automatic shared
    /// library resolution via ELF DT_NEEDED parsing.
    Shell {
        #[arg(long, help = ktstr::cli::KERNEL_HELP_NO_RAW)]
        kernel: Option<String>,
        /// Virtual topology as "numa_nodes,llcs,cores,threads".
        #[arg(long, default_value = "1,1,1,1")]
        topology: String,
        /// Files or directories to include in the guest. Repeatable.
        #[arg(short = 'i', long = "include-files", action = ArgAction::Append)]
        include_files: Vec<PathBuf>,
        /// Guest memory in MiB (minimum 128). When absent, estimated
        /// from payload and include file sizes.
        #[arg(long = "memory-mib", value_parser = clap::value_parser!(u32).range(128..))]
        memory_mib: Option<u32>,
        /// Forward kernel console (COM1/dmesg) to stderr in real-time.
        /// Sets loglevel=7 for verbose kernel output.
        #[arg(long)]
        dmesg: bool,
        /// Run a command in the VM instead of an interactive shell.
        /// The VM exits after the command completes.
        #[arg(long)]
        exec: Option<String>,

        /// Max wall-clock for a `--exec` payload before the VM is
        /// force-killed (a panic-less guest hang otherwise blocks
        /// forever). Parsed by humantime: `30s`, `5m`, `1h`. Ignored
        /// without `--exec`. Must exceed guest boot (a few seconds): a
        /// near-zero value force-kills before the payload runs.
        #[arg(long, value_parser = humantime::parse_duration, default_value = "120s")]
        exec_timeout: std::time::Duration,

        /// Disable all performance mode features (flock, pinning, RT
        /// scheduling, hugepages, NUMA mbind, KVM exit suppression).
        /// For shared runners or unprivileged containers.
        /// Also settable via KTSTR_NO_PERF_MODE env var.
        #[arg(long)]
        no_perf_mode: bool,

        /// Reserve only N host CPUs for the shell VM. Requires
        /// `--no-perf-mode` — perf-mode already holds every LLC
        /// exclusively, so capping under perf-mode would
        /// double-reserve. See `ktstr::cli::CPU_CAP_HELP` for the
        /// full contract.
        #[arg(long, requires = "no_perf_mode", help = ktstr::cli::CPU_CAP_HELP)]
        cpu_cap: Option<usize>,

        #[arg(long, help = ktstr::cli::DISK_HELP)]
        disk: Option<String>,
    },
    /// Capture or compare a host-wide per-thread state snapshot.
    ///
    /// `capture` walks `/proc` at capture time and writes every
    /// visible thread's cumulative scheduling, memory, and I/O
    /// counters as zstd-compressed JSON (`.ctprof.zst`). Every
    /// field is cumulative-from-birth so probe attachment time
    /// does not bias the reading — a diff between two captures
    /// measures exactly the activity over the window.
    ///
    /// Memory fields (`allocated_bytes` / `deallocated_bytes`)
    /// are JEMALLOC-ONLY: they read the per-thread
    /// `tsd_s.thread_allocated` / `thread_deallocated` TSD
    /// counters via ptrace + `process_vm_readv`. Targets not
    /// linked against jemalloc, or for which the probe attach
    /// fails, land their threads at zero per the best-effort
    /// capture contract. Other allocators (glibc malloc, mimalloc)
    /// expose no equivalent per-thread counter the capture
    /// pipeline can reach.
    ///
    /// PRIVILEGE: capture pulls the jemalloc counters by briefly
    /// stopping every probed thread via `ptrace(PTRACE_SEIZE)`
    /// (an unavoidable observer effect; the stop is bounded to a
    /// handful of milliseconds per thread). `PTRACE_SEIZE` requires
    /// root, the `CAP_SYS_PTRACE` capability, OR a host configured
    /// with `kernel.yama.ptrace_scope=0`. When the attach is
    /// denied, the probe falls through to the absent-counter
    /// default of zero — the rest of the snapshot still populates.
    /// See the capture summary line for a per-snapshot tally and
    /// remediation hint.
    ///
    /// `compare` joins two snapshots on the selected grouping
    /// axis (pcomm, cgroup, or comm — see `--group-by`) and
    /// renders a per-metric baseline/candidate/delta table.
    Ctprof {
        #[command(subcommand)]
        command: CtprofCommand,
    },
    /// Generate shell completions for ktstr.
    Completions {
        /// Shell to generate completions for.
        shell: clap_complete::Shell,
        /// Binary name to register the completion under. Override
        /// when invoking ktstr through a symlink with a different
        /// name (the shell looks up completions by argv[0]).
        #[arg(long, default_value = "ktstr")]
        binary: String,
    },
    /// Enumerate every ktstr flock held on this host.
    ///
    /// Troubleshooting companion for `--cpu-cap` contention. Scans
    /// `{KTSTR_LOCK_DIR}/ktstr-llc-*.lock`,
    /// `{KTSTR_LOCK_DIR}/ktstr-cpu-*.lock` (default `/tmp`), and
    /// `{cache_root}/.locks/*.lock`, cross-referenced against
    /// `/proc/locks` via [`ktstr::cli::list_locks`] to name the
    /// holder process (PID + cmdline) for each held lock. Read-only
    /// — does NOT attempt any flock acquire.
    Locks {
        /// Emit the snapshot as JSON (compact object under --watch,
        /// pretty-printed otherwise). Stable field names; schema
        /// documented at [`ktstr::cli::list_locks`].
        #[arg(long)]
        json: bool,
        /// Redraw the snapshot on the given interval until SIGINT.
        /// Value is parsed by `humantime`: `100ms`, `1s`, `5m`, `1h`.
        /// Human output clears and redraws in place; `--json` emits
        /// one line-terminated object per interval (ndjson-style).
        #[arg(long, value_parser = humantime::parse_duration)]
        watch: Option<std::time::Duration>,
    },
}

#[derive(Subcommand)]
enum CtprofCommand {
    /// Capture a host-wide per-thread snapshot to `<output>` as
    /// zstd-compressed JSON. Walks `/proc` for every live tgid,
    /// enumerates threads, records schedstat / sched / status
    /// CSW / page faults / io bytes / affinity / cgroup / identity.
    /// Per-cgroup aggregates (cpu.stat, memory.current) are
    /// captured once per distinct path.
    ///
    /// Memory fields are JEMALLOC-ONLY (per-thread TSD
    /// `thread_allocated` / `thread_deallocated` counters); other
    /// allocators land at zero. Capture briefly stops every probed
    /// thread via ptrace (observer effect, bounded ms per thread)
    /// and therefore needs root, `CAP_SYS_PTRACE`, or
    /// `kernel.yama.ptrace_scope=0` to pull the jemalloc counters.
    /// Without the privilege, the rest of the snapshot still
    /// populates and the jemalloc fields fall through to zero.
    ///
    /// PROBE SUMMARY LINE: capture emits a single info-level
    /// `tracing` event per snapshot summarising probe outcomes. The
    /// default tracing config writes to stderr, so the line lands
    /// alongside any other capture-time diagnostics; `RUST_LOG=warn`
    /// suppresses it (alongside other info), `RUST_LOG=debug`
    /// reveals the per-tgid attach events that feed it.
    ///
    /// Format depends on whether any per-tgid failures landed:
    ///
    /// 1. No failures —
    ///    `ctprof probe: <N> tgids walked, <N> jemalloc detected,
    ///    <N> probed OK, 0 failed`
    /// 2. Failures, ptrace not dominant —
    ///    `... <N> failed (dominant: <tag>)`
    /// 3. Failures, ptrace dominates (≥50% of failures attributable
    ///    to `ptrace-seize` or `ptrace-interrupt`) —
    ///    `... <N> failed (dominant: <tag>; hint: re-run as root, or
    ///    sudo setcap cap_sys_ptrace+eip $(which ktstr), or set
    ///    kernel.yama.ptrace_scope=0)`
    ///
    /// "tgids walked" counts every live tgid the procfs walk reached
    /// (the snapshot's full denominator); "jemalloc detected" counts
    /// the subset that successfully attached the TSD probe; "probed
    /// OK" counts per-thread reads that returned a counter pair;
    /// "failed" counts attach-or-probe failures whose tag is
    /// classified ACTIONABLE — see filter rule below — so a busy
    /// host typically reports `failed` ≪ (`tgids walked` − `jemalloc
    /// detected`) because the bulk of the system is not jemalloc-
    /// linked and that case does not count as a failure.
    ///
    /// ACTIONABLE TAGS (operator-facing causes for the
    /// `(dominant: <tag>)` clause):
    ///
    /// Pre-thread (attach-time) failures:
    /// - `pid-missing` — `/proc/<pid>` does not exist or is
    ///   unreachable; possible race-with-exit between enumeration
    ///   and attach.
    /// - `maps-read-failure` — `/proc/<pid>/maps` could not be read;
    ///   possible race-with-exit, uid mismatch, or sandbox/namespace
    ///   restriction.
    /// - `jemalloc-in-dso` — target's jemalloc symbol resides in a
    ///   shared object (e.g. `libjemalloc.so`). Static-TLS resolution
    ///   only reaches the main executable; DSO-linked jemalloc needs
    ///   DTV walking the engine does not implement yet.
    /// - `arch-mismatch` — target ELF arch differs from the probe
    ///   binary's. ptrace is same-arch only; rebuild the probe for
    ///   the target's arch (or filter the offending tgids).
    /// - `dwarf-parse-failure` — target binary is stripped without
    ///   reachable external debuginfo, debuglink CRC mismatched, or
    ///   the `tsd_s` struct / its member fields are absent from the
    ///   DWARF. Install matching `-debuginfo` / `-dbg` packages.
    /// - `worker-panic` — the per-tgid attach worker panicked
    ///   (caught via `catch_unwind` so the snapshot still completes).
    ///   Indicates an unexpected fault inside the parallel attach
    ///   pipeline — fd exhaustion / OOM during the ELF parse or
    ///   DWARF walk are the canonical triggers, but any panic-on-bug
    ///   under `attach_jemalloc_at` lands here. Investigate the
    ///   `tracing::error!` log for the offending tgid; the recorded
    ///   payload string identifies the panic site.
    ///
    /// Per-thread (probe-time) failures:
    /// - `ptrace-seize` — `PTRACE_SEIZE` rejected by the kernel.
    ///   ESRCH is benign (thread exited mid-snapshot); EPERM is the
    ///   privilege case the EPERM hint targets; EBUSY means another
    ///   tracer is attached.
    /// - `ptrace-interrupt` — `PTRACE_INTERRUPT` failed after a
    ///   successful seize. Race causes — privilege is already
    ///   established by the preceding successful seize. ESRCH (target
    ///   died between seize and interrupt) is the dominant case.
    /// - `waitpid` — `waitpid` after `PTRACE_INTERRUPT` returned an
    ///   unexpected status, a syscall error, or the post-interrupt
    ///   stop did not arrive within 250ms.
    /// - `get-regset` — `PTRACE_GETREGSET` failed when reading the
    ///   thread pointer. Race causes — privilege established. ESRCH
    ///   (target died mid-probe) is the dominant case.
    /// - `process-vm-readv` — `process_vm_readv` failed or returned
    ///   a short read at the resolved TSD address. Investigate
    ///   target memory protection or a corrupted TSD.
    /// - `tls-arithmetic` — TLS-address arithmetic over- or
    ///   underflowed. Indicates a malformed TLS image — file a bug
    ///   with the target binary's identity.
    ///
    /// FILTER RULE — two AttachError tags are NOT actionable and are
    /// suppressed from both the per-tgid `tracing::warn!` log and
    /// the `(dominant: <tag>)` clause:
    ///
    /// - `jemalloc-not-found` — target is not jemalloc-linked, OR
    ///   links a jemalloc build whose symbol-name prefix is not in
    ///   the recognized set. The dominant outcome on most system
    ///   processes.
    /// - `readlink-failure` — `readlink(/proc/<pid>/exe)` failed.
    ///   Typically a race-with-exit between enumeration and the
    ///   per-tgid attach, OR permission denied under the active
    ///   ptrace policy.
    ///
    /// Filtered tags do NOT contribute to the `failed` counter, so a
    /// snapshot whose only "failures" are these two will emit the
    /// no-failure variant of the line (no `(dominant: ...)` clause).
    /// If the operator sees no `(dominant: ...)` clause despite many
    /// non-jemalloc tgids on the host, that is correct: the noise
    /// floor was filtered.
    Capture {
        /// Destination path (convention: `.ctprof.zst`).
        #[arg(short, long)]
        output: PathBuf,
    },
    /// Compare two snapshots and render a per-metric diff table.
    Compare(ctprof_compare::CtprofCompareArgs),
    /// Render one snapshot as a per-group, per-metric table.
    ///
    /// Same grouping axis as `compare` (`pcomm` default, `cgroup`,
    /// `comm`, `comm-exact`) and the same per-metric aggregation
    /// rules from [`ctprof_compare::CTPROF_METRICS`], but
    /// with no baseline/candidate split — every metric renders one
    /// aggregated value per group rather than a delta against a
    /// reference snapshot. Useful for inspecting a capture in
    /// isolation: spot-check a process's per-thread CSW totals,
    /// audit the cgroup distribution, find which thread-name
    /// pattern has the highest run-time without comparing two
    /// runs.
    ///
    /// Pattern normalization (`comm` and `pcomm` token-based
    /// clustering, `cgroup` Layer-1/2/3 path tightening, and
    /// `## smaps_rollup` per-process pcomm-pattern aggregation)
    /// applies the same way it does in `compare`;
    /// `--no-thread-normalize` disables the name-family axes
    /// (comm, pcomm, smaps) and `--no-cg-normalize` disables the
    /// cgroup axis. `--cgroup-flatten` glob patterns also apply
    /// unchanged.
    Show(CtprofShowArgs),
    /// Print every registered metric with its tags and a
    /// one-line description.
    ///
    /// Discovery companion to `compare` and `show`: rendered
    /// table cells carry `[cfs-only]`, `[non-ext]`,
    /// `[fair-policy]`, `[SCHEDSTATS]`, `[SCHED_INFO]`,
    /// `[SCHED_CORE]`, `[SCHED_CLASS_EXT]`, `[TASK_DELAY_ACCT]`,
    /// `[TASK_IO_ACCOUNTING]`, and `[dead]` suffixes — this
    /// subcommand is the CLI-discoverable place to learn what
    /// each tag means and which kernel counter each metric
    /// surfaces.
    ///
    /// Output is two sections: a tag legend explaining each
    /// vocabulary element, then a `metric | tags | description`
    /// table covering every entry in the registry. Pure
    /// informational output — no snapshot is loaded; always
    /// returns 0.
    MetricList,
}

/// Arguments for the `ktstr ctprof show` subcommand. Field
/// shapes mirror [`ctprof_compare::CtprofCompareArgs`] so a
/// caller switching from compare to show keeps the same flag
/// vocabulary; the only structural difference is the single
/// positional path (one snapshot rather than baseline+candidate)
/// and the absence of any delta/percentage flags — show renders
/// one value per (group, metric) cell, no diff math.
/// Grouping axis for `ktstr ctprof show`. Show-specific
/// subset of [`ctprof_compare::GroupBy`] that OMITS the
/// compare-only `All` variant: `All` renders as a hierarchical
/// cgroup -> pcomm -> comm tree (the compound-key grouping in
/// [`ctprof_compare`]'s `build_groups` + the `write_primary_all`
/// tree renderer), and `write_show` carries no equivalent
/// hierarchy machinery. Modeling the show axis as its own
/// value-enum makes `--group-by all` unrepresentable: clap
/// rejects it at parse time with a possible-values list rather
/// than accepting a value `write_show` would panic on. The
/// four variants map 1:1 onto their [`ctprof_compare::GroupBy`]
/// counterparts via [`From`], and the kebab-case clap tokens
/// (`pcomm`, `cgroup`, `comm`, `comm-exact`) match the compare
/// subcommand's spelling for the same axes.
#[derive(Debug, Clone, Copy, PartialEq, Eq, clap::ValueEnum)]
pub enum ShowGroupBy {
    /// Group by process name (`pcomm`). See
    /// [`ctprof_compare::GroupBy::Pcomm`].
    Pcomm,
    /// Group by cgroup path. See
    /// [`ctprof_compare::GroupBy::Cgroup`].
    Cgroup,
    /// Group by thread-name pattern. See
    /// [`ctprof_compare::GroupBy::Comm`].
    Comm,
    /// Group by literal thread name. See
    /// [`ctprof_compare::GroupBy::CommExact`].
    CommExact,
}

impl From<ShowGroupBy> for ctprof_compare::GroupBy {
    fn from(g: ShowGroupBy) -> Self {
        match g {
            ShowGroupBy::Pcomm => ctprof_compare::GroupBy::Pcomm,
            ShowGroupBy::Cgroup => ctprof_compare::GroupBy::Cgroup,
            ShowGroupBy::Comm => ctprof_compare::GroupBy::Comm,
            ShowGroupBy::CommExact => ctprof_compare::GroupBy::CommExact,
        }
    }
}

#[derive(Debug, clap::Args)]
pub struct CtprofShowArgs {
    /// Snapshot path (`.ctprof.zst`) from `ktstr ctprof capture -o`.
    pub snapshot: std::path::PathBuf,
    /// Grouping key. Same semantics as
    /// `ktstr ctprof compare --group-by`. `pcomm` (default)
    /// aggregates per process name with token-based pattern
    /// normalization (so `worker-{0..N}` parents collapse into
    /// one `worker-{N}` bucket); `cgroup` per cgroup path; `comm`
    /// aggregates threads by NAME PATTERN under the same
    /// token-based normalizer; `comm-exact` is a synonym for
    /// `comm --no-thread-normalize` on the thread axis only.
    #[arg(long, value_enum, default_value_t = ShowGroupBy::Pcomm, help_heading = "Grouping")]
    pub group_by: ShowGroupBy,
    /// Glob patterns that collapse dynamic cgroup path segments —
    /// same shape as
    /// `ktstr ctprof compare --cgroup-flatten`. Repeatable.
    #[arg(long, help_heading = "Grouping")]
    pub cgroup_flatten: Vec<String>,
    /// Disable token-based pattern normalization across every
    /// name-family axis: `--group-by comm`, `--group-by pcomm`,
    /// AND the `## smaps_rollup` per-process keying (which
    /// normalizes by the pcomm pattern by default — see
    /// `ctprof_compare::collect_smaps_rollup`). With this flag
    /// set, threads / processes group by their literal name and
    /// smaps rows preserve their per-PID identity
    /// (`pcomm[tgid]`) instead of collapsing to the normalized
    /// pcomm pattern. The digit/hex/alpha-prefix placeholders
    /// are bypassed on every axis. Has no effect under
    /// `--group-by comm-exact` (already literal) or
    /// `--group-by cgroup`.
    #[arg(long, help_heading = "Grouping")]
    pub no_thread_normalize: bool,
    /// Disable token-based pattern normalization for the cgroup
    /// axis (`--group-by cgroup`). Cgroup paths group by literal
    /// post-`--cgroup-flatten` path. Has no effect under any
    /// other grouping.
    #[arg(long, help_heading = "Grouping")]
    pub no_cg_normalize: bool,
    /// Multi-key sort spec for the rendered groups. Format mirrors
    /// `ktstr ctprof compare --sort-by`:
    /// `metric1[:dir1],metric2[:dir2],...` where each `metric` is
    /// one of the primary or derived metric names (run
    /// `ctprof metric-list` for the full vocabulary) and
    /// `dir` is `asc` or `desc` (default `desc`). Show ranks
    /// groups by the tuple of the named metric's *absolute
    /// aggregated value* (not a delta — there is only one
    /// snapshot to inspect), descending by default; rows within
    /// a group keep registry order. Empty (the default) keeps
    /// the alphabetical group-key iteration. Example:
    /// `--sort-by run_time_ns,wait_sum:asc`.
    ///
    /// Affects only the per-thread metric table and the derived-
    /// metrics section. The `## smaps_rollup` sub-table sorts
    /// process rows independently by total Rss descending (its
    /// own built-in default); `--sort-by` does not propagate to
    /// it.
    #[arg(long, default_value = "", help_heading = "Display")]
    pub sort_by: String,
    /// Comma-separated column names to render. Empty (the
    /// default) emits the full `(group | threads | metric |
    /// value)` layout. Valid names: `group`, `threads`,
    /// `metric`, `value`. Order in the spec is the rendered
    /// order. Show is single-snapshot, so the
    /// `baseline`/`candidate`/`delta`/`%`/`arrow` columns
    /// (compare-only) are rejected at parse time. Example:
    /// `--columns metric,value`.
    #[arg(long, default_value = "", help_heading = "Display")]
    pub columns: String,
    /// Comma-separated section names to render. Empty (the
    /// default) renders every section that has data. When
    /// non-empty, restricts output to the listed sub-tables —
    /// every section not named is suppressed before its
    /// data-availability gate runs. Valid names: `primary`,
    /// `taskstats-delay`, `derived`, `cgroup-stats`,
    /// `cgroup-limits`, `memory-stat`, `memory-events`,
    /// `pressure`, `host-pressure`, `smaps-rollup`,
    /// `sched-ext`. Useful for narrowing a wide show to one
    /// area of interest. Example:
    /// `--sections primary,host-pressure`.
    #[arg(long, default_value = "", help_heading = "Filter")]
    pub sections: String,
    /// Comma-separated metric names to render. Empty (the
    /// default) renders every metric in the primary and
    /// derived sub-tables. When non-empty, restricts the
    /// rendered ROWS to the listed names — names must come
    /// from the `ctprof metric-list` vocabulary. Useful
    /// for zooming on a specific counter family without
    /// computing every metric: `--metrics
    /// run_time_ns,wait_sum,affine_success_ratio`. Composes
    /// with `--sections` — naming `--sections primary
    /// --metrics run_time_ns` shows a single primary row.
    #[arg(long, default_value = "", help_heading = "Filter")]
    pub metrics: String,
    /// Wrap table cells to fit the terminal width. Off by
    /// default — wide tables can spill past the terminal edge,
    /// matching the prior shell-pipeline-friendly layout. When
    /// set, cells too wide for the available width wrap inside
    /// the cell rather than overflowing, at the cost of taller
    /// rows. The wrap kicks in only when stdout is a tty (the
    /// terminal width is unknown otherwise); when piped to a
    /// file or another command, the flag is silently dropped
    /// and output stays unwrapped so awk/grep pipelines see
    /// the same byte sequence as without the flag.
    #[arg(long, help_heading = "Display")]
    pub wrap: bool,
    /// Maximum rendered lines per section. Sections whose table
    /// output exceeds this limit are truncated with a notice.
    /// Applies independently to each `## <heading>` sub-table.
    /// `0` disables truncation entirely. Default `500`.
    #[arg(long, default_value_t = 500, help_heading = "Display")]
    pub limit: usize,
}

/// Acquire source, configure, build, and cache a kernel image.
///
/// `version` accepts `MAJOR.MINOR[.PATCH][-rcN]`, a `MAJOR.MINOR`
/// prefix (resolves to the latest patch), or `START..END` for a
/// range that expands against kernel.org's `releases.json` to every
/// `stable` / `longterm` release inside the inclusive interval. A
/// range is detected via [`KernelId::parse`] and dispatched here to
/// [`kernel_build_one`] per resolved version, sharing the
/// download / cache-lookup / build pipeline that single-version
/// invocations use. Range mode collects per-version errors as a
/// best-effort summary: a build failure on one version is reported
/// and the iteration continues to the next, so a stale endpoint
/// doesn't block the rest of the range from caching. `--git` and
/// `--source` paths bypass range expansion (clap's
/// `conflicts_with` already rejects `version + source` and
/// `version + git` combinations).
#[allow(clippy::too_many_arguments)]
fn kernel_build(
    version: Option<String>,
    source: Option<PathBuf>,
    git: Option<String>,
    git_ref: Option<String>,
    force: bool,
    clean: bool,
    cpu_cap: Option<usize>,
    extra_kconfig: Option<PathBuf>,
    skip_sha256: bool,
) -> Result<()> {
    // Read the extra-kconfig fragment ONCE up front so a range
    // expansion doesn't re-read the same file per version (and so
    // a bad path surfaces before any download / build work fires).
    // [`ktstr::cli::read_extra_kconfig`] does the 4-arm error
    // classification (ENOENT/EISDIR/EACCES/UTF-8) and emits an
    // empty-file warning so a 0-byte fragment doesn't silently
    // produce an "extras present but nothing merged" build.
    let extra_content: Option<String> = match extra_kconfig.as_ref() {
        Some(p) => {
            Some(ktstr::cli::read_extra_kconfig(p, "ktstr").map_err(|e| anyhow::anyhow!("{e}"))?)
        }
        None => None,
    };

    if source.is_none()
        && git.is_none()
        && let Some(ref v) = version
    {
        use ktstr::kernel_path::KernelId;
        let id = KernelId::parse(v);
        // Validate before any I/O so an inverted range surfaces the
        // "swap the endpoints" diagnostic ahead of any download.
        id.validate()
            .map_err(|e| anyhow::anyhow!("--kernel {id}: {e}"))?;
        if let KernelId::Range { start, end, .. } = id {
            let versions = ktstr::cli::expand_kernel_range(&start, &end, "ktstr")?;
            let total = versions.len();
            let mut failures: Vec<(String, anyhow::Error)> = Vec::new();
            for (i, ver) in versions.iter().enumerate() {
                eprintln!("ktstr: [{}/{total}] kernel build {ver}", i + 1);
                if let Err(e) = kernel_build_one(
                    Some(ver.clone()),
                    None,
                    None,
                    None,
                    force,
                    clean,
                    cpu_cap,
                    extra_content.as_deref(),
                    skip_sha256,
                ) {
                    eprintln!("ktstr: {ver}: {e:#}");
                    failures.push((ver.clone(), e));
                }
            }
            if failures.is_empty() {
                Ok(())
            } else {
                anyhow::bail!(
                    "kernel build range {start}..{end}: {failed}/{total} \
                     version(s) failed: {names}",
                    failed = failures.len(),
                    names = failures
                        .iter()
                        .map(|(v, _)| v.as_str())
                        .collect::<Vec<_>>()
                        .join(", "),
                );
            }
        } else {
            kernel_build_one(
                version,
                source,
                git,
                git_ref,
                force,
                clean,
                cpu_cap,
                extra_content.as_deref(),
                skip_sha256,
            )
        }
    } else {
        kernel_build_one(
            version,
            source,
            git,
            git_ref,
            force,
            clean,
            cpu_cap,
            extra_content.as_deref(),
            skip_sha256,
        )
    }
}

/// Single-version variant of [`kernel_build`]: handles one tarball,
/// `--source`, or `--git` invocation. Carries the `kernel_build`
/// implementation as it stood before range dispatch was wired in;
/// extracted into a helper so the range loop in `kernel_build` can
/// reuse the same download + cache + build pipeline per resolved
/// version without duplicating it.
///
/// `extra_kconfig` is the pre-loaded user fragment from
/// `--extra-kconfig PATH` (read once in [`kernel_build`] before
/// fanning out to per-version invocations). `Some(content)` folds
/// into the cache key suffix via
/// [`ktstr::cache_key_suffix_with_extra`] and into the configure
/// pass via the Cow merge construction in
/// [`ktstr::cli::kernel_build_pipeline`].
#[allow(clippy::too_many_arguments)]
fn kernel_build_one(
    version: Option<String>,
    source: Option<PathBuf>,
    git: Option<String>,
    git_ref: Option<String>,
    force: bool,
    clean: bool,
    cpu_cap: Option<usize>,
    extra_kconfig: Option<&str>,
    skip_sha256: bool,
) -> Result<()> {
    use ktstr::cache::CacheDir;
    use ktstr::fetch;

    // Resolve the CLI --cpu-cap flag against KTSTR_CPU_CAP env and
    // the implicit "no cap" default. Conflict with
    // KTSTR_BYPASS_LLC_LOCKS=1 surfaces here so operators see the
    // parse-time error, not an opaque pipeline bail later.
    if cpu_cap.is_some() && ktstr::bypass_llc_locks_active() {
        anyhow::bail!(
            "--cpu-cap conflicts with KTSTR_BYPASS_LLC_LOCKS=1; unset one of them. \
             --cpu-cap is a resource contract; bypass disables the contract entirely."
        );
    }
    let resolved_cap = cli::CpuCap::resolve(cpu_cap)?;

    let cache = CacheDir::new()?;

    // Temporary directory for tarball/git source extraction.
    let tmp_dir = tempfile::TempDir::new()?;

    // Acquire source.
    let client = fetch::shared_client();
    let mut acquired = if let Some(ref src_path) = source {
        fetch::local_source(src_path)?
    } else if let Some(ref url) = git {
        let ref_name = git_ref.as_deref().expect("clap requires --ref with --git");
        fetch::git_clone(url, ref_name, tmp_dir.path(), "ktstr")?
    } else {
        // Tarball download: explicit version, prefix, or latest stable.
        let ver = match version {
            Some(v) if fetch::is_major_minor_prefix(&v) => {
                // Major.minor prefix (e.g., "6.12") — resolve latest patch.
                fetch::fetch_version_for_prefix(client, &v, "ktstr")?
            }
            Some(v) => v,
            None => fetch::fetch_latest_stable_version(client, "ktstr")?,
        };
        // Check cache before downloading. Cache key folds in the
        // merged-kconfig hash so an `--extra-kconfig` build looks
        // up a distinct slot from a vanilla baked-in-only build —
        // `cache_key_suffix_with_extra(None)` equals
        // `cache_key_suffix()` so the no-extra path is byte-
        // identical to pre-flag behavior.
        let (arch, _) = fetch::arch_info();
        let cache_key = format!(
            "{ver}-tarball-{arch}-kc{}",
            ktstr::cache_key_suffix_with_extra(extra_kconfig),
        );
        if !force && let Some(entry) = cli::cache_lookup(&cache, &cache_key, "ktstr") {
            eprintln!("ktstr: cached kernel found: {}", entry.path.display());
            eprintln!("ktstr: use --force to rebuild");
            return Ok(());
        }
        let sp = cli::Spinner::start("Downloading kernel...");
        let result = fetch::download_tarball(client, &ver, tmp_dir.path(), "ktstr", skip_sha256);
        drop(sp);
        let mut acquired = result?;
        // `download_tarball` builds its `cache_key` against the bare
        // `cache_key_suffix()` (see `fetch::download_tarball`).
        // Override with the merged-suffix key we looked up under so
        // the post-build cache store lands at the same slot we'd
        // hit on a re-run with the same `--extra-kconfig`.
        acquired.cache_key = cache_key;
        acquired
    };

    // For `--source` and `--git` paths, `local_source` and `git_clone`
    // build `acquired.cache_key` against the bare `cache_key_suffix()`
    // — already shaped `...-kc{baked_hash}`. With `--extra-kconfig`
    // set, lift the `-xkc{extra_hash}` append to
    // [`cli::append_extra_kconfig_suffix`] so both binaries share
    // one merge path.
    if source.is_some() || git.is_some() {
        cli::append_extra_kconfig_suffix(&mut acquired.cache_key, extra_kconfig);
    }

    // Check cache for --source and --git (tarball already checked above).
    if !force
        && (source.is_some() || git.is_some())
        && !acquired.is_dirty
        && let Some(entry) = cli::cache_lookup(&cache, &acquired.cache_key, "ktstr")
    {
        eprintln!("ktstr: cached kernel found: {}", entry.path.display());
        eprintln!("ktstr: use --force to rebuild");
        return Ok(());
    }

    // `--force` fail-fast pre-check: if tests are actively holding
    // the cache-entry lock, bail with the PID list instead of
    // silently waiting to stomp the in-use entry. The returned
    // guard drops at the end of this `if` scope before
    // `kernel_build_pipeline` runs; `store()` inside the pipeline
    // takes its own (now-uncontested) blocking lock. The brief
    // window between drop and re-take is acceptable for an
    // interactive `--force` operator action.
    if force {
        let _force_check = cache.try_acquire_exclusive_lock(&acquired.cache_key)?;
    }

    cli::kernel_build_pipeline(
        &acquired,
        &cache,
        "ktstr",
        clean,
        source.is_some(),
        resolved_cap,
        extra_kconfig,
    )?;

    Ok(())
}

fn run_completions(shell: clap_complete::Shell, binary: &str) {
    let mut cmd = Cli::command();
    clap_complete::generate(shell, &mut cmd, binary, &mut std::io::stdout());
}

/// Entry point for `ktstr ctprof show <snapshot>`. Loads one
/// snapshot, builds groups along the requested axis, aggregates
/// every metric per
/// [`ctprof_compare::CTPROF_METRICS`], and renders the
/// result as a `(group_key, threads, metric, value)` table on
/// stdout. Exits non-zero only on I/O / parse errors; the
/// rendered table is always considered successful output.
///
/// Reuses the grouping pipeline from
/// [`ctprof_compare`] so pattern normalization (`comm` Layer-2
/// token clustering, `cgroup` Layer-1/2/3 tightening) and
/// `--cgroup-flatten` glob handling stay identical to the compare
/// path.
fn run_show(args: &CtprofShowArgs) -> Result<i32> {
    use anyhow::Context;
    // Parse `--sort-by` BEFORE the snapshot load so an operator
    // typo fails fast without paying for disk I/O. Mirrors
    // run_compare's ordering for the same reason. `--columns`
    // and `--sections` share the fail-fast contract — an invalid
    // spec must surface before the snapshot read.
    let sort_by = ctprof_compare::parse_sort_by(&args.sort_by)
        .with_context(|| format!("parse --sort-by {:?}", args.sort_by))?;
    // `--columns` parses with compare_side=false so the
    // baseline/candidate/delta/% columns (compare-only) are
    // rejected before any snapshot work begins.
    let columns = ctprof_compare::parse_columns(&args.columns, false)
        .with_context(|| format!("parse --columns {:?}", args.columns))?;
    let sections = ctprof_compare::parse_sections(&args.sections)
        .with_context(|| format!("parse --sections {:?}", args.sections))?;
    let metrics = ctprof_compare::parse_metrics(&args.metrics)
        .with_context(|| format!("parse --metrics {:?}", args.metrics))?;

    // Convert the show-specific grouping axis into the shared
    // `ctprof_compare::GroupBy` once at the boundary. `ShowGroupBy`
    // omits the compare-only `All` variant, so this conversion
    // never produces `GroupBy::All` and the downstream
    // `write_show` `All` arm stays unreachable.
    let group_by: ctprof_compare::GroupBy = args.group_by.into();

    // Mirror run_compare's pre-load warning: explicit
    // `--sections cgroup-stats` (or any other cgroup-only
    // section) under a non-cgroup `--group-by` would render
    // zero rows for that section silently. Surface a
    // diagnostic before the snapshot load so the operator sees
    // it immediately.
    ctprof_compare::warn_cgroup_only_sections_under_non_cgroup(&sections, group_by);

    let snap = ktstr::ctprof::CtprofSnapshot::load(&args.snapshot)
        .with_context(|| format!("load snapshot {}", args.snapshot.display()))?;
    let mut out = String::new();
    // Infallible: writing into a String cannot fail.
    let _ = write_show(
        &mut out,
        &snap,
        group_by,
        &args.cgroup_flatten,
        args.no_thread_normalize,
        args.no_cg_normalize,
        &sort_by,
        &columns,
        &sections,
        &metrics,
        args.wrap,
    );
    if args.limit > 0 {
        print!("{}", ctprof_compare::limit_sections(&out, args.limit));
    } else {
        print!("{out}");
    }
    Ok(0)
}

/// Render the per-group / per-metric show table into `w`. Mirrors
/// [`ctprof_compare::write_diff`]'s shape — the formatter
/// layer is split from the I/O wrapper so unit tests can drive
/// rendering into a `String` buffer without shelling through
/// stdout. Write errors propagate as [`std::fmt::Error`]; callers
/// that write into an infallible sink (`String`) can ignore.
///
/// `sections` is a [`ctprof_compare::Section`] filter — empty
/// renders every section that has data, non-empty restricts to
/// the named entries (mirrors `--sections` on `write_diff`).
/// `metrics` is a row-level filter — empty renders every metric
/// in the primary and derived sub-tables, non-empty restricts
/// the rendered rows to the listed names. `wrap` chooses between
/// `Disabled` (fixed-width, default) and `Dynamic`
/// (terminal-width-aware) comfy-table arrangement.
#[allow(clippy::too_many_arguments)]
fn write_show<W: std::fmt::Write>(
    w: &mut W,
    snap: &ktstr::ctprof::CtprofSnapshot,
    group_by: ctprof_compare::GroupBy,
    cgroup_flatten: &[String],
    no_thread_normalize: bool,
    no_cg_normalize: bool,
    sort_by: &[ctprof_compare::SortKey],
    columns: &[ctprof_compare::Column],
    sections: &[ctprof_compare::Section],
    metrics: &[&'static str],
    wrap: bool,
) -> std::fmt::Result {
    let flatten = ctprof_compare::compile_flatten_patterns(cgroup_flatten);
    // Single-snapshot cgroup key map. compare()'s
    // `build_cgroup_key_map` walks the union of paths from two
    // snapshots; for show we feed the same snap on both arguments
    // so the union iteration produces the same key set as a
    // single-snap pass (the inner BTreeSet of paths dedups the
    // duplicate insertions). Skipped under `--no-cg-normalize`
    // and under any grouping other than Cgroup.
    let cgroup_key_map = if group_by == ctprof_compare::GroupBy::Cgroup && !no_cg_normalize {
        Some(ctprof_compare::build_cgroup_key_map(snap, snap, &flatten))
    } else {
        None
    };
    // Pattern counts: `build_groups` falls back to a per-snapshot
    // local count when None, which is exactly the right behavior
    // for a single-snapshot view (compare()'s union-over-baseline-
    // and-candidate gate doesn't apply when there's only one
    // snapshot to inspect).
    let groups = ctprof_compare::build_groups(
        snap,
        group_by,
        &flatten,
        None,
        cgroup_key_map.as_ref(),
        no_thread_normalize,
    );

    let group_header = match group_by {
        ctprof_compare::GroupBy::Pcomm => "pcomm",
        ctprof_compare::GroupBy::Cgroup => "cgroup",
        ctprof_compare::GroupBy::Comm => "comm-pattern",
        ctprof_compare::GroupBy::CommExact => "comm",
        ctprof_compare::GroupBy::All => unreachable!("All is decomposed before write_show"),
    };

    // Resolve the column set: caller-supplied override wins,
    // otherwise fall back to the show-side default
    // `(group, threads, metric, value)`. DisplayOptions is
    // marked `#[non_exhaustive]` so we mutate the
    // default-constructed struct rather than a struct
    // expression. Plumbing wrap/sections through DisplayOptions
    // means every section's table-builder call site below routes
    // through `display_options.new_table()` (the wrap-aware
    // helper) and every section emission gates on
    // `display_options.is_section_enabled(...)`.
    let mut display_options = ctprof_compare::DisplayOptions::default();
    display_options.columns = columns.to_vec();
    display_options.sections = sections.to_vec();
    display_options.metrics = metrics.to_vec();
    display_options.wrap = wrap;
    let resolved_columns = display_options.resolved_show_columns();

    // Iteration order: when `sort_by` is empty, fall through the
    // BTreeMap by-key order (alphabetical group key, registry order
    // metrics); when non-empty, rank groups by the tuple of named
    // metrics' *absolute aggregated values* (no deltas — show has
    // one snapshot, not two). Mirrors compare's
    // `sort_diff_rows_by_keys` shape (per-key direction, lex order
    // on the tuple, deterministic group_key tie-break) but on
    // `Aggregated::numeric()` rather than `DiffRow.delta`. Within a
    // group, metrics still iterate `CTPROF_METRICS` so the
    // metric column lands in registry order regardless of sort.
    //
    // Computed once outside the section gates because every
    // section that iterates groups (Primary + Derived) reuses
    // it; computing inside the Primary `if` would either
    // duplicate the work or leak the binding into a scope where
    // a `--sections derived` invocation skips the Primary block
    // entirely and then can't reach `group_order`.
    let group_order: Vec<&String> = if sort_by.is_empty() {
        groups.keys().collect()
    } else {
        let mut keys: Vec<&String> = groups.keys().collect();
        // Per-group sort tuple: missing values (a metric absent from
        // this group's `metrics` map, or one whose `Aggregated`
        // returns `None` from `numeric()` — categorical Mode etc.)
        // sink to the bottom for the requested direction, matching
        // `sort_diff_rows_by_keys`'s NEG_INFINITY/INFINITY handling.
        let group_tuple = |group_key: &str| -> Vec<f64> {
            let group = groups.get(group_key);
            sort_by
                .iter()
                .map(|k| {
                    group
                        .and_then(|g| g.metrics.get(k.metric))
                        .and_then(|a| a.numeric())
                        .unwrap_or(if k.descending {
                            f64::NEG_INFINITY
                        } else {
                            f64::INFINITY
                        })
                })
                .collect()
        };
        keys.sort_by(|a, b| {
            let ta = group_tuple(a);
            let tb = group_tuple(b);
            for (i, key) in sort_by.iter().enumerate() {
                let (va, vb) = (ta[i], tb[i]);
                let ord = if key.descending {
                    vb.partial_cmp(&va).unwrap_or(std::cmp::Ordering::Equal)
                } else {
                    va.partial_cmp(&vb).unwrap_or(std::cmp::Ordering::Equal)
                };
                if ord != std::cmp::Ordering::Equal {
                    return ord;
                }
            }
            // Final tie-break: ascending group_key for determinism.
            a.cmp(b)
        });
        keys
    };

    show_render::write_show_primary(
        w,
        &display_options,
        &groups,
        &group_order,
        &resolved_columns,
        group_header,
        group_by,
        no_thread_normalize,
    )?;

    show_render::write_show_derived(
        w,
        &display_options,
        &groups,
        &group_order,
        &resolved_columns,
        group_header,
        group_by,
        no_thread_normalize,
    )?;

    show_render::write_show_cgroup_sections(
        w,
        &display_options,
        snap,
        group_by,
        &flatten,
        cgroup_key_map.as_ref(),
    )?;

    show_render::write_show_host_pressure(w, &display_options, snap)?;

    show_render::write_show_smaps(w, &display_options, snap, no_thread_normalize)?;

    show_render::write_show_sched_ext(w, &display_options, snap)?;

    Ok(())
}

/// One entry in the [`psi_resources`] table — a display name
/// paired with the accessor that pulls one
/// [`ctprof::PsiResource`] out of a [`ctprof::Psi`]
/// bundle.
type PsiAccessor = (&'static str, fn(&ctprof::Psi) -> ctprof::PsiResource);

/// Returns the four PSI resource accessors paired with their
/// display names. Centralizing the resource list keeps the
/// host-level and per-cgroup rendering paths in lockstep — adding
/// a fifth resource (e.g. a future `net.pressure`) means one
/// edit here, not four scattered through the renderers.
fn psi_resources() -> [PsiAccessor; 4] {
    [
        ("cpu", |p| p.cpu),
        ("memory", |p| p.memory),
        ("io", |p| p.io),
        ("irq", |p| p.irq),
    ]
}

/// Render a centi-percent PSI average as `N.NN%`. The kernel
/// emits `LOAD_INT.LOAD_FRAC` from the `psi_show` seq_printf in
/// `kernel/sched/psi.c` with 2-decimal-digit precision — preserve
/// that on display.
fn format_psi_avg(centi_percent: u16) -> String {
    let int = centi_percent / 100;
    let frac = centi_percent % 100;
    format!("{int}.{frac:02}%")
}

/// Returns true when any host-level PSI resource has a non-zero
/// avg or total reading. Used to suppress the host pressure
/// section when the kernel returned all zeros (CONFIG_PSI off,
/// PSI not yet warmed up, or synthetic test fixture).
fn host_psi_has_data(psi: &ctprof::Psi) -> bool {
    [psi.cpu, psi.memory, psi.io, psi.irq]
        .iter()
        .any(psi_resource_has_data)
}

fn psi_resource_has_data(r: &ctprof::PsiResource) -> bool {
    let h =
        |h: &ctprof::PsiHalf| h.avg10 != 0 || h.avg60 != 0 || h.avg300 != 0 || h.total_usec != 0;
    h(&r.some) || h(&r.full)
}

#[cfg(test)]
mod psi_show_tests {
    //! Tests for the show-side PSI helpers
    //! (`psi_resources`, `format_psi_avg`,
    //! `host_psi_has_data`, `psi_resource_has_data`). The
    //! integration tests for the full show pipeline already
    //! exercise the rendering happy path; these unit tests pin
    //! the helper boundaries (zero-suppression edges,
    //! centi-percent → display rounding) directly so a
    //! regression in either function surfaces with a
    //! pinpointed assertion rather than a rendered-output
    //! diff.
    use super::*;
    use ktstr::ctprof::{Psi, PsiHalf, PsiResource};

    #[test]
    fn format_psi_avg_renders_centi_percent_with_two_decimal_digits() {
        // Lossless N.NN% display: integer + 2-digit fraction.
        assert_eq!(format_psi_avg(0), "0.00%");
        assert_eq!(format_psi_avg(1), "0.01%");
        assert_eq!(format_psi_avg(50), "0.50%");
        assert_eq!(format_psi_avg(1859), "18.59%");
        assert_eq!(format_psi_avg(10000), "100.00%");
        // Kernel EWMA rounding ceiling from `calc_load` in
        // include/linux/sched/loadavg.h; render the boundary
        // verbatim rather than clamping.
        assert_eq!(format_psi_avg(10099), "100.99%");
    }

    #[test]
    fn psi_resources_lists_four_in_canonical_order() {
        let names: Vec<&str> = psi_resources().iter().map(|(n, _)| *n).collect();
        assert_eq!(names, vec!["cpu", "memory", "io", "irq"]);
    }

    /// Build a `PsiResource` with one centi-percent sentinel set
    /// on either the `some` or `full` half. The bin crate is a
    /// distinct compilation unit from the lib, so the
    /// `#[non_exhaustive]` markers on `Psi` / `PsiResource` /
    /// `PsiHalf` block struct-literal construction here — same
    /// constraint that drives `tests/common/ctprof.rs`'s
    /// Default + per-field assignment pattern.
    fn psi_resource_with_some_avg10(v: u16) -> PsiResource {
        let mut half = PsiHalf::default();
        half.avg10 = v;
        let mut r = PsiResource::default();
        r.some = half;
        r
    }

    fn psi_resource_with_full_avg10(v: u16) -> PsiResource {
        let mut half = PsiHalf::default();
        half.avg10 = v;
        let mut r = PsiResource::default();
        r.full = half;
        r
    }

    #[test]
    fn psi_resources_accessors_route_to_correct_field() {
        // Distinct sentinel per resource so a swapped accessor
        // surfaces as a wrong-resource value here.
        let mut psi = Psi::default();
        psi.cpu = psi_resource_with_some_avg10(1);
        psi.memory = psi_resource_with_some_avg10(2);
        psi.io = psi_resource_with_some_avg10(3);
        psi.irq = psi_resource_with_full_avg10(4);
        let accessors = psi_resources();
        assert_eq!(accessors[0].1(&psi).some.avg10, 1, "cpu accessor");
        assert_eq!(accessors[1].1(&psi).some.avg10, 2, "memory accessor");
        assert_eq!(accessors[2].1(&psi).some.avg10, 3, "io accessor");
        assert_eq!(accessors[3].1(&psi).full.avg10, 4, "irq accessor");
    }

    #[test]
    fn host_psi_has_data_returns_false_for_all_zero_bundle() {
        assert!(!host_psi_has_data(&Psi::default()));
    }

    #[test]
    fn host_psi_has_data_returns_true_for_any_nonzero_field() {
        // Each per-half field flagged independently so a
        // regression that only checked one half (or one resource)
        // surfaces here.
        for resource_idx in 0..4 {
            for is_full in [false, true] {
                let mut psi = Psi::default();
                let target_resource = match resource_idx {
                    0 => &mut psi.cpu,
                    1 => &mut psi.memory,
                    2 => &mut psi.io,
                    _ => &mut psi.irq,
                };
                let half = if is_full {
                    &mut target_resource.full
                } else {
                    &mut target_resource.some
                };
                half.total_usec = 1;
                assert!(
                    host_psi_has_data(&psi),
                    "resource_idx={resource_idx}, is_full={is_full} should be detected"
                );
            }
        }
    }
}

fn main() -> Result<()> {
    // Restore SIGPIPE so piping ktstr output to `head` / `less` /
    // similar doesn't panic inside `print!`. Shared helper lives
    // in `cli::restore_sigpipe_default`; see that doc for the
    // rationale + SAFETY text.
    ktstr::cli::restore_sigpipe_default();
    tracing_subscriber::fmt()
        .with_env_filter(
            tracing_subscriber::EnvFilter::try_from_default_env()
                .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")),
        )
        .with_writer(std::io::stderr)
        .init();

    let args = Cli::parse();

    match args.command {
        Command::Topo => {
            let topo = TestTopology::from_system()?;
            println!("CPUs:       {}", topo.total_cpus());
            println!("LLCs:       {}", topo.num_llcs());
            println!("NUMA nodes: {}", topo.num_numa_nodes());
            for (i, llc) in topo.llcs().iter().enumerate() {
                println!("  LLC {} (node {}): {:?}", i, llc.numa_node(), llc.cpus(),);
            }
        }

        Command::Kernel { command } => match command {
            KernelCommand::List { json, range } => match range {
                Some(r) => cli::kernel_list_range_preview(json, &r)?,
                None => cli::kernel_list(json)?,
            },
            KernelCommand::Build {
                version,
                source,
                git,
                git_ref,
                force,
                clean,
                cpu_cap,
                extra_kconfig,
                skip_sha256,
            } => kernel_build(
                version,
                source,
                git,
                git_ref,
                force,
                clean,
                cpu_cap,
                extra_kconfig,
                skip_sha256,
            )?,
            KernelCommand::Clean {
                keep,
                force,
                corrupt_only,
            } => cli::kernel_clean(keep, force, corrupt_only)?,
        },

        Command::Shell {
            kernel,
            topology,
            include_files,
            memory_mib,
            dmesg,
            exec,
            exec_timeout,
            no_perf_mode,
            cpu_cap,
            disk,
        } => {
            if no_perf_mode {
                // SAFETY: single-threaded at this point — no concurrent env readers.
                unsafe { std::env::set_var(ktstr::KTSTR_NO_PERF_MODE_ENV, "1") };
            }
            if let Some(cap) = cpu_cap {
                // Parse-time conflict with KTSTR_BYPASS_LLC_LOCKS — see
                // kernel_build fn for the same check on the build path.
                if ktstr::bypass_llc_locks_active() {
                    anyhow::bail!(
                        "--cpu-cap conflicts with KTSTR_BYPASS_LLC_LOCKS=1; unset \
                         one of them. --cpu-cap is a resource contract; bypass \
                         disables the contract entirely."
                    );
                }
                // Validate the cap up front via CpuCap::new so a value
                // of 0 or a bogus KTSTR_CPU_CAP env overlay surfaces
                // at CLI-parse time, not deep inside VM build. The
                // CpuCap value itself is passed to the VMM via
                // KTSTR_CPU_CAP — KtstrVmBuilder::build re-resolves it
                // from the env there, so every resolve path (direct
                // CLI, env overlay, nested exec) agrees on precedence.
                cli::CpuCap::new(cap)?;
                // SAFETY: single-threaded at this point — no concurrent env readers.
                unsafe { std::env::set_var(ktstr::KTSTR_CPU_CAP_ENV, cap.to_string()) };
            }
            // Shared parse path with `cargo ktstr shell` — surfaces
            // size errors at CLI-argument time, never mid-VM-setup.
            let disk_cfg = cli::parse_disk_arg(disk.as_deref())?;
            cli::check_kvm()?;
            let kernel_path = cli::resolve_kernel_image(
                kernel.as_deref(),
                &cli::KernelResolvePolicy {
                    accept_raw_image: false,
                    cli_label: "ktstr",
                },
            )?;

            let (numa_nodes, llcs, cores, threads) = cli::parse_topology_string(&topology)?;

            let resolved_includes = cli::resolve_include_files(&include_files)?;

            let include_refs: Vec<(&str, &Path)> = resolved_includes
                .iter()
                .map(|(a, p)| (a.as_str(), p.as_path()))
                .collect();

            let exec_code = ktstr::run_shell(
                kernel_path,
                numa_nodes,
                llcs,
                cores,
                threads,
                &include_refs,
                memory_mib,
                dmesg,
                exec.as_deref(),
                exec_timeout,
                disk_cfg,
                // No test-entry context at the operator-driven
                // `ktstr shell` entry — scheduler enable/disable +
                // wprof_args overrides + performance_mode are
                // strictly test-descriptor-driven, so pass the
                // empty/false defaults the lib's run_shell uses on
                // no-test-entry paths.
                None,
                false,
                &[],
                &[],
            )?;
            // Exit with the guest payload's own exit code (recovered from
            // the ExecExit bulk frame in run_interactive); interactive
            // mode (None) exits 0.
            std::process::exit(exec_code.unwrap_or(0));
        }

        Command::Ctprof { command } => match command {
            CtprofCommand::Capture { output } => {
                ctprof::capture_to(&output)?;
                eprintln!("ktstr: wrote ctprof snapshot to {}", output.display());
            }
            CtprofCommand::Compare(args) => {
                let code = ctprof_compare::run_compare(&args)?;
                if code != 0 {
                    std::process::exit(code);
                }
            }
            CtprofCommand::Show(args) => {
                let code = run_show(&args)?;
                if code != 0 {
                    std::process::exit(code);
                }
            }
            CtprofCommand::MetricList => {
                let code = ctprof_compare::run_metric_list()?;
                if code != 0 {
                    std::process::exit(code);
                }
            }
        },

        Command::Completions { shell, binary } => {
            run_completions(shell, &binary);
        }

        Command::Locks { json, watch } => cli::list_locks(json, watch)?,
    }

    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use ktstr::metric_types::{MonotonicCount, MonotonicNs};

    // -- clap argument-parse pins: Shell --cpu-cap requires --no-perf-mode
    //
    // Mirror of the same constraint on the `cargo ktstr shell`
    // subcommand. ktstr and cargo-ktstr define separate Shell
    // variants (each binary has its own clap tree), so the
    // `requires = "no_perf_mode"` attribute must be pinned on BOTH
    // sides. A drift between the two — e.g. cargo-ktstr keeps the
    // requires, ktstr loses it — would surface here without waiting
    // for a runtime double-reservation bug.

    /// `ktstr shell --cpu-cap 4 --no-perf-mode` parses successfully.
    /// Positive-path pin for the standalone `ktstr` binary's Shell
    /// subcommand — complements the cargo-ktstr mirror test.
    #[test]
    fn parse_shell_cpu_cap_with_no_perf_mode_succeeds() {
        let parsed = Cli::try_parse_from(["ktstr", "shell", "--cpu-cap", "4", "--no-perf-mode"])
            .unwrap_or_else(|e| panic!("{e}"));
        match parsed.command {
            Command::Shell {
                cpu_cap,
                no_perf_mode,
                ..
            } => {
                assert_eq!(cpu_cap, Some(4));
                assert!(no_perf_mode, "--no-perf-mode must be set");
            }
            _ => panic!("expected Shell"),
        }
    }

    /// `ktstr shell --cpu-cap 4` without `--no-perf-mode` FAILS at
    /// parse time via the `requires = "no_perf_mode"` constraint.
    /// Negative-path pin — a regression that drops the requires
    /// attribute would allow the command to parse and then
    /// double-reserve under perf-mode at runtime.
    #[test]
    fn parse_shell_cpu_cap_without_no_perf_mode_fails() {
        // `Cli` intentionally has no Debug derive, so unwrap
        // helpers that format the Ok variant are unavailable.
        // Match on Err directly to extract the clap error.
        let msg = match Cli::try_parse_from(["ktstr", "shell", "--cpu-cap", "4"]) {
            Err(e) => e.to_string(),
            Ok(_) => panic!("--cpu-cap without --no-perf-mode must fail the parse"),
        };
        assert!(
            msg.to_ascii_lowercase().contains("no-perf-mode")
                || msg.to_ascii_lowercase().contains("no_perf_mode"),
            "clap error must name the missing --no-perf-mode flag, got: {msg}",
        );
    }

    /// `ktstr shell --no-perf-mode` without `--cpu-cap` parses
    /// successfully with `cpu_cap: None`. Pins the shape of the
    /// unset sentinel (expanded to the 30% default by the planner)
    /// — a regression that made --cpu-cap mandatory-with-no-perf-mode
    /// would break the shared-runner path that uses --no-perf-mode
    /// alone.
    #[test]
    fn parse_shell_no_perf_mode_without_cpu_cap_succeeds() {
        let parsed = Cli::try_parse_from(["ktstr", "shell", "--no-perf-mode"])
            .unwrap_or_else(|e| panic!("{e}"));
        match parsed.command {
            Command::Shell {
                cpu_cap,
                no_perf_mode,
                ..
            } => {
                assert_eq!(cpu_cap, None, "no --cpu-cap must produce None");
                assert!(no_perf_mode);
            }
            _ => panic!("expected Shell"),
        }
    }

    /// `ktstr shell --memory-mb 256` is rejected because the
    /// canonical flag was renamed to `--memory-mib`. Pre-1.0
    /// break-cleanly forbids a compat alias for the old form. A
    /// regression that re-added `alias = "memory-mb"` to the
    /// Shell variant's `memory_mib` field would silently accept
    /// the old spelling — this test pins the reject. Mirrors
    /// `parse_shell_memory_mb_rejected` in the cargo-ktstr binary
    /// for symmetric coverage across both binaries.
    #[test]
    fn parse_shell_memory_mb_rejected() {
        let rejected = Cli::try_parse_from(["ktstr", "shell", "--memory-mb", "256"]);
        assert!(
            rejected.is_err(),
            "`--memory-mb` (old flag name) must be rejected — the \
             canonical name is `--memory-mib`. A regression that \
             re-added an alias for the old form would surface here.",
        );
    }

    // -- ctprof show CLI tests
    //
    // Pin clap-parse shape on the `Show` variant + the rendered
    // output of `write_show` against a synthetic snapshot. The
    // CLI-parse tests guard against argument-vocabulary drift
    // (matching the pattern compare's args use) and the
    // write_show tests guard the rendered table against
    // regressions in column count, header, and group-axis routing.

    /// `ktstr ctprof show <path>` with no flags must parse
    /// into a Show variant carrying the positional path and
    /// default GroupBy::Pcomm. Positive-path pin for argv shape.
    #[test]
    fn parse_ctprof_show_positional_only_succeeds() {
        let parsed = Cli::try_parse_from(["ktstr", "ctprof", "show", "/tmp/snap.ctprof.zst"])
            .unwrap_or_else(|e| panic!("{e}"));
        match parsed.command {
            Command::Ctprof {
                command: CtprofCommand::Show(args),
            } => {
                assert_eq!(
                    args.snapshot,
                    std::path::PathBuf::from("/tmp/snap.ctprof.zst")
                );
                assert_eq!(args.group_by, ShowGroupBy::Pcomm);
                assert!(args.cgroup_flatten.is_empty());
                assert!(!args.no_thread_normalize);
                assert!(!args.no_cg_normalize);
                // `--sort-by` defaults to the empty string —
                // empty-spec sentinel for "fall through to
                // alphabetical iteration".
                assert!(args.sort_by.is_empty());
            }
            _ => panic!("expected Ctprof/Show"),
        }
    }

    /// `ktstr ctprof show --group-by all` must be REJECTED at clap
    /// parse time. `all` is a compare-only axis (it renders as a
    /// cgroup -> pcomm -> comm tree via `write_primary_all`); show
    /// has no hierarchy renderer, so `write_show`'s `GroupBy::All`
    /// arm is `unreachable!()`. Before the fix, `CtprofShowArgs`
    /// reused the full `GroupBy` value-enum, clap accepted `all`,
    /// and `run_show` -> `write_show` panicked on the unreachable
    /// arm. Modeling the show axis as `ShowGroupBy` (no `All`)
    /// makes the value unrepresentable: clap fails the parse with a
    /// possible-values diagnostic instead. This test fails (parse
    /// succeeds) on the pre-fix code and passes after.
    #[test]
    fn parse_ctprof_show_group_by_all_is_rejected() {
        let parsed = Cli::try_parse_from([
            "ktstr",
            "ctprof",
            "show",
            "/tmp/snap.ctprof.zst",
            "--group-by",
            "all",
        ]);
        // `let Err` rather than `expect_err`/`unwrap_err`: those
        // require the Ok type (`Cli`) to be `Debug`, which it is not.
        let Err(err) = parsed else {
            panic!(
                "`--group-by all` is a compare-only axis with no show \
                 renderer; it must be rejected at parse time, not \
                 accepted and panicked on in write_show"
            );
        };
        // clap's invalid-value error for a value-enum carries the
        // offending token and the accepted set; assert on the
        // offending value so the message is operator-actionable.
        let msg = err.to_string();
        assert!(
            msg.contains("all"),
            "rejection message must name the invalid value `all`, got: {msg}",
        );
    }

    /// Every axis show DOES support parses into the matching
    /// `ShowGroupBy` variant. Companion to the `all`-rejection
    /// test: pins that narrowing the value-enum did not also drop
    /// a legitimate axis. A regression that, say, removed
    /// `CommExact` from `ShowGroupBy` would surface here as a
    /// parse failure for `comm-exact`.
    #[test]
    fn parse_ctprof_show_group_by_supported_axes() {
        for (token, expected) in [
            ("pcomm", ShowGroupBy::Pcomm),
            ("cgroup", ShowGroupBy::Cgroup),
            ("comm", ShowGroupBy::Comm),
            ("comm-exact", ShowGroupBy::CommExact),
        ] {
            let parsed = Cli::try_parse_from([
                "ktstr",
                "ctprof",
                "show",
                "/tmp/snap.ctprof.zst",
                "--group-by",
                token,
            ])
            .unwrap_or_else(|e| panic!("`--group-by {token}` must parse: {e}"));
            match parsed.command {
                Command::Ctprof {
                    command: CtprofCommand::Show(args),
                } => assert_eq!(
                    args.group_by, expected,
                    "`--group-by {token}` must parse to {expected:?}",
                ),
                _ => panic!("expected Ctprof/Show for `--group-by {token}`"),
            }
        }
    }

    /// Every `ShowGroupBy` variant converts to the matching
    /// `ctprof_compare::GroupBy` and NONE maps to `All`. Pins the
    /// `From<ShowGroupBy>` boundary `run_show` relies on: because
    /// the conversion can never yield `GroupBy::All`, the
    /// `unreachable!()` arm in `write_show` is genuinely
    /// unreachable on the show path.
    #[test]
    fn show_group_by_converts_to_compare_group_by_without_all() {
        for (show, expected) in [
            (ShowGroupBy::Pcomm, ctprof_compare::GroupBy::Pcomm),
            (ShowGroupBy::Cgroup, ctprof_compare::GroupBy::Cgroup),
            (ShowGroupBy::Comm, ctprof_compare::GroupBy::Comm),
            (ShowGroupBy::CommExact, ctprof_compare::GroupBy::CommExact),
        ] {
            let got: ctprof_compare::GroupBy = show.into();
            assert_eq!(got, expected, "{show:?} must convert to {expected:?}");
            assert_ne!(
                got,
                ctprof_compare::GroupBy::All,
                "no ShowGroupBy variant may convert to GroupBy::All",
            );
        }
    }

    /// `--group-by`, `--cgroup-flatten`, `--no-thread-normalize`,
    /// `--no-cg-normalize`, and `--sort-by` propagate from clap
    /// into the `CtprofShowArgs` struct unchanged. Pins the
    /// parse shape for every flag the show subcommand surfaces.
    #[test]
    fn parse_ctprof_show_with_every_flag_succeeds() {
        let parsed = Cli::try_parse_from([
            "ktstr",
            "ctprof",
            "show",
            "/tmp/snap.ctprof.zst",
            "--group-by",
            "comm",
            "--cgroup-flatten",
            "/kubepods/*/workload",
            "--no-thread-normalize",
            "--no-cg-normalize",
            "--sort-by",
            "run_time_ns:desc,wait_sum:asc",
        ])
        .unwrap_or_else(|e| panic!("{e}"));
        match parsed.command {
            Command::Ctprof {
                command: CtprofCommand::Show(args),
            } => {
                assert_eq!(args.group_by, ShowGroupBy::Comm);
                assert_eq!(
                    args.cgroup_flatten,
                    vec!["/kubepods/*/workload".to_string()]
                );
                assert!(args.no_thread_normalize);
                assert!(args.no_cg_normalize);
                // `--sort-by` comes through verbatim — clap stores
                // the raw spec; `parse_sort_by` is the parser layer
                // and runs in `run_show`, not at clap parse time.
                assert_eq!(args.sort_by, "run_time_ns:desc,wait_sum:asc");
            }
            _ => panic!("expected Ctprof/Show"),
        }
    }

    /// `ktstr ctprof show <path> --sort-by run_time_ns` parses
    /// successfully with the spec stored verbatim on
    /// `CtprofShowArgs::sort_by`. Single-key, default-direction
    /// pin — the unmarked form ("--sort-by metric" without a `:dir`
    /// suffix) routes through `parse_sort_by`'s `None` arm in
    /// `run_show` and ranks descending.
    #[test]
    fn parse_ctprof_show_sort_by_single_key_succeeds() {
        let parsed = Cli::try_parse_from([
            "ktstr",
            "ctprof",
            "show",
            "/tmp/snap.ctprof.zst",
            "--sort-by",
            "run_time_ns",
        ])
        .unwrap_or_else(|e| panic!("{e}"));
        match parsed.command {
            Command::Ctprof {
                command: CtprofCommand::Show(args),
            } => {
                assert_eq!(args.sort_by, "run_time_ns");
            }
            _ => panic!("expected Ctprof/Show"),
        }
    }

    // -- ctprof compare CLI --sort-by clap-parse pins
    //
    // Mirror of the show-side parse tests above for the compare
    // subcommand. Pins that the `--sort-by` flag clap-parses
    // into `CtprofCompareArgs::sort_by` as a raw String
    // (parsing through `parse_sort_by` happens later, in
    // `run_compare`). A regression that drops the field or
    // re-types it as `Vec<String>` (e.g. a misguided "make it
    // repeatable" refactor) would surface here at parse time.
    /// `ktstr ctprof compare <a> <b>` with no `--sort-by`
    /// must parse with `sort_by` defaulting to the empty string.
    /// Positive-path pin for the default-sentinel contract.
    #[test]
    fn parse_ctprof_compare_sort_by_defaults_to_empty() {
        let parsed = Cli::try_parse_from([
            "ktstr",
            "ctprof",
            "compare",
            "/tmp/a.ctprof.zst",
            "/tmp/b.ctprof.zst",
        ])
        .unwrap_or_else(|e| panic!("{e}"));
        match parsed.command {
            Command::Ctprof {
                command: CtprofCommand::Compare(args),
            } => {
                assert_eq!(args.baseline, std::path::PathBuf::from("/tmp/a.ctprof.zst"));
                assert_eq!(
                    args.candidate,
                    std::path::PathBuf::from("/tmp/b.ctprof.zst")
                );
                // `--sort-by` defaults to the empty string —
                // parse_sort_by treats this as the "fall through
                // to default delta_pct sort" sentinel.
                assert!(args.sort_by.is_empty());
            }
            _ => panic!("expected Ctprof/Compare"),
        }
    }

    /// `ktstr ctprof compare <a> <b> --sort-by <spec>` parses
    /// the spec verbatim into `CtprofCompareArgs::sort_by`.
    /// Pins that clap stores the raw string — `parse_sort_by`'s
    /// validation is deferred to `run_compare`, not parse time.
    /// Use a multi-key spec with mixed directions to exercise
    /// every parser feature reachable through clap.
    #[test]
    fn parse_ctprof_compare_with_sort_by_succeeds() {
        let parsed = Cli::try_parse_from([
            "ktstr",
            "ctprof",
            "compare",
            "/tmp/a.ctprof.zst",
            "/tmp/b.ctprof.zst",
            "--sort-by",
            "run_time_ns:desc,wait_time_ns:asc",
        ])
        .unwrap_or_else(|e| panic!("{e}"));
        match parsed.command {
            Command::Ctprof {
                command: CtprofCommand::Compare(args),
            } => {
                assert_eq!(args.sort_by, "run_time_ns:desc,wait_time_ns:asc");
            }
            _ => panic!("expected Ctprof/Compare"),
        }
    }

    /// `--group-by`, `--cgroup-flatten`, `--no-thread-normalize`,
    /// `--no-cg-normalize`, and `--sort-by` propagate from clap
    /// into the `CtprofCompareArgs` struct unchanged. Mirror
    /// of `parse_ctprof_show_with_every_flag_succeeds` for
    /// the compare subcommand — pins the parse shape for every
    /// flag the compare subcommand surfaces. A regression that
    /// drops a flag from the clap struct or re-types a field
    /// (e.g. `--sort-by` to `Vec<String>`) would surface here at
    /// parse time before reaching `run_compare`.
    #[test]
    fn parse_ctprof_compare_with_every_flag() {
        let parsed = Cli::try_parse_from([
            "ktstr",
            "ctprof",
            "compare",
            "/tmp/a.ctprof.zst",
            "/tmp/b.ctprof.zst",
            "--group-by",
            "comm",
            "--cgroup-flatten",
            "/kubepods/*/workload",
            "--no-thread-normalize",
            "--no-cg-normalize",
            "--sort-by",
            "run_time_ns:desc,wait_sum:asc",
        ])
        .unwrap_or_else(|e| panic!("{e}"));
        match parsed.command {
            Command::Ctprof {
                command: CtprofCommand::Compare(args),
            } => {
                assert_eq!(args.baseline, std::path::PathBuf::from("/tmp/a.ctprof.zst"));
                assert_eq!(
                    args.candidate,
                    std::path::PathBuf::from("/tmp/b.ctprof.zst")
                );
                assert_eq!(args.group_by, ctprof_compare::GroupBy::Comm);
                assert_eq!(
                    args.cgroup_flatten,
                    vec!["/kubepods/*/workload".to_string()]
                );
                assert!(args.no_thread_normalize);
                assert!(args.no_cg_normalize);
                // `--sort-by` comes through verbatim — clap stores
                // the raw spec; `parse_sort_by` is the parser layer
                // and runs in `run_compare`, not at clap parse time.
                assert_eq!(args.sort_by, "run_time_ns:desc,wait_sum:asc");
            }
            _ => panic!("expected Ctprof/Compare"),
        }
    }

    /// `write_show` on a two-thread synthetic snapshot under
    /// `GroupBy::Pcomm` renders one row per metric per group with
    /// header `pcomm | threads | metric | value`. Pins the column
    /// vocabulary, the routing of GroupBy → header, and that
    /// every row's "threads" column matches the bucket size. A
    /// regression that swapped the Sum/Mode columns or dropped a
    /// metric would surface here.
    #[test]
    fn write_show_renders_pcomm_grouping_with_expected_columns() {
        let mut snap = ktstr::ctprof::CtprofSnapshot::default();
        // Two threads under the same pcomm so the bucket has
        // size 2 — verifies the threads column reflects bucket
        // size, not snapshot total. Use a unitless cumulative
        // counter (`nr_wakeups`) for the value-cell pin so the
        // assertion does not need to model `auto_scale`'s ladder
        // step-up: with values 1000 + 2000 = 3000, an `"ns"`
        // metric would render `3.000µs` (auto-scaled), but
        // `nr_wakeups` is unitless and the empty-unit ladder
        // (""→K→M→G) only steps up at 1e3, so `3000` (>= 1e3)
        // would also scale to `3.000K`. Pick small values
        // (1+2=3) so no scaling fires regardless of unit.
        let mut t1 = ktstr::ctprof::ThreadState::default();
        t1.pcomm = "worker-proc".to_string();
        t1.comm = "worker-0".to_string();
        t1.nr_wakeups = MonotonicCount(1);
        let mut t2 = ktstr::ctprof::ThreadState::default();
        t2.pcomm = "worker-proc".to_string();
        t2.comm = "worker-1".to_string();
        t2.nr_wakeups = MonotonicCount(2);
        snap.threads.push(t1);
        snap.threads.push(t2);

        let mut out = String::new();
        write_show(
            &mut out,
            &snap,
            ctprof_compare::GroupBy::Pcomm,
            &[],
            false,
            false,
            &[],
            &[],
            &[],
            &[],
            false,
        )
        .expect("write_show into String must not fail");

        assert!(
            out.contains("pcomm"),
            "header must include `pcomm` for GroupBy::Pcomm, got: {out}",
        );
        assert!(
            out.contains("threads"),
            "header must include `threads` column, got: {out}",
        );
        assert!(
            out.contains("metric"),
            "header must include `metric` column, got: {out}",
        );
        assert!(
            out.contains("value"),
            "header must include `value` column, got: {out}",
        );
        assert!(
            out.contains("worker-proc"),
            "group key must surface in the rendered table, got: {out}",
        );
        assert!(
            out.contains("nr_wakeups"),
            "Sum metric `nr_wakeups` must render a row, got: {out}",
        );
        // Sum across the bucket: 1 + 2 = 3. `nr_wakeups` is
        // unitless and below the 1e3 ladder step, so
        // `format_value_cell` renders the integer verbatim.
        assert!(
            out.contains(" 3 "),
            "summed nr_wakeups (1+2) must surface as 3 in a value cell, got: {out}",
        );
    }

    /// Header switches based on the GroupBy axis — `comm-pattern`
    /// for `GroupBy::Comm`, `comm` for `GroupBy::CommExact`,
    /// `cgroup` for `GroupBy::Cgroup`, `pcomm` for the default.
    /// Pins the routing in `write_show`'s match arm against
    /// vocabulary drift between this and `ctprof_compare`.
    #[test]
    fn write_show_header_switches_on_group_by() {
        let snap = ktstr::ctprof::CtprofSnapshot::default();
        for (axis, expected_header) in [
            (ctprof_compare::GroupBy::Pcomm, "pcomm"),
            (ctprof_compare::GroupBy::Cgroup, "cgroup"),
            (ctprof_compare::GroupBy::Comm, "comm-pattern"),
            (ctprof_compare::GroupBy::CommExact, "comm"),
        ] {
            let mut out = String::new();
            write_show(
                &mut out,
                &snap,
                axis,
                &[],
                false,
                false,
                &[],
                &[],
                &[],
                &[],
                false,
            )
            .expect("write_show into String must not fail");
            assert!(
                out.contains(expected_header),
                "header for {axis:?} must contain `{expected_header}`, got: {out}",
            );
        }
    }

    /// Empty snapshot renders only the header row — `write_show`
    /// must not panic on the zero-thread path. Pins that the
    /// no-data case is well-behaved (compare's no-rows path is
    /// driven by both-snapshot mismatch; show's only no-data
    /// path is the single-snap one).
    #[test]
    fn write_show_empty_snapshot_renders_header_only() {
        let snap = ktstr::ctprof::CtprofSnapshot::default();
        let mut out = String::new();
        write_show(
            &mut out,
            &snap,
            ctprof_compare::GroupBy::Pcomm,
            &[],
            false,
            false,
            &[],
            &[],
            &[],
            &[],
            false,
        )
        .expect("write_show into String must not fail on empty snapshot");
        assert!(
            out.contains("pcomm"),
            "empty snapshot must still emit the header, got: {out}",
        );
        // Run-time metric does not appear because no thread
        // contributed it — the `for metric in CTPROF_METRICS`
        // loop emits no rows when `groups` is empty.
        assert!(
            !out.contains("run_time_ns"),
            "no thread → no metric row; got run_time_ns surfaced: {out}",
        );
    }

    /// `write_show` with a non-empty `sort_by` re-orders groups
    /// by the named metric's *absolute aggregated value* in
    /// descending order — pins the new sort path against
    /// alphabetical-by-default. Three pcomm buckets (`alpha`,
    /// `bravo`, `charlie`) carry distinct `run_time_ns` totals so
    /// the bucket with the largest value lands first; that order
    /// is the inverse of the alphabetical default, so a regression
    /// dropping the sort path would surface here as the wrong
    /// first-group cell.
    #[test]
    fn write_show_sort_by_orders_groups_by_metric_descending() {
        let mut snap = ktstr::ctprof::CtprofSnapshot::default();
        let mut t_alpha = ktstr::ctprof::ThreadState::default();
        t_alpha.pcomm = "alpha".to_string();
        t_alpha.comm = "alpha-w".to_string();
        t_alpha.run_time_ns = MonotonicNs(100);
        let mut t_bravo = ktstr::ctprof::ThreadState::default();
        t_bravo.pcomm = "bravo".to_string();
        t_bravo.comm = "bravo-w".to_string();
        t_bravo.run_time_ns = MonotonicNs(500);
        let mut t_charlie = ktstr::ctprof::ThreadState::default();
        t_charlie.pcomm = "charlie".to_string();
        t_charlie.comm = "charlie-w".to_string();
        t_charlie.run_time_ns = MonotonicNs(250);
        snap.threads.push(t_alpha);
        snap.threads.push(t_bravo);
        snap.threads.push(t_charlie);

        let sort_by = vec![ctprof_compare::SortKey {
            metric: "run_time_ns",
            descending: true,
        }];
        let mut out = String::new();
        write_show(
            &mut out,
            &snap,
            ctprof_compare::GroupBy::Pcomm,
            &[],
            false,
            false,
            &sort_by,
            &[],
            &[],
            &[],
            false,
        )
        .expect("write_show into String must not fail");

        // The three group keys appear in descending run_time_ns
        // order: bravo (500) → charlie (250) → alpha (100). The
        // alphabetical default would place alpha first.
        let bravo_at = out.find("bravo").expect("bravo must surface in output");
        let charlie_at = out.find("charlie").expect("charlie must surface in output");
        let alpha_at = out.find("alpha").expect("alpha must surface in output");
        assert!(
            bravo_at < charlie_at,
            "sort_by run_time_ns:desc must place bravo (500) before charlie (250); \
             alpha={alpha_at} bravo={bravo_at} charlie={charlie_at}\nout:\n{out}",
        );
        assert!(
            charlie_at < alpha_at,
            "sort_by run_time_ns:desc must place charlie (250) before alpha (100); \
             alpha={alpha_at} bravo={bravo_at} charlie={charlie_at}\nout:\n{out}",
        );
    }

    /// `write_show` cgroup-stats secondary table renders each
    /// scalar via `format_scaled_u64` — single auto-scaled value
    /// per cell, no `→` arrow, no `(+0)` zero-delta suffix that
    /// the compare-side `cgroup_cell` helper carries when
    /// rendering a baseline→candidate→delta triple. Pins the UX
    /// fix that distinguishes show (single snapshot) from
    /// compare (paired snapshots): a 7.5 GiB `memory_current`
    /// row reads `7.500GiB`, not `7.500GiB → 7.500GiB (+0B)`.
    #[test]
    fn write_show_cgroup_stats_renders_single_value_no_delta() {
        let mut snap = ktstr::ctprof::CtprofSnapshot::default();
        // Cgroup grouping requires at least one thread carrying
        // the cgroup path so build_groups produces a bucket; the
        // secondary table renders for every cgroup_stats key
        // regardless, but the primary table needs a row to
        // surround it (write_show returns Ok early only for an
        // empty cgroup_stats map, not an empty groups map).
        let mut t = ktstr::ctprof::ThreadState::default();
        t.pcomm = "worker".to_string();
        t.comm = "w".to_string();
        t.cgroup = "/app".to_string();
        snap.threads.push(t);
        // 1 GiB exactly under IEC binary auto-scale → "1.000GiB".
        // 1_500_000 µs under the µs ladder → "1.500s". The other
        // two scalars (50 unitless, 200 µs) sit below their
        // respective ladders' first step (1e3 either way) so they
        // render as bare integers via `format_scaled_u64`'s
        // no-step-up path. This test asserts the GiB and s forms
        // since those are the load-bearing auto-scaled rows.
        let one_gib: u64 = 1024 * 1024 * 1024;
        // `CgroupStats` is `#[non_exhaustive]`, so direct struct
        // construction is rejected outside its defining module
        // (the `ktstr` binary is a separate translation unit
        // from `ctprof`). Build via Default + per-field
        // assignment instead.
        let mut cgs = ktstr::ctprof::CgroupStats::default();
        cgs.cpu.usage_usec = 1_500_000;
        cgs.cpu.nr_throttled = 50;
        cgs.cpu.throttled_usec = 200;
        cgs.memory.current = one_gib;
        snap.cgroup_stats.insert("/app".to_string(), cgs);

        let mut out = String::new();
        write_show(
            &mut out,
            &snap,
            ctprof_compare::GroupBy::Cgroup,
            &[],
            false,
            false,
            &[],
            &[],
            &[],
            &[],
            false,
        )
        .expect("write_show into String must not fail");

        // Each scaled scalar appears VERBATIM with no arrow or
        // delta tail — distinguishes show's single-value cells
        // from compare's `a → b (+d)` triples.
        assert!(
            out.contains("1.500s"),
            "cpu_usage_usec 1_500_000 µs must scale to '1.500s', got:\n{out}",
        );
        assert!(
            out.contains("1.000GiB"),
            "memory_current 1 GiB must scale to '1.000GiB', got:\n{out}",
        );
        // The arrow form would be a regression — single-snapshot
        // rendering must not pretend to carry a delta.
        assert!(
            !out.contains("→"),
            "show cgroup-stats must not emit a `→` arrow \
             (compare's two-value cell shape); got:\n{out}",
        );
        // `(+0` matches the regression signature — the
        // `format_delta_cell` zero-delta tail (`(+0)`, `(+0B)`,
        // `(+0µs)`) that compare's cgroup_cell emits when
        // baseline equals candidate. Bare `(` would over-match
        // any incidental paren in comfy_table output, so pin
        // the `(+0` prefix exactly.
        assert!(
            !out.contains("(+0"),
            "show cgroup-stats must not carry a `(+0…)` zero-delta tail; \
             got:\n{out}",
        );
    }

    /// `write_show` with the `Tags` column included surfaces the
    /// registry's bracketed tag string in the dedicated tags
    /// column. Pins that the show plumbs `metric_tags` correctly
    /// through the column dispatch and that the rendered cell
    /// carries the bracketed tag. Mirrors the compare-side
    /// `write_diff_renders_tagged_metric_cell` integration test on
    /// the show path. The default column set deliberately omits
    /// the bracketed tag string from the `metric` cell so plain
    /// listings stay narrow.
    #[test]
    fn write_show_renders_tagged_metric_cell() {
        let mut snap = ktstr::ctprof::CtprofSnapshot::default();
        let mut t = ktstr::ctprof::ThreadState::default();
        t.pcomm = "worker".to_string();
        t.comm = "w".to_string();
        t.nr_wakeups_affine = MonotonicCount(7);
        snap.threads.push(t);
        let columns = vec![
            ctprof_compare::Column::Group,
            ctprof_compare::Column::Threads,
            ctprof_compare::Column::Metric,
            ctprof_compare::Column::Tags,
            ctprof_compare::Column::Value,
        ];
        let mut out = String::new();
        write_show(
            &mut out,
            &snap,
            ctprof_compare::GroupBy::Pcomm,
            &[],
            false,
            false,
            &[],
            &columns,
            &[],
            &[],
            false,
        )
        .expect("write_show into String must not fail");
        assert!(
            out.contains("[cfs-only] [SCHEDSTATS]"),
            "tagged metric tags missing from rendered tags column:\n{out}",
        );
        assert!(
            out.contains("nr_wakeups_affine"),
            "tagged metric name missing from rendered show table:\n{out}",
        );
    }

    /// `write_show` emits the `## Derived metrics` section
    /// after the main per-group table. Mirrors the compare-side
    /// `write_diff_emits_derived_section` integration test.
    /// Sets the inputs for `avg_slice_ns` (run_time_ns /
    /// timeslices) on a single thread so the derivation
    /// resolves to a non-`-` cell, and asserts both the section
    /// header and the metric name surface in the output.
    #[test]
    fn write_show_emits_derived_section() {
        let mut snap = ktstr::ctprof::CtprofSnapshot::default();
        let mut t = ktstr::ctprof::ThreadState::default();
        t.pcomm = "worker".to_string();
        t.comm = "w".to_string();
        t.run_time_ns = MonotonicNs(4_000);
        t.timeslices = MonotonicCount(8);
        snap.threads.push(t);
        let mut out = String::new();
        write_show(
            &mut out,
            &snap,
            ctprof_compare::GroupBy::Pcomm,
            &[],
            false,
            false,
            &[],
            &[],
            &[],
            &[],
            false,
        )
        .expect("write_show into String must not fail");
        assert!(
            out.contains("## Derived metrics"),
            "missing derived section header from show output:\n{out}",
        );
        assert!(
            out.contains("avg_slice_ns"),
            "missing avg_slice_ns row in derived section of show output:\n{out}",
        );
    }

    /// `write_show` with `columns=[Metric, Value]` emits exactly
    /// those two columns — the header must contain `metric` and
    /// `value` but not `threads` (or any other compare-only
    /// column name). Pins that the show-side `--columns` override
    /// reaches the renderer and overrides the default
    /// `(group, threads, metric, value)` set.
    #[test]
    fn write_show_columns_override_emits_only_selected_columns() {
        let mut snap = ktstr::ctprof::CtprofSnapshot::default();
        let mut t = ktstr::ctprof::ThreadState::default();
        t.pcomm = "worker".to_string();
        t.comm = "w".to_string();
        t.nr_wakeups = MonotonicCount(1);
        snap.threads.push(t);

        let columns = vec![
            ctprof_compare::Column::Metric,
            ctprof_compare::Column::Value,
        ];
        let mut out = String::new();
        write_show(
            &mut out,
            &snap,
            ctprof_compare::GroupBy::Pcomm,
            &[],
            false,
            false,
            &[],
            &columns,
            &[],
            &[],
            false,
        )
        .expect("write_show into String must not fail");

        // The first line of `out` is the section heading
        // `## Primary metrics`; the table column header is the
        // first line containing the `metric` token.
        let header_line = out
            .lines()
            .find(|line| line.contains("metric") && !line.starts_with("##"))
            .unwrap_or("");
        assert!(
            header_line.contains("metric"),
            "metric column must appear in header: {header_line}",
        );
        assert!(
            header_line.contains("value"),
            "value column must appear in header: {header_line}",
        );
        assert!(
            !header_line.contains("threads"),
            "threads column must NOT appear when --columns excludes it: {header_line}",
        );
    }

    /// `write_show` with an empty `sort_by` keeps the default
    /// alphabetical iteration over the BTreeMap-keyed groups.
    /// Mirror of the descending-sort test: same three buckets,
    /// no flag set, expect ascending alphabetical order
    /// regardless of the run_time_ns spread. Pins the
    /// "fall-through to default" branch of `write_show`.
    #[test]
    fn write_show_empty_sort_by_keeps_alphabetical_default() {
        let mut snap = ktstr::ctprof::CtprofSnapshot::default();
        let mut t_zulu = ktstr::ctprof::ThreadState::default();
        t_zulu.pcomm = "zulu".to_string();
        t_zulu.comm = "zulu-w".to_string();
        t_zulu.run_time_ns = MonotonicNs(999);
        let mut t_alpha = ktstr::ctprof::ThreadState::default();
        t_alpha.pcomm = "alpha".to_string();
        t_alpha.comm = "alpha-w".to_string();
        t_alpha.run_time_ns = MonotonicNs(1);
        snap.threads.push(t_zulu);
        snap.threads.push(t_alpha);

        let mut out = String::new();
        write_show(
            &mut out,
            &snap,
            ctprof_compare::GroupBy::Pcomm,
            &[],
            false,
            false,
            &[],
            &[],
            &[],
            &[],
            false,
        )
        .expect("write_show into String must not fail");

        let alpha_at = out.find("alpha").expect("alpha must surface");
        let zulu_at = out.find("zulu").expect("zulu must surface");
        assert!(
            alpha_at < zulu_at,
            "empty sort_by must keep alphabetical order \
             (alpha before zulu); alpha={alpha_at} zulu={zulu_at}\nout:\n{out}",
        );
    }

    /// `write_show` smaps section under default normalization
    /// keys by `pattern_key(&t.pcomm)` (drops the tgid) and
    /// field-sums byte counts across PIDs sharing the same
    /// pcomm pattern. Three `worker-{0,1,2}` leaders (each with
    /// its own ephemeral tgid) collapse into one `worker-{N}`
    /// row per smaps_rollup field, with values summed.
    #[test]
    fn write_show_smaps_default_normalization_collapses_pids() {
        let mut snap = ktstr::ctprof::CtprofSnapshot::default();
        for (pcomm, tgid, rss) in [
            ("worker-0", 100, 1024_u64),
            ("worker-1", 200, 2048),
            ("worker-2", 300, 4096),
        ] {
            let mut t = ktstr::ctprof::ThreadState::default();
            t.tid = tgid;
            t.tgid = tgid;
            t.pcomm = pcomm.to_string();
            t.comm = pcomm.to_string();
            t.smaps_rollup_kib.insert("Rss".into(), rss);
            t.smaps_rollup_kib.insert("Pss".into(), rss / 2);
            snap.threads.push(t);
        }

        let mut out = String::new();
        write_show(
            &mut out,
            &snap,
            ctprof_compare::GroupBy::Pcomm,
            &[],
            false, // no_thread_normalize: false → normalize
            false,
            &[],
            &[],
            &[],
            &[],
            false,
        )
        .expect("write_show must not fail");

        // Find the smaps section.
        let smaps_at = out
            .find("## smaps_rollup")
            .expect("smaps section must render");
        let after = &out[smaps_at..];
        // Normalized key surfaces; literal per-PID keys do not.
        assert!(
            after.contains("worker-{N}"),
            "show smaps must collapse to `worker-{{N}}` under default normalization:\n{after}",
        );
        for literal in &["worker-0[100]", "worker-1[200]", "worker-2[300]"] {
            assert!(
                !after.contains(literal),
                "literal per-PID key {literal:?} must NOT appear under default \
                 normalization:\n{after}",
            );
        }
        // Summed Rss = (1024 + 2048 + 4096) * 1024 = 7168 KiB =
        // 7168*1024 bytes = 7340032 B → step-up to MiB
        // (1024*1024 = 1048576), so 7340032 / 1048576 ≈ 7.000 MiB.
        assert!(
            after.contains("7.000MiB"),
            "summed Rss must render as `7.000MiB` (3-PID collapse via field-sum):\n{after}",
        );
    }

    /// `write_show` smaps section under `--no-thread-normalize`
    /// preserves the literal `pcomm[tgid]` key shape so each
    /// PID stays attributable. Same fixture as the default-mode
    /// test, but with `no_thread_normalize: true` — the section
    /// emits three rows (one per PID), not one collapsed row.
    #[test]
    fn write_show_smaps_literal_mode_preserves_per_pid_keys() {
        let mut snap = ktstr::ctprof::CtprofSnapshot::default();
        for (pcomm, tgid, rss) in [
            ("worker-0", 100, 1024_u64),
            ("worker-1", 200, 2048),
            ("worker-2", 300, 4096),
        ] {
            let mut t = ktstr::ctprof::ThreadState::default();
            t.tid = tgid;
            t.tgid = tgid;
            t.pcomm = pcomm.to_string();
            t.comm = pcomm.to_string();
            t.smaps_rollup_kib.insert("Rss".into(), rss);
            snap.threads.push(t);
        }

        let mut out = String::new();
        write_show(
            &mut out,
            &snap,
            ctprof_compare::GroupBy::Pcomm,
            &[],
            true, // no_thread_normalize: true → literal
            false,
            &[],
            &[],
            &[],
            &[],
            false,
        )
        .expect("write_show must not fail");

        let smaps_at = out
            .find("## smaps_rollup")
            .expect("smaps section must render");
        let after = &out[smaps_at..];
        // Literal per-PID keys surface; normalized form does not.
        for literal in &["worker-0[100]", "worker-1[200]", "worker-2[300]"] {
            assert!(
                after.contains(literal),
                "literal per-PID key {literal:?} must surface under \
                 --no-thread-normalize:\n{after}",
            );
        }
        assert!(
            !after.contains("worker-{N}"),
            "normalized `worker-{{N}}` key must NOT surface under \
             --no-thread-normalize:\n{after}",
        );
    }

    /// `write_show` smaps section sorts process rows by Rss
    /// descending, mirroring the compare-side sort. Two
    /// processes — `bash` (literal pcomm, large Rss) and
    /// `worker-{N}` collapsed bucket (small per-thread Rss but
    /// summed across PIDs). The heavier process must render
    /// first regardless of alphabetical key order.
    #[test]
    fn write_show_smaps_orders_by_rss_descending() {
        let mut snap = ktstr::ctprof::CtprofSnapshot::default();
        // bash: 100 MiB Rss.
        let mut bash = ktstr::ctprof::ThreadState::default();
        bash.tid = 1;
        bash.tgid = 1;
        bash.pcomm = "bash".to_string();
        bash.comm = "bash".to_string();
        bash.smaps_rollup_kib.insert("Rss".into(), 100 * 1024);
        bash.smaps_rollup_kib.insert("Pss".into(), 50 * 1024);
        snap.threads.push(bash);
        // zulu: 1 MiB Rss only — much smaller, but
        // alphabetically AFTER bash.
        let mut zulu = ktstr::ctprof::ThreadState::default();
        zulu.tid = 2;
        zulu.tgid = 2;
        zulu.pcomm = "zulu".to_string();
        zulu.comm = "zulu".to_string();
        zulu.smaps_rollup_kib.insert("Rss".into(), 1024);
        zulu.smaps_rollup_kib.insert("Pss".into(), 512);
        snap.threads.push(zulu);

        let mut out = String::new();
        write_show(
            &mut out,
            &snap,
            ctprof_compare::GroupBy::Pcomm,
            &[],
            false,
            false,
            &[],
            &[],
            &[],
            &[],
            false,
        )
        .expect("write_show must not fail");

        let smaps_at = out
            .find("## smaps_rollup")
            .expect("smaps section must render");
        let after = &out[smaps_at..];
        let bash_pos = after.find("bash").expect("bash key must surface");
        let zulu_pos = after.find("zulu").expect("zulu key must surface");
        assert!(
            bash_pos < zulu_pos,
            "smaps must sort by Rss desc — bash (100 MiB) ahead of \
             zulu (1 MiB) regardless of alphabetical order; \
             bash@{bash_pos} zulu@{zulu_pos}\nafter:\n{after}",
        );
    }

    /// Build the canonical single-cgroup fixture: one thread
    /// pinned to `/app` so `build_groups` produces a `Cgroup`
    /// bucket and `write_show`'s cgroup-stats sub-tables render.
    /// Mirrors the per-field-assignment pattern the existing
    /// cgroup test uses (the `ktstr` binary is a separate
    /// translation unit from `ctprof`, so the `#[non_exhaustive]`
    /// markers on `CtprofSnapshot` / `ThreadState` / `CgroupStats`
    /// block struct-literal construction here). The caller mutates
    /// the returned `CgroupStats` and inserts it under `/app`.
    fn cgroup_fixture() -> (ktstr::ctprof::CtprofSnapshot, ktstr::ctprof::CgroupStats) {
        let mut snap = ktstr::ctprof::CtprofSnapshot::default();
        let mut t = ktstr::ctprof::ThreadState::default();
        t.pcomm = "worker".to_string();
        t.comm = "w".to_string();
        t.cgroup = "/app".to_string();
        snap.threads.push(t);
        (snap, ktstr::ctprof::CgroupStats::default())
    }

    /// `write_show` under `GroupBy::Cgroup` emits the
    /// `## Cgroup limits / knobs` sub-table when at least one
    /// bucketed cgroup exposes a cpu.max / weight / memory.max /
    /// pids value. Pins the `stats.values().any(...)` section gate
    /// and each cell's formatter routing:
    /// `format_cpu_max` for the cpu.max pair, `format_scaled_u64`
    /// for the unitless weight / pids.current, `format_optional_limit`
    /// for memory.max (`Some` → scaled) and for memory.high /
    /// pids.max (`None` → the `max` token). The existing
    /// cgroup-stats test sets only cpu.usage / memory.current, so
    /// this whole block is otherwise dead in tests.
    #[test]
    fn write_show_emits_cgroup_limits_table_with_scaled_knobs() {
        let (mut snap, mut cgs) = cgroup_fixture();
        // cpu.max quota=50_000µs period=100_000µs →
        // format_cpu_max → "50.000ms/100.000ms".
        cgs.cpu.max_quota_us = Some(50_000);
        cgs.cpu.max_period_us = 100_000;
        // cpu.weight=200 → format_scaled_u64(200, Unitless),
        // below the 1e3 ladder step → bare "200".
        cgs.cpu.weight = Some(200);
        // memory.max=2 GiB → format_optional_limit(Some, Bytes)
        // → "2.000GiB". memory.high=None → "max".
        cgs.memory.max = Some(2 * 1024 * 1024 * 1024);
        // memory.current must be set so the CgroupStats sub-table
        // gate (non-empty cgroup_stats) is satisfied; the limits
        // gate is independent but the fixture stays realistic.
        cgs.memory.current = 1;
        // pids.current=7 → format_scaled_u64(7, Unitless) → "7".
        // pids.max=None → format_optional_limit(None, Unitless)
        // → "max".
        cgs.pids.current = Some(7);
        snap.cgroup_stats.insert("/app".to_string(), cgs);

        let mut out = String::new();
        write_show(
            &mut out,
            &snap,
            ctprof_compare::GroupBy::Cgroup,
            &[],
            false,
            false,
            &[],
            &[],
            &[],
            &[],
            false,
        )
        .expect("write_show into String must not fail");

        assert!(
            out.contains("## Cgroup limits / knobs"),
            "limits section must render when a cgroup exposes a knob:\n{out}",
        );
        assert!(
            out.contains("50.000ms/100.000ms"),
            "cpu.max 50_000µs/100_000µs must render via format_cpu_max \
             as '50.000ms/100.000ms':\n{out}",
        );
        assert!(
            out.contains(" 200 "),
            "cpu.weight 200 must render via format_scaled_u64(Unitless) \
             as the bare integer 200:\n{out}",
        );
        assert!(
            out.contains("2.000GiB"),
            "memory.max 2 GiB must render via format_optional_limit(Bytes) \
             as '2.000GiB':\n{out}",
        );
        assert!(
            out.contains(" 7 "),
            "pids.current 7 must render via format_scaled_u64(Unitless) \
             as the bare integer 7:\n{out}",
        );
        // memory.high=None and pids.max=None both render the `max`
        // token (format_optional_limit's None arm). The column
        // headers (`cpu.max`, `memory.max`, `pids.max`) carry
        // `max` as a substring of a dotted token, so assert the
        // space-padded standalone cell `" max "` to prove the
        // None→`max` rendering rather than matching a header.
        assert!(
            out.contains(" max "),
            "memory.high=None / pids.max=None must render the standalone \
             `max` cell via format_optional_limit's None arm:\n{out}",
        );
    }

    /// `write_show` under `GroupBy::Cgroup` emits the
    /// `## memory.stat` long-table and suppresses zero-valued
    /// rows. Pins the section gate
    /// (`any(|v| *v != 0)`) and the per-row
    /// `if *stat_value == 0 { continue; }` skip: a
    /// non-zero `anon` key renders one row (`format_scaled_u64`
    /// steps `4096` up to `4.096K`), while a zero `file` key is
    /// suppressed entirely.
    #[test]
    fn write_show_emits_memory_stat_table_suppressing_zero_rows() {
        let (mut snap, mut cgs) = cgroup_fixture();
        cgs.memory.stat.insert("anon".to_string(), 4096);
        cgs.memory.stat.insert("file".to_string(), 0);
        snap.cgroup_stats.insert("/app".to_string(), cgs);

        let mut out = String::new();
        write_show(
            &mut out,
            &snap,
            ctprof_compare::GroupBy::Cgroup,
            &[],
            false,
            false,
            &[],
            &[],
            &[],
            &[],
            false,
        )
        .expect("write_show into String must not fail");

        assert!(
            out.contains("## memory.stat"),
            "memory.stat section must render when any key is non-zero:\n{out}",
        );
        assert!(
            out.contains("anon"),
            "non-zero `anon` key must render a row:\n{out}",
        );
        assert!(
            out.contains("4.096K"),
            "anon=4096 must render via format_scaled_u64(Unitless) \
             stepping up to '4.096K':\n{out}",
        );
        assert!(
            !out.contains("file"),
            "zero-valued `file` key must be suppressed by the if *stat_value == 0 \
             continue:\n{out}",
        );
    }

    /// `write_show` under `GroupBy::Cgroup` emits the
    /// `## memory.events` long-table and suppresses zero-count
    /// events. Pins the section gate and the per-row
    /// `if *event_value == 0 { continue; }` skip: a
    /// non-zero `oom_kill` event renders one row (`3` below the
    /// 1e3 ladder step → bare integer), while a zero `low` event
    /// is suppressed.
    #[test]
    fn write_show_emits_memory_events_table_suppressing_zero_rows() {
        let (mut snap, mut cgs) = cgroup_fixture();
        cgs.memory.events.insert("oom_kill".to_string(), 3);
        cgs.memory.events.insert("low".to_string(), 0);
        snap.cgroup_stats.insert("/app".to_string(), cgs);

        let mut out = String::new();
        write_show(
            &mut out,
            &snap,
            ctprof_compare::GroupBy::Cgroup,
            &[],
            false,
            false,
            &[],
            &[],
            &[],
            &[],
            false,
        )
        .expect("write_show into String must not fail");

        assert!(
            out.contains("## memory.events"),
            "memory.events section must render when any event is non-zero:\n{out}",
        );
        assert!(
            out.contains("oom_kill"),
            "non-zero `oom_kill` event must render a row:\n{out}",
        );
        assert!(
            out.contains(" 3 "),
            "oom_kill=3 must render via format_scaled_u64(Unitless) \
             as the bare integer 3:\n{out}",
        );
        assert!(
            !out.contains("low"),
            "zero-count `low` event must be suppressed by the if *event_value == 0 \
             continue:\n{out}",
        );
    }

    /// `write_show` under `GroupBy::Cgroup` emits per-cgroup
    /// `## Pressure / <resource>` sub-tables and skips resources
    /// with no data. Pins the section gate, the
    /// per-resource `if !any_data { continue; }` skip,
    /// and the `some`/`full` row rendering via `format_psi_avg`
    /// (centi-percent → `N.NN%`) + `format_scaled_u64(Us)`. Only
    /// `cpu.some` is populated, so the `cpu` sub-table renders
    /// while `memory` / `io` (all-zero) are skipped.
    #[test]
    fn write_show_emits_per_cgroup_pressure_tables_skipping_zero_resources() {
        let (mut snap, mut cgs) = cgroup_fixture();
        // cpu.some.avg10=1859 centi-percent → format_psi_avg →
        // "18.59%". cpu.some.total_usec=1_500_000 →
        // format_scaled_u64(Us) → "1.500s".
        cgs.psi.cpu.some.avg10 = 1859;
        cgs.psi.cpu.some.total_usec = 1_500_000;
        snap.cgroup_stats.insert("/app".to_string(), cgs);

        let mut out = String::new();
        write_show(
            &mut out,
            &snap,
            ctprof_compare::GroupBy::Cgroup,
            &[],
            false,
            false,
            &[],
            &[],
            &[],
            &[],
            false,
        )
        .expect("write_show into String must not fail");

        assert!(
            out.contains("## Pressure / cpu"),
            "cpu pressure sub-table must render when cpu PSI has data:\n{out}",
        );
        assert!(
            out.contains("18.59%"),
            "cpu.some.avg10=1859 must render via format_psi_avg as '18.59%':\n{out}",
        );
        assert!(
            out.contains("1.500s"),
            "cpu.some.total_usec=1_500_000 must render via \
             format_scaled_u64(Us) as '1.500s':\n{out}",
        );
        assert!(
            out.contains("some"),
            "the `some` PSI row must render:\n{out}",
        );
        assert!(
            out.contains("full"),
            "the `full` PSI row must render:\n{out}",
        );
        assert!(
            !out.contains("## Pressure / memory"),
            "all-zero memory PSI must be skipped by the if !any_data \
             continue:\n{out}",
        );
        assert!(
            !out.contains("## Pressure / io"),
            "all-zero io PSI must be skipped by the if !any_data \
             continue:\n{out}",
        );
    }

    /// `write_show` emits host-level `## Host pressure / <resource>`
    /// sub-tables (group-by-independent) and skips resources with
    /// no data. Pins the section gate
    /// (`host_psi_has_data`), the per-resource
    /// `if !psi_resource_has_data(&r) { continue; }` skip,
    /// and the `some`/`full` row render. Only
    /// `memory.full` is populated, so the `memory` sub-table
    /// renders while `cpu` (all-zero) is skipped. The host PSI
    /// header is `row | avg10 | avg60 | avg300 | total` with no
    /// `cgroup` column — assert the cgroup token is absent from
    /// the host-pressure region.
    #[test]
    fn write_show_emits_host_pressure_tables_skipping_zero_resources() {
        let mut snap = ktstr::ctprof::CtprofSnapshot::default();
        // A thread keeps the primary table populated, but the host
        // PSI block is independent of grouping.
        let mut t = ktstr::ctprof::ThreadState::default();
        t.pcomm = "worker".to_string();
        t.comm = "w".to_string();
        snap.threads.push(t);
        // memory.full.avg60=500 centi-percent → format_psi_avg →
        // "5.00%". memory.full.total_usec=2_000_000 →
        // format_scaled_u64(Us) → "2.000s".
        snap.psi.memory.full.avg60 = 500;
        snap.psi.memory.full.total_usec = 2_000_000;

        let mut out = String::new();
        write_show(
            &mut out,
            &snap,
            ctprof_compare::GroupBy::Pcomm,
            &[],
            false,
            false,
            &[],
            &[],
            &[],
            &[],
            false,
        )
        .expect("write_show into String must not fail");

        let host_at = out
            .find("## Host pressure / memory")
            .expect("host memory pressure sub-table must render");
        assert!(
            out.contains("5.00%"),
            "memory.full.avg60=500 must render via format_psi_avg as '5.00%':\n{out}",
        );
        assert!(
            out.contains("2.000s"),
            "memory.full.total_usec=2_000_000 must render via \
             format_scaled_u64(Us) as '2.000s':\n{out}",
        );
        assert!(
            !out.contains("## Host pressure / cpu"),
            "all-zero host cpu PSI must be skipped by the \
             psi_resource_has_data continue:\n{out}",
        );
        // The host-pressure header omits the `cgroup` column the
        // per-cgroup PSI tables carry. Slice from the host-pressure
        // index so a `cgroup` token elsewhere in the output (the
        // primary table never carries one under Pcomm, but be
        // precise) cannot mask a regression that re-added the
        // column to the host header.
        let host_region = &out[host_at..];
        assert!(
            !host_region.contains("cgroup"),
            "host pressure header is `row | avg10 | avg60 | avg300 | total` \
             with no cgroup column:\n{host_region}",
        );
    }

    /// `write_show` emits the `## sched_ext` table when the
    /// snapshot carries `sched_ext = Some(..)`. Pins the section
    /// gate, the non-empty-state branch of the
    /// `state_cell` ternary (`state="enabled"` renders
    /// verbatim, NOT the `-` sentinel), and the five counter rows
    /// rendered via `format_scaled_u64(Unitless)`
    /// (`enable_seq=1234` steps up to `1.234K`; `switch_all=1`
    /// stays the bare integer). `snap.sched_ext` defaults to
    /// `None`, so this block is otherwise dead in tests. The
    /// counter-cell assertions are sliced from the `## sched_ext`
    /// index so the single-digit cells (`switch_all=1`) can't
    /// match the primary table's threads count.
    #[test]
    fn write_show_emits_sched_ext_table_with_state_and_scaled_counters() {
        let mut snap = ktstr::ctprof::CtprofSnapshot::default();
        let mut t = ktstr::ctprof::ThreadState::default();
        t.pcomm = "worker".to_string();
        t.comm = "w".to_string();
        snap.threads.push(t);
        // `SchedExtSysfs` is `#[non_exhaustive]` in the lib — build
        // via Default + per-field assignment.
        let mut scx = ktstr::ctprof::SchedExtSysfs::default();
        scx.state = "enabled".to_string();
        scx.switch_all = 1;
        scx.nr_rejected = 2;
        scx.hotplug_seq = 3;
        scx.enable_seq = 1234;
        snap.sched_ext = Some(scx);

        let mut out = String::new();
        write_show(
            &mut out,
            &snap,
            ctprof_compare::GroupBy::Pcomm,
            &[],
            false,
            false,
            &[],
            &[],
            &[],
            &[],
            false,
        )
        .expect("write_show into String must not fail");

        let scx_at = out
            .find("## sched_ext")
            .expect("sched_ext section must render when snap.sched_ext is Some");
        let scx_region = &out[scx_at..];
        assert!(
            scx_region.contains("state"),
            "sched_ext table must carry the `state` attr row:\n{scx_region}",
        );
        assert!(
            scx_region.contains("enabled"),
            "non-empty state must render verbatim ('enabled'), NOT the `-` \
             sentinel:\n{scx_region}",
        );
        assert!(
            scx_region.contains("switch_all"),
            "sched_ext table must carry the `switch_all` row:\n{scx_region}",
        );
        assert!(
            scx_region.contains("nr_rejected"),
            "sched_ext table must carry the `nr_rejected` row:\n{scx_region}",
        );
        assert!(
            scx_region.contains("1.234K"),
            "enable_seq=1234 must render via format_scaled_u64(Unitless) \
             stepping up to '1.234K':\n{scx_region}",
        );
        assert!(
            scx_region.contains(" 1 "),
            "switch_all=1 must render via format_scaled_u64(Unitless) \
             as the bare integer 1:\n{scx_region}",
        );
    }

    /// `write_show` with a non-empty `sections` whitelist
    /// restricts output to the named sections. Pins the
    /// restrict-to-named-entries branch of
    /// `DisplayOptions::is_section_enabled`: `--sections derived`
    /// suppresses the `## Primary metrics` outer gate (neither
    /// Primary nor TaskstatsDelay enabled) while still
    /// emitting `## Derived metrics`; the converse
    /// `--sections primary` does the inverse. Every existing
    /// write_show test passes `&[]` (empty = all-on), so this
    /// branch is otherwise uncovered. The fixture mirrors
    /// `write_show_emits_derived_section` so `avg_slice_ns`
    /// resolves to a real derived row.
    #[test]
    fn write_show_sections_filter_suppresses_unnamed_sections() {
        let mut snap = ktstr::ctprof::CtprofSnapshot::default();
        let mut t = ktstr::ctprof::ThreadState::default();
        t.pcomm = "worker".to_string();
        t.comm = "w".to_string();
        t.run_time_ns = MonotonicNs(4_000);
        t.timeslices = MonotonicCount(8);
        snap.threads.push(t);

        // `--sections derived`: Primary outer gate is false
        // (neither Primary nor TaskstatsDelay enabled), Derived
        // gate is true.
        let mut derived_only = String::new();
        write_show(
            &mut derived_only,
            &snap,
            ctprof_compare::GroupBy::Pcomm,
            &[],
            false,
            false,
            &[],
            &[],
            &[ctprof_compare::Section::Derived],
            &[],
            false,
        )
        .expect("write_show into String must not fail");
        assert!(
            !derived_only.contains("## Primary metrics"),
            "`--sections derived` must suppress the Primary section:\n{derived_only}",
        );
        assert!(
            derived_only.contains("## Derived metrics"),
            "`--sections derived` must still emit the Derived section:\n{derived_only}",
        );

        // Converse: `--sections primary` emits Primary, suppresses
        // Derived.
        let mut primary_only = String::new();
        write_show(
            &mut primary_only,
            &snap,
            ctprof_compare::GroupBy::Pcomm,
            &[],
            false,
            false,
            &[],
            &[],
            &[ctprof_compare::Section::Primary],
            &[],
            false,
        )
        .expect("write_show into String must not fail");
        assert!(
            primary_only.contains("## Primary metrics"),
            "`--sections primary` must emit the Primary section:\n{primary_only}",
        );
        assert!(
            !primary_only.contains("## Derived metrics"),
            "`--sections primary` must suppress the Derived section:\n{primary_only}",
        );
    }

    /// `run_show` parses `--sort-by` BEFORE loading the snapshot:
    /// an invalid spec must surface its parse
    /// error even when the snapshot path does not exist, proving
    /// the parse precedes the load. Pins the fail-fast ordering
    /// contract — the flattened error chain carries the
    /// `parse --sort-by` context label and NOT the
    /// `load snapshot` label. No existing test drives
    /// `run_show`; the parse guards are otherwise uncovered.
    #[test]
    fn run_show_fails_fast_on_invalid_sort_by_before_snapshot_load() {
        let args = CtprofShowArgs {
            snapshot: std::path::PathBuf::from("/nonexistent/snap.ctprof.zst"),
            group_by: ShowGroupBy::Pcomm,
            cgroup_flatten: vec![],
            no_thread_normalize: false,
            no_cg_normalize: false,
            // Invalid direction `bogusdir` makes parse_sort_by bail
            // before the snapshot is touched.
            sort_by: "not_a_metric:bogusdir".to_string(),
            columns: String::new(),
            sections: String::new(),
            metrics: String::new(),
            wrap: false,
            limit: 0,
        };
        let err = run_show(&args).expect_err(
            "an invalid --sort-by spec must fail before the absent snapshot \
             is loaded",
        );
        // `{:#}` flattens the anyhow context chain so the
        // context-label substrings are reachable.
        let flat = format!("{err:#}");
        assert!(
            flat.contains("parse --sort-by"),
            "error must come from the sort-by parser (the `parse --sort-by` \
             with_context label), got: {flat}",
        );
        assert!(
            !flat.contains("load snapshot"),
            "the snapshot load must not run — its `load snapshot` context \
             label must be absent, proving the parse fails first, got: {flat}",
        );
    }
}