dbgscope 0.3.0

use std::collections::{HashMap, HashSet};
use std::num::NonZeroUsize;
use std::sync::Arc;
use std::time::{Duration, Instant};

use thiserror::Error;

use super::decode::{
    MAX_POOL_HEADER_CHUNK, PAGE_SIZE, PoolHeaderLayout, SpecialPoolHeader, VS_UNIT, big_page_hash,
    decode_descriptor_at, decode_large_requested_size, decode_lfh_subsegment, decode_pool_header,
    decode_rb_root_for, decode_slist_header_next, decode_special_pool_header, decode_vs_chunk,
    descriptor_backend, lfh_bitmap_state, read_u16, read_u32, read_u64,
    valid_descriptor_tree_signature, valid_page_segment_signature, valid_vs_signature,
};
use super::{
    HeapIdentity, PoolBackend, PoolKind, PoolSpan, PoolState,
    layout::{LayoutError, PoolLayout, SlotBackReference, VsShape},
};

type SnapshotSource = Box<dyn std::error::Error + Send + Sync>;

/// A set of chunk addresses shared by every region one heap's evidence covers.
///
/// Shared rather than owned per region: see [`PoolRegion::reusable_chunks`].
type SharedChunks = Arc<HashSet<u64>>;

#[derive(Debug, Error)]
pub(crate) enum SnapshotError {
    #[error("read at {address:#x}+{size:#x}: {source}")]
    Read {
        address: u64,
        size: usize,
        #[source]
        source: SnapshotSource,
    },
    #[error("valid-region query at {address:#x}+{size:#x}: {source}")]
    RegionQuery {
        address: u64,
        size: usize,
        #[source]
        source: SnapshotSource,
    },
    #[error(
        "sparse virtual range at {address:#x}+{size:#x} (valid {valid_base:#x}+{valid_size:#x})"
    )]
    RegionValidation {
        address: u64,
        size: usize,
        valid_base: u64,
        valid_size: usize,
    },
    #[error("snapshot layout lookup failed: {source}")]
    Layout {
        #[source]
        source: LayoutError,
    },
    #[error("pool snapshot interrupted by Ctrl+C")]
    Interrupted,
    #[error("pool snapshot ran out of its walk budget")]
    BudgetExpired,
    #[error("interrupt-status query failed: {source}")]
    InterruptQuery {
        #[source]
        source: SnapshotSource,
    },
    #[error("invalid snapshot data: {detail}")]
    InvalidData { detail: String },
}

impl SnapshotError {
    /// Whether this ends the whole walk rather than one step of it.
    ///
    /// Every other variant is local — a node that would not read, a region that is not
    /// committed — so the walk records it and carries on. These two are not: carrying on
    /// would ignore the operator's Ctrl+C, or run past the deadline that exists precisely
    /// so nobody is left waiting on a walk they have already given up on.
    ///
    /// **Every site that swallows an error into a diagnostic has to re-raise these**, or
    /// the stop signal is absorbed by the first enclosing loop and the walk keeps going —
    /// which is the failure this predicate exists to make hard to reintroduce.
    fn halts_walk(&self) -> bool {
        matches!(self, Self::Interrupted | Self::BudgetExpired)
    }
}

impl From<LayoutError> for SnapshotError {
    fn from(source: LayoutError) -> Self {
        Self::Layout { source }
    }
}

fn missing_layout(item: impl Into<String>) -> SnapshotError {
    LayoutError::Missing { item: item.into() }.into()
}

#[derive(Debug, Clone)]
pub(crate) struct PoolRegion {
    pub address: u64,
    pub size: usize,
    pub requested_size: Option<u64>,
    pub pool_kind: PoolKind,
    pub numa_node: u16,
    pub heap: HeapIdentity,
    pub subsegment: Option<u64>,
    pub backend: PoolBackend,
    pub unit_size: u32,
    /// An LFH subsegment's busy blocks, one bit per slot. Discovery reduces the allocator's own
    /// `BlockBitmap` to this with `LfhBitmap::normalise`, so nothing that walks a region has
    /// to know which allocator laid it out.
    pub bitmap: Vec<u8>,
    pub heap_key: u64,
    pub pool_header: PoolHeaderLayout,
    pub vs_header_size: usize,
    pub vs_sizes_offset: usize,
    /// Whether this heap's VS allocator keeps a user block out of the last [`VS_UNIT`] bytes of
    /// a page: `_HEAP_VS_CONTEXT.Config.Flags.PageAlignLargeAllocs`, read once at discovery.
    ///
    /// A VS chunk's user block ordinarily begins right after its `_HEAP_VS_CHUNK_HEADER`. With
    /// this set, `nt!RtlpHpVsContextAllocate` tests whether `chunk + 0x20` is page-aligned and
    /// hands out `chunk + 0x20` when it is — one unit later than usual, with the usable size one
    /// unit shorter. Read on x64 26100.33438 at `+0xf6..+0x12a`: `lea rdi,[rsi+10h]`, then
    /// `test byte ptr [rbp+4],1` / `lea rax,[rsi+20h]` / `test rax,0FFFh` / `lea rdi,[rsi+20h]` /
    /// `add ebx,0FFFFFFF0h`, `rbp` being the context and `+4` its `Config`.
    /// `RtlpHpVsContextAllocateFinalize+0x3d` does the same for a multi-allocation, and
    /// `RtlpHpVsContextFree+0xf2` is the mirror: under the flag, a page-aligned user block's
    /// chunk is taken from `block - 0x20`. The kernel needs it because `ExFreeHeapPool+0xd6`
    /// tells a chunk from a page allocation by `P & 0xfff`, and a `_POOL_HEADER` in the last
    /// sixteen bytes of a page would hand its caller a page-aligned `P`.
    ///
    /// What that looked like from here before the walk consulted it (windbg-mcp `FOLLOWUPS.md`
    /// item 101): every chunk whose header sat at page offset 0xfe0 had its pool header decoded
    /// at 0xff0, which holds whatever the chunk held when it was free — two tree pointers — so
    /// the tag was the top half of a kernel address, the span began sixteen bytes early and ran
    /// sixteen bytes long, and the real header at the page start went unclaimed. Measured
    /// 2026-10-14 on `ctf-vm`: 1,275 paged chunks under the "tag" `0x84e7ffff`, every one with
    /// its candidate at 0xff0; the chunk at `0xffffe78459203fe0` is `0x80` bytes, holds
    /// `ffffe78459207e18 ffffe78459213bc8` at `+0x10`, and has at `0xffffe78459204000` the
    /// `MmDT` header `!pool` names there (`size: 60`). Both pool heaps on that kernel carry the
    /// flag; no target without it has been measured, which is what gating on it is for.
    pub vs_skips_page_end: bool,
    pub known_tag: Option<u32>,
    /// Allocator-derived states for segment/page-range cells.
    pub states: Vec<PoolState>,
    /// VS chunk-header addresses present in the free tree.
    ///
    /// Shared, not owned: one region is created per page-range descriptor, so on a real
    /// kernel there are thousands of them, and they all carry the *same* evidence — the one
    /// set their heap's free tree produced. Cloning the set per region copied it thousands
    /// of times over, entirely off the wire and so invisible when the walk was blamed on
    /// debugger reads.
    pub reusable_chunks: SharedChunks,
    /// VS chunk-header addresses present in delay-free/lookaside lists. Shared for the
    /// reason [`Self::reusable_chunks`] is.
    pub cached_chunks: SharedChunks,
    /// Every `nt!PoolBigPageTable` entry whose allocation begins inside this region, by the
    /// address it begins at.
    ///
    /// Resolved during discovery, where the table is, so that walking a chunk needs no reads of
    /// its own. Empty on every region that holds no such allocation, which is most of them — and
    /// on every user-mode one, which has no such table.
    pub big_pool: Arc<HashMap<u64, BigPageEntry>>,
}

/// One run of address space the target's memory manager describes the same way; see
/// [`PoolMemory::committed_run`].
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct CommitRun {
    /// One past the last byte of the run, and always above the address it was asked about —
    /// so a caller stepping a span with it cannot fail to advance.
    pub end: u64,
    /// Whether pages exist behind it.
    pub committed: bool,
}

pub(crate) trait PoolMemory {
    fn read_exact(&self, address: u64, size: usize) -> Result<Vec<u8>, SnapshotError>;
    fn valid_region(&self, address: u64, size: usize) -> Result<(u64, usize), SnapshotError>;
    fn interrupted(&self) -> Result<bool, SnapshotError>;

    /// Whether this walk's wall-clock deadline has passed.
    ///
    /// Defaulted to "never" so a plain memory source carries no clock; [`Budgeted`] is what
    /// puts one in front of it, and [`SnapshotWalker::walk`] is what wraps it.
    fn out_of_budget(&self) -> bool {
        false
    }

    /// What the target's **memory manager** says about the run of pages `address` lies in —
    /// a different question from whether the debugger can read it, and the only one that
    /// separates *nothing was ever here* from *we could not see what is here*.
    ///
    /// `None` means this source cannot say, and the caller then keeps the conservative
    /// reading: what did not read stays a hole in the walk's coverage. **That default is
    /// deliberate.** A source that does not override this behaves exactly as it did, so no
    /// fixture starts excusing gaps by accident — the ones that exercise
    /// [`PoolState::Uncommitted`] have to say so, which is what makes their assertions mean
    /// anything.
    fn committed_run(&self, _address: u64) -> Option<CommitRun> {
        None
    }
}

/// A memory source with a deadline attached.
///
/// The walk already polls the engine's interrupt flag, which Ctrl+C sets and so does a host
/// through [`InterruptHandle`](crate::dbgeng::InterruptHandle) — but **both need somebody
/// watching**, and a caller blocked on the query is not. Without a clock its own call never
/// returns and the engine keeps walking; since one engine serves one target one call at a
/// time, every later call to that target queues behind the walk. The session is then wedged
/// until it is killed, which on a live kernel leaves the guest halted. This is the one
/// stopper that needs nobody present.
struct Budgeted<'a, M> {
    inner: &'a M,
    /// `None` runs to completion — correct only where something else can stop the walk: an
    /// operator at a prompt, or a host that both holds an interrupt handle and knows when to
    /// use it.
    deadline: Option<Instant>,
}

impl<'a, M> Budgeted<'a, M> {
    fn new(inner: &'a M, deadline: Option<Instant>) -> Self {
        Self { inner, deadline }
    }
}

impl<M: PoolMemory> PoolMemory for Budgeted<'_, M> {
    fn read_exact(&self, address: u64, size: usize) -> Result<Vec<u8>, SnapshotError> {
        self.inner.read_exact(address, size)
    }

    fn valid_region(&self, address: u64, size: usize) -> Result<(u64, usize), SnapshotError> {
        self.inner.valid_region(address, size)
    }

    fn interrupted(&self) -> Result<bool, SnapshotError> {
        self.inner.interrupted()
    }

    /// This wrapper *adds* a deadline; it does not replace whatever the source already had,
    /// so wrapping can only ever make a walk stop sooner.
    fn out_of_budget(&self) -> bool {
        self.deadline
            .is_some_and(|deadline| Instant::now() >= deadline)
            || self.inner.out_of_budget()
    }

    fn committed_run(&self, address: u64) -> Option<CommitRun> {
        self.inner.committed_run(address)
    }
}

impl PoolMemory for crate::dbgeng::DebugEngine {
    fn read_exact(&self, address: u64, size: usize) -> Result<Vec<u8>, SnapshotError> {
        self.read_memory(address, size)
            .map_err(|source| SnapshotError::Read {
                address,
                size,
                source: Box::new(source),
            })
    }

    fn valid_region(&self, address: u64, size: usize) -> Result<(u64, usize), SnapshotError> {
        self.valid_virtual_region(address, size)
            .map_err(|source| SnapshotError::RegionQuery {
                address,
                size,
                source: Box::new(source),
            })
    }

    fn interrupted(&self) -> Result<bool, SnapshotError> {
        crate::dbgeng::DebugEngine::interrupted(self).map_err(|source| {
            SnapshotError::InterruptQuery {
                source: Box::new(source),
            }
        })
    }

    /// **Not asked of a kernel session at all.** `QueryVirtual` is the engine reaching
    /// `NtQueryVirtualMemory` in the debuggee, which a kernel target has no equivalent of, so
    /// every call there fails — and a kernel walk files thousands of unreadable spans, which
    /// would be thousands of failed calls to learn the same thing each time. The kernel walk
    /// therefore keeps counting every unreadable page against its coverage, which on a target
    /// trimming paged pool is the truth: those pages are committed and the walk really did
    /// miss what was in them (`windbg-mcp` FOLLOWUPS item 100).
    ///
    /// Every other answer that is not plainly *committed* or *not committed* — a query that
    /// fails, a zero-length run, a state this crate does not name — is `None` rather than a
    /// guess, because the guess is what would make this lie: read as uncommitted it excuses a
    /// gap that is real.
    fn committed_run(&self, address: u64) -> Option<CommitRun> {
        if crate::dbgeng::DebugEngine::is_kernel_target(self).ok()? {
            return None;
        }
        let region = self.virtual_region(address).ok()?;
        if !region.contains(address) {
            return None;
        }
        Some(CommitRun {
            end: region.end()?,
            committed: region.state.committed()?,
        })
    }
}

/// The one place a walk asks whether it is still allowed to run.
///
/// **The rule: every loop that issues a debugger read calls this, first thing.** Not only the
/// unbounded ones — a bounded loop is not automatically safe. The descriptor loop in
/// [`discover_segment_context`] is capped at 4096 iterations and reads twice per LFH
/// descriptor, so polling one level out in the segment loop still left ~8k reads between the
/// deadline passing and anything noticing. On a live KD link that is thousands of round
/// trips, minutes of them: the very failure this budget exists to prevent, reintroduced
/// inside the fix for it.
///
/// Reads themselves deliberately keep working past the deadline rather than failing. Having
/// [`Budgeted`] reject them would bound the walk with no rule to remember — but nearly every
/// read here is swallowed into a "cannot read ..." diagnostic, so out-of-time would surface
/// as *unreadable memory*. That is the one lie this module is built to avoid: what the walk
/// did not reach must never be reported as what is not there.
///
/// Both answers halt (see [`SnapshotError::halts_walk`]); they differ in what the caller does
/// with the halt, which [`SnapshotWalker::walk`] decides.
fn check_budget(memory: &impl PoolMemory) -> Result<(), SnapshotError> {
    if memory.interrupted()? {
        return Err(SnapshotError::Interrupted);
    }
    if memory.out_of_budget() {
        return Err(SnapshotError::BudgetExpired);
    }
    Ok(())
}

fn guarded_read(
    memory: &impl PoolMemory,
    address: u64,
    size: usize,
) -> Result<Vec<u8>, SnapshotError> {
    if size == 0 {
        return Ok(Vec::new());
    }
    let (valid_base, valid_size) = memory.valid_region(address, size)?;
    if valid_base != address || valid_size < size {
        return Err(SnapshotError::RegionValidation {
            address,
            size,
            valid_base,
            valid_size,
        });
    }
    memory.read_exact(address, size)
}

fn scalar(memory: &impl PoolMemory, address: u64, size: usize) -> Result<u64, SnapshotError> {
    let bytes = guarded_read(memory, address, size)?;
    match size {
        1 => bytes
            .first()
            .map(|&byte| byte as u64)
            .ok_or_else(|| SnapshotError::InvalidData {
                detail: "short u8".into(),
            }),
        2 => Ok(
            u16::from_le_bytes(bytes.try_into().map_err(|_| SnapshotError::InvalidData {
                detail: "short u16".into(),
            })?) as u64,
        ),
        4 => Ok(
            u32::from_le_bytes(bytes.try_into().map_err(|_| SnapshotError::InvalidData {
                detail: "short u32".into(),
            })?) as u64,
        ),
        8 => Ok(u64::from_le_bytes(bytes.try_into().map_err(|_| {
            SnapshotError::InvalidData {
                detail: "short u64".into(),
            }
        })?)),
        _ => Err(SnapshotError::InvalidData {
            detail: format!("unsupported scalar size {size}"),
        }),
    }
}

fn walk_tree_nodes(
    memory: &impl PoolMemory,
    root: u64,
    left_offset: usize,
    right_offset: usize,
    limit: usize,
    label: &str,
    diagnostics: &mut Vec<String>,
) -> Result<Vec<u64>, SnapshotError> {
    let mut nodes = Vec::new();
    let mut stack = vec![root];
    let mut seen = HashSet::new();
    // Links are followed exactly as stored. They were masked to 16 bytes, which is harmless for
    // a page-range descriptor's node and wrong for a VS free chunk's: its `Node` is at +0x8 of a
    // 16-aligned header, so masking landed every VS node on its chunk's encoded `Sizes` word,
    // read that as the left link, and took the real left link for the right one — the
    // `unreadable VS free tree node 0x84a23d…` diagnostics on ARM64 26100.1 (2026-09-23).
    while let Some(node) = stack.pop() {
        if node == 0 {
            continue;
        }
        check_budget(memory)?;
        if !seen.insert(node) {
            diagnostics.push(format!("{label} cycle detected at {node:#x}"));
            continue;
        }
        if nodes.len() >= limit {
            diagnostics.push(format!("{label} traversal limit reached"));
            break;
        }
        let size = left_offset.max(right_offset).saturating_add(8);
        match guarded_read(memory, node, size) {
            Ok(bytes) => {
                nodes.push(node);
                if let Some(right) = read_u64(&bytes, right_offset) {
                    stack.push(right);
                }
                if let Some(left) = read_u64(&bytes, left_offset) {
                    stack.push(left);
                }
            }
            Err(error) => diagnostics.push(format!("unreadable {label} node {node:#x}: {error}")),
        }
    }
    Ok(nodes)
}

fn tree_nodes(
    memory: &impl PoolMemory,
    layout: &PoolLayout,
    tree_address: u64,
    limit: usize,
    label: &str,
    diagnostics: &mut Vec<String>,
) -> Result<Vec<u64>, SnapshotError> {
    // Polled before the root and flag reads, not only inside the loop below: every caller
    // reaches here a few unpolled reads past its own poll, which is a stretch of up to five
    // between checks, and the sampled budgets in
    // `test_the_walk_stops_reading_promptly_once_its_budget_is_gone` landed on it once the
    // per-heap config read shifted them by one (2026-10-14).
    check_budget(memory)?;
    let Ok(root_offset) = layout.field("_RTL_RB_TREE", "Root") else {
        diagnostics.push(format!("cannot resolve {label} root field"));
        return Ok(Vec::new());
    };
    let root_value = match scalar(memory, tree_address + root_offset as u64, 8) {
        Ok(value) => value,
        Err(error) => {
            diagnostics.push(format!("cannot read {label} root: {error}"));
            return Ok(Vec::new());
        }
    };
    let encoded = if let Ok(encoded_offset) = layout.field("_RTL_RB_TREE", "Encoded") {
        match scalar(memory, tree_address + encoded_offset as u64, 1) {
            Ok(value) => value & 1 != 0,
            Err(error) => {
                diagnostics.push(format!("cannot read {label} encoded flag: {error}"));
                return Ok(Vec::new());
            }
        }
    } else {
        false
    };
    let decoded_root = if layout.is_user() {
        decode_rb_root_for(root_value, tree_address, encoded, true)
    } else {
        super::decode::decode_rb_root(root_value, tree_address, encoded)
    };
    let Some(root) = decoded_root else {
        diagnostics.push(format!("rejecting corrupt {label} root {root_value:#x}"));
        return Ok(Vec::new());
    };
    let Ok(left) = layout.field("_RTL_BALANCED_NODE", "Left") else {
        return Ok(Vec::new());
    };
    let Ok(right) = layout.field("_RTL_BALANCED_NODE", "Right") else {
        return Ok(Vec::new());
    };
    walk_tree_nodes(memory, root, left, right, limit, label, diagnostics)
}

fn walk_slist_nodes(
    memory: &impl PoolMemory,
    layout: &PoolLayout,
    head: u64,
    limit: usize,
    label: &str,
    diagnostics: &mut Vec<String>,
) -> Result<Vec<u64>, SnapshotError> {
    let mut nodes = Vec::new();
    let mut seen = HashSet::new();
    let Ok(header) = layout.type_layout("_SLIST_HEADER") else {
        diagnostics.push(format!("cannot resolve {label} SLIST header type"));
        return Ok(nodes);
    };
    let Ok(alignment_offset) = layout.field("_SLIST_HEADER", "Alignment") else {
        diagnostics.push(format!("cannot resolve {label} SLIST depth field"));
        return Ok(nodes);
    };
    let Ok(region_offset) = layout.field("_SLIST_HEADER", "Region") else {
        diagnostics.push(format!("cannot resolve {label} SLIST next field"));
        return Ok(nodes);
    };
    let bytes = match guarded_read(memory, head, header.size as usize) {
        Ok(value) => value,
        Err(error) => {
            diagnostics.push(format!("cannot read {label} list head: {error}"));
            return Ok(nodes);
        }
    };
    let depth = read_u16(&bytes, alignment_offset).map_or(0, usize::from);
    let mut entry = read_u64(&bytes, region_offset)
        .map(decode_slist_header_next)
        .unwrap_or(0);
    let expected = depth.min(limit);
    while entry != 0 && nodes.len() < expected {
        check_budget(memory)?;
        if !seen.insert(entry) {
            diagnostics.push(format!("{label} list cycle detected at {entry:#x}"));
            break;
        }
        nodes.push(entry);
        match scalar(memory, entry, 8) {
            Ok(next) => entry = next & !0xf,
            Err(error) => {
                diagnostics.push(format!("unreadable {label} list entry {entry:#x}: {error}"));
                break;
            }
        }
    }
    if depth > limit {
        diagnostics.push(format!("{label} list traversal limit reached"));
    } else if nodes.len() != depth {
        diagnostics.push(format!(
            "{label} list depth is {depth}, but only {} entries were readable",
            nodes.len()
        ));
    }
    Ok(nodes)
}

fn insert_cached_chunk_candidates(
    cached: &mut HashSet<u64>,
    entry: u64,
    pool_header_size: u64,
    vs_header_size: u64,
) {
    cached.insert(entry);
    let overhead = pool_header_size.saturating_add(vs_header_size);
    if let Some(header) = entry.checked_sub(overhead) {
        cached.insert(header);
        // The lookaside's half of [`PoolRegion::vs_skips_page_end`]: data sixteen bytes into a
        // page has its pool header at the page start and, under the flag, its chunk header one
        // unit before where the subtraction above puts it. Both are kept, because this is a
        // set of candidates and only a chunk decoded at one of them can match.
        if header & (PAGE_SIZE - 1) == PAGE_SIZE - pool_header_size {
            cached.insert(header.saturating_sub(VS_UNIT));
        }
    }
}

#[derive(Default)]
struct Discovery {
    regions: Vec<PoolRegion>,
    diagnostics: Vec<String>,
    /// `nt!PoolBigPageTable`, opened once for the whole walk. `None` on a target that has
    /// none — a user-mode heap, or a kernel whose symbols do not carry it.
    big_pages: Option<BigPageTable>,
}

impl Discovery {
    /// What the big-page table says about the allocation at `address`, reading whichever
    /// part of the table that question lands in and keeping it for the next one.
    fn big_page(
        &mut self,
        memory: &impl PoolMemory,
        address: u64,
    ) -> Result<Option<BigPageEntry>, SnapshotError> {
        let Self {
            diagnostics,
            big_pages,
            ..
        } = self;
        let Some(table) = big_pages.as_mut() else {
            return Ok(None);
        };
        table.lookup(memory, address, diagnostics)
    }
}

const SPECIAL_POOL_KINDS: [PoolKind; 4] = [
    PoolKind::SpecialNonPaged,
    PoolKind::SpecialNonPagedNx,
    PoolKind::SpecialPaged,
    PoolKind::SpecialPrototypePaged,
];

/// Enumerates every pool region, appending to `discovery`.
///
/// Takes `discovery` by reference rather than returning it so that a walk which halts
/// part-way still hands back the regions it *did* find. Returning `Result<Discovery, _>`
/// threw them away on the one path where they matter most — the caller can then walk what
/// discovery reached instead of reporting an empty pool.
fn discover_pool_regions(
    memory: &impl PoolMemory,
    layout: &PoolLayout,
    traversal_limit: usize,
    discovery: &mut Discovery,
) -> Result<(), SnapshotError> {
    let state_address = *layout
        .globals
        .get("ExPoolState")
        .ok_or_else(|| missing_layout("ExPoolState"))?;
    let state = layout.type_layout("_EX_POOL_HEAP_MANAGER_STATE")?;
    let node = layout.type_layout("_EX_HEAP_POOL_NODE")?;
    let number_offset = layout.field("_EX_POOL_HEAP_MANAGER_STATE", "NumberOfPools")?;
    let node_offset = layout.field("_EX_POOL_HEAP_MANAGER_STATE", "PoolNode")?;
    let special_offset = layout.field("_EX_POOL_HEAP_MANAGER_STATE", "SpecialHeaps")?;
    let heaps_offset = layout.field("_EX_HEAP_POOL_NODE", "Heaps")?;
    let lookasides_offset = layout.field("_EX_HEAP_POOL_NODE", "Lookasides").ok();
    let dynamic_lookaside_size = layout
        .type_layout("_RTL_DYNAMIC_LOOKASIDE")
        .ok()
        .map(|lookaside| lookaside.size as u64);
    let number = scalar(memory, state_address + number_offset as u64, 4)? as usize;
    if number == 0 || number > 256 {
        return Err(SnapshotError::InvalidData {
            detail: format!("implausible ExPoolState.NumberOfPools {number}"),
        });
    }
    let mut heaps = Vec::new();
    for numa_node in 0..number {
        // Four pointer reads per node, up to 256 nodes: bounded, but a thousand round trips
        // is still seconds on a live link, and the rule is uniform — a loop that reads polls.
        check_budget(memory)?;
        let Some(node_address) = state_address
            .checked_add(node_offset as u64)
            .and_then(|address| address.checked_add(numa_node as u64 * node.size as u64))
        else {
            discovery
                .diagnostics
                .push("pool-node address overflow".into());
            continue;
        };
        for heap_index in 0..4usize {
            let pointer_address = node_address + heaps_offset as u64 + heap_index as u64 * 8;
            let heap = match scalar(memory, pointer_address, 8) {
                Ok(value) => value,
                Err(error) => {
                    discovery.diagnostics.push(format!(
                        "cannot read pool node {numa_node} heap {heap_index}: {error}"
                    ));
                    continue;
                }
            };
            if heap != 0 {
                let pool_kind = match heap_index {
                    0 => PoolKind::NonPagedExecutable,
                    1 => PoolKind::NonPagedNx,
                    2 => PoolKind::Paged,
                    _ => PoolKind::PrototypePaged,
                };
                let dynamic_lookaside =
                    lookasides_offset
                        .zip(dynamic_lookaside_size)
                        .and_then(|(offset, size)| {
                            node_address.checked_add(offset as u64).and_then(|base| {
                                base.checked_add(u64::from(pool_kind.is_paged()) * size)
                            })
                        });
                heaps.push((heap, numa_node as u16, pool_kind, false, dynamic_lookaside));
            }
        }
    }
    for (special_index, pool_kind) in SPECIAL_POOL_KINDS.into_iter().enumerate() {
        let pointer_address = state_address + special_offset as u64 + special_index as u64 * 8;
        match scalar(memory, pointer_address, 8) {
            Ok(heap) if heap != 0 => heaps.push((heap, 0, pool_kind, true, None)),
            Ok(_) => {}
            Err(error) => discovery.diagnostics.push(format!(
                "cannot read special pool heap {special_index}: {error}"
            )),
        }
    }

    let globals_address = *layout
        .globals
        .get("RtlpHpHeapGlobals")
        .ok_or_else(|| missing_layout("RtlpHpHeapGlobals"))?;
    let heap_key = scalar(
        memory,
        globals_address + layout.field("_RTLP_HP_HEAP_GLOBALS", "HeapKey")? as u64,
        8,
    )?;
    let lfh_key = scalar(
        memory,
        globals_address + layout.field("_RTLP_HP_HEAP_GLOBALS", "LfhKey")? as u64,
        8,
    )?;
    // Opened before the heaps because every one of them will ask: on a kernel, an allocated
    // page range that holds no subsegment is a big-pool allocation whose tag is only here.
    discovery.big_pages = BigPageTable::open(memory, layout, &mut discovery.diagnostics)?;
    for (heap_address, numa_node, pool_kind, special, dynamic_lookaside) in heaps {
        let identity = HeapIdentity {
            pool_state: state_address,
            heap: heap_address,
            special,
        };
        if let Err(error) = discover_heap_regions(
            memory,
            layout,
            heap_address,
            numa_node,
            pool_kind,
            identity,
            dynamic_lookaside,
            heap_key,
            lfh_key,
            traversal_limit,
            discovery,
        ) {
            // A halt is about the walk, not about this heap. Recorded as a per-heap failure
            // it would read as "heap 3 is unreadable" and — worse — the loop would carry on
            // to heaps 4..n, each stopping the same way, so the deadline would bound one
            // heap instead of the walk.
            if error.halts_walk() {
                return Err(error);
            }
            discovery.diagnostics.push(format!(
                "cannot fully discover heap {heap_address:#x}: {error}"
            ));
        }
    }
    let _ = state.size;
    Ok(())
}

/// One place carrying VS free-chunk state.
struct VsRoot {
    base: u64,
    tree_offset: usize,
    delay_offset: Option<usize>,
}

/// Resolves where a VS context keeps its free-chunk state. Every shape is supported, because a
/// debugger host does not get to choose which build it is pointed at:
///
/// * **inline family** — `FreeChunkTree`/`DelayFreeContext` are in `_HEAP_VS_CONTEXT`,
///   so there is exactly one root: the context itself.
/// * **affinity-slot families** — they live in `_HEAP_VS_AFFINITY_SLOT`s reached through a slot
///   map. `SlotMapRef` and each `SlotRef` are offsets *from the context*, scaled by 64 bytes, and
///   the map holds `AffinityMask + 1` entries. Entries routinely share a slot, so the result is
///   deduplicated. The two differ only in how a slot proves it belongs to this context — see
///   [`SlotBackReference`].
///
/// Which shape it is comes from [`PoolLayout::vs_shape`] rather than from a second reading of the
/// field table here, so the walk and the provenance a caller is told about cannot disagree.
///
/// An empty vector means no shape resolved; the caller then walks no VS evidence rather than
/// guessing at an address. An `Err` means the walk itself is over — the slot map is up to 256
/// entries plus an owner read per distinct slot, all over the wire, so it polls the deadline like
/// every other read-issuing loop (see [`check_budget`]). It cannot lean on a later poll: invalid
/// slot refs yield no roots, so nothing downstream reads at all and the next check is a whole
/// segment context away.
fn vs_roots(
    memory: &impl PoolMemory,
    layout: &PoolLayout,
    context: u64,
    diagnostics: &mut Vec<String>,
) -> Result<Vec<VsRoot>, SnapshotError> {
    let shape = match layout.vs_shape() {
        Ok(shape) => shape,
        Err(detail) => {
            diagnostics.push(format!(
                "VS free-chunk state is in neither the context nor an affinity slot: {detail}"
            ));
            return Ok(Vec::new());
        }
    };

    let (slot_map_ref_offset, affinity_offset, slot_ref_offset, tree_offset, delay_offset, back) =
        match shape {
            // A context that carries the tree has no slots.
            VsShape::Inline { tree, delay } => {
                return Ok(vec![VsRoot {
                    base: context,
                    tree_offset: tree,
                    delay_offset: delay,
                }]);
            }
            VsShape::AffinitySlots {
                slot_map_ref,
                affinity_mask,
                slot_ref,
                tree,
                delay,
                back,
            } => (slot_map_ref, affinity_mask, slot_ref, tree, delay, back),
        };

    let (Ok(slot_map_ref), Ok(affinity_mask)) = (
        scalar(memory, context + slot_map_ref_offset as u64, 2),
        scalar(memory, context + affinity_offset as u64, 1),
    ) else {
        diagnostics.push(format!(
            "cannot read the VS slot map of context {context:#x}"
        ));
        return Ok(Vec::new());
    };

    let entries = affinity_mask as usize + 1;
    // A zero ref would alias the context, and the entry count is bounded by the affinity
    // mask; both would otherwise walk arbitrary memory.
    if slot_map_ref == 0 || entries > 256 {
        diagnostics.push(format!(
            "implausible VS slot map for context {context:#x}: ref {slot_map_ref:#x}, {entries} entries"
        ));
        return Ok(Vec::new());
    }

    let entry_size = layout
        .type_layout("_HEAP_VS_SLOT_MAP")
        .map_or(4, |map| map.size as u64);
    let slot_map = context + (slot_map_ref << 6);

    let mut roots = Vec::new();
    let mut seen = HashSet::new();
    for index in 0..entries {
        check_budget(memory)?;
        let entry = slot_map + index as u64 * entry_size;
        let Ok(slot_ref) = scalar(memory, entry + slot_ref_offset as u64, 2) else {
            // Skipping silently would drop that slot's free tree and delay-free list, so
            // its chunks get misclassified with nothing to say why.
            diagnostics.push(format!(
                "cannot read VS slot map entry {index} at {entry:#x}; its affinity slot is omitted"
            ));
            continue;
        };
        if slot_ref == 0 {
            continue;
        }
        let slot = context + (slot_ref << 6);
        if !seen.insert(slot) {
            continue;
        }
        // Require the slot to name the context we came from. A misdecoded SlotRef would
        // otherwise point the tree walk at unrelated memory that happens to be readable.
        //
        // Both spellings are read as the same question — *which context does this slot claim* —
        // so that a slot that disagrees produces one diagnostic shape rather than two. What the
        // stored value means differs: an address is the answer, a displacement has to be
        // subtracted from the slot to become one. `checked_sub` because a displacement larger
        // than the slot's own address is not a context, it is a rejection.
        let back_offset = match back {
            SlotBackReference::Address(offset) | SlotBackReference::Displacement(offset) => offset,
        };
        let claimed = scalar(memory, slot + back_offset as u64, 8).map(|value| match back {
            SlotBackReference::Address(_) => Some(value),
            SlotBackReference::Displacement(_) => slot.checked_sub(value),
        });
        match claimed {
            Ok(Some(owner)) if owner == context => roots.push(VsRoot {
                base: slot,
                tree_offset,
                delay_offset,
            }),
            Ok(Some(owner)) => diagnostics.push(format!(
                "VS affinity slot {slot:#x} claims context {owner:#x}, not {context:#x}; skipped"
            )),
            // A displacement that runs off the bottom of the address space names no context at
            // all; saying so as "claims none" keeps it in the same shape as any other disagreement.
            Ok(None) => diagnostics.push(format!(
                "VS affinity slot {slot:#x} claims context none, not {context:#x}; skipped"
            )),
            Err(error) => {
                diagnostics.push(format!("cannot read VS affinity slot {slot:#x}: {error}"))
            }
        }
    }
    Ok(roots)
}

/// `_RTL_HP_VS_CONFIG.Flags.PageAlignLargeAllocs`: `Pos 0, 1 Bit` of the byte at
/// `_HEAP_VS_CONTEXT.Config` on the x64 26100.33438 PDB, and the first flag the type declares.
/// The byte's offset is resolved; the bit's position inside it is this.
const VS_CONFIG_PAGE_ALIGN: u64 = 1;

/// Whether `heap_address`'s VS allocator moves a user block off the end of a page — see
/// [`PoolRegion::vs_skips_page_end`] for the rule and where it was read.
///
/// Asked once per heap, here, because what it decides is applied to every chunk. A layout that
/// did not resolve `Config`, or a context that will not read, answers **unset** and says so: that
/// is the reading the walk made before it consulted the flag at all, and on a target that does
/// carry it the diagnostic is what points at the ~1,300 misplaced headers per boot it costs.
/// An `Err` is the budget or an interrupt, polled before the read like every other read here.
fn vs_skips_page_end(
    memory: &impl PoolMemory,
    layout: &PoolLayout,
    heap_address: u64,
    diagnostics: &mut Vec<String>,
) -> Result<bool, SnapshotError> {
    let Ok(context_offset) = layout.field("_SEGMENT_HEAP", "VsContext") else {
        return Ok(false);
    };
    let Ok(config_offset) = layout.field("_HEAP_VS_CONTEXT", "Config") else {
        diagnostics.push(format!(
            "VS context of heap {heap_address:#x} has no resolved Config: a user block ending \
             a page is read at its chunk header"
        ));
        return Ok(false);
    };
    check_budget(memory)?;
    let address = heap_address + context_offset as u64 + config_offset as u64;
    Ok(match scalar(memory, address, 1) {
        Ok(flags) => flags & VS_CONFIG_PAGE_ALIGN != 0,
        Err(error) => {
            diagnostics.push(format!(
                "cannot read VS context config at {address:#x}: {error}; a user block ending a \
                 page is read at its chunk header"
            ));
            false
        }
    })
}

fn discover_vs_evidence(
    memory: &impl PoolMemory,
    layout: &PoolLayout,
    heap_address: u64,
    dynamic_lookaside: Option<u64>,
    limit: usize,
    diagnostics: &mut Vec<String>,
) -> Result<(SharedChunks, SharedChunks), SnapshotError> {
    let Ok(vs_context_offset) = layout.field("_SEGMENT_HEAP", "VsContext") else {
        return Ok(Default::default());
    };
    let context = heap_address + vs_context_offset as u64;
    let roots = vs_roots(memory, layout, context, diagnostics)?;
    if roots.is_empty() {
        return Ok(Default::default());
    }
    let tree_node_offset = layout
        .field("_HEAP_VS_CHUNK_FREE_HEADER", "TreeNode")
        .unwrap_or(0);
    let mut reusable = HashSet::new();
    for root in &roots {
        reusable.extend(
            tree_nodes(
                memory,
                layout,
                root.base + root.tree_offset as u64,
                limit,
                "VS free tree",
                diagnostics,
            )?
            .into_iter()
            .map(|node| node.saturating_sub(tree_node_offset as u64)),
        );
    }

    let mut cached = HashSet::new();
    let pool_header_size = layout
        .type_layout("_POOL_HEADER")
        .map_or(0, |value| value.size as u64);
    let vs_header_size = layout
        .type_layout("_HEAP_VS_CHUNK_HEADER")
        .map_or(0, |value| value.size as u64);
    if let Ok(list_offset) = layout.field("_HEAP_VS_DELAY_FREE_CONTEXT", "ListHead") {
        for root in &roots {
            // In the inline family the delay-free list sits beside the tree in the context;
            // in the affinity family it is per-slot. Either way it is known from `root.base`.
            let Some(delay_offset) = root.delay_offset else {
                continue;
            };
            for entry in walk_slist_nodes(
                memory,
                layout,
                root.base + delay_offset as u64 + list_offset as u64,
                limit,
                "VS delay-free",
                diagnostics,
            )? {
                insert_cached_chunk_candidates(
                    &mut cached,
                    entry,
                    pool_header_size,
                    vs_header_size,
                );
            }
        }
    }

    if let Some(dynamic) = dynamic_lookaside {
        let bucket_count = layout
            .field("_RTL_DYNAMIC_LOOKASIDE", "BucketCount")
            .ok()
            .and_then(|offset| scalar(memory, dynamic + offset as u64, 4).ok())
            .unwrap_or(0) as usize;
        let buckets_offset = layout.field("_RTL_DYNAMIC_LOOKASIDE", "Buckets").ok();
        let lookaside = layout.type_layout("_RTL_LOOKASIDE").ok();
        let list_offset = layout.field("_RTL_LOOKASIDE", "ListHead").ok();
        if bucket_count > 64 {
            diagnostics.push(format!(
                "rejecting implausible VS dynamic-lookaside bucket count {bucket_count}"
            ));
        } else if let (Some(buckets_offset), Some(lookaside), Some(list_offset)) =
            (buckets_offset, lookaside, list_offset)
        {
            for bucket in 0..bucket_count {
                check_budget(memory)?;
                let Some(bucket_address) = dynamic
                    .checked_add(buckets_offset as u64)
                    .and_then(|value| value.checked_add(bucket as u64 * u64::from(lookaside.size)))
                else {
                    diagnostics.push("VS dynamic-lookaside bucket address overflow".into());
                    break;
                };
                if let Ok(size_offset) = layout.field("_RTL_LOOKASIDE", "Size") {
                    match scalar(memory, bucket_address + size_offset as u64, 4) {
                        Ok(0) => continue,
                        Ok(size) if size > 0x1_0000 => {
                            diagnostics.push(format!(
                                "rejecting VS dynamic-lookaside bucket {bucket} size {size:#x}"
                            ));
                            continue;
                        }
                        Ok(_) => {}
                        Err(error) => {
                            diagnostics.push(format!(
                                "cannot read VS dynamic-lookaside bucket {bucket} size: {error}"
                            ));
                            continue;
                        }
                    }
                }
                for entry in walk_slist_nodes(
                    memory,
                    layout,
                    bucket_address + list_offset as u64,
                    limit,
                    "VS dynamic-lookaside",
                    diagnostics,
                )? {
                    // Lookaside links point at usable data, while delay-free links
                    // point at the VS chunk header. Keep both page-end header
                    // candidates; only a decoded chunk at that address can match.
                    insert_cached_chunk_candidates(
                        &mut cached,
                        entry,
                        pool_header_size,
                        vs_header_size,
                    );
                }
            }
        }
    }
    Ok((Arc::new(reusable), Arc::new(cached)))
}

#[allow(clippy::too_many_arguments)]
fn discover_heap_regions(
    memory: &impl PoolMemory,
    layout: &PoolLayout,
    heap_address: u64,
    numa_node: u16,
    pool_kind: PoolKind,
    identity: HeapIdentity,
    dynamic_lookaside: Option<u64>,
    heap_key: u64,
    lfh_key: u64,
    traversal_limit: usize,
    discovery: &mut Discovery,
) -> Result<(), SnapshotError> {
    let heap = layout.type_layout("_SEGMENT_HEAP")?;
    let context = layout.type_layout("_HEAP_SEG_CONTEXT")?;
    let contexts_offset = layout.field("_SEGMENT_HEAP", "SegContexts")?;
    let (reusable_chunks, cached_chunks) = discover_vs_evidence(
        memory,
        layout,
        heap_address,
        dynamic_lookaside,
        traversal_limit,
        &mut discovery.diagnostics,
    )?;
    let vs_skips_page_end =
        vs_skips_page_end(memory, layout, heap_address, &mut discovery.diagnostics)?;

    // Touch both LFH roots through guarded reads. Descriptor metadata remains the
    // authoritative source of active subsegments, while these probes make corrupt
    // bucket/affinity metadata an isolated diagnostic instead of a fatal read.
    if let Ok(lfh_offset) = layout.field("_SEGMENT_HEAP", "LfhContext") {
        let lfh = heap_address + lfh_offset as u64;
        for (field, label) in [
            ("Buckets", "LFH buckets"),
            ("AffinitySlots", "LFH affinity slots"),
        ] {
            if let Ok(offset) = layout.field("_HEAP_LFH_CONTEXT", field)
                && let Err(error) = guarded_read(memory, lfh + offset as u64, 8)
            {
                discovery
                    .diagnostics
                    .push(format!("cannot read {label}: {error}"));
            }
        }
    }

    for context_index in 0..2usize {
        check_budget(memory)?;
        let context_address =
            heap_address + contexts_offset as u64 + context_index as u64 * context.size as u64;
        if let Err(error) = discover_segment_context(
            memory,
            layout,
            context_address,
            numa_node,
            pool_kind,
            identity,
            heap_key,
            lfh_key,
            traversal_limit,
            &reusable_chunks,
            &cached_chunks,
            vs_skips_page_end,
            discovery,
        ) {
            // As in `discover_pool_regions`: a halt is about the walk, not about this
            // context. Absorbed here it would read as "context 1 is unreadable", and on the
            // *last* context index it would be absorbed entirely — `discover_heap_regions`
            // would carry on into `discover_large_allocations` and return `Ok`, so even a
            // Ctrl+C would only stop the walk if some later site happened to raise it again.
            if error.halts_walk() {
                return Err(error);
            }
            discovery.diagnostics.push(format!(
                "cannot discover segment context {context_index} at {context_address:#x}: {error}"
            ));
        }
    }
    discover_large_allocations(
        memory,
        layout,
        heap_address,
        numa_node,
        pool_kind,
        identity,
        heap_key,
        traversal_limit,
        discovery,
    )?;
    let _ = heap.size;
    Ok(())
}

#[allow(clippy::too_many_arguments)]
fn discover_segment_context(
    memory: &impl PoolMemory,
    layout: &PoolLayout,
    context_address: u64,
    numa_node: u16,
    pool_kind: PoolKind,
    identity: HeapIdentity,
    heap_key: u64,
    lfh_key: u64,
    traversal_limit: usize,
    reusable_chunks: &SharedChunks,
    cached_chunks: &SharedChunks,
    vs_skips_page_end: bool,
    discovery: &mut Discovery,
) -> Result<(), SnapshotError> {
    let segment = layout.type_layout("_HEAP_PAGE_SEGMENT")?;
    let descriptor = layout.type_layout("_HEAP_PAGE_RANGE_DESCRIPTOR")?;
    let lfh_bitmap = layout.lfh_bitmap();
    let shift = scalar(
        memory,
        context_address + layout.field("_HEAP_SEG_CONTEXT", "UnitShift")? as u64,
        1,
    )? as u32;
    if !(12..=20).contains(&shift) {
        return Ok(());
    }
    let first_descriptor = scalar(
        memory,
        context_address + layout.field("_HEAP_SEG_CONTEXT", "FirstDescriptorIndex")? as u64,
        1,
    )? as usize;
    let segment_mask = scalar(
        memory,
        context_address + layout.field("_HEAP_SEG_CONTEXT", "SegmentMask")? as u64,
        8,
    )?;
    let segment_size = (!segment_mask).wrapping_add(1);
    let descriptor_count = (segment_size >> shift).min(4096) as usize;
    if descriptor_count == 0 || first_descriptor >= descriptor_count {
        return Ok(());
    }

    let free_tree = context_address + layout.field("_HEAP_SEG_CONTEXT", "FreePageRanges")? as u64;
    let free_nodes: HashSet<_> = tree_nodes(
        memory,
        layout,
        free_tree,
        traversal_limit,
        "free-page tree",
        &mut discovery.diagnostics,
    )?
    .into_iter()
    .collect();

    let list_head = context_address + layout.field("_HEAP_SEG_CONTEXT", "SegmentListHead")? as u64;
    let list_entry = layout.field("_HEAP_PAGE_SEGMENT", "ListEntry")?;
    let desc_array = layout.field("_HEAP_PAGE_SEGMENT", "DescArray")?;
    let signature_offset = layout.field("_HEAP_PAGE_SEGMENT", "Signature")?;
    let unit_offset = layout.field("_HEAP_PAGE_RANGE_DESCRIPTOR", "UnitSize")?;
    let flags_offset = layout.field("_HEAP_PAGE_RANGE_DESCRIPTOR", "RangeFlags")?;
    let tree_signature_offset = layout.field("_HEAP_PAGE_RANGE_DESCRIPTOR", "TreeSignature")?;
    let tree_node_offset = layout.field("_HEAP_PAGE_RANGE_DESCRIPTOR", "TreeNode")?;
    let metadata_size = descriptor_count
        .checked_mul(descriptor.size as usize)
        .ok_or_else(|| SnapshotError::InvalidData {
            detail: "descriptor metadata size overflow".into(),
        })?;
    // Plain `LIST_ENTRY` links, compared with the head exactly. They used to be masked to 16
    // bytes like a tree node's, and the head is only 8-aligned — `SegContexts` at +0x140 and
    // `SegmentListHead` at +0x48 of it, on x64 26100.8972 and ARM64 26100.1 `ntdll` alike — so
    // the last segment's link back to the head never matched it, and the walk read the heap's
    // own fields as a segment header (2026-09-23).
    let mut entry = scalar(memory, list_head, 8)?;
    let mut seen = HashSet::new();
    while entry != 0 && entry != list_head && seen.len() < traversal_limit {
        check_budget(memory)?;
        if !seen.insert(entry) {
            discovery
                .diagnostics
                .push(format!("segment-list cycle at {entry:#x}"));
            break;
        }
        let segment_address = entry.saturating_sub(list_entry as u64);
        let segment_header = match guarded_read(memory, segment_address, segment.size as usize) {
            Ok(bytes) => bytes,
            Err(error) => {
                discovery.diagnostics.push(format!(
                    "cannot read segment header {segment_address:#x}: {error}"
                ));
                match scalar(memory, entry, 8) {
                    Ok(next) => entry = next,
                    Err(_) => break,
                }
                continue;
            }
        };
        let signature = read_u64(&segment_header, signature_offset)
            .or_else(|| read_u32(&segment_header, signature_offset).map(u64::from))
            .unwrap_or(0);
        if !valid_page_segment_signature(signature, segment_address, context_address, heap_key) {
            discovery.diagnostics.push(format!(
                "rejecting page segment {segment_address:#x} with invalid signature {signature:#x}"
            ));
            entry = read_u64(&segment_header, list_entry).unwrap_or(0);
            continue;
        }
        let metadata_address = segment_address + desc_array as u64;
        let metadata = match guarded_read(memory, metadata_address, metadata_size) {
            Ok(bytes) => bytes,
            Err(error) => {
                discovery.diagnostics.push(format!(
                    "cannot read descriptors at {metadata_address:#x}: {error}"
                ));
                entry = read_u64(&segment_header, list_entry).unwrap_or(0);
                continue;
            }
        };
        let mut descriptor_index = first_descriptor;
        while descriptor_index < descriptor_count {
            // Up to 4096 descriptors per segment, and an LFH one costs a subsegment header
            // plus a bitmap read apiece. Polling only in the enclosing segment loop left ~8k
            // reads between the deadline passing and anything noticing.
            check_budget(memory)?;
            let offset = descriptor_index * descriptor.size as usize;
            let Some(decoded) = decode_descriptor_at(
                &metadata,
                offset,
                descriptor.size as usize,
                unit_offset,
                flags_offset,
            ) else {
                descriptor_index += 1;
                continue;
            };
            // `TreeSignature` shares its bytes with the range's tree node, so it only means
            // anything on the descriptor that *is* the range's node — its first — and not even
            // there once the range is free: a range in the free-page tree holds its node's links
            // in those bytes. Membership of the tree is that range's evidence, and asking it for
            // the signature as well dropped every free range there is (2026-09-23).
            let descriptor_node = metadata_address + offset as u64 + tree_node_offset as u64;
            if decoded.first
                && !free_nodes.contains(&descriptor_node)
                && !read_u32(&metadata, offset + tree_signature_offset)
                    .is_some_and(valid_descriptor_tree_signature)
            {
                discovery.diagnostics.push(format!(
                    "rejecting descriptor {descriptor_index} at {:#x} with invalid tree signature",
                    metadata_address + offset as u64
                ));
                descriptor_index += decoded.unit_size.max(1) as usize;
                continue;
            }
            let unit_size = decoded.unit_size as usize;
            let Some(address) = segment_address.checked_add((descriptor_index as u64) << shift)
            else {
                break;
            };
            let size = unit_size.checked_shl(shift).unwrap_or(0);
            if size == 0 {
                descriptor_index += unit_size.max(1);
                continue;
            }
            // Getting this wrong is expensive in the quietest way available: a range parsed as
            // the wrong kind of subsegment fails to decode and is dropped *during region
            // creation*, so its allocations never reach the walk at all and nothing downstream
            // can tell they existed. Verifier special pool is the one range whose flags cannot
            // answer the question — a special-pool page range is a plain allocated range
            // (`0x03`), but it holds a pool header plus fill laid out page-granularly, so it is
            // routed by the heap it came from. `walk_region` routes the same way.
            let backend = if identity.special {
                PoolBackend::Segment
            } else {
                descriptor_backend(decoded.flags)
            };
            let mut region_address = address;
            let mut region_size = size;
            let mut bitmap = Vec::new();
            let mut block_size = size.min(u32::MAX as usize) as u32;
            if backend == PoolBackend::Lfh {
                let subsegment = layout.type_layout("_HEAP_LFH_SUBSEGMENT")?;
                let offsets = layout.field("_HEAP_LFH_SUBSEGMENT", "BlockOffsets")?
                    + layout.field("_HEAP_LFH_SUBSEGMENT_ENCODED_OFFSETS", "EncodedData")?;
                let count_offset = layout.field("_HEAP_LFH_SUBSEGMENT", "BlockCount")?;
                let bitmap_offset = layout.field("_HEAP_LFH_SUBSEGMENT", "BlockBitmap")?;
                let header = match guarded_read(memory, address, subsegment.size as usize) {
                    Ok(bytes) => bytes,
                    Err(error) => {
                        discovery
                            .diagnostics
                            .push(format!("cannot read LFH subsegment {address:#x}: {error}"));
                        descriptor_index += unit_size;
                        continue;
                    }
                };
                let encoded = read_u32(&header, offsets).unwrap_or(0);
                let blocks = read_u16(&header, count_offset)
                    .map(usize::from)
                    .unwrap_or(0);
                let lfh = match decode_lfh_subsegment(
                    encoded,
                    address,
                    lfh_key as u32,
                    blocks,
                    region_size,
                ) {
                    Ok(lfh) => lfh,
                    Err(rejection) => {
                        discovery.diagnostics.push(format!(
                            "rejecting LFH subsegment {address:#x}: {rejection}"
                        ));
                        descriptor_index += unit_size;
                        continue;
                    }
                };
                block_size = lfh.block_size;
                let raw = match guarded_read(
                    memory,
                    address + bitmap_offset as u64,
                    lfh_bitmap.bytes_for(lfh.blocks),
                ) {
                    Ok(bytes) => bytes,
                    Err(error) => {
                        discovery
                            .diagnostics
                            .push(format!("cannot read LFH bitmap at {address:#x}: {error}"));
                        descriptor_index += unit_size;
                        continue;
                    }
                };
                bitmap = match lfh_bitmap.normalise(&raw, lfh.blocks) {
                    Some(busy) => busy,
                    None => {
                        discovery.diagnostics.push(format!(
                            "LFH bitmap at {address:#x} is {} bytes, short of its {} blocks",
                            raw.len(),
                            lfh.blocks
                        ));
                        descriptor_index += unit_size;
                        continue;
                    }
                };
                region_address += lfh.first as u64;
                region_size = lfh.blocks * block_size as usize;
            } else if backend == PoolBackend::Vs {
                let vs = layout.type_layout("_HEAP_VS_SUBSEGMENT")?;
                let header = match guarded_read(memory, address, vs.size as usize) {
                    Ok(bytes) => bytes,
                    Err(error) => {
                        discovery
                            .diagnostics
                            .push(format!("cannot read VS subsegment {address:#x}: {error}"));
                        descriptor_index += unit_size;
                        continue;
                    }
                };
                let signature =
                    read_u16(&header, layout.field("_HEAP_VS_SUBSEGMENT", "Signature")?)
                        .unwrap_or(0)
                        & 0x7fff;
                let declared =
                    read_u16(&header, layout.field("_HEAP_VS_SUBSEGMENT", "Size")?).unwrap_or(0);
                if !valid_vs_signature(signature ^ declared) {
                    discovery.diagnostics.push(format!(
                        "rejecting VS subsegment {address:#x} with invalid signature"
                    ));
                    descriptor_index += unit_size;
                    continue;
                }
                let first = (vs.size as usize).next_multiple_of(16);
                if first >= region_size {
                    descriptor_index += unit_size;
                    continue;
                }
                region_address += first as u64;
                region_size -= first;
                block_size = 0;
                // **The page range is not the subsegment, and is routinely larger.** The chunk
                // area is what `nt!RtlpHpVsSubsegmentInitialize` lays out — `Size = (bytes -
                // first) >> 4` sixteen-byte units starting at `+ first` — and the descriptor
                // sizes the *range that holds it*, which on 26100 is one unit more:
                //
                // ```text
                //   053 flags=0f UnitSize=11 UnitOffset=00   <- a VS range: 17 pages
                //   054..063     UnitSize=00 UnitOffset=01..10  <- its 16 continuation units
                //   064 flags=03 UnitSize=01                 <- and the next range begins
                // ```
                //
                // while the subsegment at 0x53000 declared Size 0xffd — 0xffd0 bytes of chunks,
                // plus `first`, is 0x10000. One spare page, every time, at both 0x10000 and
                // 0x20000 subsegment sizes.
                //
                // Bounding by the range is what raised glslang/dbgscope#103: where that spare
                // page happened to be committed, the walk decoded it, and since nothing there is
                // a chunk it refused a header every sixteen bytes to the end — 0x1000/16 = 256
                // per subsegment, against 248 measured. The remaining refusals on a live walk
                // were, to a rounding error, exactly this page.
                //
                // `declared` is preferred over the descriptor rather than merely compared with
                // it, because it is the number the chunks were laid out to *and* it is checked:
                // this subsegment is only accepted at all if `Signature ^ Size == 0x2bed`, so a
                // `Size` we resolved from the wrong offset cannot reach here. Clamped anyway —
                // it may only ever shrink the region, never point the walk outside the range it
                // was given.
                let declared_size = usize::from(declared) * 16;
                if declared_size == 0 || declared_size > region_size {
                    // Not the ordinary spare page: the subsegment is claiming no chunks at all,
                    // or more than its range can hold. Its signature checked out, so this is
                    // worth a line rather than a silent clamp.
                    discovery.diagnostics.push(format!(
                        "VS subsegment {address:#x} declares {declared_size:#x} of chunks where \
                         its page range leaves room for {region_size:#x}"
                    ));
                } else {
                    region_size = declared_size;
                }
            }
            // Two independent ways to be free, and either is enough: the range sits in the
            // segment's free-page tree, or its own flags say it is not in use.
            let state = if free_nodes.contains(&descriptor_node) || !decoded.allocated() {
                PoolState::ReusableFree
            } else {
                PoolState::Allocated
            };
            // **An allocated kernel page range with no subsegment in it is a big-pool
            // allocation, and big-pool allocations carry no `_POOL_HEADER` at all.**
            // `ExAllocatePoolWithTag` sends anything that will not fit inside a page to
            // `ExpAllocateBigPool`, which takes whole pages from this same segment allocator
            // and records the caller's tag and length in `nt!PoolBigPageTable` instead of in
            // a header — which is why `!pool` calls these "large page allocation" and reads
            // their tag out of that table rather than out of the page.
            //
            // Decoding one as though it began with a header reads the caller's *own first
            // sixteen bytes* as `PreviousSize`/`BlockSize`/`PoolType`/`PoolTag`, and then
            // reports the block as starting 0x10 in and 0x10 short. Measured on a live
            // 26100 kernel (`ctf-vm`, 2026-09-24): `ffffac09dd0f5000` is a 0x1000-byte
            // `CM25` registry hive bin, and the walk called it a 4080-byte block at
            // `+0x10` tagged `..N.` — from `hbin`'s own header bytes. Where those bytes
            // happened to be zero the tag came out `0x00000000`, which is how this
            // presented: as tens of thousands of untagged allocations (`FOLLOWUPS` item 99).
            //
            // Free ranges are deliberately not looked up: `ExpRemoveTagForBigPages` takes the
            // entry out as the allocation goes away, so a miss there would be the expected
            // answer rather than a finding.
            // Every big-pool allocation inside a VS subsegment, resolved here because the table
            // is here and a chunk walk has no reads to spare. `_POOL_HEADER.BlockSize` cannot
            // describe a chunk of `MAX_POOL_HEADER_CHUNK` or more, so `nt` gives those no header
            // at all and records them in `nt!PoolBigPageTable` — **wherever they came from**.
            // They are not only the plain page ranges below: a VS subsegment holds them too, and
            // every entry's `Va` is page-aligned, so one probe per page of the chunk area finds
            // each one. Measured on a live 26100 kernel, where `0xffffac09da29f000` is an
            // `MiRr` allocation of `0xe1c0` bytes inside the 17-page VS range at page 0x53 of
            // its segment (`windbg-mcp` FOLLOWUPS item 99).
            let mut big_pool = HashMap::new();
            if backend == PoolBackend::Vs && !identity.special {
                let mut page = region_address.next_multiple_of(PAGE_SIZE);
                let end = region_address.saturating_add(region_size as u64);
                while page < end {
                    if let Some(entry) = discovery.big_page(memory, page)? {
                        big_pool.insert(page, entry);
                    }
                    page = page.saturating_add(PAGE_SIZE);
                }
            }
            let big_pool = Arc::new(big_pool);
            let mut pool_header = layout.pool_header_layout()?;
            let mut known_tag = None;
            if backend == PoolBackend::Segment
                && !identity.special
                && state == PoolState::Allocated
                && let Some(big_page) = discovery.big_page(memory, region_address)?
            {
                known_tag = Some(big_page.tag);
                pool_header.size = 0;
                // The table holds the length the caller asked for; the range is that rounded
                // up to whole pages. Report the request, as `!pool` does, and leave the
                // rounding out of the walk rather than inflating the allocation by it.
                if big_page.size != 0 && big_page.size < region_size as u64 {
                    region_size = big_page.size as usize;
                    block_size = region_size.min(u32::MAX as usize) as u32;
                }
            }
            discovery.regions.push(PoolRegion {
                address: region_address,
                size: region_size,
                requested_size: None,
                pool_kind,
                numa_node,
                heap: identity,
                subsegment: Some(address),
                backend,
                unit_size: block_size,
                bitmap,
                heap_key,
                pool_header,
                vs_header_size: layout.type_layout("_HEAP_VS_CHUNK_HEADER")?.size as usize,
                vs_sizes_offset: layout.field("_HEAP_VS_CHUNK_HEADER", "Sizes")?,
                vs_skips_page_end,
                known_tag,
                states: vec![state],
                reusable_chunks: Arc::clone(reusable_chunks),
                cached_chunks: Arc::clone(cached_chunks),
                big_pool: Arc::clone(&big_pool),
            });
            descriptor_index += unit_size;
        }
        entry = read_u64(&segment_header, list_entry).unwrap_or(0);
    }
    Ok(())
}

#[allow(clippy::too_many_arguments)]
fn discover_large_allocations(
    memory: &impl PoolMemory,
    layout: &PoolLayout,
    heap_address: u64,
    numa_node: u16,
    pool_kind: PoolKind,
    identity: HeapIdentity,
    heap_key: u64,
    traversal_limit: usize,
    discovery: &mut Discovery,
) -> Result<(), SnapshotError> {
    let Ok(tree_offset) = layout.field("_SEGMENT_HEAP", "LargeAllocMetadata") else {
        return Ok(());
    };
    let Ok(large) = layout.type_layout("_HEAP_LARGE_ALLOC_DATA") else {
        return Ok(());
    };
    let Ok(tree_node) = layout.field("_HEAP_LARGE_ALLOC_DATA", "TreeNode") else {
        return Ok(());
    };
    let Ok(virtual_offset) = layout.field("_HEAP_LARGE_ALLOC_DATA", "VirtualAddress") else {
        return Ok(());
    };
    let Ok(pages_offset) = layout.field("_HEAP_LARGE_ALLOC_DATA", "AllocatedPages") else {
        return Ok(());
    };
    let tree_address = heap_address + tree_offset as u64;
    let nodes = tree_nodes(
        memory,
        layout,
        tree_address,
        traversal_limit,
        "large-allocation tree",
        &mut discovery.diagnostics,
    )?;
    for node in nodes {
        // One metadata read per node, and `nodes` is bounded only by `traversal_limit`. The
        // nested poll in `lookup_big_page_target` covers only nodes that get that far; one
        // whose metadata will not read continues from here, reading and never asking.
        check_budget(memory)?;
        let allocation_address = node.saturating_sub(tree_node as u64);
        let allocation = match guarded_read(memory, allocation_address, large.size as usize) {
            Ok(bytes) => bytes,
            Err(error) => {
                discovery.diagnostics.push(format!(
                    "cannot read large-allocation metadata {allocation_address:#x}: {error}"
                ));
                continue;
            }
        };
        let Some((virtual_address, pages)) = read_u64(&allocation, virtual_offset)
            .zip(read_u64(&allocation, pages_offset))
            .and_then(|(virtual_address, pages)| {
                if layout.is_user() {
                    super::decode::decode_large_allocation_for(virtual_address, pages, true)
                } else {
                    super::decode::decode_large_allocation(virtual_address, pages)
                }
            })
        else {
            continue;
        };
        if pages > 0x10_0000 {
            discovery.diagnostics.push(format!(
                "rejecting implausible large allocation at {allocation_address:#x}"
            ));
            continue;
        }
        let bytes = pages.saturating_mul(PAGE_SIZE);
        // In the validated x64 layout `UnusedBytes` aliases the low 16 bits of the
        // `VirtualAddress` word. Only recover an exact request when the PDB confirms that
        // alias and the encoded value fits inside the allocation.
        let validated_unused_bytes = layout
            .field("_HEAP_LARGE_ALLOC_DATA", "UnusedBytes")
            .is_ok_and(|offset| offset == virtual_offset);
        let requested_size = read_u64(&allocation, virtual_offset)
            .and_then(|word| decode_large_requested_size(word, bytes, validated_unused_bytes));
        let (tag, tracked_size) = if layout.is_user() {
            (0, bytes)
        } else {
            match discovery.big_page(memory, virtual_address)? {
                Some(big_page) => (big_page.tag, big_page.size),
                None => {
                    discovery.diagnostics.push(format!(
                        "no validated big-page entry for large allocation {virtual_address:#x}"
                    ));
                    (0, bytes)
                }
            }
        };
        let size = tracked_size.min(bytes).min(usize::MAX as u64) as usize;
        let Ok(pool_header) = layout.pool_header_layout() else {
            continue;
        };
        discovery.regions.push(PoolRegion {
            address: virtual_address,
            size,
            requested_size: layout.is_user().then_some(requested_size).flatten(),
            pool_kind,
            numa_node,
            heap: identity,
            subsegment: None,
            backend: PoolBackend::Large,
            unit_size: size.min(u32::MAX as usize) as u32,
            bitmap: Vec::new(),
            heap_key,
            pool_header,
            vs_header_size: 0,
            vs_sizes_offset: 0,
            vs_skips_page_end: false,
            known_tag: Some(tag),
            states: vec![PoolState::Allocated],
            reusable_chunks: Arc::default(),
            cached_chunks: Arc::default(),
            big_pool: Arc::default(),
        });
    }
    Ok(())
}

/// How many entries one read of the big-page table covers.
///
/// Batches are cut on multiples of this rather than on wherever a probe started, so that the
/// second lookup landing anywhere inside a batch the first one read costs no read at all —
/// which is what makes asking this question of *every* allocated page range affordable. At
/// 32 bytes an entry a batch is 8 KiB, and a whole 26100 table — 0x8000 entries on a 4 GiB
/// guest — is 128 of them.
const BIG_PAGE_BATCH: usize = 256;

/// What `nt!PoolBigPageTable` records about one allocation.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct BigPageEntry {
    pub tag: u32,
    /// The length the caller asked for, which is *not* the length of the pages it was given:
    /// a 0xa270-byte request occupies 0xb000 of pool and this field says 0xa270.
    pub size: u64,
}

/// `nt!PoolBigPageTable`: where the kernel keeps the tag of every allocation too large to
/// carry a `_POOL_HEADER`, read as `nt` reads it and cached a batch at a time.
///
/// **Placement.** `nt!ExpAddTagForBigPages` hashes the address and linear-probes forward from
/// there, wrapping at the end of the table, claiming the first entry whose `Va` has bit 0
/// set; `nt!ExpRemoveTagForBigPages` repeats the same walk and compares `Va` against the
/// address **exactly**. Both were read on x64 26100.33438 (2026-09-24), and the hash is the
/// one [`big_page_hash`] mirrors.
///
/// **Bit 0 of `Va` is the free flag, and the address stays behind it.** A free is a
/// `lock inc` of the whole `Va` word, so a slot that held `0xffffac09e50f5000` reads
/// `0xffffac09e50f5001` afterwards and keeps its old tag and size until something else claims
/// it. Matching on `Va & !1` therefore answers for a *freed* allocation with the tag it used
/// to have — measured on the same kernel, where `ffffac09e50f5001` sat one slot from a live
/// entry and its page would not even read. The comparison here is `nt`'s: exact.
///
/// **A never-used slot is `1`, not `0`**, which is why the probe below stops at `Va <= 1`
/// rather than at zero, and why the zero test it used to have never once fired: a miss
/// scanned all 32,768 entries. Stopping at a never-used slot is sound because the allocator
/// would have claimed it — it is bit-0-set, and the probe that placed the entry being looked
/// for passed over this slot to reach wherever it went. Stopping at a *freed* slot would not
/// be sound, since it may well have been occupied when that entry was placed.
struct BigPageTable {
    base: u64,
    count: usize,
    entry_size: usize,
    va_offset: usize,
    tag_offset: usize,
    size_offset: usize,
    /// One slot per batch, filled on first touch. A walk that asks about nothing reads
    /// nothing; one that asks about thousands of ranges converges on the whole table.
    batches: Vec<Option<Vec<u8>>>,
}

impl BigPageTable {
    /// Resolve the table, or `None` when this target has no such thing to read — a user-mode
    /// Segment Heap, or a kernel whose symbols do not carry the tracker type or the two
    /// globals. `None` is not an error: the caller falls back to whatever the page itself
    /// says, which is what a build without the table leaves it with.
    fn open(
        memory: &impl PoolMemory,
        layout: &PoolLayout,
        diagnostics: &mut Vec<String>,
    ) -> Result<Option<Self>, SnapshotError> {
        let (Ok(entry), Ok(va_offset), Ok(tag_offset), Ok(size_offset)) = (
            layout.type_layout("_POOL_TRACKER_BIG_PAGES"),
            layout.field("_POOL_TRACKER_BIG_PAGES", "Va"),
            layout.field("_POOL_TRACKER_BIG_PAGES", "Key"),
            layout.field("_POOL_TRACKER_BIG_PAGES", "NumberOfBytes"),
        ) else {
            return Ok(None);
        };
        let (Some(&table_pointer_address), Some(&size_address)) = (
            layout.globals.get("PoolBigPageTable"),
            layout.globals.get("PoolBigPageTableSize"),
        ) else {
            return Ok(None);
        };
        // Two reads of its own, so they are polled for like any others: this runs at the head
        // of discovery, where a budget already spent on resolving the heaps would otherwise
        // buy two more reads before anything asked.
        check_budget(memory)?;
        let base = match scalar(memory, table_pointer_address, 8) {
            Ok(value) => value,
            Err(error) => {
                diagnostics.push(format!("cannot read big-page table pointer: {error}"));
                return Ok(None);
            }
        };
        let count =
            match scalar(memory, size_address, 4).or_else(|_| scalar(memory, size_address, 8)) {
                Ok(value) => value as usize,
                Err(error) => {
                    diagnostics.push(format!("cannot read big-page table size: {error}"));
                    return Ok(None);
                }
            };
        if base == 0 || count == 0 || count > 0x10_0000 || !count.is_power_of_two() {
            diagnostics.push(format!("rejecting implausible big-page table size {count}"));
            return Ok(None);
        }
        Ok(Some(Self {
            base,
            count,
            entry_size: entry.size as usize,
            va_offset,
            tag_offset,
            size_offset,
            batches: vec![None; count.div_ceil(BIG_PAGE_BATCH)],
        }))
    }

    /// The entry for `address`, or `None` when the table does not name it.
    fn lookup(
        &mut self,
        memory: &impl PoolMemory,
        address: u64,
        diagnostics: &mut Vec<String>,
    ) -> Result<Option<BigPageEntry>, SnapshotError> {
        let mut index =
            big_page_hash(address, self.count).ok_or_else(|| SnapshotError::InvalidData {
                detail: format!("invalid big-page table size {}", self.count),
            })?;
        for _ in 0..self.count {
            let entry_size = self.entry_size;
            let offset = (index % BIG_PAGE_BATCH) * entry_size;
            let (va_offset, tag_offset, size_offset) =
                (self.va_offset, self.tag_offset, self.size_offset);
            let Some(bytes) = self.batch(memory, index / BIG_PAGE_BATCH, diagnostics)? else {
                return Ok(None);
            };
            let Some(candidate) = read_u64(bytes, offset + va_offset) else {
                // The batch is short of this entry: either the read failed, or the table
                // runs past what the target will give us. Either way the chain ends here.
                return Ok(None);
            };
            // Never used, so the allocator would have put our entry here rather than past it.
            if candidate <= 1 {
                return Ok(None);
            }
            if candidate == address {
                let (Some(tag), Some(size)) = (
                    read_u32(bytes, offset + tag_offset),
                    read_u64(bytes, offset + size_offset),
                ) else {
                    diagnostics.push(format!("truncated big-page entry {index}"));
                    return Ok(None);
                };
                return Ok(Some(BigPageEntry { tag, size }));
            }
            index = (index + 1) % self.count;
        }
        Ok(None)
    }

    /// One batch of entries, read on first touch and kept.
    fn batch(
        &mut self,
        memory: &impl PoolMemory,
        batch: usize,
        diagnostics: &mut Vec<String>,
    ) -> Result<Option<&[u8]>, SnapshotError> {
        if self.batches.get(batch).is_none_or(Option::is_some) {
            return Ok(self.batches.get(batch).and_then(Option::as_deref));
        }
        check_budget(memory)?;
        let first = batch * BIG_PAGE_BATCH;
        let len = BIG_PAGE_BATCH.min(self.count - first);
        let (Some(byte_len), Some(entry_address)) = (
            self.entry_size.checked_mul(len),
            self.base.checked_add(first as u64 * self.entry_size as u64),
        ) else {
            return Err(SnapshotError::InvalidData {
                detail: "big-page batch address overflow".into(),
            });
        };
        match guarded_read(memory, entry_address, byte_len) {
            Ok(bytes) => self.batches[batch] = Some(bytes),
            Err(error) => {
                diagnostics.push(format!(
                    "cannot read big-page entries {first}..{} at {entry_address:#x}: {error}",
                    first + len
                ));
                // An empty batch is still a batch: it is cached so that the next thousand
                // allocations hashing into it do not each pay for the same failed read.
                self.batches[batch] = Some(Vec::new());
            }
        }
        Ok(self.batches[batch].as_deref())
    }
}

/// How many verbatim examples of one kind of diagnostic to keep.
pub const DIAGNOSTIC_EXAMPLES: usize = 8;

/// The shape of a diagnostic: the message with every number standing in for itself.
///
/// "unreadable VS free tree node 0xffffc00f6ec02f90" and the same complaint about a
/// different node share a shape; a genuinely different complaint does not.
fn diagnostic_shape(message: &str) -> String {
    message
        .split_whitespace()
        .map(|token| {
            if token.contains(|character: char| character.is_ascii_digit()) {
                "#"
            } else {
                token
            }
        })
        .collect::<Vec<_>>()
        .join(" ")
}

/// One kind of complaint, and how many times the walk made it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DiagnosticShape {
    /// The message with every number standing in for itself, so that the same complaint
    /// about different addresses is one shape and a different complaint is not.
    pub shape: String,
    /// How many messages of this shape the walk emitted, including any kept verbatim in
    /// [`PoolDiagnostics::examples`].
    pub total: usize,
}

/// What a walk complained about: a bounded verbatim sample, and the per-shape totals that
/// say what the sample left out.
///
/// A live target keeps allocating between the reads that make up one walk, so a single
/// stale list pointer yields one "unreadable ... node" line per node — 14k of them measured
/// on a busy kernel. Keeping them all buries the handful of *distinct* problems a reader
/// needs, and every consumer truncates the list anyway, so which ones survive gets decided
/// by position rather than by significance. Hence the cap of [`DIAGNOSTIC_EXAMPLES`] kept
/// messages per shape.
///
/// The totals are why this is a type rather than a `Vec<String>`. Flattened to text they
/// survive only as prose inside a summary line, and a consumer that counts the lines it
/// received is measuring this cap — then printing the answer as a property of the target.
/// A walk that emitted 7,700 complaints reporting "71 diagnostics" is the specific failure
/// the module exists to avoid, so the numbers travel as numbers.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct PoolDiagnostics {
    examples: Vec<String>,
    shapes: Vec<DiagnosticShape>,
    /// Where each seen shape sits in `shapes`. Derived from it, not state of its own.
    positions: HashMap<String, usize>,
}

impl PoolDiagnostics {
    /// Records one complaint, keeping it verbatim only while its shape has room.
    pub(crate) fn push(&mut self, message: String) {
        let shape = diagnostic_shape(&message);
        let position = match self.positions.get(&shape) {
            Some(&position) => position,
            None => {
                self.shapes.push(DiagnosticShape {
                    shape: shape.clone(),
                    total: 0,
                });
                let position = self.shapes.len() - 1;
                self.positions.insert(shape, position);
                position
            }
        };
        let seen = &mut self.shapes[position];
        seen.total += 1;
        if seen.total <= DIAGNOSTIC_EXAMPLES {
            self.examples.push(message);
        }
    }

    /// Whether the walk complained at all.
    pub fn is_empty(&self) -> bool {
        self.shapes.is_empty()
    }

    /// How many messages the walk emitted, the ones collapsed away included.
    ///
    /// This is the number that describes the *walk*; `examples().len()` and `lines().len()`
    /// describe this struct, and on a busy target the two differ by two orders of magnitude.
    pub fn emitted(&self) -> usize {
        self.shapes.iter().map(|seen| seen.total).sum()
    }

    /// The messages kept verbatim, in the order the walk emitted them.
    ///
    /// Emission order rather than by shape because heaps are walked in order, so the tail is
    /// the most recently discovered ones — which is what a caller printing "the last few" is
    /// reaching for.
    pub fn examples(&self) -> &[String] {
        &self.examples
    }

    /// The shapes, in the order they were first seen.
    pub fn shapes(&self) -> &[DiagnosticShape] {
        &self.shapes
    }

    /// The whole thing as text: every kept example, then one summary line per shape that had
    /// more than the cap.
    ///
    /// Summaries go at the end, in the order the shapes first appeared, rather than
    /// interleaved where a shape's last kept member happened to land. For display only —
    /// [`Self::emitted`] is the count, and the length of this is not.
    pub fn lines(&self) -> Vec<String> {
        let mut lines = self.examples.clone();
        for seen in &self.shapes {
            if let Some(collapsed) = seen
                .total
                .checked_sub(DIAGNOSTIC_EXAMPLES)
                .filter(|more| *more > 0)
            {
                lines.push(format!("... and {collapsed} more like `{}`", seen.shape));
            }
        }
        lines
    }
}

impl Extend<String> for PoolDiagnostics {
    fn extend<T: IntoIterator<Item = String>>(&mut self, messages: T) {
        for message in messages {
            self.push(message);
        }
    }
}

impl FromIterator<String> for PoolDiagnostics {
    fn from_iter<T: IntoIterator<Item = String>>(messages: T) -> Self {
        let mut diagnostics = Self::default();
        diagnostics.extend(messages);
        diagnostics
    }
}

#[derive(Debug, Clone, Default)]
pub(crate) struct PoolSnapshot {
    pub layout: crate::allocator::LayoutProvenance,
    pub spans: Vec<PoolSpan>,
    pub diagnostics: PoolDiagnostics,
    pub complete: bool,
    /// Whether what stopped this walk short was its **deadline** rather than anything it found.
    ///
    /// `complete` says a walk did not cover the pool; this says why, and the two answers need
    /// opposite responses. A walk that ran out of time reaches more of the pool if it is given
    /// more of it; one that hit unreadable regions or a traversal cap will report the same gaps
    /// however long it runs. Only the walk itself can tell them apart — by the time a caller has
    /// the snapshot, both look like a short list — so the reason travels as a field rather than
    /// being inferred downstream from the diagnostic it also writes.
    pub budget_expired: bool,
    /// The match threshold that intentionally ended this walk, when one fired.
    ///
    /// This is separate from `budget_expired` and from diagnostics because all three make the
    /// snapshot partial for different reasons. A caller that asked only whether a tag exists has
    /// its answer when this is present; a caller seeing either of the others does not.
    pub stopped_after_matches: Option<usize>,
    /// Query-specific stop policy for a snapshot currently being built. It never survives into
    /// the cache: a walk that fires it clears `complete`, and the cache stores complete snapshots
    /// only.
    pub match_limit: Option<PoolMatchLimit>,
    pub matched_allocations: usize,
    /// What the walk stepped over rather than gave up on; see [`WalkStalls`].
    pub stalls: WalkStalls,
    /// Chunk headers a backend decoder refused and resynchronised past, across the whole walk.
    ///
    /// A count, because the diagnostic that names them cannot be one. `PoolDiagnostics`
    /// collapses messages by shape, and a refusal is reported once per extent — so the figure
    /// beside that line counts *extents that contained a refusal*, which reads like a chunk
    /// count and is not one. On the walk that raised glslang/dbgscope#93 the difference was
    /// between "884" and a number nothing recorded.
    ///
    /// Only `walk_vs` feeds it today; LFH subsegments are refused during discovery, one per
    /// subsegment, where the count and the message already agree.
    pub refused_chunks: u64,
    /// Committed bytes of a VS subsegment the walk declined to decode, because it could not say
    /// where a chunk began in them.
    ///
    /// The cost of [`SnapshotWalker::walk_vs`] refusing to guess, and the number that keeps that
    /// refusal honest: what it buys is that nothing fabricated enters `spans`, and what it costs
    /// is coverage, which is invisible unless it is sized. A walk that reports no refusals *and*
    /// no unplaced bytes decoded every committed byte of every subsegment it reached; one that
    /// reports a large figure here has lost the chunk chain somewhere, and the chain is the only
    /// thing that can find a variable-size header.
    pub unplaced_bytes: u64,
}

/// An explicitly partial tag query: stop after this many matching allocated chunks.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct PoolMatchLimit {
    raw_tag: u32,
    /// `Some(true)` means paged only, `Some(false)` nonpaged only, and `None` means both.
    paged: Option<bool>,
    matches: NonZeroUsize,
}

impl PoolMatchLimit {
    pub(crate) fn new(raw_tag: u32, paged: Option<bool>, matches: NonZeroUsize) -> Self {
        Self {
            raw_tag,
            paged,
            matches,
        }
    }

    fn accepts(self, span: &PoolSpan) -> bool {
        span.state == PoolState::Allocated
            && span.raw_tag == self.raw_tag
            && self
                .paged
                .is_none_or(|paged| span.pool_kind.is_paged() == paged)
    }
}

impl PoolSnapshot {
    fn record_span(&mut self, span: PoolSpan) {
        let matched = self.match_limit.is_some_and(|limit| limit.accepts(&span));
        self.spans.push(span);
        if !matched {
            return;
        }
        self.matched_allocations = self.matched_allocations.saturating_add(1);
        let Some(limit) = self.match_limit else {
            return;
        };
        if self.matched_allocations >= limit.matches.get() {
            self.complete = false;
            self.stopped_after_matches = Some(limit.matches.get());
        }
    }

    fn match_limit_reached(&self) -> bool {
        self.stopped_after_matches.is_some()
    }
}

/// What valid-region queries that could not advance cost the walk, and what stepping over them
/// bought.
///
/// Numbers rather than another diagnostic, because a diagnostic can only say how *often* the
/// walk stalled and that was never the question. A walk that steps over one page and a walk
/// that abandons everything behind it emit the same line the same number of times; what
/// separates them is how much memory the decision covered — which the walk knows at the moment
/// it decides, and used to throw away.
///
/// `recovered_bytes` is what the change is judged by: committed memory read *after* a stall in
/// the same region, which is precisely the coverage a walk that gave up at the first stall
/// reported as nothing at all.
///
/// **On live 26100 it has measured zero** — 1,619 stalls, 6,627,520 bytes stepped over, nothing
/// read behind any of them (glslang/dbgscope#104). That is the number saying what it says, not a
/// counter that was never wired up: `stalled_here` latches for the rest of the region and any
/// later extent adds to `recovered_bytes`, which
/// [`SnapshotWalker::walk_region`]'s own test pins at two pages. So on that target every stall
/// sits at the end of its region's readable content, and the page-stepping buys nothing there.
/// It is kept because it is bounded — at most [`MAX_CONSECUTIVE_STALLS`] queries per dead region
/// — and because the failure it replaced was losing every committed page behind one bad one. The
/// diagnostic now carries the engine's own answer at each stall, which is what a later run needs
/// to decide whether stepping can be replaced by stopping.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct WalkStalls {
    /// How many times a query could not advance and the walk stepped over a page.
    pub pages: u64,
    /// Bytes filed as unreadable by those steps.
    pub skipped_bytes: u64,
    /// Bytes of committed memory read after a stall, in the regions that stalled.
    pub recovered_bytes: u64,
}

/// What one committed extent of a VS subsegment hands the next.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct VsResume {
    /// Where the chain names the next header — past this extent when the last chunk ran
    /// over its end, which is how the chain crosses a hole at all.
    next: u64,
    /// How far the spans filed in this extent reach, when the last of them runs past its
    /// end: a big-pool chunk the table named, emitted whole although its tail is in the hole
    /// ahead. Bytes below this were accounted for, and the extent loop must not file them as
    /// a gap on its way to the next readable extent — a gap span over a chunk the index
    /// already holds would answer for its addresses in the gap's place. At most `next`.
    covered_through: u64,
}

pub(crate) struct SnapshotWalker<'a, M> {
    pub memory: &'a M,
    pub layout: &'a PoolLayout,
    pub traversal_limit: usize,
}

/// Where a special-pool block sits, and whether that could be corroborated.
struct SpecialPlacement {
    usable: u64,
    size: u64,
    /// The size could not be confirmed, so `size` is a conservative upper bound covering
    /// the rest of the page rather than the allocation's real length. Callers must not
    /// treat it as a measurement — it inflates byte totals if they do.
    approximate: bool,
}

/// Where a special-pool block sits inside its page, and whether the page corroborates it.
///
/// The block is butted against the page end at `page + PAGE_SIZE - align_up(size, 16)`; the
/// 16-byte rounding matters, since a plain `PAGE_SIZE - size` lands 8 bytes past the real
/// block for a 0x68 allocation.
///
/// The size arrives already decoded, from the thirteen bits the kernel keeps it in — see
/// [`decode_special_pool_header`]. That is what makes this a corroboration rather than the
/// load-bearing check it used to be: read as `_POOL_HEADER.PreviousSize`'s eight bits, a 0x140
/// allocation reported 0x40 and nothing but the fill could tell that from a genuine 0x40. A
/// value that cannot wrap has no such twin, so a page that fails the check below is a page we
/// misread or one being written as we read it, not an ambiguity to resolve.
///
/// Both runs of fill the kernel checks on free are checked here, against the fill byte the page
/// itself records: from the header's end up to the block, and from the block's end to the page
/// end. The second is the alignment padding — up to 15 bytes, and evidence the old check left
/// on the floor.
///
/// It declines rather than lies in the two cases that break the placement rule: a page read
/// only in part (nothing corroborates the tail), and a start-aligned page
/// (`nt!MmSpecialPoolCatchOverruns` clear), where the block sits against the *preceding* guard
/// page instead.
///
/// The fallback reports the whole page after the header. An over-broad chunk still answers
/// "was this freed, and which tag owned it"; a confidently wrong offset does not.
fn special_pool_placement(
    page: u64,
    header: SpecialPoolHeader,
    page_bytes: &[u8],
) -> SpecialPlacement {
    let available = page_bytes.len() as u64;
    let header_size = header.header_size as u64;
    // Both hold by construction — `decode_special_pool_header` returns `None` otherwise — so
    // the slicing below cannot be out of range once the page is known to be whole.
    let aligned = u64::from(header.requested).next_multiple_of(16);
    let start = (PAGE_SIZE - aligned) as usize;
    if available == PAGE_SIZE {
        let leading = &page_bytes[header.header_size..start];
        let padding = &page_bytes[start + header.requested as usize..];
        if leading
            .iter()
            .chain(padding)
            .all(|byte| *byte == header.fill)
        {
            return SpecialPlacement {
                usable: page + PAGE_SIZE - aligned,
                size: u64::from(header.requested),
                approximate: false,
            };
        }
    }
    SpecialPlacement {
        usable: page + header_size,
        size: available.saturating_sub(header_size),
        approximate: true,
    }
}

/// The share of the budget discovery may spend before the region walk claims the rest.
///
/// The two halves are useless apart: discovery finds the regions, the walk reads what is in
/// them, and a snapshot with neither half finished is an empty pool. Discovery is also the
/// pointer-chasing half — one read per free-tree node, per list entry, per subsegment — so
/// on a live link it is the half that runs long, and left unchecked it would spend the whole
/// budget and leave nothing to walk. Holding it to two thirds means a walk that runs out of
/// time still reports the chunks in the regions it did reach, which is the answer a caller
/// asked a pool query for.
const DISCOVERY_BUDGET_SHARE: (u32, u32) = (2, 3);

/// How much of one committed extent to ask for per debugger read.
///
/// Sized so a chunk is a fraction of a second even on a slow KD link — small enough that the
/// deadline is observed at a useful granularity, large enough that a megabyte-scale extent is
/// still a handful of transfers rather than hundreds. Extents at or below this are read whole,
/// so the ordinary small region pays nothing for this.
///
/// **This sets the residual overshoot, and that is the guarantee.** Checks sit *before* reads,
/// never after, so a walk can finish one transfer — at most this many bytes — past its
/// ceiling. Checking after a successful read would not improve on that: mid-walk the next
/// iteration's check fires anyway, so the only case it would change is the one where the walk
/// has just finished everything, and there it would clear `complete` and announce that "what
/// is missing is unknown, not absent" about a walk with nothing missing. Manufacturing a false
/// incompleteness is the same class of lie as reporting an unread region as empty, reached
/// from the other side. The budget is a ceiling on effort, not a promise about return latency:
/// it exists so the engine cannot run for *minutes* after its caller has given up.
const EXTENT_READ_CHUNK: usize = 256 * 1024;

/// How many pages in a row a region may refuse to advance over before the walk gives up on
/// the rest of it.
///
/// Every one of these is a debugger round trip, and on a live KD link a round trip is the
/// expensive unit — so this is not sized for how large a hole might be, but for how much
/// probing a *dead* region is worth before its remainder is written off. Eight covers the
/// shape the live samples actually show, a single unreadable page part-way through a
/// committed region, with room to spare; a region that is dead all the way down costs eight
/// queries rather than one per page of its length.
const MAX_CONSECUTIVE_STALLS: u32 = 8;

/// Splits a walk budget into the deadline discovery works to and the one the whole walk
/// works to. `None` in gives `None` out for both: no budget, no deadlines.
///
/// A budget too large for `Instant` to represent — `PoolWalk::within` takes any `Duration`,
/// so a caller can hand over `Duration::MAX` — also yields `None` for both. Adding it would
/// panic, and "longer than this machine can measure" is a request to run unbounded, so
/// answering it that way is the caller's own meaning rather than a substitution for it.
///
/// Both halves stand or fall together, keyed on the whole-walk deadline. Deciding them
/// independently would pass `Duration::MAX` as a *bounded* discovery share (two thirds of it
/// still fits) against an unbounded walk — a state with no meaning behind it.
fn budget_deadlines(
    start: Instant,
    budget: Option<Duration>,
) -> (Option<Instant>, Option<Instant>) {
    let (numerator, denominator) = DISCOVERY_BUDGET_SHARE;
    let Some(whole) = budget.and_then(|full| start.checked_add(full)) else {
        return (None, None);
    };
    (
        budget.and_then(|full| start.checked_add(full / denominator * numerator)),
        Some(whole),
    )
}

impl<'a, M: PoolMemory> SnapshotWalker<'a, M> {
    /// Walks the pool, giving up after `budget` rather than running as long as it takes.
    ///
    /// `None` runs to completion. That is right only where something *else* can stop the
    /// walk — an operator's Ctrl+C at a WinDbg prompt — and wrong for a programmatic caller,
    /// where nothing sets that flag; see [`Budgeted`].
    ///
    /// Running out of time is **not** an error. It comes back as a snapshot that says what
    /// it reached, with `complete` cleared and a diagnostic naming the budget, because a
    /// partial walk still answers "here is a chunk carrying that tag" — and the alternative,
    /// an error, discards minutes of work to say nothing at all. A Ctrl+C is the opposite
    /// and still errors: an operator who interrupts is asking for the walk to stop, not for
    /// whatever it happened to have.
    pub(crate) fn walk(&self, budget: Option<Duration>) -> Result<PoolSnapshot, SnapshotError> {
        self.walk_until(budget, None)
    }

    /// Walk with an optional query-specific match threshold.
    pub(crate) fn walk_until(
        &self,
        budget: Option<Duration>,
        match_limit: Option<PoolMatchLimit>,
    ) -> Result<PoolSnapshot, SnapshotError> {
        let (discovery_deadline, walk_deadline) = budget_deadlines(Instant::now(), budget);
        let mut snapshot = PoolSnapshot {
            diagnostics: PoolDiagnostics::from_iter([
                "per-session paged heaps are not included".to_string()
            ]),
            complete: true,
            match_limit,
            ..PoolSnapshot::default()
        };

        let discovery_clock = Budgeted::new(self.memory, discovery_deadline);
        let mut discovery = Discovery::default();
        let expired = match discover_pool_regions(
            &discovery_clock,
            self.layout,
            self.traversal_limit,
            &mut discovery,
        ) {
            Ok(()) => false,
            Err(SnapshotError::BudgetExpired) => true,
            Err(error) => return Err(error),
        };
        if !discovery.diagnostics.is_empty() {
            snapshot.complete = false;
        }
        snapshot
            .diagnostics
            .extend(std::mem::take(&mut discovery.diagnostics));

        // The full deadline, so regions found before discovery ran out of its share still get
        // walked with the time discovery did not use.
        let walk_clock = Budgeted::new(self.memory, walk_deadline);
        let walker = SnapshotWalker {
            memory: &walk_clock,
            layout: self.layout,
            traversal_limit: self.traversal_limit,
        };
        walker.walk_discovered_regions(discovery.regions, budget, expired, &mut snapshot)?;
        Ok(snapshot)
    }

    /// Walk every discovered region and record how much of the discovery was covered.
    fn walk_discovered_regions(
        &self,
        regions: Vec<PoolRegion>,
        budget: Option<Duration>,
        mut expired: bool,
        snapshot: &mut PoolSnapshot,
    ) -> Result<(), SnapshotError> {
        let discovered = regions.len();
        let mut walked = 0usize;
        for region in regions {
            let outcome = if region.backend == PoolBackend::Large {
                // No reads of its own: the span comes from metadata discovery already did.
                self.walk_large(&region, snapshot);
                Ok(())
            } else {
                self.walk_region(&region, snapshot)
            };
            match outcome {
                // Counted only when the region was walked *through*. A region abandoned
                // part-way is exactly what the caveat below is for.
                Ok(()) if snapshot.match_limit_reached() => break,
                Ok(()) => walked += 1,
                Err(SnapshotError::BudgetExpired) => {
                    expired = true;
                    break;
                }
                Err(error) => return Err(error),
            }
        }

        if expired {
            snapshot.complete = false;
            snapshot.budget_expired = true;
            let allowed = match budget {
                Some(budget) => format!("{budget:?} budget"),
                None => "walk budget".to_string(),
            };
            snapshot.diagnostics.push(format!(
                "the walk ran out of its {allowed}: {walked} of {discovered} discovered regions \
                 were walked, and region discovery itself may not have finished. What is \
                 reported was really there; what is missing is unknown, not absent. Allow a \
                 longer budget for full coverage."
            ));
        }
        snapshot
            .spans
            .sort_by_key(|span| (span.heap, span.usable_address));
        Ok(())
    }

    /// Reads one region and decodes its chunks, stopping if the walk's time runs out.
    ///
    /// Checked per *extent*, not just per region: a region the allocator left with holes
    /// costs one `valid_region` and one `read_exact` per committed run, so a fragmented
    /// region is many round trips and checking only on the way in would let one region
    /// overrun the deadline by an unbounded amount. The backend decoders below need no such
    /// check — they run over bytes already in hand, off the wire entirely.
    fn walk_region(
        &self,
        region: &PoolRegion,
        snapshot: &mut PoolSnapshot,
    ) -> Result<(), SnapshotError> {
        // **A page range `nt!PoolBigPageTable` names needs no reading at all.** It is one
        // allocation with no header: discovery took its tag and length from the table, and
        // nothing on its pages adds to that — `walk_page_ranges` emits the one span and skips
        // the header decode. Reading the pages anyway cost two things. On a live 29671 kernel
        // (lab, 2026-10-07) the live entries summed to 111 MB, every byte of it read and then
        // discarded, over a link that moves a few MB a second. And on a page the memory
        // manager had trimmed the read failed, the range was filed as an unreadable gap with
        // no tag, and the name the table had for it went unused: `!pool` called
        // 0xffffa4b05e1b5000 an 8 KB `Gcac` and this walk called it `....` (`windbg-mcp`
        // FOLLOWUPS item 99, the paged-out half). Resident or not, the table's answer is the
        // same, so it is given without asking the pages.
        if region.backend == PoolBackend::Segment
            && let Some(tag) = region.known_tag
        {
            let state = region
                .states
                .first()
                .copied()
                .unwrap_or(PoolState::Allocated);
            snapshot.record_span(self.base_span(
                region,
                region.address,
                region.address,
                region.size as u64,
                tag,
                state,
            ));
            return Ok(());
        }
        let requested_end = region.address.saturating_add(region.size as u64);
        let mut cursor = region.address;
        let mut consecutive_stalls = 0u32;
        let mut stalled_here = false;
        // Where the next VS chunk header is, carried across the committed extents of one
        // subsegment. `walk_lfh` and `walk_page_ranges` need no such thing — their slots are a
        // fixed size, so they index off the region base and are correct wherever an extent starts.
        // A VS chunk is only findable from the end of the one before it, which is a fact about
        // the region and not about the extent, so it has to live out here.
        let mut vs_chunk = Some(region.address);
        // How far the VS spans already filed reach past the extent that filed them; see
        // [`VsResume::covered_through`]. A high-water mark, because a chunk that crosses one
        // hole may cross the next as well.
        let mut vs_covered = region.address;
        while cursor < requested_end {
            check_budget(self.memory)?;
            let remaining = requested_end.saturating_sub(cursor).min(usize::MAX as u64) as usize;
            let (reported_base, reported_size) = match self.memory.valid_region(cursor, remaining) {
                Ok(valid) => valid,
                Err(error) => {
                    snapshot.diagnostics.push(format!(
                        "cannot query region {cursor:#x}+{remaining:#x}: {error}"
                    ));
                    self.unreadable(region, cursor, requested_end - cursor, snapshot);
                    break;
                }
            };
            let valid_base = reported_base.max(cursor).min(requested_end);
            // Reported only when some of that space is memory the target **has** — which,
            // before `committed_run` could be asked, was the only reading available. On a
            // healthy live user-mode heap the whole of it is reserved subsegment tails, and
            // saying "unreadable space extends" about address space holding nothing is a
            // diagnostic that both describes nothing wrong and is wrong about what it
            // describes. Emitting it after the spans rather than before them is what lets it
            // be conditional at all: only filing them establishes which kind this is.
            // Less whatever a span already covers: a VS chunk the big-page table named can
            // end inside this hole, and its span was filed from the table with nothing left
            // to learn from the pages.
            let gap_start = cursor.max(vs_covered).min(valid_base);
            if valid_base > gap_start
                && self.unreadable(region, gap_start, valid_base - gap_start, snapshot)
            {
                snapshot.diagnostics.push(format!(
                    "region {:#x}+{:#x} is only committed through {gap_start:#x}; unreadable space extends {:#x} bytes",
                    region.address,
                    region.size,
                    valid_base - gap_start
                ));
            }
            let valid_end = reported_base
                .saturating_add(reported_size as u64)
                .min(requested_end);
            if valid_end <= valid_base {
                // Two states used to arrive here as one, and only the second is a problem.
                //
                // The engine placing the next valid region beyond the end of what was asked
                // about is the region *finishing*: its tail has already been filed as
                // unreadable by the branch above, there is nothing behind it to lose, and the
                // loop is done. Reporting that as a failure to advance is what made this the
                // walk's largest diagnostic category — 3,285 lines on a live 26100 walk —
                // while describing nothing wrong.
                if valid_base >= requested_end {
                    break;
                }
                // The same answer in the engine's other encoding, and the same response. A query
                // that finds nothing valid in the span it was given may say so either by naming a
                // base past the end — handled just above — or by answering `0x0+0x0`, which is
                // not a region at zero but the absence of one. Either way the engine has already
                // reported on every byte from here to `requested_end`, so stepping a page and
                // asking again asks a question that has been answered.
                //
                // Measured on live 26100 (glslang/dbgscope#104): every stall sample carried
                // `0x0+0x0`, eight of them consecutive — one dead region stepping itself into the
                // give-up bound, eight KD round trips to learn what the first answer said. Across
                // the walk that was ~1,100 stalled pages, and `recovered_bytes` was zero on four
                // separate runs, which is what this predicts and what it costs nothing to keep
                // reporting.
                //
                // Quiet, like its sibling above and for the same reason: a region running out of
                // readable content is not a fault, and #94 was raised because saying so 3,285
                // times drowned out the diagnostics that meant something. The bytes are still
                // filed as unreadable and `complete` still clears — the tail has not been walked,
                // and nothing here claims otherwise.
                if reported_base == 0 && reported_size == 0 {
                    let gap_start = valid_base.max(vs_covered).min(requested_end);
                    self.unreadable(region, gap_start, requested_end - gap_start, snapshot);
                    break;
                }
                // The rest is a real stall: a valid region was reported inside the span and it
                // ends at or before where the cursor already is. Step over the page it stalled
                // on and ask again, rather than giving up on every committed page behind it —
                // the samples show these firing part-way through regions of 0x10fd0 and 0x3afc0
                // bytes, not at their tails.
                //
                // To the next page **boundary**, not a page's worth of bytes: a region begins
                // wherever the allocator put it (`…de6030`, `…e02130` in those same samples),
                // so a stall at a page offset of 0xfd0 would otherwise write off 0x30 bytes of
                // the page that stalled and 0xfd0 bytes of the page after it — discarding a
                // healthy page to skip a bad one, which is this whole branch's mistake in
                // miniature. Every stall after the first is aligned by construction.
                //
                // The advance is **unconditional**: the boundary is always past `valid_base`,
                // and `requested_end` is too — the check above returned otherwise — so this
                // cannot be zero. A query that keeps answering the same way would spin here
                // without end, which is the reason this used to abandon the region outright.
                let page_end = (valid_base & !(PAGE_SIZE - 1)).saturating_add(PAGE_SIZE);
                let skip = page_end.min(requested_end) - valid_base;
                // The engine's own answer travels with the complaint, because the two shapes
                // that arrive here need opposite fixes and nothing else can tell them apart: a
                // region reported *behind* the cursor (`reported_base` below `valid_base`) means
                // the query is answering about memory already walked, while a zero-length region
                // reported ahead of it means the engine found something and could not size it.
                // `PoolDiagnostics` folds numbers into `#`, so this stays one shape however many
                // times it fires, and the verbatim sample carries the real values.
                snapshot.diagnostics.push(format!(
                    "valid-region query made no progress at {valid_base:#x}: the engine answered \
                     {reported_base:#x}+{reported_size:#x}; stepping over the rest of the page"
                ));
                let gap_start = valid_base.max(vs_covered).min(valid_base + skip);
                self.unreadable(region, gap_start, valid_base + skip - gap_start, snapshot);
                snapshot.stalls.pages += 1;
                snapshot.stalls.skipped_bytes = snapshot.stalls.skipped_bytes.saturating_add(skip);
                stalled_here = true;
                consecutive_stalls += 1;
                cursor = valid_base.saturating_add(skip);
                // Bounded, so a region that is dead all the way down still costs a fixed
                // number of queries rather than one per page of its length. Consecutive:
                // the single unreadable page in the middle of a live region — the shape these
                // samples actually have — resets it and costs one extra query.
                //
                // `>=`, so the count is the number of queries the limit permits and not one
                // short of it: the check runs after the page it counts has been stepped over.
                if consecutive_stalls >= MAX_CONSECUTIVE_STALLS {
                    snapshot.diagnostics.push(format!(
                        "region {:#x}+{:#x}: giving up after consecutive pages that would not advance",
                        region.address, region.size
                    ));
                    self.unreadable(region, cursor, requested_end - cursor, snapshot);
                    break;
                }
                continue;
            }
            consecutive_stalls = 0;
            let bytes = match self.read_extent(valid_base, (valid_end - valid_base) as usize) {
                Ok(bytes) => bytes,
                // Out of time is not "this memory would not read". Left to the arm below it
                // would be filed as an unreadable span — the walk asserting a fact about the
                // target out of its own failure to look.
                Err(error) if error.halts_walk() => return Err(error),
                Err(error) => {
                    snapshot
                        .diagnostics
                        .push(format!("cannot read region {valid_base:#x}: {error}"));
                    self.unreadable(region, valid_base, valid_end - valid_base, snapshot);
                    cursor = valid_end;
                    continue;
                }
            };
            if stalled_here {
                // Committed memory read on the far side of a stall — the coverage the old
                // `break` discarded, counted so the change can be judged by what it recovers
                // rather than by whether a diagnostic category shrank.
                snapshot.stalls.recovered_bytes = snapshot
                    .stalls
                    .recovered_bytes
                    .saturating_add(bytes.len() as u64);
            }
            // Verifier special pool is page-granular and its ranges are indistinguishable
            // from any other plain page range by their flags, so this has to come *before*
            // the backend dispatch. Walked as page ranges instead, the pool header plus
            // its fill decodes as garbage and every allocation in the heap silently
            // disappears from the snapshot.
            if region.heap.special {
                self.walk_special_pool(region, valid_base, &bytes, snapshot);
                if snapshot.match_limit_reached() {
                    return Ok(());
                }
                cursor = valid_end;
                continue;
            }
            match region.backend {
                PoolBackend::Lfh => self.walk_lfh(region, valid_base, &bytes, snapshot),
                PoolBackend::Vs => {
                    let resumed = self.walk_vs(region, valid_base, &bytes, vs_chunk, snapshot);
                    vs_chunk = resumed.map(|resume| resume.next);
                    if let Some(resume) = resumed {
                        vs_covered = vs_covered.max(resume.covered_through);
                    }
                }
                PoolBackend::Segment => self.walk_page_ranges(region, valid_base, &bytes, snapshot),
                PoolBackend::Large => return Ok(()),
            }
            if snapshot.match_limit_reached() {
                return Ok(());
            }
            cursor = valid_end;
        }
        Ok(())
    }

    /// Reads one committed extent, polling the deadline as the bytes come in.
    ///
    /// A single `read_exact` of a whole extent is one synchronous transfer that cannot
    /// observe the deadline while it runs, so a large committed page range could carry the
    /// walk past its ceiling however carefully the loops around it polled.
    ///
    /// **The chunking is of the transfer, not of the decoding.** Handing each chunk to the
    /// backend decoders instead would be much simpler and quietly wrong: `walk_lfh` skips any
    /// slot straddling the end of its slice and `walk_vs` stops at a chunk that runs past it
    /// and clears `complete`, so every chunk boundary would lose an allocation or invent an
    /// incomplete walk. The decoders still see one contiguous extent, exactly as before.
    fn read_extent(&self, base: u64, size: usize) -> Result<Vec<u8>, SnapshotError> {
        if size <= EXTENT_READ_CHUNK {
            return self.memory.read_exact(base, size);
        }
        let mut bytes = Vec::with_capacity(size);
        while bytes.len() < size {
            check_budget(self.memory)?;
            let take = EXTENT_READ_CHUNK.min(size - bytes.len());
            // A failed chunk fails the extent, which is what a failed whole-extent read did.
            //
            // A *short* chunk has to fail it too, and that is specific to reading in pieces.
            // Two things go wrong without this. The next chunk's address comes from how much
            // has been collected, so a short one shifts every later chunk and assembles a
            // buffer of contiguous bytes taken from discontiguous addresses — which the
            // decoders cannot detect, and which makes them report allocations that were never
            // at those addresses. Worse, the loop then cannot finish: the last chunk asks for
            // the remainder, comes back short, and `bytes.len()` never reaches `size`.
            // Deleting this check hangs the walk rather than failing it — verified. Refusing
            // the extent reports it unreadable, which is true, and is what a short
            // whole-extent read did before.
            let chunk = self.memory.read_exact(base + bytes.len() as u64, take)?;
            if chunk.len() != take {
                return Err(SnapshotError::InvalidData {
                    detail: format!(
                        "short extent read at {:#x}: asked {take:#x}, got {:#x}",
                        base + bytes.len() as u64,
                        chunk.len()
                    ),
                });
            }
            bytes.extend_from_slice(&chunk);
        }
        Ok(bytes)
    }

    fn walk_large(&self, region: &PoolRegion, snapshot: &mut PoolSnapshot) {
        if region.size == 0 {
            return;
        }
        snapshot.record_span(
            self.base_span(
                region,
                region.address,
                region.address,
                region.size as u64,
                region.known_tag.unwrap_or(0),
                region
                    .states
                    .first()
                    .copied()
                    .unwrap_or(PoolState::Allocated),
            ),
        );
    }

    fn base_span(
        &self,
        region: &PoolRegion,
        header: u64,
        usable: u64,
        size: u64,
        tag: u32,
        state: PoolState,
    ) -> PoolSpan {
        PoolSpan {
            header_address: header,
            usable_address: usable,
            size,
            requested_size: region.requested_size,
            raw_tag: tag,
            display_tag: super::decode::display_tag(tag),
            pool_kind: region.pool_kind,
            numa_node: region.numa_node,
            heap: region.heap,
            subsegment: region.subsegment,
            backend: region.backend,
            state,
            size_class: region.unit_size,
        }
    }

    /// Files address space the walk did not decode, split by what the memory manager says is
    /// behind it.
    ///
    /// Every caller reaches this having failed to read something, which is where the walk used
    /// to stop reasoning: it recorded one [`PoolState::Unreadable`] span and cleared
    /// `complete`. That is right for a page the target has and wrong for address space that
    /// holds nothing, and the two are not distinguishable from the failure — a reserved tail
    /// and a trimmed page fail a read identically. [`PoolMemory::committed_run`] is what
    /// separates them, and it is asked **per run** rather than per page, so one reserved tail
    /// of any size costs one query.
    ///
    /// Split rather than classified whole: a span can cross the boundary — the tail of a
    /// committed run and the reserved space after it arrive here as one failure — and calling
    /// the whole of it by either name misreports the other half. Splitting also keeps the byte
    /// totals addable, which is the only way the figure downstream means anything.
    ///
    /// Answers **whether any of it counted against coverage**, so a caller whose diagnostic is
    /// about missed memory can stay quiet when nothing was missed.
    fn unreadable(
        &self,
        region: &PoolRegion,
        address: u64,
        size: u64,
        snapshot: &mut PoolSnapshot,
    ) -> bool {
        let Some(end) = (size != 0).then(|| address.saturating_add(size)) else {
            return false;
        };
        let mut cursor = address;
        let mut missed = false;
        while cursor < end {
            // `> cursor` is not a formality: a run that does not advance would spin here, and
            // the engine answering about a *preceding* run is exactly the shape `walk_region`
            // already has to defend against. An answer that cannot move the cursor is no
            // answer, so the rest of the span is filed the conservative way and the loop ends.
            let Some(run) = self
                .memory
                .committed_run(cursor)
                .filter(|run| run.end > cursor)
            else {
                return self.gap(
                    region,
                    cursor,
                    end - cursor,
                    PoolState::Unreadable,
                    snapshot,
                ) || missed;
            };
            let stop = run.end.min(end);
            let state = if run.committed {
                PoolState::Unreadable
            } else {
                PoolState::Uncommitted
            };
            missed |= self.gap(region, cursor, stop - cursor, state, snapshot);
            cursor = stop;
        }
        missed
    }

    /// Records one gap span and, when it is a gap in *coverage*, clears `complete`; answers
    /// whether it was one.
    ///
    /// The clearing is here rather than at the call sites so the two can never disagree:
    /// `complete` means "no span says something may have been missed", and
    /// [`PoolState::is_coverage_gap`] is the one definition of which states say that.
    fn gap(
        &self,
        region: &PoolRegion,
        address: u64,
        size: u64,
        state: PoolState,
        snapshot: &mut PoolSnapshot,
    ) -> bool {
        if size == 0 {
            return false;
        }
        let missed = state.is_coverage_gap();
        if missed {
            snapshot.complete = false;
        }
        snapshot.record_span(self.base_span(region, address, address, size, 0, state));
        missed
    }

    /// Decodes a Driver Verifier special-pool region, one allocation per page.
    ///
    /// Special pool gives every allocation its own page: a header at the page start, fill,
    /// then the caller's block pushed up against the page end so an overrun lands on the
    /// following guard page. The guard page is unmapped, which is why `valid_region` stops
    /// after the data page and why a *freed* allocation simply vanishes — both are the
    /// mechanism working, not read failures.
    ///
    /// The block is placed at `page + PAGE_SIZE - align_up(size, 16)`; the 16-byte
    /// alignment is easy to miss, and dropping it puts the reported address 8 bytes past
    /// the real one (`0xf98` instead of `0xf90` for a 0x68 allocation).
    ///
    /// The page is read by [`decode_special_pool_header`] rather than by the ordinary
    /// [`decode_pool_header`]: special pool reuses those header bytes for a size, a fill
    /// pattern and a tracking flag, so the generic reading's plausibility guard is testing
    /// fields that are not there.
    fn walk_special_pool(
        &self,
        region: &PoolRegion,
        base: u64,
        bytes: &[u8],
        snapshot: &mut PoolSnapshot,
    ) {
        let mut page = base.next_multiple_of(PAGE_SIZE);
        while let Some(offset) = page.checked_sub(base).map(|delta| delta as usize) {
            if offset >= bytes.len() {
                break;
            }
            let available = (bytes.len() - offset).min(PAGE_SIZE as usize);
            let page_bytes = &bytes[offset..offset + available];
            let Some(header) = decode_special_pool_header(bytes, offset, region.pool_header) else {
                // Special pool is page-granular: one undecodable page says nothing about
                // the next, so stopping here would silently drop the rest of the region —
                // and leaving `complete` set would cache that truncated result and re-serve
                // it. Every other rejection path in this file reports and continues.
                snapshot.diagnostics.push(format!(
                    "special-pool page {page:#x}: header describes no block the page could \
                     hold; page skipped"
                ));
                snapshot.complete = false;
                let Some(next) = page.checked_add(PAGE_SIZE) else {
                    break;
                };
                page = next;
                continue;
            };
            // An untagged page is not an allocation. Freed special-pool pages are
            // unmapped rather than zeroed, so anything readable here should carry a tag.
            if header.tag != 0 {
                let placement = special_pool_placement(page, header, page_bytes);
                // Say so when the size is a bound rather than a measurement. Otherwise a
                // conservative page-sized range is indistinguishable from a real
                // allocation: `tag_census` would fold it into `total_bytes` — reporting a
                // 0x68 block as 0xff0 — while the report still claimed to be complete.
                if placement.approximate {
                    snapshot.diagnostics.push(format!(
                        "special-pool page {page:#x} tag `{}`: size not corroborated by the \
                         fill pattern; reporting the rest of the page as an upper bound",
                        super::decode::display_tag(header.tag)
                    ));
                    snapshot.complete = false;
                }
                let (usable, size) = (placement.usable, placement.size);
                snapshot.record_span(self.base_span(
                    region,
                    page,
                    usable,
                    size,
                    header.tag,
                    PoolState::Allocated,
                ));
                if snapshot.match_limit_reached() {
                    break;
                }
            }
            let Some(next) = page.checked_add(PAGE_SIZE) else {
                break;
            };
            page = next;
        }
    }

    fn walk_lfh(&self, region: &PoolRegion, base: u64, bytes: &[u8], snapshot: &mut PoolSnapshot) {
        let unit = region.unit_size as usize;
        if unit < region.pool_header.size {
            snapshot.diagnostics.push(format!(
                "rejecting implausible LFH unit size {} at {base:#x}",
                region.unit_size
            ));
            snapshot.complete = false;
            return;
        }
        let slice_offset = base.saturating_sub(region.address) as usize;
        let first_slot = slice_offset.div_ceil(unit);
        let slice_end = slice_offset.saturating_add(bytes.len());
        let mut slot = first_slot;
        while let Some(slot_offset) = slot.checked_mul(unit) {
            if slot_offset.saturating_add(unit) > slice_end {
                break;
            }
            let offset = slot_offset - slice_offset;
            let address = region.address + slot_offset as u64;
            // A block never straddles a page: the allocator skips the slot that would and
            // resumes at the next one, leaving the tail of the page as slack. Verified
            // against `!pool` on Server 26100 — with a 0x1a0 unit, blocks fill a page to
            // +0xef0 and the next block sits at +0x1090, which is exactly the following
            // slot, so linear indexing stays correct across the gap.
            //
            // Ordinary layout, then, not damage. Reporting it cost ~11.6k diagnostics on
            // an idle machine — drowning out the ones that meant something — and wrongly
            // cleared `complete` on essentially every snapshot taken.
            if address / PAGE_SIZE != (address + unit as u64 - 1) / PAGE_SIZE {
                slot += 1;
                continue;
            }
            let Some(state) = lfh_bitmap_state(&region.bitmap, slot) else {
                snapshot
                    .diagnostics
                    .push(format!("truncated LFH bitmap at slot {slot}"));
                snapshot.complete = false;
                break;
            };
            let tag = decode_pool_header(bytes, offset, region.pool_header)
                .map_or(0, |header| header.tag);
            let usable = address + region.pool_header.size as u64;
            snapshot.record_span(self.base_span(
                region,
                address,
                usable,
                unit as u64 - region.pool_header.size as u64,
                tag,
                state,
            ));
            if snapshot.match_limit_reached() {
                break;
            }
            slot += 1;
        }
    }

    /// Decodes one committed extent of a VS subsegment, chunk by chunk, and says where the
    /// chunk after the last one it read begins.
    ///
    /// **A VS extent can only be decoded from a known chunk boundary.** Chunks vary in size, so
    /// unlike [`Self::walk_lfh`] and [`Self::walk_page_ranges`] — whose slots are a fixed size,
    /// making them correct at whatever offset into the region an extent happens to start — the
    /// only way to know where a header is, is to have walked the one before it. `walk_region`
    /// hands this one *committed extent* at a time, and a subsegment is routinely committed in
    /// pieces: `RtlpHpVsSubsegmentCommitPages` commits and decommits page ranges anywhere inside
    /// the subsegment and records them in `_HEAP_VS_SUBSEGMENT.CommitBitmap`, so holes between
    /// committed extents are the allocator's steady state and not damage.
    ///
    /// Starting each extent at its own first sixteen-byte boundary — which is what this did — is
    /// therefore a guess on every extent after a hole, and one of the two sources of
    /// glslang/dbgscope#103's 106,516 refusals: one per sixteen bytes, until the scan wandered
    /// onto a word that decoded plausibly. The refusals were the harmless half. Whatever the
    /// scan wandered onto was pushed into `spans` as an allocation, tag and all.
    ///
    /// It was the smaller half. A live 26100 walk with this fixed still refused 76,398 headers,
    /// and the rest were the *bound*: the page range holding a VS subsegment is a unit larger
    /// than the subsegment, so the walk was decoding a page that holds no chunks. See
    /// [`discover_segment_context`], which now sizes the region from the subsegment.
    ///
    /// The chain is what crosses the hole. The chunk before one records a size that reaches past
    /// it and so names the header on the far side, at an address this walk already knows.
    /// `expected` carries that address between extents. `None` means the walk lost it, and a walk
    /// that has lost it does not guess again.
    ///
    /// It can lose it, and the reason is worth knowing before reading `unplaced_bytes` as damage.
    /// A *decommitted* range is the interior of a free chunk — the allocator has to keep its own
    /// headers reachable — so the chain crosses those. But a hole here is not only decommitted
    /// memory: on paged pool it is also memory that is committed and **paged out**, and nothing
    /// stops a chunk header being in it. Every unreadable header on a live 26100 walk was exactly
    /// that — `!pte` on each named address gave a pagefile PTE, `Protect: 4 - ReadWrite`, not a
    /// zero one — and a KD link cannot fault a page in. So those bytes are unreachable from a
    /// live target rather than absent from the pool, and the walk reports them instead of
    /// inventing chunks in them.
    ///
    /// A refused header costs more than itself: the walk no longer knows where the next one
    /// starts, so it advances sixteen bytes and tries again, and every header after it in the
    /// extent is decoded at a *guessed* offset rather than at the end of the previous chunk.
    /// One header rewritten while we read it and one systematically misdecoded field therefore
    /// look identical from a count — which is why what is reported here is the count of
    /// **chunks** refused rather than of extents that contained a refusal, the failing
    /// predicate in its own words, and the `Sizes` word as read. It is also why a refusal ends
    /// the chain for the whole region: every offset after it rests on that guess.
    fn walk_vs(
        &self,
        region: &PoolRegion,
        base: u64,
        bytes: &[u8],
        expected: Option<u64>,
        snapshot: &mut PoolSnapshot,
    ) -> Option<VsResume> {
        let extent_end = base.saturating_add(bytes.len() as u64);
        let Some(next) = expected.filter(|next| *next >= base) else {
            // Either an earlier extent of this region lost the chain, or the header it pointed
            // at fell inside the hole just crossed. Both come to the same thing: nothing in
            // these bytes can be placed. Sized and not merely counted, because what this costs
            // is coverage — and the alternative, decoding from a guess, costs correctness.
            snapshot.unplaced_bytes = snapshot.unplaced_bytes.saturating_add(bytes.len() as u64);
            // Two causes, kept as two shapes so a walk says which — `PoolDiagnostics` folds
            // numbers but not words, so this is the only way the split survives to a live run.
            snapshot.diagnostics.push(match expected {
                Some(next) => format!(
                    "VS extent at {base:#x} does not begin on a chunk boundary: the chunk chain \
                     names {next:#x}, {:#x} bytes back inside unreadable memory; {:#x} bytes not \
                     decoded",
                    base - next,
                    bytes.len()
                ),
                None => format!(
                    "VS extent at {base:#x} cannot be placed: the chain was already lost earlier \
                     in this region; {:#x} bytes not decoded",
                    bytes.len()
                ),
            });
            snapshot.complete = false;
            return None;
        };
        if next >= extent_end {
            // A chunk that began before this extent covers all of it — an ordinary large free
            // chunk with its interior decommitted. Nothing to decode and nothing lost: the
            // expectation still names a header further on.
            return Some(VsResume {
                next,
                covered_through: base,
            });
        }
        let mut offset = (next - base) as usize;
        // Where the *next* extent resumes, kept in step with `offset` so every way out of the
        // loop below leaves it pointing at a header rather than at wherever the bytes ran out.
        let mut resume = next;
        let mut lost = false;
        let mut chunks = 0usize;
        let header_bytes = region.vs_header_size + region.pool_header.size;
        let subsegment_end = region.address.saturating_add(region.size as u64);
        let mut refused = 0u64;
        let mut resync_from = None;
        let mut chain_breaks = 0u64;
        // What the *next* header must record as its previous size, or `None` where there is
        // nothing to compare against: at the start of an extent, and after a resynchronisation,
        // where the walk no longer knows which chunk it is standing on.
        let mut previous_chunk = None;
        // See [`VsResume::covered_through`]: raised only by a span filed past the extent.
        let mut covered_through = base;
        while offset
            .saturating_add(region.vs_header_size)
            .saturating_add(region.pool_header.size)
            <= bytes.len()
            && chunks < self.traversal_limit
        {
            let header_address = base + offset as u64;
            let Some(encoded) = read_u64(bytes, offset + region.vs_sizes_offset) else {
                break;
            };
            let chunk = match decode_vs_chunk(
                encoded,
                header_address,
                region.heap_key,
                header_bytes,
                subsegment_end,
            ) {
                Ok(chunk) => chunk,
                Err(rejection) => {
                    if refused == 0 {
                        snapshot.diagnostics.push(format!(
                            "refusing VS chunk at {header_address:#x}: {rejection}"
                        ));
                    }
                    refused += 1;
                    resync_from.get_or_insert(header_address);
                    previous_chunk = None;
                    // The sixteen-byte scan below may find its way back onto real headers
                    // inside this extent, but it cannot *know* that it has — so whatever it
                    // ends on is not an address to hand the next extent. Sticky, because a
                    // chunk decoded after the scan resumes is exactly the guess in question.
                    lost = true;
                    snapshot.complete = false;
                    offset = offset.saturating_add(16);
                    continue;
                }
            };
            if let Some(expected) = previous_chunk
                && chunk.previous_size != expected
            {
                // Not a refusal: the chunk itself is plausible, and the chain disagreeing says
                // the *stride* that reached it is wrong. Reported so one live run can tell a
                // header rewritten under us from a walk decoding the wrong field, which is the
                // question a refusal count on its own cannot answer.
                if chain_breaks == 0 {
                    snapshot.diagnostics.push(format!(
                        "VS chunk at {header_address:#x} records a previous size of \
                         {:#x} where the chunk before it measured {expected:#x}",
                        chunk.previous_size
                    ));
                }
                chain_breaks += 1;
                snapshot.complete = false;
            }
            let chunk_size = chunk.size;
            let state = if region.cached_chunks.contains(&header_address) {
                PoolState::CachedFree
            } else if region.reusable_chunks.contains(&header_address) {
                PoolState::ReusableFree
            } else if chunk.allocated {
                PoolState::Allocated
            } else {
                PoolState::ReusableFree
            };
            // **A chunk too large for `_POOL_HEADER.BlockSize` carries no header**, so decoding
            // one out of it reads the caller's own first sixteen bytes as a tag. Its name is in
            // `nt!PoolBigPageTable`, which discovery resolved into `region.big_pool`.
            //
            // Matched by **containment** rather than by arithmetic on the chunk header: the
            // table's `Va` is where the allocation starts, and taking it as given costs nothing
            // and assumes nothing about what sits between the two — which is the one thing here
            // that is measured and not yet explained (`windbg-mcp` FOLLOWUPS item 99). The size
            // has to fit inside the chunk as well, so an entry can only be claimed by a chunk
            // that could really hold it.
            //
            // **Asked before the extent check below, not after it.** Everything the span
            // carries is in hand once the header has decoded: the tag and the length are the
            // table's, the state is the free tree's, and the chunk's own bytes were never going
            // to be read. So a tail that runs into a page the memory manager has trimmed takes
            // nothing from the answer — and asked afterwards, the chunk was dropped for that
            // tail and its pages filed as an untagged gap instead, which is what `pool_chunk`
            // then answered for the allocation's own address. Measured on a live 26100 kernel
            // (`ctf-vm`, 2026-10-07): the `CIcr` allocation at 0xffffa9099c86f000 sits 0x20
            // into a 0x12c0-byte chunk whose header is resident and whose tail is not, and
            // `!pool` named it from the table while this walk answered `....`.
            let big_pool = (chunk_size as u64 > MAX_POOL_HEADER_CHUNK)
                .then(|| {
                    let end = header_address.saturating_add(chunk_size as u64);
                    region.big_pool.iter().find(|(address, entry)| {
                        (header_address..end).contains(*address)
                            && address.saturating_add(entry.size) <= end
                    })
                })
                .flatten()
                .map(|(address, entry)| (*address, *entry));
            if let Some((address, entry)) = big_pool {
                let mut span =
                    self.base_span(region, address, address, entry.size, entry.tag, state);
                span.size_class = chunk_size.min(u32::MAX as usize) as u32;
                snapshot.record_span(span);
                previous_chunk = Some(chunk_size);
                offset += chunk_size;
                resume = base + offset as u64;
                // Past the extent when the tail is in the hole ahead, which is what tells the
                // extent loop not to file that hole over this span.
                covered_through = resume;
                chunks += 1;
                if snapshot.match_limit_reached() {
                    break;
                }
                continue;
            }
            // The chunk reaches past the committed extent, which after the bound check in
            // `decode_vs_chunk` can only mean a hole ahead of it inside the subsegment — so
            // this is the ordinary free chunk with a decommitted interior, not a walk running
            // out of bytes. It carries the chain over the hole, which is the whole reason the
            // expectation is returned rather than recomputed per extent.
            //
            // **What the walk loses by it depends on what is in the hole**, and the same
            // question the gap spans ask answers this one. A chunk's span is geometry and
            // state, not contents: its header was read, its size came out of that header and
            // passed the bound check, and its state comes from the free tree — so where the
            // tail holds nothing, nothing about the chunk is unknown and the span is emitted
            // below like any other. Where the tail is memory the process has, the walk really
            // did fail to see part of this chunk, and then no span is emitted and `complete`
            // clears.
            //
            // Measured on `sihost` (26200, 2026-09-24): five free chunks of 0x1660 to 0xfe70
            // bytes, each with a decommitted middle, were the *only* thing left holding that
            // walk at `Partial` once the gap spans were classified — and this site cleared
            // `complete` without a diagnostic, so nothing in the answer said why.
            if offset.saturating_add(chunk_size) > bytes.len()
                && !self
                    .memory
                    .committed_run(base.saturating_add(bytes.len() as u64))
                    .is_some_and(|run| !run.committed)
            {
                resume = header_address.saturating_add(chunk_size as u64);
                snapshot.diagnostics.push(format!(
                    "VS chunk at {header_address:#x} is {chunk_size:#x} bytes and runs \
                     {:#x} past the committed extent at {:#x}; no span is emitted for it",
                    offset.saturating_add(chunk_size) - bytes.len(),
                    base.saturating_add(bytes.len() as u64)
                ));
                snapshot.complete = false;
                break;
            }
            // Where the chunk's user block begins: the pool header on a kernel target, the
            // caller's own bytes on a user-mode one. Right after the chunk header, except where
            // that is the last unit of a page and the allocator hands out the next page start
            // instead — see [`PoolRegion::vs_skips_page_end`]. The chain is not involved: the
            // chunk's extent is what its header says, and only where the user block sits in it
            // moves.
            let mut physical_header = header_address + region.vs_header_size as u64;
            if region.vs_skips_page_end
                && physical_header.saturating_add(VS_UNIT) & (PAGE_SIZE - 1) == 0
            {
                // Only where the chunk has the unit to skip. `decode_vs_chunk` admitted this
                // chunk against its two headers alone, and a chunk exactly that size — the
                // smallest free remainder a split leaves — would otherwise have its user block
                // placed at its own end, which is the next chunk's header. The allocator asks
                // for one more unit than a request needs under the flag
                // (`RtlpHpVsContextAllocate+0x1a..+0x3c`: `and eax,1` / `inc eax` /
                // `add eax,r10d`), so a chunk here with no room is one nothing was allocated
                // from, and it is read in place; an *allocated* one contradicts that and is
                // reported.
                if chunk_size >= header_bytes.saturating_add(VS_UNIT as usize) {
                    physical_header += VS_UNIT;
                } else if chunk.allocated {
                    snapshot.diagnostics.push(format!(
                        "VS chunk at {header_address:#x} is {chunk_size:#x} bytes, too short to \
                         hold a user block moved off the page end; read in place"
                    ));
                }
            }
            let pool_offset = physical_header.saturating_sub(base) as usize;
            let tag = decode_pool_header(bytes, pool_offset, region.pool_header)
                .map_or(0, |header| header.tag);
            let overhead = physical_header
                .saturating_sub(header_address)
                .saturating_add(region.pool_header.size as u64);
            let span_header = if region.pool_header.size == 0 {
                header_address
            } else {
                physical_header
            };
            let mut span = self.base_span(
                region,
                span_header,
                physical_header + region.pool_header.size as u64,
                (chunk_size as u64).saturating_sub(overhead),
                tag,
                state,
            );
            span.size_class = chunk_size.min(u32::MAX as usize) as u32;
            snapshot.record_span(span);
            previous_chunk = Some(chunk_size);
            offset += chunk_size;
            resume = base + offset as u64;
            chunks += 1;
            if snapshot.match_limit_reached() {
                break;
            }
        }
        if let Some(from) = resync_from {
            snapshot.refused_chunks = snapshot.refused_chunks.saturating_add(refused);
            snapshot.diagnostics.push(format!(
                "{refused} VS chunk headers refused, resynchronising from {from:#x}"
            ));
        }
        if chain_breaks > 0 {
            snapshot.diagnostics.push(format!(
                "{chain_breaks} VS chunks disagreed with the size of the chunk before them, \
                 from {base:#x}"
            ));
        }
        if !snapshot.match_limit_reached() && chunks >= self.traversal_limit {
            snapshot.complete = false;
            snapshot
                .diagnostics
                .push(format!("VS traversal limit reached at {base:#x}"));
        }
        (!lost).then_some(VsResume {
            next: resume,
            covered_through,
        })
    }

    fn walk_page_ranges(
        &self,
        region: &PoolRegion,
        base: u64,
        bytes: &[u8],
        snapshot: &mut PoolSnapshot,
    ) {
        let unit = region.unit_size.max(1) as usize;
        let slice_offset = base.saturating_sub(region.address) as usize;
        let first_slot = slice_offset.div_ceil(unit);
        let slice_end = slice_offset.saturating_add(bytes.len());
        let mut slot = first_slot;
        while let Some(slot_offset) = slot.checked_mul(unit) {
            if slot_offset >= slice_end {
                break;
            }
            let offset = slot_offset - slice_offset;
            let remaining = bytes.len() - offset;
            let size = unit.min(remaining);
            let state = region
                .states
                .get(slot)
                .copied()
                .or_else(|| region.states.first().copied())
                .unwrap_or(PoolState::Unreadable);
            let tag = region.known_tag.or_else(|| {
                decode_pool_header(bytes, offset, region.pool_header).map(|header| header.tag)
            });
            let address = region.address + slot_offset as u64;
            let header_size = region.pool_header.size.min(size);
            snapshot.record_span(self.base_span(
                region,
                address,
                address + header_size as u64,
                size.saturating_sub(header_size) as u64,
                tag.unwrap_or(0),
                state,
            ));
            if snapshot.match_limit_reached() {
                break;
            }
            slot += 1;
        }
    }
}

/// Walk user-mode Segment Heap roots through the same region discovery and backend decoders
/// used for kernel pool heaps.
///
/// The adapter supplies user addresses and a zero-sized pool header through `layout`; everything
/// below the root enumeration—page segments, descriptors, LFH bitmaps, VS chains, free-state
/// evidence, large allocations, diagnostics, caps, and deadlines—is shared.
pub(crate) fn walk_user_segment_heaps<M: PoolMemory>(
    memory: &M,
    layout: &PoolLayout,
    peb: u64,
    heaps: &[u64],
    budget: Option<Duration>,
    traversal_limit: usize,
) -> Result<PoolSnapshot, SnapshotError> {
    let (discovery_deadline, walk_deadline) = budget_deadlines(Instant::now(), budget);
    let discovery_clock = Budgeted::new(memory, discovery_deadline);
    let globals_address = *layout
        .globals
        .get("RtlpHpHeapGlobals")
        .ok_or_else(|| missing_layout("RtlpHpHeapGlobals"))?;
    let heap_key = scalar(
        &discovery_clock,
        globals_address + layout.field("_RTLP_HP_HEAP_GLOBALS", "HeapKey")? as u64,
        8,
    )?;
    let lfh_key = scalar(
        &discovery_clock,
        globals_address + layout.field("_RTLP_HP_HEAP_GLOBALS", "LfhKey")? as u64,
        8,
    )?;

    let mut discovery = Discovery::default();
    let mut expired = false;
    for &heap in heaps {
        let identity = HeapIdentity {
            pool_state: peb,
            heap,
            special: false,
        };
        match discover_heap_regions(
            &discovery_clock,
            layout,
            heap,
            0,
            PoolKind::NonPagedNx,
            identity,
            None,
            heap_key,
            lfh_key,
            traversal_limit,
            &mut discovery,
        ) {
            Ok(()) => {}
            Err(SnapshotError::BudgetExpired) => {
                expired = true;
                break;
            }
            Err(error) if error.halts_walk() => return Err(error),
            Err(error) => discovery
                .diagnostics
                .push(format!("cannot fully discover heap {heap:#x}: {error}")),
        }
    }

    let mut snapshot = PoolSnapshot {
        complete: discovery.diagnostics.is_empty(),
        diagnostics: PoolDiagnostics::from_iter(std::mem::take(&mut discovery.diagnostics)),
        ..PoolSnapshot::default()
    };
    let walk_clock = Budgeted::new(memory, walk_deadline);
    let walker = SnapshotWalker {
        memory: &walk_clock,
        layout,
        traversal_limit,
    };
    walker.walk_discovered_regions(discovery.regions, budget, expired, &mut snapshot)?;
    Ok(snapshot)
}

#[cfg(test)]
mod tests {
    use std::{cell::Cell, collections::HashMap};

    use super::*;

    // ---- LFH slots that would straddle a page ---------------------------------------

    fn lfh_region(address: u64, unit_size: u32) -> PoolRegion {
        PoolRegion {
            address,
            size: 0x1000,
            requested_size: None,
            pool_kind: PoolKind::NonPagedNx,
            numa_node: 0,
            heap: HeapIdentity {
                pool_state: 0,
                heap: 0,
                special: false,
            },
            subsegment: None,
            backend: PoolBackend::Lfh,
            unit_size,
            // Eight slots' worth, all marked allocated.
            bitmap: vec![0xff],
            heap_key: 0,
            pool_header: PoolHeaderLayout {
                size: 0x10,
                previous_size: 0,
                pool_index: 1,
                block_size: 2,
                pool_type: 3,
                tag: 4,
            },
            vs_header_size: 0,
            vs_sizes_offset: 0,
            vs_skips_page_end: false,
            known_tag: None,
            states: Vec::new(),
            reusable_chunks: Arc::default(),
            cached_chunks: Arc::default(),
            big_pool: Arc::default(),
        }
    }

    /// The allocator refuses to let a block straddle a page: it skips that slot, leaves the
    /// page tail as slack, and resumes at the next slot. Confirmed against `!pool` on
    /// Server 26100. So the straddling slot must be skipped **silently** — it is ordinary
    /// layout, and reporting it once cost ~11.6k diagnostics on an idle machine.
    #[test]
    fn test_page_straddling_lfh_slot_is_skipped_without_complaint() {
        // Slot 0 at 0x1f80 spans 0x1f80..0x2080 and crosses; slot 1 at 0x2080 does not.
        let region = lfh_region(0x1f80, 0x100);
        let memory = FlatMemory::new(0x1f80, 0x200);
        let layout = vs_layout(VsFixture::Inline);
        let walker = SnapshotWalker {
            memory: &memory,
            layout: &layout,
            traversal_limit: 1000,
        };
        let mut snapshot = PoolSnapshot {
            complete: true,
            ..PoolSnapshot::default()
        };
        walker.walk_lfh(&region, 0x1f80, &memory.bytes, &mut snapshot);

        // The straddling slot is dropped, the following one is still reported: skipping
        // must not desynchronise the linear slot indexing.
        assert_eq!(snapshot.spans.len(), 1);
        assert_eq!(snapshot.spans[0].header_address, 0x2080);
        // The two properties that were wrong: silence, and an intact `complete`.
        assert!(
            snapshot.diagnostics.is_empty(),
            "normal layout must not be reported: {:?}",
            snapshot.diagnostics
        );
        assert!(
            snapshot.complete,
            "a straddling slot is expected layout and must not mark the snapshot incomplete"
        );
    }

    fn match_limited_snapshot(tag: &[u8; 4], matches: usize) -> PoolSnapshot {
        PoolSnapshot {
            complete: true,
            match_limit: Some(PoolMatchLimit::new(
                u32::from_le_bytes(*tag),
                None,
                NonZeroUsize::new(matches).expect("test match limits are nonzero"),
            )),
            ..PoolSnapshot::default()
        }
    }

    fn walk_page_tag(tag: [u8; 4], slots: usize, stop_after: usize) -> PoolSnapshot {
        let bytes = vec![0; slots * 0x20];
        let mut region = lfh_region(0x4000, 0x20);
        region.backend = PoolBackend::Segment;
        region.size = bytes.len();
        region.known_tag = Some(u32::from_le_bytes(tag));
        region.states = vec![PoolState::Allocated; slots];
        let memory = FlatMemory::new(region.address, bytes.len());
        let layout = vs_layout(VsFixture::Inline);
        let walker = SnapshotWalker {
            memory: &memory,
            layout: &layout,
            traversal_limit: 1000,
        };
        let mut snapshot = match_limited_snapshot(b"Tgsm", stop_after);
        walker.walk_page_ranges(&region, region.address, &bytes, &mut snapshot);
        snapshot
    }

    #[test]
    fn test_match_limit_does_not_stop_a_walk_with_zero_matches() {
        let snapshot = walk_page_tag(*b"Othr", 3, 1);
        assert_eq!(snapshot.spans.len(), 3);
        assert_eq!(snapshot.stopped_after_matches, None);
        assert!(snapshot.complete);
    }

    #[test]
    fn test_match_limit_stops_on_the_first_match() {
        let snapshot = walk_page_tag(*b"Tgsm", 3, 1);
        assert_eq!(snapshot.spans.len(), 1);
        assert_eq!(snapshot.stopped_after_matches, Some(1));
        assert_eq!(snapshot.matched_allocations, 1);
        assert!(!snapshot.complete);
    }

    #[test]
    fn test_match_limit_stops_on_the_requested_nth_match() {
        let snapshot = walk_page_tag(*b"Tgsm", 4, 2);
        assert_eq!(snapshot.spans.len(), 2);
        assert_eq!(snapshot.stopped_after_matches, Some(2));
        assert_eq!(snapshot.matched_allocations, 2);
        assert_eq!(snapshot.spans.last().unwrap().display_tag, "Tgsm");
    }

    #[test]
    fn test_match_limit_counts_only_allocated_chunks_in_scope() {
        let raw_tag = u32::from_le_bytes(*b"Tgsm");
        let mut snapshot = PoolSnapshot {
            complete: true,
            match_limit: Some(PoolMatchLimit::new(
                raw_tag,
                Some(false),
                NonZeroUsize::new(2).unwrap(),
            )),
            ..PoolSnapshot::default()
        };
        let span = |address, kind, state| {
            let mut span = PoolSpan::allocation(
                address,
                0x20,
                raw_tag,
                kind,
                HeapIdentity {
                    pool_state: 1,
                    heap: 1,
                    special: false,
                },
                PoolBackend::Lfh,
            );
            span.state = state;
            span
        };

        snapshot.record_span(span(0x1000, PoolKind::Paged, PoolState::Allocated));
        snapshot.record_span(span(0x1020, PoolKind::NonPagedNx, PoolState::ReusableFree));
        snapshot.record_span(span(0x1040, PoolKind::NonPagedNx, PoolState::Allocated));
        assert_eq!(snapshot.stopped_after_matches, None);
        snapshot.record_span(span(0x1060, PoolKind::NonPagedNx, PoolState::Allocated));
        assert_eq!(snapshot.stopped_after_matches, Some(2));
    }

    #[test]
    fn test_user_regions_never_report_payload_bytes_as_pool_tags() {
        let no_pool_header = PoolHeaderLayout {
            size: 0,
            previous_size: 0,
            pool_index: 0,
            block_size: 0,
            pool_type: 0,
            tag: 0,
        };
        let memory = FlatMemory::new(0x1000, 0x1000);
        let layout = vs_layout(VsFixture::Inline);
        let walker = SnapshotWalker {
            memory: &memory,
            layout: &layout,
            traversal_limit: 1000,
        };
        let mut snapshot = PoolSnapshot {
            complete: true,
            ..PoolSnapshot::default()
        };

        let mut lfh = lfh_region(0x1000, 0x20);
        lfh.size = 0x20;
        lfh.pool_header = no_pool_header;
        let mut lfh_bytes = vec![0; 0x20];
        lfh_bytes[..4].copy_from_slice(b"LFH!");
        walker.walk_lfh(&lfh, lfh.address, &lfh_bytes, &mut snapshot);

        let mut vs = vs_region(0x40);
        vs.pool_header = no_pool_header;
        let mut vs_bytes = vs_extent(&[(0x40, 0)]);
        vs_bytes[0x10..0x14].copy_from_slice(b"VS!!");
        walker.walk_vs(&vs, vs.address, &vs_bytes, Some(vs.address), &mut snapshot);

        let mut page = lfh_region(0x3000, 0x20);
        page.size = 0x20;
        page.backend = PoolBackend::Segment;
        page.pool_header = no_pool_header;
        page.states = vec![PoolState::Allocated];
        let mut page_bytes = vec![0; 0x20];
        page_bytes[..4].copy_from_slice(b"PAGE");
        walker.walk_page_ranges(&page, page.address, &page_bytes, &mut snapshot);

        assert_eq!(snapshot.spans.len(), 3);
        assert!(
            snapshot.spans.iter().all(|span| span.raw_tag == 0),
            "user payload bytes must not be decoded as kernel pool tags: {:?}",
            snapshot.spans
        );
    }

    // ---- VS roots: inline context vs affinity slots ----------------------------------

    /// Memory backed by one contiguous buffer. Anything outside it fails to read, the way
    /// an unmapped page does.
    struct FlatMemory {
        base: u64,
        bytes: Vec<u8>,
    }

    impl FlatMemory {
        fn new(base: u64, len: usize) -> Self {
            Self {
                base,
                bytes: vec![0; len],
            }
        }

        fn put(&mut self, address: u64, data: &[u8]) {
            let offset = (address - self.base) as usize;
            self.bytes[offset..offset + data.len()].copy_from_slice(data);
        }

        fn put_u16(&mut self, address: u64, value: u16) {
            self.put(address, &value.to_le_bytes());
        }

        fn put_u64(&mut self, address: u64, value: u64) {
            self.put(address, &value.to_le_bytes());
        }
    }

    impl PoolMemory for FlatMemory {
        fn read_exact(&self, address: u64, size: usize) -> Result<Vec<u8>, SnapshotError> {
            let offset = address
                .checked_sub(self.base)
                .and_then(|offset| usize::try_from(offset).ok())
                .ok_or_else(|| SnapshotError::InvalidData {
                    detail: format!("read below the fixture at {address:#x}"),
                })?;
            self.bytes
                .get(offset..offset + size)
                .map(<[u8]>::to_vec)
                .ok_or_else(|| SnapshotError::InvalidData {
                    detail: format!("read past the fixture at {address:#x}+{size:#x}"),
                })
        }

        fn valid_region(&self, address: u64, size: usize) -> Result<(u64, usize), SnapshotError> {
            Ok((address, size))
        }

        fn interrupted(&self) -> Result<bool, SnapshotError> {
            Ok(false)
        }
    }

    const VS_CONTEXT: u64 = 0x1000;

    /// Which shape a `vs_layout` fixture describes.
    #[derive(Clone, Copy, PartialEq, Eq)]
    enum VsFixture {
        /// The free tree is in `_HEAP_VS_CONTEXT`.
        Inline,
        /// Affinity slots, each naming its context by address (`VsContext`).
        SlotsByAddress,
        /// Affinity slots, each naming its context by displacement (`VsContextOffset`).
        SlotsByDisplacement,
    }

    /// A layout carrying only what `vs_roots` consults, in one of the three shapes a target's
    /// PDB can present. The two slot shapes differ by **one field name**, which is the whole
    /// substance of the change they cover: the same slot map, the same arithmetic, and a
    /// back-reference that is an address in one and `slot - context` in the other.
    fn vs_layout(shape: VsFixture) -> PoolLayout {
        let mut types = HashMap::new();
        types.insert(
            "_HEAP_VS_CONTEXT",
            match shape {
                VsFixture::Inline => {
                    type_layout(0x80, &[("FreeChunkTree", 0x10), ("DelayFreeContext", 0x30)])
                }
                _ => type_layout(0x60, &[("SlotMapRef", 0), ("AffinityMask", 2)]),
            },
        );
        if shape != VsFixture::Inline {
            let back = match shape {
                VsFixture::SlotsByDisplacement => "VsContextOffset",
                _ => "VsContext",
            };
            types.insert(
                "_HEAP_VS_AFFINITY_SLOT",
                type_layout(
                    0x80,
                    &[
                        (back, 0),
                        ("FreeChunkTree", 0x10),
                        ("DelayFreeContext", 0x40),
                    ],
                ),
            );
            types.insert("_HEAP_VS_SLOT_MAP", type_layout(4, &[("SlotRef", 0)]));
        }
        PoolLayout {
            key: crate::pool::layout::LayoutKey {
                image: crate::dbgeng::KernelImage::default(),
                session: 1,
            },
            globals: HashMap::new(),
            types,
        }
    }

    /// Four affinity entries over three distinct slots: two whose back-pointer agrees (one
    /// of them referenced twice, so dedup is exercised) and one that names another context.
    fn affinity_fixture() -> FlatMemory {
        let mut memory = FlatMemory::new(VS_CONTEXT, 0x1200);
        memory.put_u16(VS_CONTEXT, 0x10); // SlotMapRef -> map at ctx + 0x10*64
        memory.put(VS_CONTEXT + 2, &[3]); // AffinityMask -> 4 entries
        let map = VS_CONTEXT + 0x400;
        for (index, slot_ref) in [0x20u16, 0x30, 0x20, 0x40].into_iter().enumerate() {
            memory.put_u16(map + index as u64 * 4, slot_ref);
        }
        memory.put_u64(VS_CONTEXT + 0x800, VS_CONTEXT); // 0x20 << 6, owner agrees
        memory.put_u64(VS_CONTEXT + 0xc00, VS_CONTEXT); // 0x30 << 6, owner agrees
        memory.put_u64(VS_CONTEXT + 0x1000, 0xdead_beef); // 0x40 << 6, owner disagrees
        memory
    }

    /// The same slots, the same map, the same arithmetic — with each slot naming its context by
    /// displacement instead of by address, as `_HEAP_VS_AFFINITY_SLOT::VsContextOffset` does.
    ///
    /// Deliberately the same geometry as [`affinity_fixture`], so a test that swaps one for the
    /// other is measuring the back-reference rule and nothing else. The third slot disagrees the
    /// way a displacement can: `0x1234` is a real displacement to somewhere that is not this
    /// context.
    fn displacement_fixture() -> FlatMemory {
        let mut memory = FlatMemory::new(VS_CONTEXT, 0x1200);
        memory.put_u16(VS_CONTEXT, 0x10); // SlotMapRef -> map at ctx + 0x10*64
        memory.put(VS_CONTEXT + 2, &[3]); // AffinityMask -> 4 entries
        let map = VS_CONTEXT + 0x400;
        for (index, slot_ref) in [0x20u16, 0x30, 0x20, 0x40].into_iter().enumerate() {
            memory.put_u16(map + index as u64 * 4, slot_ref);
        }
        memory.put_u64(VS_CONTEXT + 0x800, 0x800); // 0x20 << 6, displacement agrees
        memory.put_u64(VS_CONTEXT + 0xc00, 0xc00); // 0x30 << 6, displacement agrees
        memory.put_u64(VS_CONTEXT + 0x1000, 0x1234); // 0x40 << 6, displacement disagrees
        memory
    }

    /// The slot map is up to 256 entries plus an owner read per distinct slot, all over the
    /// wire. It cannot lean on a later poll either: invalid slot refs produce no roots, so
    /// nothing downstream reads at all and the next check is a whole segment context away.
    #[test]
    fn test_vs_roots_polls_the_budget_between_slot_map_entries() {
        // Two reads resolve the slot map itself; the loop below them gets nothing.
        let memory = Impatient::over(affinity_fixture(), 2);

        let outcome = vs_roots(
            &memory,
            &vs_layout(VsFixture::SlotsByAddress),
            VS_CONTEXT,
            &mut Vec::new(),
        );

        assert!(
            matches!(outcome, Err(SnapshotError::BudgetExpired)),
            "the slot-map loop read on past the deadline, resolving {} roots",
            outcome.map_or(0, |roots| roots.len())
        );
    }

    /// Inline family: the tree is in the context, so there is exactly one root and no
    /// slot map is consulted at all.
    #[test]
    fn test_vs_roots_reads_the_legacy_in_context_shape() {
        let memory = FlatMemory::new(VS_CONTEXT, 0x100);
        let mut diagnostics = Vec::new();
        let roots = vs_roots(
            &memory,
            &vs_layout(VsFixture::Inline),
            VS_CONTEXT,
            &mut diagnostics,
        )
        .unwrap();
        assert_eq!(roots.len(), 1);
        assert_eq!(roots[0].base, VS_CONTEXT);
        assert_eq!(roots[0].tree_offset, 0x10);
        assert_eq!(roots[0].delay_offset, Some(0x30));
        assert!(diagnostics.is_empty());
    }

    /// Affinity family: `slot = VsContext + (SlotRef << 6)`, and entries routinely share a slot.
    #[test]
    fn test_vs_roots_walks_the_affinity_slots_and_dedups_them() {
        let memory = affinity_fixture();
        let mut diagnostics = Vec::new();
        let roots = vs_roots(
            &memory,
            &vs_layout(VsFixture::SlotsByAddress),
            VS_CONTEXT,
            &mut diagnostics,
        )
        .unwrap();
        let bases: Vec<u64> = roots.iter().map(|root| root.base).collect();
        // Four entries, one repeat, one rejected -> two roots.
        assert_eq!(bases, vec![VS_CONTEXT + 0x800, VS_CONTEXT + 0xc00]);
        assert!(roots.iter().all(|root| root.tree_offset == 0x10));
        assert!(roots.iter().all(|root| root.delay_offset == Some(0x40)));
    }

    /// Self-relative family: a slot names its context as `slot - context`, so the same map and
    /// the same arithmetic must produce the same roots as the address-bearing shape.
    ///
    /// The geometry is identical to [`affinity_fixture`] on purpose. What differs is one field
    /// name in the layout and the value each slot stores, which is the whole substance of the
    /// change (`windbg-mcp` FOLLOWUPS item 78).
    #[test]
    fn test_vs_roots_walks_affinity_slots_that_name_their_context_by_displacement() {
        let memory = displacement_fixture();
        let mut diagnostics = Vec::new();

        let roots = vs_roots(
            &memory,
            &vs_layout(VsFixture::SlotsByDisplacement),
            VS_CONTEXT,
            &mut diagnostics,
        )
        .unwrap();

        let bases: Vec<u64> = roots.iter().map(|root| root.base).collect();
        // The same four entries, one repeat and one rejected, as the address shape resolves.
        assert_eq!(bases, vec![VS_CONTEXT + 0x800, VS_CONTEXT + 0xc00]);
        assert!(roots.iter().all(|root| root.tree_offset == 0x10));
        assert!(roots.iter().all(|root| root.delay_offset == Some(0x40)));
    }

    /// The two back-reference rules are not interchangeable, and getting them the wrong way round
    /// does not fail loudly — it rejects every slot and walks no VS evidence at all.
    ///
    /// This is the measurement the fix rests on, as a test: the *same memory* read under the
    /// other build's rule resolves nothing. It is also what says an alias would have been the
    /// wrong fix. Confirmed live against `ntdll` on 26200, where the displacement `0xa80` read as
    /// an address produced `claims context 0xa80`.
    #[test]
    fn test_the_two_affinity_back_reference_rules_do_not_substitute_for_each_other() {
        for (fixture, wrong_shape, what) in [
            (
                displacement_fixture(),
                VsFixture::SlotsByAddress,
                "displacements read as addresses",
            ),
            (
                affinity_fixture(),
                VsFixture::SlotsByDisplacement,
                "addresses read as displacements",
            ),
        ] {
            let mut diagnostics = Vec::new();

            let roots = vs_roots(
                &fixture,
                &vs_layout(wrong_shape),
                VS_CONTEXT,
                &mut diagnostics,
            )
            .unwrap();

            assert!(
                roots.is_empty(),
                "{what}: {} slot(s) were accepted by the wrong rule, so the tree walk would read \
                 unrelated memory",
                roots.len()
            );
            assert!(
                diagnostics
                    .iter()
                    .all(|line| line.contains("claims context")),
                "{what}: every slot should be rejected for naming another context, not skipped \
                 for some other reason: {diagnostics:?}"
            );
        }
    }

    /// A displacement larger than the slot's own address names no context at all. Wrapping would
    /// hand the tree walk an address near the top of the space, which may well be readable.
    #[test]
    fn test_vs_roots_rejects_a_displacement_that_runs_off_the_bottom() {
        let mut memory = displacement_fixture();
        memory.put_u64(VS_CONTEXT + 0x800, u64::MAX);
        let mut diagnostics = Vec::new();

        let roots = vs_roots(
            &memory,
            &vs_layout(VsFixture::SlotsByDisplacement),
            VS_CONTEXT,
            &mut diagnostics,
        )
        .unwrap();

        assert_eq!(
            roots.iter().map(|root| root.base).collect::<Vec<_>>(),
            vec![VS_CONTEXT + 0xc00],
            "the slot with the impossible displacement should be the only one dropped"
        );
        assert!(
            diagnostics
                .iter()
                .any(|line| line.contains("claims context none")),
            "a displacement that cannot name a context should say so: {diagnostics:?}"
        );
    }

    /// A slot whose back-pointer names a different context is not this context's slot.
    /// Trusting it would aim the tree walk at unrelated but readable memory.
    #[test]
    fn test_vs_roots_rejects_a_slot_whose_back_pointer_disagrees() {
        let memory = affinity_fixture();
        let mut diagnostics = Vec::new();
        let roots = vs_roots(
            &memory,
            &vs_layout(VsFixture::SlotsByAddress),
            VS_CONTEXT,
            &mut diagnostics,
        )
        .unwrap();
        assert!(roots.iter().all(|root| root.base != VS_CONTEXT + 0x1000));
        assert_eq!(diagnostics.len(), 1);
        assert!(diagnostics[0].contains("claims context"));
    }

    /// A zero `SlotMapRef` would alias the context itself; refuse rather than walk it.
    #[test]
    fn test_vs_roots_refuses_an_implausible_slot_map() {
        let mut memory = affinity_fixture();
        memory.put_u16(VS_CONTEXT, 0);
        let mut diagnostics = Vec::new();
        let roots = vs_roots(
            &memory,
            &vs_layout(VsFixture::SlotsByAddress),
            VS_CONTEXT,
            &mut diagnostics,
        )
        .unwrap();
        assert!(roots.is_empty());
        assert_eq!(diagnostics.len(), 1);
        assert!(diagnostics[0].contains("implausible VS slot map"));
    }

    /// Neither shape resolving is a reportable condition, not a silently empty walk — that
    /// ambiguity is exactly what made the 26100 breakage so hard to see.
    #[test]
    fn test_vs_roots_reports_when_neither_shape_resolves() {
        let memory = FlatMemory::new(VS_CONTEXT, 0x100);
        let layout = PoolLayout {
            key: crate::pool::layout::LayoutKey {
                image: crate::dbgeng::KernelImage::default(),
                session: 1,
            },
            globals: HashMap::new(),
            types: HashMap::new(),
        };
        let mut diagnostics = Vec::new();
        let roots = vs_roots(&memory, &layout, VS_CONTEXT, &mut diagnostics).unwrap();
        assert!(roots.is_empty());
        assert_eq!(diagnostics.len(), 1);
        assert!(diagnostics[0].contains("neither the context nor an affinity slot"));
    }

    const SPECIAL_PAGE: u64 = 0xffff_8c8f_13a0_2000;
    /// The byte Driver Verifier fills unused special-pool space with. A *fixture* value now,
    /// not a constant the walker knows: the page records its own fill byte, and the decoder
    /// reads it from there.
    const VERIFIER_FILL: u8 = 0xfd;
    /// `_POOL_HEADER` as Server 26100 x64 declares it: `PreviousSize`/`PoolIndex` share the
    /// USHORT at +0, `BlockSize`/`PoolType` the one at +2, and `Ulong1` overlays all four.
    const X64_POOL_HEADER: PoolHeaderLayout = PoolHeaderLayout {
        size: 0x10,
        previous_size: 0,
        pool_index: 0,
        block_size: 2,
        pool_type: 2,
        tag: 4,
    };

    /// `(usable, size, approximate)` — flattened so the assertions below stay readable and
    /// so every case states whether the size was corroborated or merely bounded.
    fn placed(placement: SpecialPlacement) -> (u64, u64, bool) {
        (placement.usable, placement.size, placement.approximate)
    }

    /// A special-pool page as the kernel writes one: the requested size in the low thirteen
    /// bits of `Ulong1`, the fill byte in `BlockSize`'s, the tag at +4, fill from the header's
    /// end to the block, the block against the page end, and fill again over the alignment
    /// padding behind it.
    fn special_page(requested: u32, tracked: bool) -> Vec<u8> {
        let mut page = vec![VERIFIER_FILL; PAGE_SIZE as usize];
        let word = requested | if tracked { 0x4000 } else { 0 } | (u32::from(VERIFIER_FILL) << 16);
        page[..4].copy_from_slice(&word.to_le_bytes());
        page[4..8].copy_from_slice(b"Tsp1");
        let start = PAGE_SIZE as usize - (requested as usize).next_multiple_of(16);
        page[start..start + requested as usize].fill(0x41);
        page
    }

    /// Decodes a fixture page and places its block, which is the path the walk takes.
    fn place(page: &[u8]) -> SpecialPlacement {
        let header =
            decode_special_pool_header(page, 0, X64_POOL_HEADER).expect("page describes a block");
        special_pool_placement(SPECIAL_PAGE, header, page)
    }

    /// The real values read off Server 26100.32995: a 0x68 message in special pool sits at
    /// page+0xf90, not page+0xf98. Rounding the request up to 16 is what makes the
    /// difference, and getting it wrong reports an address 8 bytes into the block.
    #[test]
    fn test_special_pool_block_is_pushed_against_the_page_end() {
        assert_eq!(
            placed(place(&special_page(0x68, false))),
            (SPECIAL_PAGE + 0xf90, 0x68, false)
        );
        // Already a multiple of 16: no rounding, block ends exactly at the page end.
        assert_eq!(
            placed(place(&special_page(0x40, false))),
            (SPECIAL_PAGE + 0xfc0, 0x40, false)
        );
    }

    /// The bug behind glslang/dbgscope#83, and the reason the size is no longer read as
    /// `_POOL_HEADER.PreviousSize`: eight bits cannot hold 0x140, so that allocation used to
    /// present as a 0x40 one — a value that fits the page perfectly well and that no range
    /// check could tell from a genuine 0x40. Thirteen bits hold anything a page can, so the
    /// two are now different numbers rather than the same one.
    #[test]
    fn test_special_pool_reads_a_size_that_does_not_fit_eight_bits() {
        let page = special_page(0x140, false);
        assert_eq!(
            page[0], 0x40,
            "the low byte alone is the value that used to be read"
        );
        assert_eq!(
            placed(place(&page)),
            (SPECIAL_PAGE + 0xec0, 0x140, false),
            "the block is 0x140 bytes at page+0xec0, not 0x40 bytes at page+0xfc0"
        );
        // The largest block a page can hold, which is where the mask has to stop being generous.
        assert_eq!(
            placed(place(&special_page(0xff0, false))),
            (SPECIAL_PAGE + 0x10, 0xff0, false)
        );
    }

    /// Verifier tracking puts eight bytes of its own between the pool header and the fill, and
    /// the page says so in `Ulong1`. Reading the fill from +0x10 there would compare tracking
    /// data against the fill byte and give up on a page that is perfectly well formed.
    #[test]
    fn test_special_pool_skips_verifiers_tracking_block() {
        let mut page = special_page(0x68, true);
        page[0x10..0x18].copy_from_slice(&0xdead_beef_u64.to_le_bytes());
        let header = decode_special_pool_header(&page, 0, X64_POOL_HEADER).unwrap();
        assert_eq!(header.header_size, 0x18);
        assert_eq!(header.requested, 0x68);
        assert_eq!(header.tag, u32::from_le_bytes(*b"Tsp1"));
        assert_eq!(
            placed(special_pool_placement(SPECIAL_PAGE, header, &page)),
            (SPECIAL_PAGE + 0xf90, 0x68, false)
        );
    }

    /// The corroboration the old check left on the floor: the kernel compares the alignment
    /// padding behind the block against the fill too, and up to fifteen bytes of evidence is
    /// worth having when the alternative is trusting the leading run alone.
    #[test]
    fn test_special_pool_checks_the_padding_behind_the_block() {
        let mut page = special_page(0x68, false);
        page[PAGE_SIZE as usize - 1] = 0x41;
        assert_eq!(
            placed(place(&page)),
            (SPECIAL_PAGE + 0x10, PAGE_SIZE - 0x10, true)
        );
    }

    /// The fill byte comes off the page, so a page filled with something else still resolves
    /// precisely — and one whose fill disagrees with its own recorded byte does not.
    #[test]
    fn test_special_pool_uses_the_fill_byte_the_page_records() {
        let mut page = special_page(0x68, false);
        page[0x10..0xf90].fill(0xc0);
        page[0xff8..].fill(0xc0);
        page[2] = 0xc0;
        assert_eq!(
            placed(place(&page)),
            (SPECIAL_PAGE + 0xf90, 0x68, false),
            "0xfd is Verifier's default, not part of the format"
        );
        page[2] = VERIFIER_FILL;
        assert_eq!(
            placed(place(&page)),
            (SPECIAL_PAGE + 0x10, PAGE_SIZE - 0x10, true)
        );
    }

    /// A start-aligned page (`MmSpecialPoolCatchOverruns` clear) puts the block next to the
    /// preceding guard page, so the end-of-page formula would be wrong. The fill check
    /// declines instead of reporting a bogus address.
    #[test]
    fn test_special_pool_declines_a_start_aligned_page() {
        let mut page = special_page(0x68, false);
        page[0xf90..].fill(VERIFIER_FILL);
        page[0x10..0x78].fill(0x41); // block immediately after the header
        assert_eq!(
            placed(place(&page)),
            (SPECIAL_PAGE + 0x10, PAGE_SIZE - 0x10, true)
        );
    }

    /// A page that describes no block a page could hold is not one we read correctly, and the
    /// walk skips it rather than reporting a span from it.
    #[test]
    fn test_special_pool_header_refuses_an_impossible_block() {
        let mut page = special_page(0x68, false);
        page[..4].copy_from_slice(&(u32::from(VERIFIER_FILL) << 16).to_le_bytes());
        assert!(decode_special_pool_header(&page, 0, X64_POOL_HEADER).is_none());
        // 0x1000 bytes cannot share a 0x1000 byte page with the header in front of them, and
        // the mask reaches 0x1fff, so this is a value the field can really hold.
        page[..4].copy_from_slice(&(0x1000 | (u32::from(VERIFIER_FILL) << 16)).to_le_bytes());
        assert!(decode_special_pool_header(&page, 0, X64_POOL_HEADER).is_none());
        // Tracked, and one alignment step too large to leave room for the tracking block.
        page[..4]
            .copy_from_slice(&(0xfe8 | 0x4000 | (u32::from(VERIFIER_FILL) << 16)).to_le_bytes());
        assert!(decode_special_pool_header(&page, 0, X64_POOL_HEADER).is_none());
    }

    /// A partially readable page corroborates nothing about where the block sits, so the
    /// end-of-page placement must not be used and the size must not exceed what was read.
    #[test]
    fn test_special_pool_fallback_respects_the_readable_length() {
        let mut partial = special_page(0x68, false);
        partial.truncate(0x200);
        assert_eq!(placed(place(&partial)), (SPECIAL_PAGE + 0x10, 0x1f0, true));
    }

    fn special_region(address: u64, pages: usize) -> PoolRegion {
        PoolRegion {
            address,
            size: pages * PAGE_SIZE as usize,
            requested_size: None,
            pool_kind: PoolKind::SpecialNonPagedNx,
            numa_node: 0,
            heap: HeapIdentity {
                pool_state: 0,
                heap: 0,
                special: true,
            },
            subsegment: None,
            backend: PoolBackend::Segment,
            unit_size: PAGE_SIZE as u32,
            bitmap: Vec::new(),
            heap_key: 0,
            pool_header: X64_POOL_HEADER,
            vs_header_size: 0,
            vs_sizes_offset: 0,
            vs_skips_page_end: false,
            known_tag: None,
            states: Vec::new(),
            reusable_chunks: Arc::default(),
            cached_chunks: Arc::default(),
            big_pool: Arc::default(),
        }
    }

    const VS_BASE: u64 = 0x2000;
    const VS_HEAP_KEY: u64 = 0x1234_5678_9abc_def0;

    fn vs_region(size: usize) -> PoolRegion {
        PoolRegion {
            address: VS_BASE,
            size,
            requested_size: None,
            pool_kind: PoolKind::NonPagedNx,
            numa_node: 0,
            heap: HeapIdentity {
                pool_state: 0,
                heap: 0,
                special: false,
            },
            subsegment: Some(VS_BASE),
            backend: PoolBackend::Vs,
            unit_size: 0,
            bitmap: Vec::new(),
            heap_key: VS_HEAP_KEY,
            pool_header: X64_POOL_HEADER,
            vs_header_size: 0x10,
            vs_sizes_offset: 0,
            vs_skips_page_end: false,
            known_tag: None,
            states: Vec::new(),
            reusable_chunks: Arc::default(),
            cached_chunks: Arc::default(),
            big_pool: Arc::default(),
        }
    }

    /// VS chunks laid end to end from [`VS_BASE`]: each carries its `Sizes` word, then the pool
    /// header, then its data. `(size, previous)` are both bytes, and `previous` is written as
    /// given so the chain can be broken on purpose.
    fn vs_extent(chunks: &[(usize, usize)]) -> Vec<u8> {
        let mut bytes = vec![0u8; chunks.iter().map(|(size, _)| size).sum()];
        let mut offset = 0;
        for &(size, previous) in chunks {
            let decoded =
                ((size as u64 / 16) << 16) | ((previous as u64 / 16) << 32) | (1u64 << 48);
            let encoded = decoded ^ VS_HEAP_KEY ^ (VS_BASE + offset as u64);
            bytes[offset..offset + 8].copy_from_slice(&encoded.to_le_bytes());
            let pool = offset + 0x10;
            bytes[pool..pool + 4].copy_from_slice(&0x0001_0000u32.to_le_bytes());
            bytes[pool + 4..pool + 8].copy_from_slice(b"VS!!");
            offset += size;
        }
        bytes
    }

    fn walk_vs_extent(bytes: &[u8]) -> PoolSnapshot {
        walk_vs_region(vs_region(bytes.len()), bytes)
    }

    fn walk_vs_region(region: PoolRegion, bytes: &[u8]) -> PoolSnapshot {
        let memory = FlatMemory::new(VS_BASE, bytes.len());
        let layout = vs_layout(VsFixture::Inline);
        let walker = SnapshotWalker {
            memory: &memory,
            layout: &layout,
            traversal_limit: 1000,
        };
        let mut snapshot = PoolSnapshot {
            complete: true,
            ..PoolSnapshot::default()
        };
        walker.walk_vs(&region, VS_BASE, bytes, Some(VS_BASE), &mut snapshot);
        snapshot
    }

    /// The measured shape of a VS chunk whose header sits at page offset 0xfe0 under
    /// `PageAlignLargeAllocs` ([`PoolRegion::vs_skips_page_end`]): its user block is the next
    /// page start, and the unit before that holds whatever the chunk held while it was free —
    /// on `ctf-vm`, two tree pointers whose top half reads as a tag.
    ///
    /// Three chunks from [`VS_BASE`], the middle one at `0x2fe0` with its pool header at
    /// `0x3000` and the stale pointers at `0x2ff0`, as `dq 0xffffe78459203fe0` read them on
    /// 2026-10-14 (`c1914c0d038da756 ffffe78400000056 | ffffe78459207e18 ffffe78459213bc8 |
    /// 54446d4d03060000`).
    fn vs_extent_with_a_user_block_moved_off_the_page_end() -> Vec<u8> {
        let mut bytes = vs_extent(&[(0xfe0, 0), (0x80, 0xfe0), (0x40, 0x80)]);
        bytes[0xff0..0xff8].copy_from_slice(&0xffff_e784_5920_7e18u64.to_le_bytes());
        bytes[0xff8..0x1000].copy_from_slice(&0xffff_e784_5921_3bc8u64.to_le_bytes());
        bytes[0x1000..0x1004].copy_from_slice(&0x0306_0000u32.to_le_bytes());
        bytes[0x1004..0x1008].copy_from_slice(b"MmDT");
        bytes
    }

    fn span_of_class(snapshot: &PoolSnapshot, size_class: u32) -> &PoolSpan {
        snapshot
            .spans
            .iter()
            .find(|span| span.size_class == size_class)
            .unwrap_or_else(|| panic!("no span of class {size_class:#x}: {:?}", snapshot.spans))
    }

    /// Under the flag, a user block that would begin in a page's last unit begins the next page:
    /// the header is where `!pool` says (`size: 60`, so 0x50 usable), the tag is the real one,
    /// and the chain is untouched — the chunk after is found where the sizes put it.
    #[test]
    fn test_a_vs_user_block_that_would_end_a_page_begins_the_next_one() {
        let bytes = vs_extent_with_a_user_block_moved_off_the_page_end();
        let mut region = vs_region(bytes.len());
        region.vs_skips_page_end = true;

        let snapshot = walk_vs_region(region, &bytes);

        let moved = span_of_class(&snapshot, 0x80);
        assert_eq!(moved.header_address, VS_BASE + 0x1000);
        assert_eq!(moved.usable_address, VS_BASE + 0x1010);
        assert_eq!(moved.size, 0x50);
        assert_eq!(moved.raw_tag, u32::from_le_bytes(*b"MmDT"));
        let after = span_of_class(&snapshot, 0x40);
        assert_eq!(after.header_address, VS_BASE + 0x1060 + 0x10);
        assert_eq!(after.raw_tag, u32::from_le_bytes(*b"VS!!"));
        assert_eq!(snapshot.spans.len(), 3);
        assert!(snapshot.complete, "{:?}", snapshot.diagnostics);
    }

    /// The control, and what the flag gates: without it the same bytes read as they did before
    /// the walk consulted it — the stale pointer as the header, its top half as the tag, a span
    /// a unit early and a unit long. That is what a target **without** the flag would hold, so
    /// it is what the walk has to go on reading there.
    #[test]
    fn test_without_the_flag_a_page_end_user_block_is_read_at_its_chunk_header() {
        let bytes = vs_extent_with_a_user_block_moved_off_the_page_end();

        let snapshot = walk_vs_extent(&bytes);

        let unmoved = span_of_class(&snapshot, 0x80);
        assert_eq!(unmoved.header_address, VS_BASE + 0xff0);
        assert_eq!(unmoved.size, 0x60);
        assert_eq!(unmoved.raw_tag, 0xffff_e784);
    }

    /// The allocator's test is `chunk + 0x20` page-aligned, not `chunk + 0x10`: a user block
    /// that begins exactly on a page boundary is not in the last unit of the page before, and
    /// stays where it is.
    #[test]
    fn test_a_vs_user_block_on_a_page_boundary_is_left_where_it_is() {
        let bytes = vs_extent(&[(0xff0, 0), (0x40, 0xff0)]);
        let mut region = vs_region(bytes.len());
        region.vs_skips_page_end = true;

        let snapshot = walk_vs_region(region, &bytes);

        let on_boundary = span_of_class(&snapshot, 0x40);
        assert_eq!(on_boundary.header_address, VS_BASE + 0x1000);
        assert_eq!(on_boundary.size, 0x20);
        assert_eq!(on_boundary.raw_tag, u32::from_le_bytes(*b"VS!!"));
    }

    /// A chunk at offset 0xfe0 with no unit to skip — two headers' worth, the smallest remainder
    /// a split leaves — keeps its user block in place rather than having it placed at its own
    /// end, which is the next chunk's header; and since the fixture marks every chunk allocated,
    /// which the allocator would never do to one this short under the flag, the walk says so.
    /// Raised by Codex on dbgscope#199.
    #[test]
    fn test_a_vs_chunk_too_short_to_move_its_user_block_is_read_in_place() {
        let bytes = vs_extent(&[(0xfe0, 0), (0x20, 0xfe0), (0x40, 0x20)]);
        let mut region = vs_region(bytes.len());
        region.vs_skips_page_end = true;

        let snapshot = walk_vs_region(region, &bytes);

        let short = span_of_class(&snapshot, 0x20);
        assert_eq!(short.header_address, VS_BASE + 0xff0);
        assert_eq!(short.usable_address, VS_BASE + 0x1000);
        assert_eq!(short.size, 0);
        let after = span_of_class(&snapshot, 0x40);
        assert_eq!(after.header_address, VS_BASE + 0x1010);
        assert_eq!(after.raw_tag, u32::from_le_bytes(*b"VS!!"));
        assert_eq!(snapshot.spans.len(), 3);
        assert!(
            snapshot
                .diagnostics
                .examples()
                .iter()
                .any(|line| line.contains("too short to hold a user block moved off the page end")),
            "{:?}",
            snapshot.diagnostics
        );
    }

    /// A VS chunk too large for `_POOL_HEADER.BlockSize` carries no header, and its tag is in
    /// the big-page table like any other big-pool allocation.
    ///
    /// `BlockSize` is eight bits of sixteen-byte units, so 4080 bytes -- header included -- is
    /// the most it can describe and anything needing a page has no header at all. The two
    /// allocators that produce such a chunk are indistinguishable by their page range
    /// descriptors, which is why the size answers this rather than the backend: measured on a
    /// live 26100 kernel, `!pool` calls `0xffffac09da29f000` an `MiRr` allocation of `0xe1c0`
    /// bytes and the walk called it 57,792 untagged bytes -- the same length, with the name
    /// lost (`windbg-mcp` FOLLOWUPS item 99).
    ///
    /// The fixture writes a *valid* pool header into the big chunk on purpose. Without one the
    /// wrong answer would be an absent tag, which a fixture of zeroes cannot tell from the right
    /// one -- and the tag it writes is the one the small chunk beside it legitimately carries,
    /// so a walk that reads the header anyway reports something entirely plausible.
    #[test]
    fn test_a_vs_chunk_too_large_for_a_pool_header_is_tagged_from_the_big_page_table() {
        // One chunk of 0x2000 -- far past what `BlockSize` can encode -- then an ordinary one.
        let mut bytes = vs_extent(&[(0x2000, 0), (0x40, 0x2000)]);
        let mut region = vs_region(bytes.len());
        // The allocation the table names: page-aligned, as every live entry's `Va` is, and
        // short of the chunk that holds it by the header space `nt` did not need.
        let allocation = (VS_BASE + 0x10).next_multiple_of(PAGE_SIZE);
        region.big_pool = Arc::new(HashMap::from([(
            allocation,
            BigPageEntry {
                tag: u32::from_le_bytes(*b"BIGV"),
                size: 0x1000,
            },
        )]));
        // And a second entry, for an allocation this extent does not hold, so the match has to
        // be by containment rather than by "the region names exactly one".
        Arc::get_mut(&mut region.big_pool)
            .unwrap()
            .insert(VS_BASE + 0x9000, BigPageEntry { tag: 0, size: 0x10 });
        let _ = &mut bytes;
        let memory = FlatMemory::new(VS_BASE, bytes.len());
        let layout = vs_layout(VsFixture::Inline);
        let walker = SnapshotWalker {
            memory: &memory,
            layout: &layout,
            traversal_limit: 1000,
        };
        let mut snapshot = PoolSnapshot {
            complete: true,
            ..PoolSnapshot::default()
        };

        walker.walk_vs(&region, VS_BASE, &bytes, Some(VS_BASE), &mut snapshot);

        let big = snapshot
            .spans
            .iter()
            .find(|span| span.size_class == 0x2000)
            .unwrap_or_else(|| panic!("the large chunk is missing: {:?}", snapshot.spans));
        assert_eq!(super::super::decode::display_tag(big.raw_tag), "BIGV");
        // The table's `Va` is where the allocation begins and `NumberOfBytes` is how long it is;
        // neither is the chunk, and the phantom header is gone from both.
        assert_eq!(big.header_address, allocation);
        assert_eq!(big.usable_address, allocation);
        assert_eq!(big.size, 0x1000);
        // The chunk beside it is under the limit, so it keeps the header it really has.
        let small = snapshot
            .spans
            .iter()
            .find(|span| span.size_class == 0x40)
            .unwrap_or_else(|| panic!("the small chunk is missing: {:?}", snapshot.spans));
        assert_eq!(super::super::decode::display_tag(small.raw_tag), "VS!!");
    }

    /// An entry a chunk could not really hold is not that chunk's, however well it lines up.
    ///
    /// Containment alone is the cheap half of the match: an address inside the chunk. The length
    /// is what says the two are the same allocation, and without it a chunk would adopt the
    /// entry of whatever happens to begin inside it -- reporting somebody else's tag against a
    /// length that does not fit, which is worse than the untagged answer it replaces.
    #[test]
    fn test_a_big_page_entry_too_long_for_its_chunk_is_not_claimed_by_it() {
        let bytes = vs_extent(&[(0x2000, 0), (0x40, 0x2000)]);
        let mut region = vs_region(bytes.len());
        let allocation = (VS_BASE + 0x10).next_multiple_of(PAGE_SIZE);
        region.big_pool = Arc::new(HashMap::from([(
            allocation,
            BigPageEntry {
                tag: u32::from_le_bytes(*b"BIGV"),
                // One byte past what the chunk holding `allocation` can contain.
                size: (VS_BASE + 0x2000 - allocation) + 1,
            },
        )]));
        let memory = FlatMemory::new(VS_BASE, bytes.len());
        let layout = vs_layout(VsFixture::Inline);
        let walker = SnapshotWalker {
            memory: &memory,
            layout: &layout,
            traversal_limit: 1000,
        };
        let mut snapshot = PoolSnapshot {
            complete: true,
            ..PoolSnapshot::default()
        };

        walker.walk_vs(&region, VS_BASE, &bytes, Some(VS_BASE), &mut snapshot);

        let big = snapshot
            .spans
            .iter()
            .find(|span| span.size_class == 0x2000)
            .unwrap_or_else(|| panic!("the large chunk is missing: {:?}", snapshot.spans));
        assert_ne!(super::super::decode::display_tag(big.raw_tag), "BIGV");
        assert_eq!(big.header_address, VS_BASE + 0x10);
    }

    /// glslang/dbgscope#93: "884x rejecting implausible VS chunk size # at #" counted the
    /// *extents* that contained a refusal, because the message was latched behind a bool. How
    /// many chunks were refused was reported nowhere — and a refusal is not free, since the
    /// walk then advances sixteen bytes at a time and decodes every later header in the extent
    /// at a guessed offset.
    #[test]
    fn test_vs_refusals_are_counted_as_chunks_not_as_extents() {
        let mut bytes = vs_extent(&[(0x40, 0), (0x40, 0x40), (0x40, 0x40)]);
        bytes.extend_from_slice(&[0u8; 0x40]); // three more header-sized strides of nothing
        let snapshot = walk_vs_extent(&bytes);

        assert_eq!(snapshot.spans.len(), 3);
        assert_eq!(snapshot.refused_chunks, 3);
        assert!(!snapshot.complete);
        let examples = snapshot.diagnostics.examples();
        assert_eq!(
            examples
                .iter()
                .filter(|message| message.starts_with("refusing VS chunk at"))
                .count(),
            1,
            "the detail is a sample, not one line per refusal: {examples:?}"
        );
        assert!(
            examples.iter().any(|message| message
                == "3 VS chunk headers refused, resynchronising from 0x20c0"),
            "{examples:?}"
        );
    }

    /// One predicate covered two different failures and printed one sentence for both, so a
    /// live run could not say which was happening. Distinct prose keeps them distinct shapes
    /// under diagnostic collapse while the numbers still fold.
    #[test]
    fn test_vs_says_which_check_a_chunk_failed() {
        let refuse = |size: usize, end: u64| {
            let decoded = (size as u64 / 16) << 16;
            decode_vs_chunk(
                decoded ^ VS_HEAP_KEY ^ VS_BASE,
                VS_BASE,
                VS_HEAP_KEY,
                0x20,
                end,
            )
            .unwrap_err()
        };
        assert_eq!(refuse(0, 0x3000).reason, "the size word decodes to zero");
        assert_eq!(
            refuse(0x10, 0x3000).reason,
            "the chunk is smaller than its own headers"
        );
        assert_eq!(
            refuse(0x100, 0x2080).reason,
            "the chunk runs past the end of its subsegment"
        );
        // The word as read travels with the refusal, so a header can be re-decoded by hand
        // against another candidate mix without another walk of the target.
        assert_eq!(
            refuse(0x10, 0x3000).encoded,
            (1u64 << 16) ^ VS_HEAP_KEY ^ VS_BASE
        );
        // A subsegment's last chunk reaches its boundary exactly, and refusing that would drop
        // a well-formed chunk from every subsegment that is fully occupied.
        assert_eq!(
            decode_vs_chunk(
                ((0x80u64 / 16) << 16) ^ VS_HEAP_KEY ^ VS_BASE,
                VS_BASE,
                VS_HEAP_KEY,
                0x20,
                VS_BASE + 0x80,
            )
            .unwrap()
            .size,
            0x80
        );
    }

    /// The corroboration the decoder had in hand and threw away (`let _ = sizes.previous_size`).
    /// It is what tells the two explanations for a refusal apart: a chain that holds either
    /// side of one bad chunk is a header rewritten while we read it, while a chain that never
    /// holds says the stride itself is wrong and the chunks behind it are fiction.
    #[test]
    fn test_vs_reports_a_chunk_that_disagrees_with_its_predecessor() {
        let snapshot = walk_vs_extent(&vs_extent(&[(0x40, 0), (0x40, 0x40), (0x40, 0x20)]));

        assert_eq!(
            snapshot.spans.len(),
            3,
            "the chunk itself is still plausible"
        );
        assert_eq!(snapshot.refused_chunks, 0);
        assert!(!snapshot.complete);
        let examples = snapshot.diagnostics.examples();
        assert!(
            examples.iter().any(|message| message
                == "VS chunk at 0x2080 records a previous size of 0x20 where the chunk before it measured 0x40"),
            "{examples:?}"
        );
        assert!(
            examples
                .iter()
                .any(|message| message.starts_with("1 VS chunks disagreed")),
            "{examples:?}"
        );
        // A chain that holds says nothing at all.
        let quiet = walk_vs_extent(&vs_extent(&[(0x40, 0), (0x40, 0x40), (0x40, 0x40)]));
        assert!(quiet.complete);
        assert!(quiet.diagnostics.is_empty(), "{:?}", quiet.diagnostics);
    }

    /// A memory source with gaps in it, answering `valid_region` two ways the debugger does.
    ///
    /// A **hole** is a page with nothing committed: the query looks past it and names where the
    /// next committed run begins, which the walk records and steps over. A **stall** is the
    /// query naming the address it was given and reporting no length with it — no answer the
    /// walk can advance on, and the shape behind the 3,285 `valid-region query made no
    /// progress` lines a live 26100 walk emitted.
    struct HoleyMemory {
        base: u64,
        bytes: Vec<u8>,
        holes: HashSet<u64>,
        stalls: HashSet<u64>,
        /// Pages where the query answers `0x0+0x0` — the engine's way of saying it found
        /// nothing valid in the whole span, as opposed to `stalls`, which model it finding
        /// something and failing to size it. The two look alike at the call site and mean
        /// opposite things about what lies behind them.
        blind: HashSet<u64>,
        /// What the *memory manager* says, which is a separate axis from every field above:
        /// those decide what reads, this decides whether there was anything there to read.
        ///
        /// `None` is a source that cannot answer — the debugger against a kernel target, or a
        /// dump recording no memory information — and is the default, so every test written
        /// before this existed keeps its old meaning. `Some` names the pages with no pages
        /// behind them.
        uncommitted: Option<HashSet<u64>>,
        queries: Cell<usize>,
        commit_queries: Cell<usize>,
    }

    impl HoleyMemory {
        fn new(base: u64, bytes: Vec<u8>) -> Self {
            Self {
                base,
                bytes,
                holes: HashSet::new(),
                stalls: HashSet::new(),
                blind: HashSet::new(),
                uncommitted: None,
                queries: Cell::new(0),
                commit_queries: Cell::new(0),
            }
        }

        /// Give this source a memory manager, telling it which pages hold nothing.
        fn with_memory_manager(mut self, uncommitted: &[u64]) -> Self {
            self.uncommitted = Some(uncommitted.iter().copied().collect());
            self
        }

        fn end(&self) -> u64 {
            self.base + self.bytes.len() as u64
        }

        fn committed(&self, page: u64) -> bool {
            page >= self.base
                && page < self.end()
                && !self.holes.contains(&page)
                && !self.stalls.contains(&page)
                && !self.blind.contains(&page)
        }
    }

    impl PoolMemory for HoleyMemory {
        fn read_exact(&self, address: u64, size: usize) -> Result<Vec<u8>, SnapshotError> {
            let offset = (address - self.base) as usize;
            self.bytes
                .get(offset..offset + size)
                .map(<[u8]>::to_vec)
                .ok_or_else(|| SnapshotError::InvalidData {
                    detail: format!("read past the fixture at {address:#x}+{size:#x}"),
                })
        }

        fn valid_region(&self, address: u64, size: usize) -> Result<(u64, usize), SnapshotError> {
            self.queries.set(self.queries.get() + 1);
            let page = address & !(PAGE_SIZE - 1);
            if self.blind.contains(&page) {
                return Ok((0, 0));
            }
            if self.stalls.contains(&page) {
                return Ok((address, 0));
            }
            let mut next = page;
            while next < self.end() && !self.committed(next) {
                next += PAGE_SIZE;
            }
            if next >= self.end() {
                // Nothing committed from here on, so the engine names a base past the span
                // that was asked about. That is the region *ending*, not a failure to advance.
                return Ok((self.end(), 0));
            }
            let base = next.max(address);
            let mut run = next;
            while self.committed(run) {
                run += PAGE_SIZE;
            }
            Ok((base, (run.min(address + size as u64) - base) as usize))
        }

        fn interrupted(&self) -> Result<bool, SnapshotError> {
            Ok(false)
        }

        /// Answers in **runs**, as the memory manager does: like pages are merged, so a walk
        /// stepping a gap of any length costs one query per run and not one per page. The
        /// counter is what lets a test say so.
        fn committed_run(&self, address: u64) -> Option<CommitRun> {
            let uncommitted = self.uncommitted.as_ref()?;
            self.commit_queries.set(self.commit_queries.get() + 1);
            let mut page = address & !(PAGE_SIZE - 1);
            let committed = !uncommitted.contains(&page);
            loop {
                page += PAGE_SIZE;
                if page >= self.end() {
                    // Past the fixture the whole rest of the address space is one run of the
                    // same kind, which keeps a run always able to advance the caller.
                    return Some(CommitRun {
                        end: u64::MAX,
                        committed,
                    });
                }
                if !uncommitted.contains(&page) != committed {
                    return Some(CommitRun {
                        end: page,
                        committed,
                    });
                }
            }
        }
    }

    /// Three special-pool pages, so each committed page the walk reaches shows up as exactly
    /// one span at a known address — which is what makes "did it walk past the gap" checkable.
    fn special_pages(count: usize) -> Vec<u8> {
        let mut bytes = Vec::new();
        for _ in 0..count {
            bytes.extend_from_slice(&special_page(0x68, false));
        }
        bytes
    }

    fn walk_holey(memory: &HoleyMemory, region: &PoolRegion) -> PoolSnapshot {
        let layout = vs_layout(VsFixture::Inline);
        let walker = SnapshotWalker {
            memory,
            layout: &layout,
            traversal_limit: 1000,
        };
        let mut snapshot = PoolSnapshot {
            complete: true,
            ..PoolSnapshot::default()
        };
        walker.walk_region(region, &mut snapshot).unwrap();
        snapshot
    }

    /// glslang/dbgscope#94. One page the query cannot advance over used to cost every
    /// committed page behind it: the walk filed the rest of the region as unreadable and
    /// `break`ed, while the sibling case one branch above — a query reporting a *gap* —
    /// recorded the hole and carried on. The samples show these firing part-way through
    /// regions tens of pages long, so what was given up on was not a tail.
    #[test]
    fn test_a_stalled_query_costs_a_page_not_the_region() {
        let mut memory = HoleyMemory::new(SPECIAL_PAGE, special_pages(4));
        memory.stalls.insert(SPECIAL_PAGE + 0x1000);
        let snapshot = walk_holey(&memory, &special_region(SPECIAL_PAGE, 4));

        let allocated: Vec<_> = snapshot
            .spans
            .iter()
            .filter(|span| span.state == PoolState::Allocated)
            .map(|span| span.usable_address)
            .collect();
        assert_eq!(
            allocated,
            [
                SPECIAL_PAGE + 0xf90,
                SPECIAL_PAGE + 0x2f90,
                SPECIAL_PAGE + 0x3f90
            ],
            "the pages behind the stalled one must still be walked"
        );
        assert!(
            snapshot
                .spans
                .iter()
                .any(|span| span.state == PoolState::Unreadable
                    && span.header_address == SPECIAL_PAGE + 0x1000
                    && span.size == PAGE_SIZE),
            "the page that stalled is still filed as unreadable"
        );
        assert!(!snapshot.complete);
        assert_eq!(
            snapshot.stalls,
            WalkStalls {
                pages: 1,
                skipped_bytes: PAGE_SIZE,
                // Two pages of committed memory on the far side of the stall — what the old
                // `break` reported as nothing at all.
                recovered_bytes: 2 * PAGE_SIZE,
            }
        );
    }

    /// A region begins wherever the allocator put it — `…de6030` and `…e02130` in the live
    /// samples — so a stall can land at any page offset. Advancing a page's *worth of bytes*
    /// from there writes off the tail of the page that stalled and the head of the page after
    /// it, discarding a healthy page in order to skip a bad one: this branch's own mistake, in
    /// miniature. The other stall tests miss it because their region is page-aligned.
    #[test]
    fn test_a_stall_part_way_through_a_page_does_not_swallow_the_next_one() {
        let mut memory = HoleyMemory::new(SPECIAL_PAGE, special_pages(3));
        memory.stalls.insert(SPECIAL_PAGE);
        // Starting 0x30 bytes short of the second page, as an unaligned region does.
        let mut region = special_region(SPECIAL_PAGE, 3);
        region.address = SPECIAL_PAGE + 0xfd0;
        region.size = 3 * PAGE_SIZE as usize - 0xfd0;
        let snapshot = walk_holey(&memory, &region);

        assert_eq!(
            snapshot.stalls.skipped_bytes, 0x30,
            "only the rest of the page that stalled is written off"
        );
        let allocated: Vec<_> = snapshot
            .spans
            .iter()
            .filter(|span| span.state == PoolState::Allocated)
            .map(|span| span.usable_address)
            .collect();
        assert_eq!(
            allocated,
            [SPECIAL_PAGE + 0x1f90, SPECIAL_PAGE + 0x2f90],
            "the page immediately after the stall is healthy and must still be walked"
        );
    }

    /// The gap spans of a walk, as `(state, address, size)`, in address order.
    fn gaps(snapshot: &PoolSnapshot) -> Vec<(PoolState, u64, u64)> {
        let mut gaps: Vec<_> = snapshot
            .spans
            .iter()
            .filter(|span| !span.state.is_chunk())
            .map(|span| (span.state, span.header_address, span.size))
            .collect();
        gaps.sort_by_key(|&(_, address, _)| address);
        gaps
    }

    /// `windbg-mcp` FOLLOWUPS item 98. The same page that does not read is a hole in the walk's
    /// coverage or is nothing at all, and **only the memory manager can say which** — so this
    /// runs one fixture past three memory managers and changes nothing else.
    ///
    /// The three answers are the whole rule. A source that says the page holds nothing files
    /// [`PoolState::Uncommitted`] and leaves `complete` alone; one that says the page is
    /// committed files [`PoolState::Unreadable`] and clears it; and one that cannot say keeps
    /// the second reading, because *unknown* has to fall on the conservative side or the
    /// excuse is being granted by the absence of evidence.
    ///
    /// Measured live before it was written: on `sihost` (26200, 4 Segment Heaps, 19,459
    /// chunks, 2026-09-24) every one of the 45 unreadable spans holding the walk at `Partial`
    /// was `MEM_RESERVE`, and both controls — an allocated chunk and a free one — were
    /// `MEM_COMMIT`.
    #[test]
    fn test_only_the_memory_manager_decides_whether_a_gap_costs_coverage() {
        let gap = SPECIAL_PAGE + PAGE_SIZE;
        let walk = |memory: HoleyMemory| {
            let mut memory = memory;
            memory.holes.insert(gap);
            let snapshot = walk_holey(&memory, &special_region(SPECIAL_PAGE, 4));
            // The pages on either side are walked whatever the answer is: this changes how a
            // gap is *described*, never how much is read.
            assert_eq!(
                snapshot
                    .spans
                    .iter()
                    .filter(|span| span.state == PoolState::Allocated)
                    .count(),
                3,
                "the gap is one page of four"
            );
            (snapshot, memory.commit_queries.get())
        };

        let (nothing_there, queries) =
            walk(HoleyMemory::new(SPECIAL_PAGE, special_pages(4)).with_memory_manager(&[gap]));
        assert_eq!(
            gaps(&nothing_there),
            [(PoolState::Uncommitted, gap, PAGE_SIZE)],
            "a page with nothing behind it is not a hole in what the walk saw"
        );
        assert!(
            nothing_there.complete,
            "and so the walk covered everything it set out to"
        );
        assert_eq!(
            queries, 1,
            "asked per run, not per page — a reserved tail of any length costs one query"
        );

        let (really_missed, _) =
            walk(HoleyMemory::new(SPECIAL_PAGE, special_pages(4)).with_memory_manager(&[]));
        assert_eq!(
            gaps(&really_missed),
            [(PoolState::Unreadable, gap, PAGE_SIZE)],
            "a committed page that will not read is memory the target has and we did not see"
        );
        assert!(!really_missed.complete);

        let (cannot_say, _) = walk(HoleyMemory::new(SPECIAL_PAGE, special_pages(4)));
        assert_eq!(
            gaps(&cannot_say),
            gaps(&really_missed),
            "a source with no answer keeps the conservative reading, not the convenient one"
        );
        assert!(!cannot_say.complete);
    }

    /// The diagnostic that goes with the gap, which is about *missed memory* and so has to be
    /// as conditional as the coverage is.
    ///
    /// It fired on every reserved tail before this, which on a live user-mode heap is all of
    /// them — a line per gap describing nothing wrong, and wrong about what it describes
    /// ("unreadable space" for address space that holds nothing). The same failure mode as
    /// glslang/dbgscope#94, where 3,285 such lines drowned out the diagnostics that meant
    /// something.
    #[test]
    fn test_a_gap_with_nothing_behind_it_is_not_worth_a_diagnostic() {
        let gap = SPECIAL_PAGE + PAGE_SIZE;
        let complains = |memory: HoleyMemory| {
            let mut memory = memory;
            memory.holes.insert(gap);
            let snapshot = walk_holey(&memory, &special_region(SPECIAL_PAGE, 4));
            snapshot
                .diagnostics
                .examples()
                .iter()
                .any(|message| message.contains("only committed through"))
        };

        assert!(
            !complains(
                HoleyMemory::new(SPECIAL_PAGE, special_pages(4)).with_memory_manager(&[gap])
            ),
            "nothing was missed, so there is nothing to report"
        );
        assert!(
            complains(HoleyMemory::new(SPECIAL_PAGE, special_pages(4)).with_memory_manager(&[])),
            "a committed page that would not read is still worth saying"
        );
        assert!(
            complains(HoleyMemory::new(SPECIAL_PAGE, special_pages(4))),
            "and so is one nothing could be asked about"
        );
    }

    /// A gap that crosses the boundary is **split**, because calling the whole of it by either
    /// name misreports the other half — and the half that is really missed still costs the
    /// walk its `complete`.
    ///
    /// This is the shape a live target actually produces: a subsegment's committed pages end
    /// part-way through the range that holds it, so the tail of a committed run and the
    /// reserved space after it reach the walk as one failed read.
    #[test]
    fn test_a_gap_straddling_the_commit_boundary_is_split_at_it() {
        let mut memory = HoleyMemory::new(SPECIAL_PAGE, special_pages(4))
            // Only the second of the two pages that will not read holds nothing.
            .with_memory_manager(&[SPECIAL_PAGE + 2 * PAGE_SIZE]);
        memory.holes.insert(SPECIAL_PAGE + PAGE_SIZE);
        memory.holes.insert(SPECIAL_PAGE + 2 * PAGE_SIZE);
        let snapshot = walk_holey(&memory, &special_region(SPECIAL_PAGE, 4));

        assert_eq!(
            gaps(&snapshot),
            [
                (PoolState::Unreadable, SPECIAL_PAGE + PAGE_SIZE, PAGE_SIZE),
                (
                    PoolState::Uncommitted,
                    SPECIAL_PAGE + 2 * PAGE_SIZE,
                    PAGE_SIZE
                ),
            ],
            "two spans, each the size of what it describes"
        );
        assert!(
            !snapshot.complete,
            "half of it was memory the target has, and that half is still missing"
        );
    }

    /// A memory manager whose answer cannot advance the walk, which is the one way this loop
    /// could hang rather than misreport.
    ///
    /// Not hypothetical: `walk_region` already defends against the engine naming a region
    /// *behind* the cursor, and the same answer reaches here. The rest of the span is filed
    /// the conservative way and the loop ends.
    ///
    /// **Deleting the guard hangs this test rather than failing it** — verified 2026-09-24,
    /// which is the point: without it a walk that met such an answer would never return, and
    /// a wedged engine process is the one failure this module cannot report its way out of.
    #[test]
    fn test_a_commit_run_that_cannot_advance_is_no_answer_at_all() {
        struct Stuck(HoleyMemory);
        impl PoolMemory for Stuck {
            fn read_exact(&self, address: u64, size: usize) -> Result<Vec<u8>, SnapshotError> {
                self.0.read_exact(address, size)
            }
            fn valid_region(
                &self,
                address: u64,
                size: usize,
            ) -> Result<(u64, usize), SnapshotError> {
                self.0.valid_region(address, size)
            }
            fn interrupted(&self) -> Result<bool, SnapshotError> {
                Ok(false)
            }
            fn committed_run(&self, address: u64) -> Option<CommitRun> {
                // Names the address it was given, so a caller that trusted it would ask the
                // same question forever.
                Some(CommitRun {
                    end: address,
                    committed: false,
                })
            }
        }

        let mut inner = HoleyMemory::new(SPECIAL_PAGE, special_pages(4));
        inner.holes.insert(SPECIAL_PAGE + PAGE_SIZE);
        let memory = Stuck(inner);
        let layout = vs_layout(VsFixture::Inline);
        let walker = SnapshotWalker {
            memory: &memory,
            layout: &layout,
            traversal_limit: 1000,
        };
        let mut snapshot = PoolSnapshot {
            complete: true,
            ..PoolSnapshot::default()
        };
        walker
            .walk_region(&special_region(SPECIAL_PAGE, 4), &mut snapshot)
            .unwrap();

        assert_eq!(
            gaps(&snapshot),
            [(PoolState::Unreadable, SPECIAL_PAGE + PAGE_SIZE, PAGE_SIZE)],
            "an answer that cannot advance is discarded, not acted on"
        );
        assert!(!snapshot.complete);
    }

    /// A VS subsegment with a hole in it, walked in two committed extents.
    ///
    /// `bytes` tiles the whole region with chunks the way the allocator does; `holes` names the
    /// pages `valid_region` will report as uncommitted. That is what
    /// `RtlpHpVsSubsegmentCommitPages` produces on a real target — it commits and decommits page
    /// ranges anywhere inside the subsegment and tracks them in `_HEAP_VS_SUBSEGMENT.CommitBitmap`
    /// — and it is the shape the old walk had no way to survive.
    fn walk_vs_with_holes(chunks: &[(usize, usize)], holes: &[u64]) -> PoolSnapshot {
        walk_vs_holes(chunks, holes, false)
    }

    /// `walk_vs_with_holes`, plus a memory manager that confirms the holes hold nothing —
    /// which is what a live target's does, and what the walk needs before it can treat a
    /// chunk running into one as fully understood.
    fn walk_vs_holes(chunks: &[(usize, usize)], holes: &[u64], asked: bool) -> PoolSnapshot {
        let bytes = vs_extent(chunks);
        let region = vs_region(bytes.len());
        let mut memory = HoleyMemory::new(VS_BASE, bytes);
        memory.holes.extend(holes.iter().copied());
        if asked {
            memory = memory.with_memory_manager(holes);
        }
        walk_holey(&memory, &region)
    }

    /// glslang/dbgscope#103: 106,516 VS chunk headers refused on one live 26100 walk, ~196 per
    /// extent, every one of them failing the same subsegment-bound check — which read as a bound
    /// that was wrong. It was not. Nothing was wrong with any of the three predicates: the walk
    /// was handing them a *guess*, because it started every committed extent at that extent's own
    /// first sixteen-byte boundary, and only the first extent of a subsegment begins on a chunk.
    ///
    /// Here the chunk before the hole is 0x2800 bytes and lands the next header at 0x5800, in the
    /// middle of the page the walk resumes on. Starting at 0x5000 costs 128 refusals to scan back
    /// onto it; following the chain costs none.
    #[test]
    fn test_a_vs_extent_after_a_hole_resumes_on_the_chunk_the_chain_names() {
        let snapshot = walk_vs_with_holes(
            &[
                (0x800, 0),
                (0x800, 0x800),
                // Spans the decommitted page, as the free chunk whose interior was decommitted
                // always does, and ends part-way into the page after it.
                (0x2800, 0x800),
                (0x800, 0x2800),
            ],
            &[VS_BASE + 0x2000],
        );

        assert_eq!(
            snapshot.refused_chunks,
            0,
            "the chain names the header, so nothing has to be scanned for: {:?}",
            snapshot.diagnostics.examples()
        );
        assert_eq!(snapshot.unplaced_bytes, 0);
        let allocated: Vec<_> = snapshot
            .spans
            .iter()
            .filter(|span| span.state == PoolState::Allocated)
            .map(|span| span.header_address)
            .collect();
        // The chunk that spans the hole is not among them: its header was read and understood,
        // but the walk emits no span for a chunk it could not read to the end of.
        assert_eq!(
            allocated,
            [VS_BASE + 0x10, VS_BASE + 0x810, VS_BASE + 0x3810],
            "the chunk on the far side of the hole is found, and nothing else is invented"
        );
    }

    /// The other half of the same rule, and the price of it. When the header the chain names
    /// falls *inside* the hole, the walk has no way to know where a chunk begins in the extent
    /// after it — so it decodes none of it and says how much that cost. The extent here holds two
    /// perfectly good chunks; declining them loses coverage, and decoding from a guess would put
    /// whatever a garbage word decoded to into `spans` as an allocation, which is worse.
    #[test]
    fn test_a_vs_extent_whose_chunk_boundary_fell_in_the_hole_is_not_guessed_at() {
        let snapshot = walk_vs_with_holes(
            &[
                (0x1000, 0),
                (0x1000, 0x1000),
                (0x1000, 0x1000),
                (0x1000, 0x1000),
            ],
            &[VS_BASE + 0x2000],
        );

        assert_eq!(snapshot.refused_chunks, 0);
        assert_eq!(
            snapshot.unplaced_bytes, 0x1000,
            "the whole undecodable extent is sized, not just noted"
        );
        assert!(
            snapshot
                .diagnostics
                .examples()
                .iter()
                .any(|message| message.contains("does not begin on a chunk boundary")),
            "{:?}",
            snapshot.diagnostics.examples()
        );
        let allocated: Vec<_> = snapshot
            .spans
            .iter()
            .filter(|span| span.state == PoolState::Allocated)
            .map(|span| span.header_address)
            .collect();
        assert_eq!(
            allocated,
            [VS_BASE + 0x10, VS_BASE + 0x1010],
            "nothing from the extent the walk could not place reaches the snapshot"
        );
    }

    /// An extent entirely inside one chunk. Nothing to decode there and nothing lost: the
    /// expectation still names a header further on, so this must not be billed as coverage the
    /// walk gave up — which is the difference between a hole in a large free chunk (ordinary) and
    /// a chain the walk dropped (not).
    #[test]
    fn test_an_extent_inside_a_single_chunk_costs_nothing() {
        let snapshot = walk_vs_with_holes(&[(0x1000, 0), (0x3000, 0x1000)], &[VS_BASE + 0x2000]);

        assert_eq!(snapshot.unplaced_bytes, 0);
        assert_eq!(snapshot.refused_chunks, 0);
        assert!(
            !snapshot
                .diagnostics
                .examples()
                .iter()
                .any(|message| message.contains("does not begin on a chunk boundary")),
            "{:?}",
            snapshot.diagnostics.examples()
        );
    }

    /// The chunk in the test above **is** billed as coverage the walk gave up, and that is the
    /// other half of `windbg-mcp` FOLLOWUPS item 98: a free chunk whose middle the allocator
    /// decommitted runs past the committed extent, so no span was emitted for it and
    /// `complete` cleared — silently, with no diagnostic saying which chunk or why.
    ///
    /// A span is geometry and state, and both are known here: the header was read, the size
    /// came out of it and passed the subsegment bound, and the state comes from the free tree.
    /// The only thing the walk could not see is the chunk's *contents*, which no span carries
    /// — so where the tail holds nothing the chunk is emitted like any other.
    ///
    /// The control is the same fixture with no memory manager, and it must still refuse: a
    /// tail the walk cannot ask about may be memory the target has, and a chunk that large
    /// with an unread middle is a real hole in what was seen.
    ///
    /// On `sihost` (26200, 2026-09-24) this was the last thing between a healthy live walk and
    /// `Complete` — five chunks, 0x1660 to 0xfe70 bytes, after all 48 gap spans had been
    /// classified.
    #[test]
    fn test_a_chunk_running_into_nothing_is_still_a_chunk() {
        let chunks = [(0x1000usize, 0usize), (0x3000, 0x1000)];
        let holes = [VS_BASE + 0x2000];

        let asked = walk_vs_holes(&chunks, &holes, true);
        assert_eq!(
            asked
                .spans
                .iter()
                .filter(|span| span.state.is_chunk())
                .map(|span| (span.header_address, span.size))
                .collect::<Vec<_>>(),
            [(VS_BASE + 0x10, 0xfe0), (VS_BASE + 0x1010, 0x2fe0)],
            "the chunk that spans the hole is understood, so it is reported"
        );
        assert!(
            asked.complete,
            "and nothing about it is unknown, so the walk covered everything: {:?}",
            asked.diagnostics.examples()
        );

        let unasked = walk_vs_holes(&chunks, &holes, false);
        assert_eq!(
            unasked
                .spans
                .iter()
                .filter(|span| span.state.is_chunk())
                .map(|span| span.header_address)
                .collect::<Vec<_>>(),
            [VS_BASE + 0x10],
            "with no answer about the tail the chunk is not invented"
        );
        assert!(!unasked.complete);
        assert!(
            unasked
                .diagnostics
                .examples()
                .iter()
                .any(|message| message.contains("runs")
                    && message.contains("past the committed extent")),
            "and the walk says which chunk it dropped: {:?}",
            unasked.diagnostics.examples()
        );
    }

    /// The paged-out half of `windbg-mcp` FOLLOWUPS item 99, VS side. A big-pool chunk's tag
    /// and length are the table's and its state is the free tree's; its own pages were never
    /// going to be read. So a tail in a page the memory manager has trimmed takes nothing from
    /// the span — and the walk used to drop the chunk for that tail and file its pages as an
    /// untagged gap, which is what `pool_chunk` then answered for the allocation's own
    /// address. Measured on `ctf-vm` (26100.33438, 2026-10-07): `CIcr` at 0xffffa9099c86f000,
    /// 0x20 into a 0x12c0-byte chunk with a resident header and a trimmed tail, `!pool` naming
    /// it and this walk answering `....`, 4096 bytes, `unreadable`.
    ///
    /// The control is the same fixture with no table entry, and it must still drop the chunk:
    /// without the table, a chunk that large with an unread tail is a real hole in what was
    /// seen, and a span over it would be the walk asserting what it did not read.
    #[test]
    fn test_a_big_pool_chunk_whose_tail_is_paged_out_is_still_named_from_the_table() {
        let chunks = [(0x2000usize, 0usize), (0x40, 0x2000)];
        let hole = VS_BASE + 0x1000;
        let allocation = hole;
        let walk = |named: bool| {
            let bytes = vs_extent(&chunks);
            let mut region = vs_region(bytes.len());
            if named {
                region.big_pool = Arc::new(HashMap::from([(
                    allocation,
                    BigPageEntry {
                        tag: u32::from_le_bytes(*b"BIGV"),
                        size: 0x1000,
                    },
                )]));
            }
            let mut memory = HoleyMemory::new(VS_BASE, bytes);
            memory.holes.insert(hole);
            // Committed, so the hole is memory the target has and cannot read — a trimmed
            // page, not a decommitted interior.
            let memory = memory.with_memory_manager(&[]);
            walk_holey(&memory, &region)
        };

        let named = walk(true);
        let big = named
            .spans
            .iter()
            .find(|span| span.size_class == 0x2000)
            .unwrap_or_else(|| panic!("the chunk the table names is missing: {:?}", named.spans));
        assert_eq!(super::super::decode::display_tag(big.raw_tag), "BIGV");
        assert_eq!(big.header_address, allocation);
        assert_eq!(big.usable_address, allocation);
        assert_eq!(big.size, 0x1000);
        assert_eq!(big.state, PoolState::Allocated);
        assert!(
            !named
                .spans
                .iter()
                .any(|span| span.state == PoolState::Unreadable),
            "the trimmed page is inside a span the index holds, so it is not also a gap: {:?}",
            named.spans
        );
        assert!(
            named.spans.iter().any(|span| span.size_class == 0x40),
            "the chain crosses the hole to the chunk after it: {:?}",
            named.spans
        );
        assert_eq!(named.unplaced_bytes, 0);
        assert!(
            named.complete,
            "nothing about the chunk was unknown: {:?}",
            named.diagnostics.examples()
        );

        let unnamed = walk(false);
        assert!(
            !unnamed.spans.iter().any(|span| span.size_class == 0x2000),
            "without the table the chunk is not invented: {:?}",
            unnamed.spans
        );
        assert!(
            unnamed
                .spans
                .iter()
                .any(|span| span.state == PoolState::Unreadable && span.header_address == hole),
            "and its trimmed page is filed as the gap it is: {:?}",
            unnamed.spans
        );
        assert!(!unnamed.complete);
    }

    /// The same half, segment side. Discovery names an allocated page range from the table,
    /// and there is nothing on its pages to add: `walk_page_ranges` emits the one span and
    /// skips the header decode. Reading the pages anyway cost the allocation when they were
    /// trimmed — the range was filed as an untagged gap and the name went unused (`Gcac` at
    /// 0xffffa4b05e1b5000, lab 29671, 2026-10-07) — and cost every resident one a read whose
    /// bytes were then discarded: 111 MB of them on that kernel, summed over the live entries.
    /// So a named range is answered without asking the memory at all, which the query count
    /// pins; the control is the same range unnamed, which has to be read and cannot be.
    #[test]
    fn test_a_page_range_the_table_names_is_answered_without_reading_it() {
        let size = 0x2000usize;
        let walk = |named: bool| {
            let mut region = vs_region(size);
            region.backend = PoolBackend::Segment;
            region.unit_size = size as u32;
            region.states = vec![PoolState::Allocated];
            region.pool_header.size = 0;
            region.known_tag = named.then_some(u32::from_le_bytes(*b"BIGP"));
            let mut memory = HoleyMemory::new(VS_BASE, vec![0; size]);
            memory.holes.extend([VS_BASE, VS_BASE + 0x1000]);
            let memory = memory.with_memory_manager(&[]);
            let snapshot = walk_holey(&memory, &region);
            (snapshot, memory.queries.get())
        };

        let (named, queries) = walk(true);
        assert_eq!(queries, 0, "the table answered; the pages were not asked");
        let [span] = named.spans.as_slice() else {
            panic!("one span for one allocation: {:?}", named.spans);
        };
        assert_eq!(super::super::decode::display_tag(span.raw_tag), "BIGP");
        assert_eq!(span.header_address, VS_BASE);
        assert_eq!(span.usable_address, VS_BASE);
        assert_eq!(span.size, size as u64);
        assert_eq!(span.state, PoolState::Allocated);
        assert!(named.complete);

        let (unnamed, queries) = walk(false);
        assert!(queries > 0, "an unnamed range has to be read");
        assert!(
            unnamed
                .spans
                .iter()
                .all(|span| span.state == PoolState::Unreadable),
            "and when it cannot be, that is all the walk can say: {:?}",
            unnamed.spans
        );
        assert!(!unnamed.complete);
    }

    /// glslang/dbgscope#104, settled on live 26100: every stall the walk met answered `0x0+0x0`,
    /// and the eight verbatim samples were eight *consecutive* pages — one dead region stepping
    /// itself into the give-up bound, paying eight KD round trips to be told eight times what the
    /// first answer already said. `0x0+0x0` is not a region at zero; it is the engine reporting
    /// nothing valid in the span it was asked about, which is a statement about every byte
    /// through to the end of it.
    ///
    /// So it ends the region, exactly like the sibling answer that names a base past the end.
    /// The distinction that matters is with [`Self::test_a_stalled_query_costs_a_page_not_the_region`]:
    /// a query that reports a region and cannot size it has said nothing about what lies behind
    /// it, and there the page step stays.
    #[test]
    fn test_a_query_that_finds_nothing_ends_the_region_rather_than_stepping() {
        let mut memory = HoleyMemory::new(SPECIAL_PAGE, special_pages(8));
        for page in 4..8 {
            memory.blind.insert(SPECIAL_PAGE + page * PAGE_SIZE);
        }
        let snapshot = walk_holey(&memory, &special_region(SPECIAL_PAGE, 8));

        assert_eq!(
            memory.queries.get(),
            2,
            "one query to walk the committed run, one to be told nothing follows it — a page \
             step would have cost four more and learnt nothing"
        );
        assert_eq!(
            snapshot.stalls,
            WalkStalls::default(),
            "nothing was stepped over, so nothing is billed as stepped over"
        );
        assert!(
            !snapshot
                .diagnostics
                .examples()
                .iter()
                .any(|message| message.contains("made no progress")),
            "a region running out of readable content is not a fault: {:?}",
            snapshot.diagnostics.examples()
        );
        // Still accounted for, and still incomplete: those pages were not walked, and the walk
        // says so rather than reporting the region as fully covered.
        assert!(!snapshot.complete);
        let unreadable: u64 = snapshot
            .spans
            .iter()
            .filter(|span| span.state == PoolState::Unreadable)
            .map(|span| span.size)
            .sum();
        assert_eq!(unreadable, 4 * PAGE_SIZE);
        let allocated = snapshot
            .spans
            .iter()
            .filter(|span| span.state == PoolState::Allocated)
            .count();
        assert_eq!(
            allocated, 4,
            "the readable pages in front of it are still walked"
        );
    }

    /// The reason the `break` was there in the first place: the advance has to be
    /// unconditional or a query that keeps answering the same way spins forever. Unconditional
    /// *and* bounded — a region that is dead all the way down costs a fixed number of queries,
    /// not one per page of its length, which on a live KD link is the cost that matters.
    #[test]
    fn test_a_region_that_never_advances_is_given_up_on() {
        let pages = 4096;
        let mut memory = HoleyMemory::new(SPECIAL_PAGE, vec![0u8; pages * PAGE_SIZE as usize]);
        for page in 0..pages as u64 {
            memory.stalls.insert(SPECIAL_PAGE + page * PAGE_SIZE);
        }
        let snapshot = walk_holey(&memory, &special_region(SPECIAL_PAGE, pages));

        assert_eq!(
            memory.queries.get(),
            MAX_CONSECUTIVE_STALLS as usize,
            "a dead region costs the limit in queries, and not one more"
        );
        assert!(!snapshot.complete);
        assert_eq!(snapshot.stalls.pages, u64::from(MAX_CONSECUTIVE_STALLS));
        // Every byte of it is accounted for as unreadable, whether stepped over or written off.
        let unreadable: u64 = snapshot
            .spans
            .iter()
            .filter(|span| span.state == PoolState::Unreadable)
            .map(|span| span.size)
            .sum();
        assert_eq!(unreadable, pages as u64 * PAGE_SIZE);
    }

    /// The other half of the fix, and the larger half of the 3,285: a region simply running
    /// out of committed pages arrived at the same branch and was reported as a failure to
    /// advance. Nothing is behind it to lose — the hole branch has already filed the tail —
    /// so there is nothing to report.
    #[test]
    fn test_a_region_running_out_of_committed_pages_is_not_a_stall() {
        let mut memory = HoleyMemory::new(SPECIAL_PAGE, special_pages(4));
        memory.holes.insert(SPECIAL_PAGE + 0x3000);
        let snapshot = walk_holey(&memory, &special_region(SPECIAL_PAGE, 4));

        assert_eq!(snapshot.stalls, WalkStalls::default());
        assert!(
            !snapshot
                .diagnostics
                .examples()
                .iter()
                .any(|message| message.contains("made no progress")),
            "{:?}",
            snapshot.diagnostics.examples()
        );
        assert!(
            snapshot
                .diagnostics
                .examples()
                .iter()
                .any(|message| message.contains("only committed through")),
            "the tail is still reported as the unreadable space it is"
        );
        assert_eq!(
            snapshot
                .spans
                .iter()
                .filter(|span| span.state == PoolState::Allocated)
                .count(),
            3
        );
    }

    /// Special pool is page-granular, so one page that does not decode says nothing about the
    /// next. Abandoning the region there — which is what this did before glslang/dbgscope#86 —
    /// dropped every later allocation *and* left `complete` set, so the truncated snapshot was
    /// the one cached and re-served.
    #[test]
    fn test_a_bad_special_pool_page_costs_that_page_only() {
        let mut bytes = Vec::new();
        bytes.extend_from_slice(&special_page(0x68, false));
        bytes.extend_from_slice(&vec![0u8; PAGE_SIZE as usize]); // decodes as nothing
        bytes.extend_from_slice(&special_page(0x140, false));
        let region = special_region(SPECIAL_PAGE, 3);
        let memory = FlatMemory::new(SPECIAL_PAGE, bytes.len());
        let layout = vs_layout(VsFixture::Inline);
        let walker = SnapshotWalker {
            memory: &memory,
            layout: &layout,
            traversal_limit: 1000,
        };
        let mut snapshot = PoolSnapshot {
            complete: true,
            ..PoolSnapshot::default()
        };
        walker.walk_special_pool(&region, SPECIAL_PAGE, &bytes, &mut snapshot);

        let found: Vec<_> = snapshot
            .spans
            .iter()
            .map(|span| (span.usable_address, span.size))
            .collect();
        assert_eq!(
            found,
            [
                (SPECIAL_PAGE + 0xf90, 0x68),
                (SPECIAL_PAGE + 0x2000 + 0xec0, 0x140)
            ],
            "the page after the bad one must still be walked"
        );
        assert!(!snapshot.complete, "a skipped page is not a complete walk");
        assert!(
            snapshot
                .diagnostics
                .examples()
                .iter()
                .any(|message| message.contains(&format!("{:#x}", SPECIAL_PAGE + 0x1000))),
            "the diagnostic has to name the page that was skipped: {:?}",
            snapshot.diagnostics.examples()
        );
    }
    use crate::pool::{
        decode::{
            DESCRIPTOR_FLAG_ALLOCATED, DESCRIPTOR_FLAG_FIRST, DESCRIPTOR_FLAG_SUBSEGMENT,
            DESCRIPTOR_FLAG_VS,
        },
        layout::{LayoutKey, TypeLayout},
    };

    /// `RangeFlags` for an in-use page range: allocated, and the range's first descriptor.
    /// The kernel's own free path insists on both before it looks at anything else.
    const RANGE_IN_USE: u8 = DESCRIPTOR_FLAG_ALLOCATED | DESCRIPTOR_FLAG_FIRST;
    /// An LFH subsegment range, as `nt!RtlpHpFreeHeap` requires it: `0x0b`.
    const RANGE_LFH: u8 = RANGE_IN_USE | DESCRIPTOR_FLAG_SUBSEGMENT;
    /// A VS subsegment range: `0x0f`.
    const RANGE_VS: u8 = RANGE_LFH | DESCRIPTOR_FLAG_VS;

    const K: u64 = 0xffff_8000_0000_0000;
    const STATE: u64 = K + 0x10_0000;
    const GLOBALS: u64 = K + 0x11_0000;
    const BIG_TABLE_POINTER: u64 = K + 0x12_0000;
    const BIG_TABLE_COUNT: u64 = K + 0x12_0010;
    const HEAP: u64 = K + 0x20_0000;
    const POOL_NODE: u64 = STATE + 0x40;
    const DYNAMIC_LOOKASIDE: u64 = POOL_NODE + 0x20;
    const SEGMENT: u64 = K + 0x30_0000;
    const LARGE_META: u64 = K + 0x80_0000;
    const LARGE_VA: u64 = K + 0x90_0000;
    const BIG_TABLE: u64 = K + 0xa0_0000;

    /// Sparse synthetic memory, stored as coalesced contiguous runs.
    ///
    /// Both the read and the write path are shaped to keep Miri off per-byte work, and the two
    /// were found in separate rounds.
    ///
    /// Reads resolve through a binary search over runs, not the `BTreeMap<u64, u8>` this began
    /// as with one `get` per byte. Semantics are unchanged — a byte is readable iff it was
    /// written, so a read spanning a gap still fails — and that was worth 2.0x.
    ///
    /// Writes go through [`Writes`], a log replayed in bulk, rather than building that byte map
    /// at all. The read fix left the two pool snapshot tests at 398s and 369s of an 849s CI job,
    /// and the second of them is three assertions over a fixture — so what remained was never
    /// "Miri interpreting the walk", which is what the previous round of this comment concluded.
    /// It was the ~22k `BTreeMap` inserts `fill`/`put` performed to build the fixture in the
    /// first place, each a fully interpreted, borrow-checked trip through `NodeRef` and raw
    /// pointers. Bulk `copy_from_slice` over a handful of runs is one Miri operation per write.
    ///
    /// Measured, not assumed: on one machine that second test alone took 178.4s before the
    /// write fix; both tests together now take 6.1s.
    struct SyntheticMemory {
        /// `(start, bytes)` for each contiguous written range, sorted by `start` and never
        /// overlapping or abutting.
        runs: Vec<(u64, Vec<u8>)>,
        holes: Vec<(u64, u64)>,
    }

    /// The fixture's write log: `(address, bytes)` in the order written, later writes winning.
    ///
    /// A plain append-only `Vec`, one entry per `put`/`fill` call rather than per byte, so the
    /// whole fixture is a few dozen pushes and bulk copies instead of ~22k `BTreeMap` inserts.
    #[derive(Default)]
    struct Writes(Vec<(u64, Vec<u8>)>);

    impl SyntheticMemory {
        /// Overwrites one byte inside a range an earlier write already covers.
        fn poke(&mut self, address: u64, byte: u8) {
            let (start, data) = self
                .runs
                .iter_mut()
                .find(|(start, data)| address >= *start && address < *start + data.len() as u64)
                .expect("the address is inside a written run");
            data[(address - *start) as usize] = byte;
        }

        fn new(writes: Writes, holes: Vec<(u64, u64)>) -> Self {
            // Extents first: merge every written interval that overlaps or abuts a neighbour,
            // which yields exactly the runs the byte map used to produce by iterating in key
            // order. Sorting is safe here because an extent carries no value to be overwritten.
            let mut extents: Vec<(u64, u64)> = writes
                .0
                .iter()
                .filter(|(_, data)| !data.is_empty())
                .map(|(address, data)| (*address, address + data.len() as u64))
                .collect();
            extents.sort_unstable();
            let mut runs: Vec<(u64, Vec<u8>)> = Vec::new();
            for (start, end) in extents {
                match runs.last_mut() {
                    Some((run_start, data)) if start <= *run_start + data.len() as u64 => {
                        let len = (end - *run_start) as usize;
                        // A contained write leaves the run as it is; only a write reaching past
                        // the current end grows it.
                        if len > data.len() {
                            data.resize(len, 0);
                        }
                    }
                    _ => runs.push((start, vec![0; (end - start) as usize])),
                }
            }
            // Then replay in write order, so the last write to an address wins — the property
            // that re-inserting into the byte map used to provide.
            for (address, data) in writes.0.into_iter().filter(|(_, data)| !data.is_empty()) {
                let index = runs
                    .partition_point(|(start, _)| *start <= address)
                    .checked_sub(1)
                    .expect("every non-empty write contributed an extent");
                let (start, run) = &mut runs[index];
                let offset = (address - *start) as usize;
                run[offset..offset + data.len()].copy_from_slice(&data);
            }
            Self { runs, holes }
        }

        /// The run containing `address`, if any.
        fn run_at(&self, address: u64) -> Option<&(u64, Vec<u8>)> {
            let index = self
                .runs
                .partition_point(|(start, _)| *start <= address)
                .checked_sub(1)?;
            let run = &self.runs[index];
            (address < run.0 + run.1.len() as u64).then_some(run)
        }

        /// Whether a single byte was written — the fixture's own sanity checks ask this.
        fn contains(&self, address: u64) -> bool {
            self.run_at(address).is_some()
        }
    }

    struct ShortMemory;

    struct BigPageMemory {
        table: Vec<u8>,
        count: usize,
        read_calls: Cell<usize>,
        interrupt_checks: Cell<usize>,
        interrupt_after_checks: Option<usize>,
    }

    impl PoolMemory for ShortMemory {
        fn read_exact(&self, _address: u64, _size: usize) -> Result<Vec<u8>, SnapshotError> {
            Ok(Vec::new())
        }

        fn valid_region(&self, address: u64, size: usize) -> Result<(u64, usize), SnapshotError> {
            Ok((address, size))
        }

        fn interrupted(&self) -> Result<bool, SnapshotError> {
            Ok(false)
        }
    }

    impl PoolMemory for BigPageMemory {
        fn read_exact(&self, address: u64, size: usize) -> Result<Vec<u8>, SnapshotError> {
            self.read_calls.set(self.read_calls.get() + 1);
            if address == BIG_TABLE_POINTER && size == 8 {
                return Ok(BIG_TABLE.to_le_bytes().to_vec());
            }
            if address == BIG_TABLE_COUNT && size == 4 {
                return Ok((self.count as u32).to_le_bytes().to_vec());
            }
            if address == BIG_TABLE_COUNT && size == 8 {
                return Ok(((1u64 << 32) | self.count as u64).to_le_bytes().to_vec());
            }
            let offset = address
                .checked_sub(BIG_TABLE)
                .and_then(|value| usize::try_from(value).ok());
            if let Some(bytes) = offset
                .and_then(|offset| offset.checked_add(size).map(|end| (offset, end)))
                .and_then(|(offset, end)| self.table.get(offset..end))
            {
                return Ok(bytes.to_vec());
            }
            Err(SnapshotError::Read {
                address,
                size,
                source: Box::new(std::io::Error::other("sparse big-page memory")),
            })
        }

        fn valid_region(&self, address: u64, size: usize) -> Result<(u64, usize), SnapshotError> {
            Ok((address, size))
        }

        fn interrupted(&self) -> Result<bool, SnapshotError> {
            let checks = self.interrupt_checks.get();
            self.interrupt_checks.set(checks + 1);
            Ok(self
                .interrupt_after_checks
                .is_some_and(|limit| checks >= limit))
        }
    }

    impl PoolMemory for SyntheticMemory {
        fn read_exact(&self, address: u64, size: usize) -> Result<Vec<u8>, SnapshotError> {
            let unreadable = || SnapshotError::Read {
                address,
                size,
                source: Box::new(std::io::Error::other("sparse synthetic memory")),
            };
            let (start, data) = self.run_at(address).ok_or_else(unreadable)?;
            let offset = (address - start) as usize;
            // One run must cover the whole read: runs never abut, so a read that runs off the
            // end of one has hit a gap, exactly as the per-byte lookup used to report.
            data.get(offset..offset + size)
                .map(<[u8]>::to_vec)
                .ok_or_else(unreadable)
        }

        fn valid_region(&self, address: u64, size: usize) -> Result<(u64, usize), SnapshotError> {
            let end = address.saturating_add(size as u64);
            for &(hole_start, hole_end) in &self.holes {
                if address >= hole_start && address < hole_end {
                    return Ok((hole_end, end.saturating_sub(hole_end) as usize));
                }
                if address < hole_start && end > hole_start {
                    return Ok((address, hole_start.saturating_sub(address) as usize));
                }
            }
            Ok((address, size))
        }

        fn interrupted(&self) -> Result<bool, SnapshotError> {
            Ok(false)
        }
    }

    fn type_layout(size: u32, fields: &[(&'static str, u32)]) -> TypeLayout {
        TypeLayout {
            size,
            fields: fields.iter().copied().collect(),
        }
    }

    fn synthetic_layout() -> PoolLayout {
        let mut types = HashMap::new();
        types.insert(
            "_EX_POOL_HEAP_MANAGER_STATE",
            type_layout(
                0x200,
                &[
                    ("HeapManager", 8),
                    ("PoolNode", 0x40),
                    ("NumberOfPools", 0),
                    ("SpecialHeaps", 0x20),
                ],
            ),
        );
        types.insert(
            "_EX_HEAP_POOL_NODE",
            type_layout(0x120, &[("Heaps", 0), ("Lookasides", 0x20)]),
        );
        types.insert(
            "_SEGMENT_HEAP",
            type_layout(
                0x800,
                &[
                    ("SegContexts", 0x100),
                    ("VsContext", 0x300),
                    ("LfhContext", 0x380),
                    ("LargeAllocMetadata", 0x400),
                ],
            ),
        );
        types.insert(
            "_HEAP_SEG_CONTEXT",
            type_layout(
                0x80,
                &[
                    ("SegmentListHead", 0),
                    ("FreePageRanges", 0x10),
                    ("UnitShift", 0x20),
                    ("FirstDescriptorIndex", 0x21),
                    ("SegmentMask", 0x28),
                    ("PagesPerUnitShift", 0x30),
                ],
            ),
        );
        types.insert(
            "_HEAP_PAGE_SEGMENT",
            type_layout(
                0x100,
                &[("ListEntry", 0), ("Signature", 0x20), ("DescArray", 0x100)],
            ),
        );
        types.insert(
            "_HEAP_PAGE_RANGE_DESCRIPTOR",
            type_layout(
                0x20,
                &[
                    ("UnitSize", 0x1f),
                    ("RangeFlags", 0x18),
                    ("TreeNode", 0),
                    ("TreeSignature", 0),
                ],
            ),
        );
        types.insert(
            "_HEAP_VS_CONTEXT",
            type_layout(
                0x40,
                &[
                    ("FreeChunkTree", 0),
                    ("DelayFreeContext", 0x10),
                    ("SubsegmentList", 0x30),
                    ("Config", 0x38),
                ],
            ),
        );
        types.insert(
            "_HEAP_VS_DELAY_FREE_CONTEXT",
            type_layout(0x10, &[("ListHead", 0)]),
        );
        types.insert(
            "_HEAP_VS_SUBSEGMENT",
            type_layout(0xfe0, &[("Signature", 0), ("Size", 2), ("ListEntry", 8)]),
        );
        types.insert(
            "_HEAP_VS_CHUNK_HEADER",
            type_layout(0x10, &[("Sizes", 0), ("EncodedSegmentPageOffset", 8)]),
        );
        types.insert(
            "_HEAP_VS_CHUNK_FREE_HEADER",
            type_layout(0x20, &[("TreeNode", 8)]),
        );
        types.insert("_HEAP_LFH_CONTEXT", type_layout(0x20, &[("Buckets", 0)]));
        types.insert(
            "_HEAP_LFH_SUBSEGMENT",
            type_layout(
                0x20,
                &[
                    ("BlockOffsets", 0),
                    ("BlockCount", 4),
                    ("BlockBitmap", 8),
                    ("ListEntry", 0x10),
                ],
            ),
        );
        types.insert(
            "_HEAP_LFH_SUBSEGMENT_ENCODED_OFFSETS",
            type_layout(4, &[("EncodedData", 0)]),
        );
        types.insert(
            "_RTLP_HP_HEAP_GLOBALS",
            type_layout(0x10, &[("HeapKey", 0), ("LfhKey", 8)]),
        );
        types.insert(
            "_RTL_RB_TREE",
            type_layout(0x10, &[("Root", 0), ("Encoded", 8)]),
        );
        types.insert(
            "_RTL_BALANCED_NODE",
            type_layout(0x10, &[("Left", 0), ("Right", 8)]),
        );
        types.insert(
            "_RTL_DYNAMIC_LOOKASIDE",
            type_layout(0x80, &[("BucketCount", 8), ("Buckets", 0x40)]),
        );
        types.insert(
            "_RTL_LOOKASIDE",
            type_layout(0x40, &[("ListHead", 0), ("Size", 0x30)]),
        );
        types.insert(
            "_SLIST_HEADER",
            type_layout(0x10, &[("Alignment", 0), ("Region", 8)]),
        );
        types.insert(
            "_POOL_HEADER",
            type_layout(
                0x10,
                &[
                    ("PreviousSize", 0),
                    ("PoolIndex", 0),
                    ("BlockSize", 2),
                    ("PoolType", 2),
                    ("PoolTag", 8),
                ],
            ),
        );
        types.insert(
            "_HEAP_LARGE_ALLOC_DATA",
            type_layout(
                0x28,
                &[
                    ("TreeNode", 0),
                    ("VirtualAddress", 0x18),
                    ("AllocatedPages", 0x20),
                ],
            ),
        );
        types.insert(
            "_POOL_TRACKER_BIG_PAGES",
            type_layout(0x20, &[("Va", 0), ("Key", 8), ("NumberOfBytes", 0x10)]),
        );
        PoolLayout {
            key: LayoutKey {
                image: crate::dbgeng::KernelImage {
                    base: K,
                    ..crate::dbgeng::KernelImage::default()
                },
                session: 1,
            },
            globals: [
                ("ExPoolState", STATE),
                ("RtlpHpHeapGlobals", GLOBALS),
                ("PoolBigPageTable", BIG_TABLE_POINTER),
                ("PoolBigPageTableSize", BIG_TABLE_COUNT),
            ]
            .into_iter()
            .collect(),
            types,
        }
    }

    fn put(bytes: &mut Writes, address: u64, data: &[u8]) {
        bytes.0.push((address, data.to_vec()));
    }

    fn fill(bytes: &mut Writes, address: u64, size: usize) {
        bytes.0.push((address, vec![0; size]));
    }

    fn put_u16(bytes: &mut Writes, address: u64, value: u16) {
        put(bytes, address, &value.to_le_bytes());
    }

    fn put_u32(bytes: &mut Writes, address: u64, value: u32) {
        put(bytes, address, &value.to_le_bytes());
    }

    fn put_u64(bytes: &mut Writes, address: u64, value: u64) {
        put(bytes, address, &value.to_le_bytes());
    }

    fn packed_slist_next(entry: u64) -> u64 {
        (entry << 4) | 3
    }

    fn pool_header(bytes: &mut Writes, address: u64, tag: &[u8; 4]) {
        fill(bytes, address, 0x10);
        put(bytes, address, &[1, 0, 4, 1]);
        put(bytes, address + 8, tag);
    }

    /// The fixture, with its VS subsegment declaring the chunk area its page range actually
    /// leaves — `(2 pages - sizeof(_HEAP_VS_SUBSEGMENT rounded)) / 16`, which is what
    /// `RtlpHpVsSubsegmentCreate` writes. Kept in step deliberately: the walk now cross-checks
    /// the two, so a fixture that disagreed would put a complaint in every other test's output
    /// and hide the one case that should raise it.
    fn synthetic_memory() -> SyntheticMemory {
        synthetic_memory_declaring((0x2000 - 0xfe0) / 16)
    }

    fn synthetic_memory_declaring(declared: u16) -> SyntheticMemory {
        let mut bytes = Writes::default();
        fill(&mut bytes, STATE, 0x200);
        put_u32(&mut bytes, STATE, 1);
        for heap_index in 0..4 {
            put_u64(&mut bytes, STATE + 0x40 + heap_index * 8, HEAP);
        }
        fill(&mut bytes, GLOBALS, 0x10);
        let heap_key = 0x55aa_1234_9876_0000;
        let lfh_key = 0xa5c3_1357;
        put_u64(&mut bytes, GLOBALS, heap_key);
        put_u64(&mut bytes, GLOBALS + 8, lfh_key);
        fill(&mut bytes, HEAP, 0x800);

        let context = HEAP + 0x100;
        let list_head = context;
        put_u64(&mut bytes, list_head, SEGMENT);
        put_u64(&mut bytes, context + 0x10, SEGMENT + 0x100 + 6 * 0x20);
        put(&mut bytes, context + 0x20, &[12, 1]);
        put_u64(&mut bytes, context + 0x28, !0xffffu64);

        fill(&mut bytes, SEGMENT, 0x100);
        put_u64(&mut bytes, SEGMENT, list_head);
        put_u64(
            &mut bytes,
            SEGMENT + 0x20,
            SEGMENT ^ context ^ heap_key ^ super::super::decode::PAGE_SEGMENT_SIGNATURE,
        );
        fill(&mut bytes, SEGMENT + 0x100, 16 * 0x20);
        for (index, units, flags) in [
            (1u64, 2u8, RANGE_LFH),
            (3, 2, RANGE_VS),
            (5, 1, RANGE_IN_USE),
            (6, 1, 0x00),
            (7, 1, RANGE_IN_USE),
        ] {
            let descriptor = SEGMENT + 0x100 + index * 0x20;
            if flags & DESCRIPTOR_FLAG_FIRST != 0 {
                put_u32(
                    &mut bytes,
                    descriptor,
                    super::super::decode::DESCRIPTOR_TREE_SIGNATURE,
                );
            }
            put(&mut bytes, descriptor + 0x18, &[flags]);
            put(&mut bytes, descriptor + 0x1f, &[units]);
        }

        let lfh = SEGMENT + 0x1000;
        fill(&mut bytes, lfh, 0x20);
        let decoded_offsets = u32::from(0x40u16) | (u32::from(0x40u16) << 16);
        put_u32(
            &mut bytes,
            lfh,
            decoded_offsets ^ lfh_key as u32 ^ (lfh >> 12) as u32,
        );
        put_u16(&mut bytes, lfh + 4, 4);
        put(&mut bytes, lfh + 8, &[0x49]);
        fill(&mut bytes, lfh + 0x40, 0x100);
        pool_header(&mut bytes, lfh + 0x40, b"LFH!");
        pool_header(&mut bytes, lfh + 0x80, b"LFHC");
        pool_header(&mut bytes, lfh + 0xc0, b"LFHF");
        pool_header(&mut bytes, lfh + 0x100, b"LFH2");

        let vs = SEGMENT + 0x3000;
        fill(&mut bytes, vs, 0x2000);
        put_u16(&mut bytes, vs, 0x8000 | (0x2bed ^ declared));
        put_u16(&mut bytes, vs + 2, declared);
        let first_chunk = vs + 0xfe0;
        let cached_chunk = first_chunk + 0x40;
        let free_chunk = cached_chunk + 0x40;
        for (address, allocated) in [
            (first_chunk, true),
            (cached_chunk, false),
            (free_chunk, false),
        ] {
            let decoded = (4u64 << 16) | (u64::from(allocated) << 48);
            put_u64(&mut bytes, address, decoded ^ address ^ heap_key);
        }
        pool_header(&mut bytes, first_chunk + 0x10, b"VS!!");
        let vs_context = HEAP + 0x300;
        put_u64(&mut bytes, vs_context, free_chunk + 8);
        put_u64(&mut bytes, free_chunk + 8, 0);
        put_u64(&mut bytes, free_chunk + 16, 0);
        let delay_head = vs_context + 0x10;
        put_u16(&mut bytes, delay_head, 1);
        put_u64(
            &mut bytes,
            delay_head + 8,
            packed_slist_next(cached_chunk + 0x20),
        );
        put_u64(&mut bytes, cached_chunk + 0x20, 0);

        fill(&mut bytes, DYNAMIC_LOOKASIDE, 0x80);
        put_u32(&mut bytes, DYNAMIC_LOOKASIDE + 8, 1);
        let lookaside_chunk = free_chunk + 0x40;
        let decoded = 4u64 << 16;
        put_u64(
            &mut bytes,
            lookaside_chunk,
            decoded ^ lookaside_chunk ^ heap_key,
        );
        let lookaside_head = DYNAMIC_LOOKASIDE + 0x40;
        put_u16(&mut bytes, lookaside_head, 1);
        put_u64(
            &mut bytes,
            lookaside_head + 8,
            packed_slist_next(lookaside_chunk + 0x20),
        );
        put_u64(&mut bytes, lookaside_chunk + 0x20, 0);
        put_u32(&mut bytes, lookaside_head + 0x30, 0x40);

        fill(&mut bytes, SEGMENT + 0x5000, 0x1000);
        pool_header(&mut bytes, SEGMENT + 0x5000, b"SEGM");
        fill(&mut bytes, SEGMENT + 0x6000, 0x1000);
        fill(&mut bytes, SEGMENT + 0x7000, 0x800);
        pool_header(&mut bytes, SEGMENT + 0x7000, b"SPRS");

        let large_tree = HEAP + 0x400;
        put_u64(&mut bytes, large_tree, LARGE_META ^ large_tree);
        put(&mut bytes, large_tree + 8, &[1]);
        fill(&mut bytes, LARGE_META, 0x28);
        put_u64(&mut bytes, LARGE_META + 0x18, LARGE_VA | 0x1800);
        put_u64(&mut bytes, LARGE_META + 0x20, (2u64 << 12) | 0x5a5);
        // Large allocations are described entirely by allocator metadata; leave
        // their payload absent so the walker cannot accidentally read it.
        put_u64(&mut bytes, BIG_TABLE_POINTER, BIG_TABLE);
        put_u64(&mut bytes, BIG_TABLE_COUNT, 4);
        fill(&mut bytes, BIG_TABLE, 4 * 0x20);
        let first = super::super::decode::big_page_hash(LARGE_VA, 4).unwrap();
        let adjacent = (first + 1) % 4;
        let collision = BIG_TABLE + first as u64 * 0x20;
        put_u64(&mut bytes, collision, LARGE_VA + 0x10_0000);
        let entry = BIG_TABLE + adjacent as u64 * 0x20;
        put_u64(&mut bytes, entry, LARGE_VA);
        put(&mut bytes, entry + 8, b"BIG!");
        put_u64(&mut bytes, entry + 0x10, 0x1800);

        SyntheticMemory::new(bytes, vec![(SEGMENT + 0x7800, SEGMENT + 0x8000)])
    }

    fn big_page_memory(address: u64, count: usize, collision_distance: usize) -> BigPageMemory {
        let mut table = vec![0; count * 0x20];
        let first = super::super::decode::big_page_hash(address, count).unwrap();
        for distance in 0..collision_distance {
            let index = (first + distance) % count;
            let offset = index * 0x20;
            table[offset..offset + 8]
                .copy_from_slice(&(address + (distance as u64 + 1) * 0x10_0000).to_le_bytes());
        }
        let index = (first + collision_distance) % count;
        let offset = index * 0x20;
        table[offset..offset + 8].copy_from_slice(&address.to_le_bytes());
        table[offset + 8..offset + 12].copy_from_slice(b"BTCH");
        table[offset + 0x10..offset + 0x18].copy_from_slice(&0x9000u64.to_le_bytes());
        BigPageMemory {
            table,
            count,
            read_calls: Cell::new(0),
            interrupt_checks: Cell::new(0),
            interrupt_after_checks: None,
        }
    }

    /// glslang/dbgscope#103, settled on a live 26100 kernel: the page range holding a VS
    /// subsegment is **larger than the subsegment**, by one unit every time, so a walk bounded by
    /// the descriptor decodes a page that holds no chunks and refuses a header every sixteen
    /// bytes across it. The subsegment's own `Size` is the chunk area, and it is only reachable
    /// through a signature check that would have rejected a misread — so it is preferred, and
    /// clamped so it can never point the walk outside the range it was given.
    #[test]
    fn test_a_vs_subsegment_is_bounded_by_its_own_declared_size() {
        let complaint = "declares";
        let quiet = walk_synthetic(synthetic_memory());
        assert!(
            !quiet
                .diagnostics
                .examples()
                .iter()
                .any(|message| message.contains(complaint)),
            "a subsegment sized like the real ones draws no complaint: {:?}",
            quiet.diagnostics.examples()
        );
        let all = vs_chunks(&quiet);

        // The live shape: chunks end before the range does. What sits past the declared end is
        // not part of this subsegment and must not be decoded — which is the entire finding.
        let bounded = walk_synthetic(synthetic_memory_declaring(8));
        // `first` is the fixture's `_HEAP_VS_SUBSEGMENT` size rounded up, and eight units past it
        // is where the chunk area now ends — well short of the page range, exactly as a real one
        // does. Nothing at or beyond that address is part of this subsegment.
        let declared_end = SEGMENT + 0x3000 + 0xfe0 + 8 * 16;
        assert!(
            vs_chunks(&bounded) < all,
            "bounding the subsegment has to cost the chunks past its end: {:?}",
            bounded.diagnostics.examples()
        );
        assert!(
            snapshot_vs_spans(&bounded).all(|span| span.header_address < declared_end),
            "nothing past {declared_end:#x} may be decoded: {:?}",
            snapshot_vs_spans(&bounded)
                .map(|span| span.header_address)
                .collect::<Vec<_>>()
        );
        assert!(
            snapshot_vs_spans(&quiet).any(|span| span.header_address >= declared_end),
            "and the check is only worth anything if the wider bound did reach past it"
        );
        assert!(
            !bounded
                .diagnostics
                .examples()
                .iter()
                .any(|message| message.contains(complaint)),
            "and that is the ordinary shape, not something to complain about once per \
             subsegment: {:?}",
            bounded.diagnostics.examples()
        );

        // Neither of these can be the chunk area, and both would be silent damage if clamped
        // without a word: no chunks at all, and more than the range can hold.
        for declared in [0, 0x200] {
            let bogus = walk_synthetic(synthetic_memory_declaring(declared));
            assert!(
                bogus
                    .diagnostics
                    .examples()
                    .iter()
                    .any(|message| message.contains(complaint)),
                "declared {declared:#x}: {:?}",
                bogus.diagnostics.examples()
            );
            assert_eq!(
                vs_chunks(&bogus),
                all,
                "an unusable declaration falls back to the descriptor's bound"
            );
        }
    }

    fn snapshot_vs_spans(snapshot: &PoolSnapshot) -> impl Iterator<Item = &PoolSpan> {
        snapshot
            .spans
            .iter()
            .filter(|span| span.backend == PoolBackend::Vs && span.state != PoolState::Unreadable)
    }

    fn vs_chunks(snapshot: &PoolSnapshot) -> usize {
        snapshot_vs_spans(snapshot).count()
    }

    fn walk_synthetic(memory: SyntheticMemory) -> PoolSnapshot {
        let layout = synthetic_layout();
        SnapshotWalker {
            memory: &memory,
            layout: &layout,
            traversal_limit: 1024,
        }
        .walk(None)
        .unwrap()
    }

    #[test]
    fn test_pool_snapshot_walks_all_backends() {
        let memory = synthetic_memory();
        assert!(!memory.contains(LARGE_VA));
        let layout = synthetic_layout();
        let walker = SnapshotWalker {
            memory: &memory,
            layout: &layout,
            traversal_limit: 1024,
        };
        let snapshot = walker.walk(None).unwrap();
        for backend in [
            PoolBackend::Lfh,
            PoolBackend::Vs,
            PoolBackend::Segment,
            PoolBackend::Large,
        ] {
            assert!(
                snapshot.spans.iter().any(|span| span.backend == backend),
                "missing {backend:?}"
            );
        }
        assert!(snapshot.spans.iter().any(|span| span.pool_kind.is_paged()));
        assert!(snapshot.spans.iter().any(|span| {
            span.raw_tag == u32::from_le_bytes(*b"LFH!") && span.state == PoolState::Allocated
        }));
        assert!(snapshot.spans.iter().any(|span| {
            span.backend == PoolBackend::Vs && span.state == PoolState::CachedFree
        }));
        assert!(snapshot.spans.iter().any(|span| {
            span.backend == PoolBackend::Vs
                && span.header_address == SEGMENT + 0x40b0
                && span.state == PoolState::CachedFree
        }));
        assert!(
            snapshot
                .spans
                .iter()
                .filter(|span| span.backend == PoolBackend::Vs)
                .all(|span| span.size_class == 0x40)
        );
        assert!(snapshot.spans.iter().any(|span| {
            span.backend == PoolBackend::Segment && span.state == PoolState::ReusableFree
        }));
        assert!(
            snapshot
                .spans
                .iter()
                .any(|span| span.state == PoolState::Unreadable)
        );
        assert!(snapshot.spans.iter().any(|span| {
            span.backend == PoolBackend::Vs
                && span.header_address == SEGMENT + 0x3ff0
                && span.usable_address == SEGMENT + 0x4000
                && span.raw_tag == u32::from_le_bytes(*b"VS!!")
        }));
        assert!(snapshot.spans.iter().any(|span| {
            span.backend == PoolBackend::Large
                && span.raw_tag == u32::from_le_bytes(*b"BIG!")
                && span.size == 0x1800
        }));
        assert!(
            snapshot
                .diagnostics
                .examples()
                .iter()
                .any(|message| { message.contains("per-session paged heaps are not included") })
        );
        assert!(
            snapshot
                .diagnostics
                .examples()
                .iter()
                .any(|message| message.contains("only committed through"))
        );
    }

    // ---- the walk budget --------------------------------------------------------------

    /// The synthetic fixture, with a budget that runs out after a fixed number of reads.
    ///
    /// The real budget is wall-clock, which is untestable without making the test a race
    /// against the machine it runs on. Reads are the honest stand-in: they are what the
    /// budget exists to bound — over a live KD link every one crosses the wire — and their
    /// count is exactly what differs between a dump that walks in a second and a live kernel
    /// that takes minutes.
    struct Impatient<M> {
        inner: M,
        reads: Cell<usize>,
        allowance: usize,
    }

    impl Impatient<SyntheticMemory> {
        fn new(allowance: usize) -> Self {
            Self::over(synthetic_memory(), allowance)
        }
    }

    impl<M> Impatient<M> {
        fn over(inner: M, allowance: usize) -> Self {
            Self {
                inner,
                reads: Cell::new(0),
                allowance,
            }
        }
    }

    impl<M: PoolMemory> PoolMemory for Impatient<M> {
        fn read_exact(&self, address: u64, size: usize) -> Result<Vec<u8>, SnapshotError> {
            self.reads.set(self.reads.get() + 1);
            self.inner.read_exact(address, size)
        }

        fn valid_region(&self, address: u64, size: usize) -> Result<(u64, usize), SnapshotError> {
            self.inner.valid_region(address, size)
        }

        fn interrupted(&self) -> Result<bool, SnapshotError> {
            Ok(false)
        }

        fn out_of_budget(&self) -> bool {
            self.reads.get() >= self.allowance
        }
    }

    fn walk_impatiently(allowance: usize) -> PoolSnapshot {
        let memory = Impatient::new(allowance);
        let layout = synthetic_layout();
        SnapshotWalker {
            memory: &memory,
            layout: &layout,
            traversal_limit: 1024,
        }
        .walk(None)
        .expect("running out of budget is an outcome, not an error")
    }

    /// The wedge this budget exists to prevent: with nothing to stop it, the walk ran on
    /// past the caller's timeout and every later call to that target queued behind it. So
    /// the deadline has to produce an *answer* — a snapshot that says what it reached and
    /// admits what it did not. An `Err` would discard the work and say nothing.
    #[test]
    fn test_a_walk_that_runs_out_of_budget_reports_what_it_reached() {
        let snapshot = walk_impatiently(40);

        assert!(
            !snapshot.complete,
            "a truncated walk that claims completeness is the lie this whole API guards against"
        );
        assert!(
            snapshot
                .diagnostics
                .examples()
                .iter()
                .any(|message| message.contains("ran out of its")
                    && message.contains("discovered regions")),
            "the snapshot has to say why it is short: {:?}",
            snapshot.diagnostics
        );
    }

    /// Running out of time is about the walk, not about whichever unit happened to be under
    /// the cursor. Recorded as a per-unit failure it reads as "heap 3 is unreadable", and —
    /// the real damage — the enclosing loop swallows it and carries on to the next unit,
    /// each stopping the same way, so the deadline bounds one unit instead of the walk. On
    /// the *last* iteration it is absorbed outright and the walk simply continues.
    ///
    /// Swept over budgets because where the halt lands decides which handler sees it: an
    /// early one stops inside a heap's VS evidence, a later one inside a segment context.
    /// Both swallow errors into diagnostics, so both have to re-raise.
    #[test]
    fn test_running_out_of_budget_is_never_reported_as_a_broken_unit() {
        for allowance in [10, 25, 40, 80, 120, 200, 400] {
            let snapshot = walk_impatiently(allowance);
            for absorbed in [
                "cannot fully discover heap",
                "cannot discover segment context",
            ] {
                assert!(
                    !snapshot
                        .diagnostics
                        .examples()
                        .iter()
                        .any(|message| message.contains(absorbed)),
                    "budget {allowance}: a per-unit handler absorbed the halt as \
                     `{absorbed}`: {:?}",
                    snapshot.diagnostics
                );
            }
        }
    }

    /// How many reads a walk may still issue after its budget is spent.
    ///
    /// Not zero: the check sits at the top of each read-issuing loop, so the iteration
    /// already under way finishes and a few of those loops nest. Measured at 3 against this
    /// fixture, and pinned there rather than rounded up — the number this guards against is
    /// the thousands a single unpolled loop costs, so slack here buys nothing and hides the
    /// next missing check.
    const OVERSHOOT_ALLOWED: usize = 3;

    #[test]
    fn test_the_walk_stops_reading_promptly_once_its_budget_is_gone() {
        let overshoots: Vec<(usize, usize)> = [10, 25, 40, 80, 120, 200, 400]
            .into_iter()
            .map(|allowance| {
                let memory = Impatient::new(allowance);
                let layout = synthetic_layout();
                SnapshotWalker {
                    memory: &memory,
                    layout: &layout,
                    traversal_limit: 1024,
                }
                .walk(None)
                .expect("running out of budget is an outcome, not an error");
                (allowance, memory.reads.get().saturating_sub(allowance))
            })
            .collect();

        let worst = overshoots.iter().map(|(_, over)| *over).max().unwrap_or(0);
        assert!(
            worst <= OVERSHOOT_ALLOWED,
            "{worst} reads issued after the deadline (limit {OVERSHOOT_ALLOWED}) — some \
             read-issuing loop is not polling the budget. (budget, overshoot): {overshoots:?}"
        );
    }

    /// A segment holding one plain allocated page range and one free range, and nothing else.
    ///
    /// Neither is a subsegment, and that is the point: with `RangeFlags` bit `0x01` read as
    /// "LFH", both look like LFH subsegment ranges, their page contents decode as a subsegment
    /// header, and they are discarded before the walk ever sees them. The free range is
    /// deliberately *not* in the free-page tree, so its state has to come from its own flags.
    fn segment_of_plain_page_ranges(heap_key: u64) -> SyntheticMemory {
        SyntheticMemory::new(plain_page_range_writes(heap_key), Vec::new())
    }

    fn plain_page_range_writes(heap_key: u64) -> Writes {
        let mut bytes = Writes::default();
        let context = HEAP + 0x100;

        fill(&mut bytes, HEAP, 0x800);
        put_u64(&mut bytes, context, SEGMENT); // SegmentListHead
        put(&mut bytes, context + 0x20, &[12, 1]); // UnitShift, FirstDescriptorIndex
        put_u64(&mut bytes, context + 0x28, !0xffffu64); // SegmentMask: 16 descriptors

        fill(&mut bytes, SEGMENT, 0x300);
        put_u64(&mut bytes, SEGMENT, context); // ListEntry back to the head: one segment
        put_u64(
            &mut bytes,
            SEGMENT + 0x20,
            SEGMENT ^ context ^ heap_key ^ super::super::decode::PAGE_SEGMENT_SIGNATURE,
        );
        for (index, flags) in [(1u64, RANGE_IN_USE), (2, DESCRIPTOR_FLAG_FIRST)] {
            let descriptor = SEGMENT + 0x100 + index * 0x20;
            put_u32(
                &mut bytes,
                descriptor,
                super::super::decode::DESCRIPTOR_TREE_SIGNATURE,
            );
            put(&mut bytes, descriptor + 0x18, &[flags]);
            put(&mut bytes, descriptor + 0x1f, &[1]);
        }
        bytes
    }

    /// A kernel page range that `nt!PoolBigPageTable` names is a big-pool allocation, and it
    /// has no `_POOL_HEADER` for the walk to read.
    ///
    /// `ExpAllocateBigPool` takes whole pages from this same segment allocator and puts the
    /// caller's tag and length in that table instead of in a header, which is why the range
    /// descriptor cannot tell one from a plain page-range allocation: both are `RangeFlags`
    /// `0x03`. Decoding the page as though a header were there reads the caller's own first
    /// sixteen bytes as one — measured on a live 26100 kernel (`FOLLOWUPS` item 99), where a
    /// `CM25` registry hive bin came back as a 4080-byte block at `+0x10` tagged `..N.`, from
    /// the bytes of `hbin`'s own header.
    ///
    /// The fixture writes a *valid-looking* pool header at the page start on purpose: without
    /// it the wrong answer would be an empty tag, which a fixture of zeroes cannot tell from
    /// the right one.
    #[test]
    fn test_a_page_range_the_big_page_table_names_carries_no_pool_header() {
        let heap_key = 0x55aa_1234_9876_0000;
        let allocated = SEGMENT + 0x1000;
        let mut writes = plain_page_range_writes(heap_key);
        fill(&mut writes, allocated, 0x1000);
        pool_header(&mut writes, allocated, b"HBIN");
        put_u64(&mut writes, BIG_TABLE_POINTER, BIG_TABLE);
        put_u64(&mut writes, BIG_TABLE_COUNT, 8);
        fill(&mut writes, BIG_TABLE, 8 * 0x20);
        let index = super::super::decode::big_page_hash(allocated, 8).unwrap();
        let entry = BIG_TABLE + index as u64 * 0x20;
        put_u64(&mut writes, entry, allocated);
        put(&mut writes, entry + 8, b"BIGP");
        put_u64(&mut writes, entry + 0x10, 0xfc0);
        let memory = SyntheticMemory::new(writes, Vec::new());
        let layout = synthetic_layout();
        let mut discovery = Discovery {
            big_pages: BigPageTable::open(&memory, &layout, &mut Vec::new()).unwrap(),
            ..Discovery::default()
        };
        assert!(discovery.big_pages.is_some(), "the fixture has a table");

        discover_segment_context(
            &memory,
            &layout,
            HEAP + 0x100,
            0,
            PoolKind::NonPagedNx,
            HeapIdentity {
                pool_state: STATE,
                heap: HEAP,
                special: false,
            },
            heap_key,
            0xa5c3_1357,
            1024,
            &SharedChunks::default(),
            &SharedChunks::default(),
            false,
            &mut discovery,
        )
        .expect("the fixture is readable throughout");

        let walker = SnapshotWalker {
            memory: &memory,
            layout: &layout,
            traversal_limit: 1024,
        };
        let mut snapshot = PoolSnapshot {
            complete: true,
            ..PoolSnapshot::default()
        };
        walker
            .walk_discovered_regions(discovery.regions, None, false, &mut snapshot)
            .expect("the fixture is readable throughout");

        let span = snapshot
            .spans
            .iter()
            .find(|span| span.header_address == allocated)
            .unwrap_or_else(|| panic!("no span at the allocated range: {:?}", snapshot.spans));
        assert_eq!(super::super::decode::display_tag(span.raw_tag), "BIGP");
        // The block starts at the page, not 0x10 into it, and is the length the table
        // records rather than the page less a header that was never there.
        assert_eq!(span.usable_address, allocated);
        assert_eq!(span.size, 0xfc0);
    }

    /// A page range that is not a subsegment must be walked as page ranges, not rejected.
    ///
    /// This is glslang/dbgscope#90 at the smallest scale that shows it. Every allocated range
    /// carries `RangeFlags` bit `0x01`, so reading that bit as "this is an LFH subsegment"
    /// sends plain page-range allocations, VS subsegments and Verifier special pool alike
    /// through the LFH decoder, which refuses the result and drops the range — silently, and
    /// before any of its allocations can be indexed.
    #[test]
    fn test_a_plain_page_range_is_not_taken_for_an_lfh_subsegment() {
        let heap_key = 0x55aa_1234_9876_0000;
        let memory = segment_of_plain_page_ranges(heap_key);
        let layout = synthetic_layout();
        let mut discovery = Discovery::default();

        discover_segment_context(
            &memory,
            &layout,
            HEAP + 0x100,
            0,
            PoolKind::NonPagedNx,
            HeapIdentity {
                pool_state: STATE,
                heap: HEAP,
                special: false,
            },
            heap_key,
            0xa5c3_1357,
            1024,
            &SharedChunks::default(),
            &SharedChunks::default(),
            false,
            &mut discovery,
        )
        .expect("the fixture is readable throughout");

        assert!(
            discovery.diagnostics.is_empty(),
            "nothing here is a subsegment, so nothing should have been decoded as one: {:?}",
            discovery.diagnostics
        );
        let described: Vec<_> = discovery
            .regions
            .iter()
            .map(|region| (region.address, region.backend, region.states.as_slice()))
            .collect();
        assert_eq!(
            described,
            [
                (
                    SEGMENT + 0x1000,
                    PoolBackend::Segment,
                    [PoolState::Allocated].as_slice()
                ),
                (
                    SEGMENT + 0x2000,
                    PoolBackend::Segment,
                    [PoolState::ReusableFree].as_slice()
                ),
            ]
        );
    }

    /// A segment laid out the way ARM64 26100.1 `ntdll` lays one out, where two shapes the
    /// walker mishandled meet: a segment list whose head is only 8-aligned, and a free range
    /// that is a node of the free-page tree.
    ///
    /// Measured 2026-09-23 on `user_heap_smoke`'s child. `SegContexts` is at +0x140 of the heap
    /// and `SegmentListHead` at +0x48 of a context, so the head sits at +0x188; x64 26100.8972
    /// has the same offsets. And the segment's trailing free range carried no `TreeSignature`,
    /// because it is the free-page tree's only node and those bytes are its (null) links.
    fn segment_with_an_unaligned_head_and_a_free_tree_node(heap_key: u64) -> SyntheticMemory {
        let mut bytes = Writes::default();
        let context = HEAP + 0x108;
        let free_node = SEGMENT + 0x100 + 2 * 0x20;

        fill(&mut bytes, HEAP, 0x800);
        put_u64(&mut bytes, context, SEGMENT); // SegmentListHead, 8 bytes past a 16 boundary
        put_u64(&mut bytes, context + 0x10, free_node); // FreePageRanges.Root, not encoded
        put(&mut bytes, context + 0x20, &[12, 1]); // UnitShift, FirstDescriptorIndex
        put_u64(&mut bytes, context + 0x28, !0xffffu64); // SegmentMask: 16 descriptors

        fill(&mut bytes, SEGMENT, 0x300);
        put_u64(&mut bytes, SEGMENT, context); // ListEntry back to the head, exactly
        put_u64(
            &mut bytes,
            SEGMENT + 0x20,
            SEGMENT ^ context ^ heap_key ^ super::super::decode::PAGE_SEGMENT_SIGNATURE,
        );
        let allocated = SEGMENT + 0x100 + 0x20;
        put_u32(
            &mut bytes,
            allocated,
            super::super::decode::DESCRIPTOR_TREE_SIGNATURE,
        );
        put(&mut bytes, allocated + 0x18, &[RANGE_IN_USE]);
        put(&mut bytes, allocated + 0x1f, &[1]);
        // The free range: first, not in use, and a tree node whose links are all null — so
        // where an allocated range has its signature, this one has zeroes.
        put(&mut bytes, free_node + 0x18, &[DESCRIPTOR_FLAG_FIRST]);
        put(&mut bytes, free_node + 0x1f, &[1]);
        SyntheticMemory::new(bytes, Vec::new())
    }

    /// Both ranges of that segment are found, the free one as free, and nothing is said about
    /// either — where the walker used to read the heap's own fields as a second segment and
    /// drop the free range for the signature its node bytes cannot carry.
    #[test]
    fn test_an_unaligned_segment_list_head_and_a_free_tree_node_are_walked_as_ntdll_lays_them_out()
    {
        let heap_key = 0x55aa_1234_9876_0000;
        let memory = segment_with_an_unaligned_head_and_a_free_tree_node(heap_key);
        let layout = synthetic_layout();
        let mut discovery = Discovery::default();

        discover_segment_context(
            &memory,
            &layout,
            HEAP + 0x108,
            0,
            PoolKind::NonPagedNx,
            HeapIdentity {
                pool_state: STATE,
                heap: HEAP,
                special: false,
            },
            heap_key,
            0xa5c3_1357,
            1024,
            &SharedChunks::default(),
            &SharedChunks::default(),
            false,
            &mut discovery,
        )
        .expect("the fixture is readable throughout");

        assert!(
            discovery.diagnostics.is_empty(),
            "one segment, one allocated range and one free one, all well formed: {:?}",
            discovery.diagnostics
        );
        let described: Vec<_> = discovery
            .regions
            .iter()
            .map(|region| (region.address, region.backend, region.states.as_slice()))
            .collect();
        assert_eq!(
            described,
            [
                (
                    SEGMENT + 0x1000,
                    PoolBackend::Segment,
                    [PoolState::Allocated].as_slice()
                ),
                (
                    SEGMENT + 0x2000,
                    PoolBackend::Segment,
                    [PoolState::ReusableFree].as_slice()
                ),
            ]
        );
    }

    /// A tree whose nodes sit 8 bytes past a 16-byte boundary is walked node for node.
    ///
    /// That is every VS free tree: a free chunk's `Node` is at +0x8 of its 16-aligned header,
    /// and the eight bytes below it are the chunk's encoded `Sizes`. Each node here has such a
    /// word below it, pointing nowhere readable, which is what a walk that masks its links to
    /// 16 bytes follows instead of the real ones — the `unreadable VS free tree node 0x84a23d…`
    /// diagnostics measured on ARM64 26100.1, 2026-09-23.
    #[test]
    fn test_a_tree_of_nodes_eight_past_a_boundary_is_walked_exactly() {
        const ROOT: u64 = K + 0x40_0008;
        const LEFT: u64 = K + 0x40_1008;
        const RIGHT: u64 = K + 0x40_2008;
        let mut bytes = Writes::default();
        for (node, left, right) in [(ROOT, LEFT, RIGHT), (LEFT, 0, 0), (RIGHT, 0, 0)] {
            put_u64(&mut bytes, node - 8, 0x84a2_3d0f_b239_be00); // the chunk's encoded Sizes
            put_u64(&mut bytes, node, left);
            put_u64(&mut bytes, node + 8, right);
        }
        let memory = SyntheticMemory::new(bytes, Vec::new());
        let mut diagnostics = Vec::new();

        let mut nodes = walk_tree_nodes(&memory, ROOT, 0, 8, 16, "VS free tree", &mut diagnostics)
            .expect("the fixture is readable throughout");
        nodes.sort_unstable();

        assert_eq!(nodes, [ROOT, LEFT, RIGHT]);
        assert!(diagnostics.is_empty(), "{diagnostics:?}");
        let tree = K + 0x50_0000;
        assert_eq!(
            decode_rb_root_for(ROOT ^ tree, tree, true, false),
            Some(ROOT)
        );
    }

    /// One page segment whose every descriptor is a committed LFH range.
    ///
    /// The shared fixture has five descriptors in a sixteen-slot array and only one of them
    /// is LFH, so its descriptor loop issues almost no reads — a walk over it stays inside
    /// the overshoot limit *even with the loop's budget check deleted*, which is exactly how
    /// that missing check went unnoticed. A real kernel segment is the opposite: descriptors
    /// packed with committed subsegments, one header read each.
    ///
    /// The subsegment bodies are deliberately left unwritten. Reading one still costs the
    /// round trip this is counting, and what the walk makes of the bytes afterwards is not
    /// what is under test.
    fn segment_of_lfh_descriptors(heap_key: u64) -> SyntheticMemory {
        let mut bytes = Writes::default();
        let context = HEAP + 0x100;

        fill(&mut bytes, HEAP, 0x800);
        put_u64(&mut bytes, context, SEGMENT); // SegmentListHead
        put(&mut bytes, context + 0x20, &[12, 1]); // UnitShift, FirstDescriptorIndex
        put_u64(&mut bytes, context + 0x28, !0xffffu64); // SegmentMask: 16 descriptors

        fill(&mut bytes, SEGMENT, 0x300);
        put_u64(&mut bytes, SEGMENT, context); // ListEntry back to the head: one segment
        put_u64(
            &mut bytes,
            SEGMENT + 0x20,
            SEGMENT ^ context ^ heap_key ^ super::super::decode::PAGE_SEGMENT_SIGNATURE,
        );
        for index in 1..16u64 {
            let descriptor = SEGMENT + 0x100 + index * 0x20;
            put_u32(
                &mut bytes,
                descriptor,
                super::super::decode::DESCRIPTOR_TREE_SIGNATURE,
            );
            put(&mut bytes, descriptor + 0x18, &[RANGE_LFH]);
            put(&mut bytes, descriptor + 0x1f, &[1]);
        }
        SyntheticMemory::new(bytes, Vec::new())
    }

    /// The descriptor loop is bounded — 4096 iterations — and that is precisely why it was
    /// missed: "bounded" is not "cheap" when each iteration reads. Polling one level out, in
    /// the segment loop, let a single segment issue thousands of round trips after the
    /// deadline, which on a live KD link is minutes: the wedge, rebuilt inside its own fix.
    #[test]
    fn test_a_segment_full_of_descriptors_polls_the_budget_between_them() {
        // Enough to get through the context fields, the free-page tree, the segment header
        // and the descriptor array, so the halt lands *inside* the descriptor loop — which
        // is the loop under test.
        const ALLOWANCE: usize = 10;
        let heap_key = 0x55aa_1234_9876_0000;
        let memory = Impatient::over(segment_of_lfh_descriptors(heap_key), ALLOWANCE);
        let layout = synthetic_layout();
        let mut discovery = Discovery::default();

        let outcome = discover_segment_context(
            &memory,
            &layout,
            HEAP + 0x100,
            0,
            PoolKind::NonPagedNx,
            HeapIdentity {
                pool_state: STATE,
                heap: HEAP,
                special: false,
            },
            heap_key,
            0xa5c3_1357,
            1024,
            &SharedChunks::default(),
            &SharedChunks::default(),
            false,
            &mut discovery,
        );

        assert!(
            matches!(outcome, Err(SnapshotError::BudgetExpired)),
            "the descriptor loop has to report the halt: {outcome:?}"
        );
        assert!(
            memory.reads.get() <= ALLOWANCE + OVERSHOOT_ALLOWED,
            "{} reads for a budget of {ALLOWANCE}: the descriptor loop ran on past the \
             deadline instead of polling it",
            memory.reads.get()
        );
    }

    /// A large-allocation tree of `count` nodes whose metadata will not read.
    ///
    /// Each node carries only its `_RTL_BALANCED_NODE` links, so the tree walk finds them all
    /// and the metadata read that follows fails — which is the path under test. A node whose
    /// metadata *does* read reaches `lookup_big_page_target`, and that polls the budget
    /// itself, so it could never demonstrate the missing check.
    fn large_allocation_tree(count: u64) -> SyntheticMemory {
        let mut bytes = Writes::default();
        let tree = HEAP + 0x400;
        let node = |index: u64| LARGE_META + index * 0x40;

        fill(&mut bytes, tree, 0x10);
        put_u64(&mut bytes, tree, node(0)); // Root
        for index in 0..count {
            // Left chains to the next node, Right terminates. Only 0x10 bytes per node, so
            // the 0x28-byte metadata read spans into a gap and fails.
            fill(&mut bytes, node(index), 0x10);
            let next = if index + 1 < count {
                node(index + 1)
            } else {
                0
            };
            put_u64(&mut bytes, node(index), next);
        }
        SyntheticMemory::new(bytes, Vec::new())
    }

    /// The large-allocation loop reads once per node and `nodes` is capped only by
    /// `traversal_limit` — a million. Without a poll here the deadline is not consulted again
    /// until the enclosing heap.
    #[test]
    fn test_the_large_allocation_loop_polls_the_budget_between_nodes() {
        // The tree walk itself costs a read per node plus the root and encoded flag, so the
        // budget has to clear that for the halt to land in the node loop below it.
        const NODES: u64 = 12;
        const ALLOWANCE: usize = 16;
        let memory = Impatient::over(large_allocation_tree(NODES), ALLOWANCE);
        let layout = synthetic_layout();
        let mut discovery = Discovery::default();

        let outcome = discover_large_allocations(
            &memory,
            &layout,
            HEAP,
            0,
            PoolKind::NonPagedNx,
            HeapIdentity {
                pool_state: STATE,
                heap: HEAP,
                special: false,
            },
            0,
            1024,
            &mut discovery,
        );

        assert!(
            matches!(outcome, Err(SnapshotError::BudgetExpired)),
            "the node loop has to report the halt: {outcome:?}"
        );
        assert!(
            memory.reads.get() <= ALLOWANCE + OVERSHOOT_ALLOWED,
            "{} reads for a budget of {ALLOWANCE}: the node loop ran on past the deadline",
            memory.reads.get()
        );
    }

    /// One extent far larger than a single read should be, reported as entirely committed.
    struct OneHugeExtent {
        reads: Cell<usize>,
        allowance: usize,
    }

    impl PoolMemory for OneHugeExtent {
        fn read_exact(&self, _address: u64, size: usize) -> Result<Vec<u8>, SnapshotError> {
            self.reads.set(self.reads.get() + 1);
            Ok(vec![0; size])
        }

        fn valid_region(&self, address: u64, size: usize) -> Result<(u64, usize), SnapshotError> {
            Ok((address, size))
        }

        fn interrupted(&self) -> Result<bool, SnapshotError> {
            Ok(false)
        }

        fn out_of_budget(&self) -> bool {
            self.reads.get() >= self.allowance
        }
    }

    /// The loops around a read can poll perfectly and still overrun, because the read itself
    /// is one synchronous transfer: a large committed page range is megabytes in flight with
    /// no opportunity to notice the deadline. Chunking the transfer is what makes the ceiling
    /// hold whatever the pool's geometry turns out to be.
    #[test]
    fn test_a_large_extent_is_read_in_chunks_that_observe_the_deadline() {
        const ALLOWANCE: usize = 2;
        let memory = OneHugeExtent {
            reads: Cell::new(0),
            allowance: ALLOWANCE,
        };
        let layout = vs_layout(VsFixture::Inline);
        let mut region = lfh_region(0x1000, 0x100);
        region.size = EXTENT_READ_CHUNK * 8;
        region.bitmap = vec![0xff; 64];
        let walker = SnapshotWalker {
            memory: &memory,
            layout: &layout,
            traversal_limit: 1000,
        };

        let outcome = walker.walk_region(&region, &mut PoolSnapshot::default());

        assert!(
            matches!(outcome, Err(SnapshotError::BudgetExpired)),
            "an extent that outlives the deadline has to stop mid-transfer: {outcome:?}"
        );
        assert_eq!(
            memory.reads.get(),
            ALLOWANCE,
            "the extent was not chunked, so the deadline could not be observed while it \
             transferred"
        );
    }

    /// A source that answers every read one byte short.
    struct ShortChunks {
        reads: Cell<usize>,
    }

    impl PoolMemory for ShortChunks {
        fn read_exact(&self, _address: u64, size: usize) -> Result<Vec<u8>, SnapshotError> {
            self.reads.set(self.reads.get() + 1);
            Ok(vec![0; size.saturating_sub(1)])
        }

        fn valid_region(&self, address: u64, size: usize) -> Result<(u64, usize), SnapshotError> {
            Ok((address, size))
        }

        fn interrupted(&self) -> Result<bool, SnapshotError> {
            Ok(false)
        }
    }

    /// Reading in pieces makes a short chunk dangerous in a way a short whole-extent read
    /// never was: the next chunk's address comes from how much has been collected, so
    /// accepting a short one shifts every later chunk and assembles contiguous bytes out of
    /// discontiguous addresses. The decoders cannot tell — they would report allocations that
    /// were never at those addresses. So the extent fails, and stays honestly unreadable.
    #[test]
    fn test_a_short_chunk_fails_the_extent_instead_of_misassembling_it() {
        let memory = ShortChunks {
            reads: Cell::new(0),
        };
        let layout = vs_layout(VsFixture::Inline);
        let mut region = lfh_region(0x1000, 0x100);
        region.size = EXTENT_READ_CHUNK * 3;
        region.bitmap = vec![0xff; 64];
        let walker = SnapshotWalker {
            memory: &memory,
            layout: &layout,
            traversal_limit: 1000,
        };

        assert!(
            matches!(
                walker.read_extent(region.address, region.size),
                Err(SnapshotError::InvalidData { .. })
            ),
            "a short chunk has to fail the extent"
        );
        assert_eq!(
            memory.reads.get(),
            1,
            "it must not read on after the short chunk"
        );

        // And through the region walk: reported as unreadable, with nothing decoded out of a
        // buffer that was never assembled.
        let mut snapshot = PoolSnapshot::default();
        walker.walk_region(&region, &mut snapshot).unwrap();

        assert!(
            snapshot
                .spans
                .iter()
                .all(|span| span.state == PoolState::Unreadable),
            "a misassembled extent must not yield decoded allocations: {:?}",
            snapshot.spans
        );
        assert!(
            snapshot
                .diagnostics
                .examples()
                .iter()
                .any(|message| message.contains("cannot read region")),
            "the failure has to be said out loud: {:?}",
            snapshot.diagnostics
        );
    }

    /// The common case must not pay for the uncommon one: an extent that fits in a chunk is
    /// still exactly one read.
    #[test]
    fn test_a_small_extent_is_still_a_single_read() {
        let memory = OneHugeExtent {
            reads: Cell::new(0),
            allowance: usize::MAX,
        };
        let layout = vs_layout(VsFixture::Inline);
        let region = lfh_region(0x1000, 0x100);
        let walker = SnapshotWalker {
            memory: &memory,
            layout: &layout,
            traversal_limit: 1000,
        };

        walker
            .walk_region(&region, &mut PoolSnapshot::default())
            .unwrap();

        assert_eq!(memory.reads.get(), 1);
    }

    /// Discovery finds the regions and the walk reads what is in them, so a budget spent
    /// entirely on the first half yields an empty pool. Two things keep that from happening:
    /// discovery is held to a share of the budget, and — pinned here — the regions it found
    /// before stopping survive the halt instead of being dropped with the error.
    #[test]
    fn test_regions_found_before_the_halt_are_not_thrown_away() {
        // Enough reads to get past VS evidence and into the segment walk, not enough to
        // finish it.
        let snapshot = walk_impatiently(120);
        let summary = snapshot
            .diagnostics
            .examples()
            .iter()
            .find(|message| message.contains("discovered regions"))
            .expect("the walk stopped short and must say so");
        let tokens: Vec<&str> = summary.split_whitespace().collect();
        let discovered: usize = tokens
            .iter()
            .position(|token| *token == "discovered")
            .and_then(|index| index.checked_sub(1))
            .and_then(|index| tokens[index].parse().ok())
            .unwrap_or_else(|| panic!("no region count in `{summary}`"));

        assert!(
            discovered > 0,
            "discovery's partial results were discarded with the halt: `{summary}`"
        );
    }

    /// Neither half of the walk is useful without the other, so discovery gets a share of
    /// the budget rather than all of it — and the whole walk still stops at the deadline it
    /// was given.
    #[test]
    fn test_discovery_gets_a_share_of_the_budget_and_the_walk_the_rest() {
        let start = Instant::now();
        let budget = Duration::from_secs(120);
        let (discovery, whole) = budget_deadlines(start, Some(budget));
        let (discovery, whole) = (discovery.unwrap(), whole.unwrap());

        assert_eq!(whole, start + budget);
        assert!(
            discovery < whole,
            "discovery must yield before the deadline"
        );
        assert!(
            discovery > start,
            "discovery must get real time, not a token slice"
        );
        assert_eq!(budget_deadlines(start, None), (None, None));
        // `PoolWalk::within` accepts any `Duration`, so a budget longer than `Instant` can
        // represent is reachable from safe public API. Adding it would panic; a budget
        // beyond measuring is a request to run unbounded, and is answered as one.
        assert_eq!(budget_deadlines(start, Some(Duration::MAX)), (None, None));
    }

    /// A budget and a Ctrl+C both halt the walk, and they must not be confused. Running out
    /// of time means "here is what I got"; an operator interrupting means "stop" — handing
    /// them a snapshot they cancelled, which then renders as a full table, is not that.
    #[test]
    fn test_an_interrupt_still_stops_the_walk_outright() {
        struct Interrupting(SyntheticMemory);

        impl PoolMemory for Interrupting {
            fn read_exact(&self, address: u64, size: usize) -> Result<Vec<u8>, SnapshotError> {
                self.0.read_exact(address, size)
            }

            fn valid_region(
                &self,
                address: u64,
                size: usize,
            ) -> Result<(u64, usize), SnapshotError> {
                self.0.valid_region(address, size)
            }

            fn interrupted(&self) -> Result<bool, SnapshotError> {
                Ok(true)
            }
        }

        let memory = Interrupting(synthetic_memory());
        let layout = synthetic_layout();

        assert!(matches!(
            SnapshotWalker {
                memory: &memory,
                layout: &layout,
                traversal_limit: 1024,
            }
            .walk(Some(Duration::from_secs(600))),
            Err(SnapshotError::Interrupted)
        ));
    }

    /// Memory that hands back one small extent at a time and runs out of budget after a
    /// fixed number of them.
    ///
    /// This is the shape that makes a per-*region* check insufficient. A region the
    /// allocator left with holes is read one committed run at a time — a `valid_region` and
    /// a `read_exact` each, both over the wire on a live target — so checking only on the
    /// way in lets a single region overrun the deadline by as many round trips as it has
    /// extents.
    struct Fragmented {
        extent: usize,
        allowance: usize,
        extents_read: Cell<usize>,
    }

    impl PoolMemory for Fragmented {
        fn read_exact(&self, _address: u64, size: usize) -> Result<Vec<u8>, SnapshotError> {
            Ok(vec![0; size])
        }

        fn valid_region(&self, address: u64, size: usize) -> Result<(u64, usize), SnapshotError> {
            self.extents_read.set(self.extents_read.get() + 1);
            Ok((address, size.min(self.extent)))
        }

        fn interrupted(&self) -> Result<bool, SnapshotError> {
            Ok(false)
        }

        fn out_of_budget(&self) -> bool {
            self.extents_read.get() >= self.allowance
        }
    }

    #[test]
    fn test_a_fragmented_region_checks_the_budget_between_extents() {
        // 0x1000 of region in 0x400 extents is four round trips; the budget allows three.
        let memory = Fragmented {
            extent: 0x400,
            allowance: 3,
            extents_read: Cell::new(0),
        };
        let layout = vs_layout(VsFixture::Inline);
        let mut region = lfh_region(0x1000, 0x100);
        region.bitmap = vec![0xff; 16];
        let walker = SnapshotWalker {
            memory: &memory,
            layout: &layout,
            traversal_limit: 1000,
        };

        let outcome = walker.walk_region(&region, &mut PoolSnapshot::default());

        assert!(
            matches!(outcome, Err(SnapshotError::BudgetExpired)),
            "the region walk has to report the halt, not absorb it: {outcome:?}"
        );
        assert_eq!(
            memory.extents_read.get(),
            3,
            "the region went on reading past its deadline"
        );
    }

    /// A budget nobody comes close to spending must not change the answer.
    #[test]
    fn test_a_generous_budget_walks_exactly_as_an_unbounded_one_does() {
        let memory = synthetic_memory();
        let layout = synthetic_layout();
        let walker = SnapshotWalker {
            memory: &memory,
            layout: &layout,
            traversal_limit: 1024,
        };

        let unbounded = walker.walk(None).unwrap();
        let budgeted = walker.walk(Some(Duration::from_secs(600))).unwrap();

        assert_eq!(budgeted.spans, unbounded.spans);
        assert_eq!(budgeted.complete, unbounded.complete);
        assert_eq!(budgeted.diagnostics, unbounded.diagnostics);
    }

    // ---- diagnostic floods -------------------------------------------------------------

    /// A thousand of one complaint and one of another, the shape of a real flood.
    fn flooded() -> PoolDiagnostics {
        let mut diagnostics: Vec<String> = (0..1000)
            .map(|node| format!("unreadable VS free tree node {node:#x}: sparse memory"))
            .collect();
        diagnostics.push("rejecting implausible LFH unit size 3 at 0x1000".into());
        PoolDiagnostics::from_iter(diagnostics)
    }

    /// A live target keeps allocating between the reads of one walk, so a single stale list
    /// pointer yields one complaint per node — 14k of them measured on a busy kernel, which
    /// buries the few distinct problems and leaves every consumer truncating by position.
    /// Collapsing keeps examples and, crucially, the *count*: the volume is itself the
    /// finding, so it must survive.
    #[test]
    fn test_a_flood_of_one_complaint_becomes_examples_plus_a_count() {
        let diagnostics = flooded();

        assert_eq!(
            diagnostics.examples().len(),
            DIAGNOSTIC_EXAMPLES + 1,
            "expected {DIAGNOSTIC_EXAMPLES} examples of the flood, plus the distinct message"
        );
        assert!(diagnostics.examples()[0].contains("unreadable VS free tree node 0x0"));
        assert!(
            diagnostics
                .examples()
                .iter()
                .any(|message| message.contains("rejecting implausible LFH unit size")),
            "a distinct complaint must not be collapsed away by a flood of another"
        );

        let totals: Vec<usize> = diagnostics.shapes().iter().map(|seen| seen.total).collect();
        assert_eq!(
            totals,
            vec![1000, 1],
            "shapes carry the whole count each, in the order first seen: {:?}",
            diagnostics.shapes()
        );
        assert!(
            diagnostics.lines().last().is_some_and(
                |line| line.contains(&format!("and {} more", 1000 - DIAGNOSTIC_EXAMPLES))
            ),
            "the rendering owes the reader the count too: {:?}",
            diagnostics.lines()
        );
    }

    /// The count of the *walk* is not the count of what survived the cap, and the gap is two
    /// orders of magnitude on a real target. Reporting the latter as the former — measured
    /// once as "71 diagnostics" for a walk that made ~7,700 complaints — describes the
    /// collapsing as though it were a property of the pool, which is precisely the lie this
    /// module is built to avoid.
    #[test]
    fn test_the_emitted_count_is_the_walk_not_the_sample() {
        let diagnostics = flooded();

        assert_eq!(diagnostics.emitted(), 1001, "every message is counted");
        assert!(
            diagnostics.emitted() > diagnostics.lines().len() * 50,
            "a caller that counted lines ({}) instead of messages ({}) would be off by two \
             orders of magnitude and never know",
            diagnostics.lines().len(),
            diagnostics.emitted()
        );
    }

    /// Merging is by message, not by wholesale concatenation: discovery's complaints join the
    /// walk's own shapes rather than starting fresh ones, or a flood split across the two
    /// phases would keep twice the examples and report itself as two findings.
    #[test]
    fn test_extending_merges_into_the_shapes_already_seen() {
        let mut diagnostics = flooded();
        diagnostics.extend(
            (1000..1010)
                .map(|node| format!("unreadable VS free tree node {node:#x}: sparse memory")),
        );

        assert_eq!(diagnostics.shapes().len(), 2, "no shape was duplicated");
        assert_eq!(diagnostics.shapes()[0].total, 1010);
        assert_eq!(diagnostics.emitted(), 1011);
        assert_eq!(
            diagnostics.examples().len(),
            DIAGNOSTIC_EXAMPLES + 1,
            "the cap holds across the merge"
        );
    }

    /// Grouping is by shape, so the same complaint about different addresses is one group
    /// while genuinely different complaints stay apart.
    #[test]
    fn test_diagnostic_shape_ignores_numbers_but_not_wording() {
        assert_eq!(
            diagnostic_shape("unreadable VS free tree node 0xdeadbeef: sparse"),
            diagnostic_shape("unreadable VS free tree node 0x41414141: sparse")
        );
        assert_ne!(
            diagnostic_shape("unreadable VS free tree node 0xdeadbeef"),
            diagnostic_shape("unreadable VS delay-free node 0xdeadbeef")
        );
    }

    #[test]
    fn test_special_pool_kinds_cover_all_heap_slots() {
        assert_eq!(SPECIAL_POOL_KINDS.len(), 4);
        assert_eq!(SPECIAL_POOL_KINDS[3], PoolKind::SpecialPrototypePaged);
        assert!(SPECIAL_POOL_KINDS[3].is_paged());
    }

    /// Whether a VS user block skips a page end is read off the target's own context, once per
    /// heap, and lands on every region of that heap — bit 0 of `Config` and nothing else, since
    /// bit 1 (`EnableDelayFree`) shares the byte. A layout with no `Config` reads unset and says
    /// so, which is the one way a target carrying the flag can be walked without it.
    #[test]
    fn test_discovery_reads_whether_a_vs_user_block_skips_a_page_end_from_the_context() {
        let identity = HeapIdentity {
            pool_state: STATE,
            heap: HEAP,
            special: false,
        };
        let discover = |memory: &SyntheticMemory, layout: &PoolLayout| {
            let mut discovery = Discovery::default();
            discover_heap_regions(
                memory,
                layout,
                HEAP,
                0,
                PoolKind::NonPagedNx,
                identity,
                None,
                0x55aa_1234_9876_0000,
                0xa5c3_1357,
                1024,
                &mut discovery,
            )
            .expect("the fixture is readable throughout");
            discovery
        };
        let config = HEAP + 0x300 + 0x38;
        for (flags, expected) in [(0u8, false), (1, true), (2, false), (3, true)] {
            let mut memory = synthetic_memory();
            memory.poke(config, flags);

            let discovery = discover(&memory, &synthetic_layout());

            let vs: Vec<_> = discovery
                .regions
                .iter()
                .filter(|region| region.backend == PoolBackend::Vs)
                .collect();
            assert!(!vs.is_empty(), "the fixture has a VS subsegment");
            assert!(
                vs.iter().all(|region| region.vs_skips_page_end == expected),
                "config {flags:#x} should read {expected}"
            );
            assert!(
                !discovery
                    .diagnostics
                    .iter()
                    .any(|line| line.contains("Config")),
                "{:?}",
                discovery.diagnostics
            );
        }

        let mut memory = synthetic_memory();
        memory.poke(config, 1);
        let mut layout = synthetic_layout();
        layout
            .types
            .get_mut("_HEAP_VS_CONTEXT")
            .unwrap()
            .fields
            .remove("Config");

        let discovery = discover(&memory, &layout);

        assert!(
            discovery
                .regions
                .iter()
                .all(|region| !region.vs_skips_page_end)
        );
        assert!(
            discovery
                .diagnostics
                .iter()
                .any(|line| line.contains("has no resolved Config")),
            "{:?}",
            discovery.diagnostics
        );
    }

    #[test]
    fn test_discovery_skips_missing_optional_large_layout() {
        let memory = big_page_memory(LARGE_VA, 4, 0);
        let mut layout = synthetic_layout();
        layout.types.remove("_HEAP_LARGE_ALLOC_DATA");
        let mut discovery = Discovery::default();

        discover_large_allocations(
            &memory,
            &layout,
            HEAP,
            0,
            PoolKind::NonPagedNx,
            HeapIdentity {
                pool_state: STATE,
                heap: HEAP,
                special: false,
            },
            0,
            1024,
            &mut discovery,
        )
        .unwrap();

        assert!(discovery.regions.is_empty());
        assert_eq!(memory.read_calls.get(), 0);
    }

    #[test]
    fn test_big_page_lookup_skips_missing_optional_layout() {
        let memory = big_page_memory(LARGE_VA, 4, 0);
        let mut layout = synthetic_layout();
        layout.types.remove("_POOL_TRACKER_BIG_PAGES");

        assert!(
            BigPageTable::open(&memory, &layout, &mut Vec::new())
                .unwrap()
                .is_none()
        );
        assert_eq!(memory.read_calls.get(), 0);
    }

    #[test]
    fn test_big_page_lookup_batches_collision_chain() {
        let memory = big_page_memory(LARGE_VA, 512, 300);
        let layout = synthetic_layout();
        let mut diagnostics = Vec::new();
        let mut table = BigPageTable::open(&memory, &layout, &mut diagnostics)
            .unwrap()
            .unwrap();

        assert_eq!(
            table.lookup(&memory, LARGE_VA, &mut diagnostics).unwrap(),
            Some(BigPageEntry {
                tag: u32::from_le_bytes(*b"BTCH"),
                size: 0x9000,
            })
        );
        assert!(memory.read_calls.get() <= 5);
        assert!(diagnostics.is_empty());

        // The chain crossed both batches, so asking again is pure arithmetic. This is what
        // makes the lookup affordable per *page range* rather than per large allocation.
        let reads = memory.read_calls.get();
        assert!(
            table
                .lookup(&memory, LARGE_VA, &mut diagnostics)
                .unwrap()
                .is_some()
        );
        assert_eq!(memory.read_calls.get(), reads);
    }

    #[test]
    fn test_big_page_lookup_honors_interrupt_between_batches() {
        let mut memory = big_page_memory(LARGE_VA, 512, 300);
        let layout = synthetic_layout();
        let mut table = BigPageTable::open(&memory, &layout, &mut Vec::new())
            .unwrap()
            .unwrap();
        memory.interrupt_after_checks = Some(1);

        assert!(matches!(
            table.lookup(&memory, LARGE_VA, &mut Vec::new()),
            Err(SnapshotError::Interrupted)
        ));
        assert_eq!(memory.interrupt_checks.get(), 2);
    }

    /// `nt!ExpRemoveTagForBigPages` frees an entry with `lock inc [Va]`, so a freed
    /// allocation is still in the table under `address | 1` with its old tag and size, and
    /// `nt`'s own comparison is `cmp rcx,rdi` — exact. Matching on `Va & !1`, as this did,
    /// answers for a page that has been given back with the tag it used to carry.
    #[test]
    fn test_big_page_lookup_walks_past_a_freed_entry_to_the_live_one() {
        let count = 8;
        let layout = synthetic_layout();
        let mut bytes = vec![0u8; count * 0x20];
        let first = super::super::decode::big_page_hash(LARGE_VA, count).unwrap();
        let put = |bytes: &mut Vec<u8>, index: usize, va: u64, tag: &[u8; 4], size: u64| {
            let offset = (index % count) * 0x20;
            bytes[offset..offset + 8].copy_from_slice(&va.to_le_bytes());
            bytes[offset + 8..offset + 12].copy_from_slice(tag);
            bytes[offset + 0x10..offset + 0x18].copy_from_slice(&size.to_le_bytes());
        };
        // The same address, freed, sitting where the hash lands. Then the live entry.
        put(&mut bytes, first, LARGE_VA | 1, b"DEAD", 0x3000);
        put(&mut bytes, first + 1, LARGE_VA, b"LIVE", 0x1800);
        let memory = BigPageMemory {
            table: bytes,
            count,
            read_calls: Cell::new(0),
            interrupt_checks: Cell::new(0),
            interrupt_after_checks: None,
        };
        let mut table = BigPageTable::open(&memory, &layout, &mut Vec::new())
            .unwrap()
            .unwrap();

        assert_eq!(
            table.lookup(&memory, LARGE_VA, &mut Vec::new()).unwrap(),
            Some(BigPageEntry {
                tag: u32::from_le_bytes(*b"LIVE"),
                size: 0x1800,
            })
        );
    }

    /// A slot the allocator has never used reads `1` — `POOL_BIG_TABLE_ENTRY_FREE` with no
    /// address behind it — and ends the chain, because `ExpAddTagForBigPages` claims the
    /// first bit-0-set slot from the hash and so could not have walked past this one. The
    /// test that matters is the *miss*: a lookup that scans the whole table instead costs a
    /// read of every batch in it.
    #[test]
    fn test_big_page_lookup_stops_at_a_never_used_slot() {
        let count = 512;
        let layout = synthetic_layout();
        let mut bytes = vec![0u8; count * 0x20];
        let absent = LARGE_VA + 0x20_0000;
        let first = super::super::decode::big_page_hash(absent, count).unwrap();
        // Every slot occupied, so that a walk which does not stop where it should has nothing
        // else to stop it and runs into the next batch. Each occupant is distinct from
        // `absent`, which the chain must not accidentally contain.
        for index in 0..count {
            let offset = index * 0x20;
            let occupant = LARGE_VA + 0x100_0000 + index as u64 * 0x1000;
            bytes[offset..offset + 8].copy_from_slice(&occupant.to_le_bytes());
        }
        let never_used = ((first + 3) % count) * 0x20;
        bytes[never_used..never_used + 8].copy_from_slice(&1u64.to_le_bytes());
        assert!(
            first % BIG_PAGE_BATCH + 3 < BIG_PAGE_BATCH,
            "the fixture's chain has to stay inside one batch for the read count to mean              anything"
        );
        let memory = BigPageMemory {
            table: bytes,
            count,
            read_calls: Cell::new(0),
            interrupt_checks: Cell::new(0),
            interrupt_after_checks: None,
        };
        let mut table = BigPageTable::open(&memory, &layout, &mut Vec::new())
            .unwrap()
            .unwrap();
        let reads = memory.read_calls.get();

        assert_eq!(
            table.lookup(&memory, absent, &mut Vec::new()).unwrap(),
            None
        );
        // One batch, not both: the chain ended before the walk left it.
        assert_eq!(memory.read_calls.get() - reads, 1);
    }

    #[test]
    fn test_snapshot_errors_preserve_category_and_source() {
        let memory = synthetic_memory();
        let error = memory.read_exact(0, 1).unwrap_err();
        assert!(matches!(&error, SnapshotError::Read { .. }));
        assert!(std::error::Error::source(&error).is_some());

        let error = guarded_read(&memory, SEGMENT + 0x7800, 0x10).unwrap_err();
        assert!(matches!(error, SnapshotError::RegionValidation { .. }));

        let error = scalar(&ShortMemory, 0x1000, 1).unwrap_err();
        assert!(matches!(
            error,
            SnapshotError::InvalidData { detail } if detail == "short u8"
        ));
    }
}