hprof-analyzer 0.2.0

//! Pass 2: the reference-graph core (RSS- and parity-critical).
//!
//! Reads the heap a second time to build the object reference graph: a forward
//! CSR (`fwd_offsets`/`fwd_targets`) and a deferred, blocked+delta-encoded
//! inbound-referrer CSR (`InboundBuilder`). It resolves real GC roots and adds
//! synthetic system-class roots to mirror MAT's `addSystemClassRootsIfMissing`,
//! builds thread stacks/frames and bounded thread-local / alloc-site samples,
//! then hands off to the dominator/retained stages. This is the most
//! memory-sensitive file in the crate (hard peak-RSS budget) and the most
//! parity-sensitive (byte-exact, MAT-frozen counts) — most edits here should be
//! comments, not behavior changes.

#[cfg(not(target_arch = "wasm32"))]
use std::time::Instant;
use std::{
    collections::HashMap,
    io::{self, ErrorKind},
};

use crate::{
    pass1::Pass1,
    reader::{HEAP_DUMP_END_KIND, HprofReader},
    types::{HprofType, heap, tags},
};

mod boxed;
mod dup_prim_arrays;
mod fielddecode;
pub mod framework_scan;
mod meta;
mod model;
mod scan;
pub(crate) mod sizing;
mod strings;

pub(crate) use boxed::compute_boxed_holders;
pub(crate) use dup_prim_arrays::{
    DupPrimArrays, compute_dup_array_holders, compute_dup_prim_arrays,
};
pub(crate) use fielddecode::ATTRIBUTION_TOP_N;
pub(crate) use fielddecode::{CollDesc, CollKind, builtin_coll_descs};
pub use framework_scan::scan_frameworks;
pub(crate) use meta::*;
pub use model::*;
pub(crate) use scan::*;
pub use sizing::*;
pub use strings::*;

// ── Pass2 main logic ───────────────────────────────────────────────────────

/// Zero-sized entry point for the second parse pass; see [`Pass2::build`].
pub struct Pass2;

impl Pass2 {
    /// Run pass 2 over the dump at `path`, consuming pass1's tables. Detects
    /// ref size, computes MAT shallow sizes, interns the class histogram, scans
    /// the heap twice (degree-count then forward-CSR fill), resolves real +
    /// synthetic GC roots, and captures thread/alloc metadata. Returns the
    /// `Graph`, a deferred `InboundBuilder` (the inbound CSR is built later to
    /// keep its ~5.5GB off the rpo peak), and the early-compressed `shallow` /
    /// `class_idx` blobs. `compress` selects the codec for those cold arrays.
    #[allow(clippy::type_complexity)]
    pub fn build(
        source: &crate::source::HprofSource,
        mut p1: Pass1,
        compress: crate::cvec::Codec,
        opts: &crate::AnalyzeOptions,
        queries: &[(crate::query::ast::Query, crate::query::plan::QueryPlan)],
        in_sets_by_slot: &mut std::collections::HashMap<usize, Vec<crate::query::execute::InSet>>,
        exists_bools_by_slot: &mut std::collections::HashMap<usize, Vec<bool>>,
    ) -> io::Result<(
        Graph,
        InboundBuilder,
        crate::cvec::CompressedU32,
        crate::cvec::CompressedU32,
        Option<crate::cvec::CompressedU32>,
        crate::query::execute::QueryExecState,
        Option<crate::query::refwalk::RefWalkCsr>,
        // Decoded toString(s) values: dense_idx → String.
        // Empty when no toString(s) query ran.
        std::collections::HashMap<u32, String>,
        // True when the string-capture table overflowed its cap during the scan;
        // toString(s) results may be partial. Separate from the map so callers can
        // surface `QueryResult.truncated` even when the map is non-empty.
        bool,
    )> {
        #[cfg(not(target_arch = "wasm32"))]
        let _t_build = Instant::now();
        macro_rules! t_phase {
            ($label:expr) => {
                #[cfg(not(target_arch = "wasm32"))]
                if std::env::var_os("HPROF_TIMING").is_some() {
                    eprintln!(
                        "[timing] {}: {:.3}s",
                        $label,
                        _t_build.elapsed().as_secs_f64()
                    );
                }
                #[cfg(target_arch = "wasm32")]
                let _ = $label;
            };
        }

        let n = p1.id_map.len();
        let id_size = p1.id_size;
        let ptr_size = id_size as usize;

        let source_owned = source.clone();
        let open = move || source_owned.open();

        // ── Phase 0: detect ref_size ─────────────────────────────────────
        // Reuse the object-array (addr, count) data already collected in pass1
        // instead of re-scanning the whole file. Array addresses are the id_map
        // entries whose kind == 1 (object array).
        let ref_size = if id_size == 8 {
            let mut array_addr_counts: Vec<(u64, u64)> = Vec::new();
            for i in 0..n {
                if p1.kind[i] == 1 {
                    array_addr_counts.push((p1.id_map.addr_at(i), p1.elem_count[i] as u64));
                }
            }
            detect_ref_size(id_size, &array_addr_counts)
        } else {
            id_size
        } as usize;

        // ── Phase 0b: compute shallow sizes with MAT formula ─────────────
        // Uses per-object kind (0=instance,1=obj_array,2=prim_array,3=class_obj)
        // and raw element counts collected in pass1 — authoritative, no heuristics.
        let mut size_cache: HashMap<u64, usize> = HashMap::new();

        // Set of class-object addresses (used later for edge/class-obj resolution).
        let class_addrs: std::collections::HashSet<u64> = p1.class_map.keys().cloned().collect();

        let mut shallow: Vec<u32> = Vec::with_capacity(n);
        for i in 0..n {
            let cid = p1.class_ids[i];
            let sz = match p1.kind[i] {
                3 => {
                    // Class object: shallow from static fields only, attributed to java.lang.Class.
                    let addr = p1.class_addr_table.get(cid as usize).copied().unwrap_or(0);
                    match p1.class_map.get(&addr) {
                        Some(ci) => class_obj_shallow(ci, ptr_size, ref_size),
                        None => align_up(ptr_size + ref_size, 8) as u32,
                    }
                }
                1 => {
                    // Object array: cid is the array class index (elem count from pass1).
                    obj_array_shallow(p1.elem_count[i] as u64, ptr_size, ref_size)
                }
                2 => {
                    // Primitive array: cid is the raw element type code.
                    let elem_size = HprofType::from_code(cid as u8)
                        .map(|t| t.byte_size())
                        .unwrap_or(1);
                    prim_array_shallow(p1.elem_count[i] as u64, elem_size, ptr_size, ref_size)
                }
                _ => {
                    // Instance: MAT calculateSizeRecursive over the super chain.
                    let addr = p1.class_addr_table.get(cid as usize).copied().unwrap_or(0);
                    if p1.class_map.contains_key(&addr) {
                        instance_shallow_size(
                            addr,
                            &p1.class_map,
                            ptr_size,
                            ref_size,
                            &mut size_cache,
                        )
                    } else {
                        align_up(ptr_size + ref_size, 8) as u32
                    }
                }
            };
            shallow.push(sz);
        }

        // ── Phase 0c: Build class names ──────────────────────────────────
        // MAT keys the class histogram by CLASS-OBJECT identity, not by name: a
        // class loaded by two different loaders yields two histogram rows even
        // though the names are identical. We therefore intern by a u64 key:
        //   - instances / object arrays: the class-object address (loader-distinct)
        //   - primitive arrays: PRIM_KEY_BASE | type_code (boot-loaded, single row)
        //   - class objects (java.lang.Class): the JLC_KEY sentinel (single row)
        const PRIM_KEY_BASE: u64 = 0xFFFF_0000_0000_0000;
        const JLC_KEY: u64 = 0xFFFF_FFFF_FFFF_FFFF;
        let mut class_key_to_idx: HashMap<u64, u32> = HashMap::new();
        let mut class_names: Vec<String> = Vec::new();
        let mut class_loader_id: Vec<u64> = Vec::new();

        let mut get_or_insert_class =
            |key: u64, name: &dyn Fn() -> String, loader: &dyn Fn() -> u64| -> u32 {
                if let Some(&idx) = class_key_to_idx.get(&key) {
                    return idx;
                }
                let idx = class_names.len() as u32;
                class_key_to_idx.insert(key, idx);
                class_names.push(name());
                class_loader_id.push(loader());
                idx
            };

        // Build class_idx array
        let mut class_idx: Vec<u32> = vec![0u32; n];
        // class_addr_to_hist: class object address → histogram idx, for instances
        // (and object arrays). Used to build field_plans_dense after the closure drops.
        let mut class_addr_to_hist: HashMap<u64, u32> = HashMap::new();

        // First pass: populate class_idx for all objects (kind-driven, no heuristics)
        for i in 0..n {
            let cid = p1.class_ids[i];

            match p1.kind[i] {
                3 => {
                    // Class object → single java/lang/Class row (MAT parity). Boot-loaded.
                    class_idx[i] =
                        get_or_insert_class(JLC_KEY, &|| "java/lang/Class".to_string(), &|| 0);
                }
                2 => {
                    // Primitive array: cid is the raw element type code. Boot-loaded.
                    let tc = cid as u8;
                    class_idx[i] = get_or_insert_class(
                        PRIM_KEY_BASE | tc as u64,
                        &|| prim_array_class_name(tc).to_string(),
                        &|| 0,
                    );
                }
                1 => {
                    // Object array: cid indexes the array-class address (loader-distinct).
                    let addr = p1.class_addr_table.get(cid as usize).copied().unwrap_or(0);
                    class_idx[i] = get_or_insert_class(
                        addr,
                        &|| {
                            p1.class_map
                                .get(&addr)
                                .and_then(|ci| p1.strings.get(&ci.name_id).cloned())
                                .unwrap_or_else(|| "[Ljava/lang/Object;".to_string())
                        },
                        &|| p1.class_map.get(&addr).map(|ci| ci.loader_id).unwrap_or(0),
                    );
                    class_addr_to_hist.entry(addr).or_insert(class_idx[i]);
                }
                _ => {
                    // Instance: cid indexes the class-object address (loader-distinct).
                    let addr = p1.class_addr_table.get(cid as usize).copied().unwrap_or(0);
                    class_idx[i] = get_or_insert_class(
                        addr,
                        &|| {
                            p1.class_map
                                .get(&addr)
                                .and_then(|ci| p1.strings.get(&ci.name_id).cloned())
                                .unwrap_or_else(|| format!("unknown@{addr:#x}"))
                        },
                        &|| p1.class_map.get(&addr).map(|ci| ci.loader_id).unwrap_or(0),
                    );
                    class_addr_to_hist.entry(addr).or_insert(class_idx[i]);
                }
            }
        }

        // Free pass1 per-object arrays that are dead after Phase 0b/0c: they
        // are only read to derive `shallow` and `class_idx` above. Releasing
        // them here (~173 MB for a 11 M-object heap) shrinks peak RSS before
        // the edge-scan allocations (inb_flat / fwd_targets).
        //
        // NOTE: `class_ids`/`kind` are also read by the loader-label resolution
        // loop below, which must run AFTER the `get_or_insert_class` closure is
        // last used (~line 913, it holds a mutable borrow of `class_loader_id`).
        // We therefore keep `class_ids` alive until after that loop and free it
        // there; only the vec not needed by the loop is freed here.
        // ── Fold: arrays-by-size histogram ───────────────────────────────
        // Bucket every array (kind 1=obj, 2=prim) by power-of-two element
        // length into a per-kind BTreeMap keyed by `upper_len`, accumulating
        // object count + shallow bytes. Zero-length arrays are tallied
        // separately. Reuses data already in memory (`shallow` is authoritative
        // for arrays here — Phase 0b computed it with the same MAT formulas the
        // sub-pass 2a scan later re-derives) and runs BEFORE `p1.elem_count` is
        // freed on the next line: no extra scan, RSS/runtime-neutral.
        let arrays_by_size = {
            use crate::report::{ArraysBySize, SizeHistogramBucket};
            use std::collections::BTreeMap;
            let mut obj: BTreeMap<u64, (u64, u64)> = BTreeMap::new();
            let mut prim: BTreeMap<u64, (u64, u64)> = BTreeMap::new();
            let mut zero_length_count: u64 = 0;
            for i in 0..n {
                let k = p1.kind[i];
                if k != 1 && k != 2 {
                    continue;
                }
                let len = p1.elem_count[i] as u64;
                if len == 0 {
                    zero_length_count += 1;
                    continue;
                }
                let upper_len = len.next_power_of_two();
                let map = if k == 1 { &mut obj } else { &mut prim };
                let e = map.entry(upper_len).or_insert((0, 0));
                e.0 += 1;
                e.1 += shallow[i] as u64;
            }
            let to_vec = |m: BTreeMap<u64, (u64, u64)>| -> Vec<SizeHistogramBucket> {
                m.into_iter()
                    .map(|(upper_len, (objects, shallow))| SizeHistogramBucket {
                        upper_len,
                        objects,
                        shallow,
                    })
                    .collect()
            };
            ArraysBySize {
                obj_array_buckets: to_vec(obj),
                prim_array_buckets: to_vec(prim),
                zero_length_count,
            }
        };
        p1.elem_count = Vec::new();

        // Precompute per-class instance-field plans once (offset + excluded flag).
        // Borrowed immutably in the hot scan loop — no per-instance allocation.
        let field_plans = build_field_plans(&p1.class_map, &p1.strings, id_size as usize);

        // Dense field_plans indexed by histogram class idx — replaces per-object HashMap
        // lookups in the hot scan loops. Sized by the max instance-class histogram index
        // (built from class_addr_to_hist, which is independent of the get_or_insert_class
        // closure so we don't need to borrow class_names or class_key_to_idx here).
        let n_dense_classes = class_idx
            .iter()
            .copied()
            .max()
            .map(|m| m as usize + 1)
            .unwrap_or(0);
        let mut field_plans_dense: Vec<FieldPlan> = vec![Vec::new(); n_dense_classes];
        for (&class_addr, &hidx) in &class_addr_to_hist {
            if let Some(plan) = field_plans.get(&class_addr) {
                if !plan.is_empty() {
                    field_plans_dense[hidx as usize] = plan.clone();
                }
            }
        }
        drop(field_plans);

        let has_histogram = queries
            .iter()
            .any(|(_, p)| p.kind == crate::query::plan::StageKind::HistogramOnly);
        // Tally per-class counts/shallow NOW (class_idx & shallow are live and about
        // to be compressed). Vectors are sized by the max histogram index seen and
        // padded to class_names.len() in the deferred build below; we cannot read
        // class_names.len() here because the `get_or_insert_class` closure still
        // holds a mutable borrow of `class_names`.
        let hist_tally: Option<(Vec<u64>, Vec<u64>)> = if has_histogram {
            let cap = class_idx
                .iter()
                .copied()
                .max()
                .map(|m| m as usize + 1)
                .unwrap_or(0);
            let mut counts = vec![0u64; cap];
            let mut shallow_totals = vec![0u64; cap];
            for i in 0..n {
                // Skip class objects (kind 3): the OQL SingleScan path only
                // delivers instances/arrays to the visitor (CLASS_DUMP records
                // are never sent to `visit_instance`/`visit_array`), so counting
                // them here would make `SELECT COUNT(*)` over-report relative to
                // `SELECT *` for any pattern matching `java.lang.Class`. Excluding
                // them keeps the histogram (aggregate) path consistent with the
                // scan (projection) path over the same object universe.
                if p1.kind[i] == 3 {
                    continue;
                }
                let ci = class_idx[i] as usize;
                counts[ci] += 1;
                shallow_totals[ci] += shallow[i] as u64;
            }
            Some((counts, shallow_totals))
        } else {
            None
        };

        // Compress class_idx and alloc_stack_serial BEFORE allocating out_degree
        // and in_degree (~4 GB). Both arrays are final at this point and not read
        // again until the retained/report phases. Freeing their ~2 GB dense Vecs
        // here removes ~2 GB from the 2a-scan binding peak (previously class_idx
        // was freed only after the degree arrays were already live).
        let class_idx_c = if compress != crate::cvec::Codec::None {
            let c = crate::cvec::CompressedU32::compress(&class_idx, compress)?;
            class_idx = Vec::new();
            c
        } else {
            crate::cvec::CompressedU32::compress(&class_idx, crate::cvec::Codec::None)?
        };
        let alloc_serial_c = if compress != crate::cvec::Codec::None {
            let c = crate::cvec::CompressedU32::compress(&p1.alloc_stack_serial, compress)?;
            p1.alloc_stack_serial = Vec::new();
            Some(c)
        } else {
            None
        };
        crate::trace::probe(
            "pass2: after early-compress class_idx+alloc_serial (before degree alloc)",
        );

        // ── Phase 1: Sub-pass 2a — count degrees ────────────────────────
        // class_idx is now compressed (freed above); only shallow(~2GB) + id_map(~2GB)
        // remain alongside the new degree arrays (~4 GB total), lowering the scan peak.
        let mut out_degree: Vec<u32> = vec![0u32; n];
        let mut in_degree: Vec<u32> = vec![0u32; n];
        crate::trace::probe("pass2: after out/in_degree alloc");

        // ── OQL: build live resolver, answer HistogramOnly now, arm SingleScan ──
        // Built here (not at the return) because shallow is used by the resolver
        // and compressed/emptied below (after the scan), while class_map/strings
        // are freed only much later. class_names is the live Vec that moves into
        // Graph at the end.
        let query_resolver = crate::query::run::LiveResolver::new(
            &p1.class_map,
            &p1.strings,
            id_size as usize,
            &p1.id_map,
            &shallow,
        );
        // Arm one executor per valid SingleScan query. Plan-time field
        // validation runs here (earliest point a live schema exists): a query
        // referencing a field absent from its FROM class's super-chain is
        // rejected up front with a pre-set error QueryResult, and its executor
        // is not armed. `scan_outcomes` records, in `queries` input order for
        // SingleScan entries, either Ok(slot) (armed → routed via the driver's
        // QueryExecState) or Err(pre-set validation-error result at that slot).
        //
        // Phase-1 queries (`finalize_at == P1`) are armed as row executors;
        // cross-phase queries (`finalize_at == P2` for RefWalk, `P3` for
        // @retainedHeapSize/dominators) are armed in carry mode so their matched
        // dense indices are carried to the late stage instead of finalized here
        // (the reference CSR / retained sizes don't exist yet during the scan).
        let mut scan_execs: Vec<(usize, crate::query::execute::SingleScanExecutor<_>)> = Vec::new();
        let mut scan_outcomes: Vec<Result<usize, (usize, crate::query::model::QueryResult)>> =
            Vec::new();
        for (slot, (q, plan)) in queries.iter().enumerate() {
            if plan.kind != crate::query::plan::StageKind::SingleScan
                && plan.kind != crate::query::plan::StageKind::GroupBy
            {
                continue;
            }
            match crate::query::plan::validate_fields(q, &query_resolver) {
                Ok(()) => {
                    let cross_phase = plan.finalize_at != crate::query::plan::Phase::P1;
                    let mut exec = if cross_phase {
                        crate::query::execute::SingleScanExecutor::new_carry(
                            q,
                            plan,
                            &query_resolver,
                            crate::query::carry::Carry::index_only(
                                crate::query::carry::DEFAULT_CARRY_CAP,
                            ),
                        )
                    } else {
                        crate::query::execute::SingleScanExecutor::new(q, plan, &query_resolver)
                    };
                    // Inject any resolved IN-subquery membership sets for this
                    // slot (computed by an earlier inner scan). Absent for a
                    // query without IN-subqueries.
                    if let Some(sets) = in_sets_by_slot.remove(&slot) {
                        exec.set_in_subquery_sets(sets);
                    }
                    // Inject pre-evaluated EXISTS/NOT EXISTS boolean results for
                    // this slot (computed by an earlier inner scan). Absent for a
                    // query without EXISTS subqueries.
                    if let Some(bools) = exists_bools_by_slot.remove(&slot) {
                        exec.set_exists_results(bools);
                    }
                    scan_execs.push((slot, exec));
                    scan_outcomes.push(Ok(slot));
                }
                Err(e) => {
                    scan_outcomes.push(Err((
                        slot,
                        crate::query::model::QueryResult {
                            name: String::new(),
                            oql: String::new(),
                            columns: Vec::new(),
                            rows: Vec::new(),
                            row_count: 0,
                            truncated: false,
                            error: Some(e.0),
                            note: None,
                            viz: None,
                            elapsed_ms: None,
                        },
                    )));
                }
            }
        }
        let mut scan_driver =
            crate::query::run::ScanDriver::new(scan_execs).with_src_capture(opts.reachable_only);
        // Computed once: does any armed query target an array class? Gates the
        // per-array name construction in the 2a scan so instance-only query sets
        // pay zero extra allocation on the array path.
        let scan_wants_arrays = scan_driver.wants_arrays();

        // When --ref-paths or --field-stats is set, also build named plans (field name strings).
        // For --ref-paths: used during the forward-CSR fill to annotate each edge with its field name.
        // For --field-stats: used to populate class_ref_field_names on the Graph.
        // Gated: the extra allocations are acceptable only under these explicit flags.
        let field_plans_named_dense: Vec<FieldPlanNamed> =
            if opts.ref_paths || opts.field_stats || opts.obj_graph {
                let named = build_field_plans_named(&p1.class_map, &p1.strings, id_size as usize);
                let mut dense: Vec<FieldPlanNamed> = vec![Vec::new(); n_dense_classes];
                for (&class_addr, &hidx) in &class_addr_to_hist {
                    if let Some(plan) = named.get(&class_addr) {
                        if !plan.is_empty() {
                            dense[hidx as usize] = plan.clone();
                        }
                    }
                }
                dense
            } else {
                Vec::new()
            };

        // Collect thread object addresses for capture during 2a scan.
        let capture_thread_addrs: std::collections::HashSet<u64> =
            p1.thread_serial_to_obj_id.values().copied().collect();
        let mut captured_thread_blobs: HashMap<u64, (u64, Vec<u8>)> = HashMap::new();

        // Pre-locate java/lang/System class addr and "props" name_id so we can
        // capture the props static field address during the 2a CLASS_DUMP scan.
        let (system_class_addr, props_name_id) = {
            let props_nid = p1
                .strings
                .iter()
                .find(|(_, v)| v.as_str() == "props")
                .map(|(&k, _)| k)
                .unwrap_or(0);
            let system_caddr = p1
                .class_map
                .iter()
                .find(|(_, ci)| {
                    p1.strings
                        .get(&ci.name_id)
                        .map(|s| s.as_str() == "java/lang/System")
                        .unwrap_or(false)
                })
                .map(|(&a, _)| a)
                .unwrap_or(0);
            (system_caddr, props_nid)
        };
        let mut captured_props_addr: u64 = 0;

        // ── Sub-pass 2a scan ─────────────────────────────────────────────
        // Create field-decode state before the scan so it can be fused into
        // the single 2a pass, eliminating the separate build_field_decode_views
        // rescan.  shallow is fully populated before this point (sized in
        // Phase 0b), and p1/opts live through the end of this block.
        let mut fd_state = fielddecode::FieldDecodeState::new(
            id_size,
            opts.collections,
            fielddecode::CollCaps::from_size(opts.report_size),
        );
        {
            let mut r = source.open()?;
            // Scratch buffer reused across INSTANCE_DUMP and OBJ_ARRAY_DUMP reads (fix #6)
            let mut scratch: Vec<u8> = Vec::with_capacity(4096);

            loop {
                let (tag, length) = match r.next_record()? {
                    None => break,
                    Some(h) => h,
                };
                let result: io::Result<()> = (|| match tag {
                    tags::HEAP_DUMP | tags::HEAP_DUMP_SEGMENT => Self::scan_heap_2a(
                        &mut r,
                        id_size,
                        length,
                        &p1.id_map,
                        &class_addr_to_hist,
                        &field_plans_dense,
                        &mut out_degree,
                        &mut in_degree,
                        &mut scratch,
                        if scan_driver.is_empty() {
                            None
                        } else {
                            Some(&mut scan_driver as &mut dyn crate::query::ObjectVisitor)
                        },
                        scan_wants_arrays,
                        &p1.class_map,
                        &p1.strings,
                        &capture_thread_addrs,
                        &mut captured_thread_blobs,
                        &std::collections::HashSet::new(),
                        &mut HashMap::new(),
                        system_class_addr,
                        props_name_id,
                        &mut captured_props_addr,
                        Some(&mut fd_state),
                        &p1,
                        &shallow,
                        &opts.coll_descs,
                    ),
                    tags::HEAP_DUMP_END => Err(io::Error::new(HEAP_DUMP_END_KIND, "heap_dump_end")),
                    _ => r.skip(length),
                })();
                match result {
                    Ok(()) => {}
                    Err(e) if e.kind() == HEAP_DUMP_END_KIND => break,
                    Err(e)
                        if e.kind() == ErrorKind::UnexpectedEof
                            || e.kind() == ErrorKind::InvalidData =>
                    {
                        break;
                    }
                    Err(e) => return Err(e),
                }
            }
        }
        crate::trace::probe("pass2: after 2a scan (out+in_degree filled)");
        t_phase!("2a scan done");

        // Always-on field-decode views (collections, arrays, references). The
        // state was populated inline during the 2a scan above (fused pass),
        // so this just runs the post-scan fold without any additional I/O.
        // Must be called here — fd_state borrows &p1 and &shallow which are
        // mutated below (kind/class_ids freed, shallow zero-size patched).
        let (
            fd_collections,
            fd_references,
            fd_referent_idx,
            fd_null_referent_count,
            fd_attribution_raw,
            fd_fields_by_size_raw,
            fd_coll_values_raw,
            fd_node_kv,
            fd_attribution_trunc,
            fd_dbb_capacity_sum,
            fd_tl_null_key_count,
            fd_tl_entry_records,
        ) = fd_state.finish(&p1)?;
        crate::trace::probe("pass2: after field_decode_views (fused into 2a, no extra scan)");
        crate::trace::trim();
        crate::trace::probe("pass2: after trim post-fielddecode");
        t_phase!("fielddecode done");

        // Finalize SingleScan query results while class metadata is still live.
        // (No name/OQL source yet — the CLI wiring task fills these; empty slices
        // are safe.) Dropping the driver + resolver here ends their borrows of
        // p1.class_map / p1.strings before those maps are freed below.
        // Cross-phase-aware query state: the driver returns armed executors as
        // finished (Phase-1) or pending carries (Phase-3 @retainedHeapSize),
        // each tagged with its `slot` (input index in `queries`). Validation
        // errors and histogram results are pushed as finished at their slots.
        // The caller reassembles by slot after the late stage, so no positional
        // `query_order` reorder is needed here.
        // Build the query-gated RefWalk CSR + tail table BEFORE `finish_state`
        // consumes the driver. `None` on a non-RefWalk run — the late window
        // then keeps its empty slices (byte/RSS-identical to before). `n` is the
        // dense object count, so `fwd_off` has the standard `n+1` length.
        let refwalk_csr = scan_driver.take_refwalk_csr(n).map(|(off, tgt, fid)| {
            let truncated = scan_driver.refwalk_truncated();
            let field_names = scan_driver.take_refwalk_field_names().unwrap_or_default();
            let tails = scan_driver.take_refwalk_tails().unwrap_or_default();
            crate::query::refwalk::RefWalkCsr {
                fwd_off: off,
                fwd_tgt: tgt,
                fwd_field: fid,
                field_names,
                tails,
                truncated,
            }
        });

        // Decode the query-gated toString(s) capture into dense_idx → String.
        // One `scan_prim_arrays` pass over the dump, then dropped. Empty (no I/O)
        // when no toString(s) query ran — non-toString runs are byte/RSS-identical.
        // Capture the truncated flag BEFORE taking the capture state (take replaces
        // the capture with an empty one; reading after would always yield false).
        let string_values_truncated = scan_driver.string_capture_truncated();
        let string_values: std::collections::HashMap<u32, String> =
            if let Some(capture) = scan_driver.take_string_capture() {
                capture.decode_all(&open, id_size)?
            } else {
                std::collections::HashMap::new()
            };

        let mut query_state = scan_driver.finish_state();
        for outcome in scan_outcomes {
            if let Err((slot, err_result)) = outcome {
                query_state.push_finished(slot, err_result);
            }
        }
        drop(query_resolver);

        // Class objects already map to the java/lang/Class row (JLC_KEY) from Phase 0c.
        let jlc_idx = get_or_insert_class(JLC_KEY, &|| "java/lang/Class".to_string(), &|| 0);

        // ── Build class_obj_class_idx ─────────────────────────────────────
        // For each class object, record the histogram row of the class it
        // represents. Under identity keying, that row is keyed by the class
        // object's own address (the same key instances of that class use).
        let mut class_obj_class_idx: HashMap<u32, u32> = HashMap::new();
        for i in 0..n {
            let addr = p1.id_map.addr_at(i);
            if class_addrs.contains(&addr) {
                let ci = p1.class_map.get(&addr);
                // Determine the histogram row this class-object represents.
                // Must use the SAME key that instances of the class use:
                //   - Primitive array class-objects ([I, [B, ...): PRIM_KEY_BASE|tc
                //   - java/lang/Class class-object: JLC_KEY
                //   - All other class-objects: addr-based key (same as instances)
                let name = ci.and_then(|c| p1.strings.get(&c.name_id));
                let idx = if let Some(n) = name {
                    if let Some(tc) = prim_array_type_code(n) {
                        // Class-object for a primitive array: register under PRIM_KEY
                        get_or_insert_class(
                            PRIM_KEY_BASE | tc as u64,
                            &|| prim_array_class_name(tc).to_string(),
                            &|| 0,
                        )
                    } else if n == "java/lang/Class" {
                        // Class-object for java.lang.Class: register under JLC_KEY
                        get_or_insert_class(JLC_KEY, &|| "java/lang/Class".to_string(), &|| 0)
                    } else {
                        get_or_insert_class(addr, &|| n.to_string(), &|| {
                            ci.map(|c| c.loader_id).unwrap_or(0)
                        })
                    }
                } else {
                    get_or_insert_class(addr, &|| format!("unknown@{addr:#x}"), &|| {
                        ci.map(|c| c.loader_id).unwrap_or(0)
                    })
                };
                class_obj_class_idx.insert(i as u32, idx);
            }
        }
        let _ = jlc_idx;
        let _ = jlc_idx;

        // ── OQL HistogramOnly: build ClassSummary + run now ──────────────────
        // Deferred to here (after `get_or_insert_class`'s last use) because that
        // closure mutably borrows `class_names`; the per-class tally was captured
        // above while class_idx/shallow were still live. Results are appended so
        // histogram queries land alongside the SingleScan results in the return.
        if let Some((mut counts, mut shallow_totals)) = hist_tally {
            let n_classes = class_names.len();
            counts.resize(n_classes, 0);
            shallow_totals.resize(n_classes, 0);
            // Normalize each class name to the pretty/dotted display form the OQL
            // FROM patterns use (`[C` -> `char[]`, `java/lang/String` ->
            // `java.lang.String`). class_names holds RAW JVM descriptors for
            // primitive/object arrays, so matching those directly against a
            // `FROM char[]` pattern would fail (raw-vs-pretty asymmetry) and
            // silently return COUNT 0. The scan path already projects the pretty
            // name via visit_array, so this makes the histogram/aggregate path
            // agree with it. `pretty_names` outlives `summaries` and the
            // `run_histogram` loop below (all in this block).
            let pretty_names: Vec<String> = (0..n_classes)
                .map(|ci| crate::report::pretty_class_name(&class_names[ci]))
                .collect();
            let summaries: Vec<crate::query::histogram::ClassSummary> = (0..n_classes)
                .map(|ci| crate::query::histogram::ClassSummary {
                    name: pretty_names[ci].as_str(),
                    count: counts[ci],
                    shallow_total: shallow_totals[ci],
                })
                .collect();
            for (slot, (q, plan)) in queries.iter().enumerate() {
                if plan.kind == crate::query::plan::StageKind::HistogramOnly {
                    let r = crate::query::histogram::run_histogram(q, plan, &summaries);
                    query_state.push_finished(slot, r);
                }
            }
        }

        // Resolve each distinct non-boot class-loader OBJECT address to the
        // class NAME of that loader object (e.g.
        // "jdk/internal/loader/ClassLoaders$AppClassLoader"), so the report
        // layer can label loaders instead of showing a raw address. Runs here,
        // AFTER the last `get_or_insert_class` use (that closure mutably borrows
        // `class_loader_id`), and BEFORE class_map/strings/id_map are freed or
        // moved (~lines 1013-1014, ~1198) and before `class_ids`/`kind` are
        // freed just below. Bounded by #distinct loaders (tens to low
        // hundreds), so it costs no per-object RSS. Boot loader (addr 0) is
        // labeled `<boot>` in the report layer.
        let mut loader_labels: std::collections::HashMap<u64, String> =
            std::collections::HashMap::new();
        for &loader_addr in &class_loader_id {
            if loader_addr == 0 {
                continue; // boot loader handled in report layer
            }
            if loader_labels.contains_key(&loader_addr) {
                continue;
            }
            // Resolve: loader_addr -> object index -> its class-obj addr -> name.
            if let Some(idx) = p1.id_map.index_of(loader_addr) {
                // Only plain instances (kind 0) are real loader objects.
                if p1.kind[idx] == 0 {
                    let cid = p1.class_ids[idx];
                    let class_addr = p1.class_addr_table.get(cid as usize).copied().unwrap_or(0);
                    if let Some(name) = p1
                        .class_map
                        .get(&class_addr)
                        .and_then(|ci| p1.strings.get(&ci.name_id))
                    {
                        loader_labels.insert(loader_addr, name.clone());
                    }
                }
            }
        }

        // Ensure no zero shallow sizes for instances/arrays (fall back to minimum).
        // Class objects (kind==3) are exempt: MAT reports 0 shallow for a class
        // whose static-field bytes sum to 0 (e.g. array classes like `[I`), so we
        // must not bump those to the object minimum.
        let min_obj = align_up(ptr_size + ref_size, 8) as u32;
        for (i, s) in shallow.iter_mut().enumerate() {
            if *s == 0 && p1.kind[i] != 3 {
                *s = min_obj;
            }
        }
        // kind is now dead (last user was the zero-shallow loop above and the
        // loader-label loop earlier). Free before the GC-root and fielddecode
        // phases to lower peak RSS on large dumps.
        p1.kind = Vec::new();
        crate::trace::trim();

        // ── Phase 2: Build GC root indices ───────────────────────────────
        let mut gc_root_set: std::collections::HashSet<u32> = std::collections::HashSet::new();
        // Per-index representative root type (minimum sub-tag when an index has
        // several root records), carried into Graph for B1 grouping + why-alive.
        let mut root_type_of: std::collections::HashMap<u32, u8> = std::collections::HashMap::new();
        let note_type = |m: &mut std::collections::HashMap<u32, u8>, idx: u32, ty: u8| {
            m.entry(idx).and_modify(|e| *e = (*e).min(ty)).or_insert(ty);
        };
        for (&addr, &ty) in p1.gc_root_addrs.iter().zip(p1.gc_root_types.iter()) {
            if let Some(idx) = p1.id_map.index_of(addr) {
                gc_root_set.insert(idx as u32);
                note_type(&mut root_type_of, idx as u32, ty);
            }
        }
        // Add implicit roots: non-array boot-loader classes (loader_id==0) if no sticky roots
        if !p1.has_sticky_class_roots {
            for (&caddr, ci) in &p1.class_map {
                if ci.loader_id == 0 {
                    // Check it's not an array class (name doesn't start with '[')
                    let is_array = p1
                        .strings
                        .get(&ci.name_id)
                        .map(|n| n.starts_with('['))
                        .unwrap_or(false);
                    if !is_array {
                        if let Some(idx) = p1.id_map.index_of(caddr) {
                            gc_root_set.insert(idx as u32);
                            note_type(&mut root_type_of, idx as u32, heap::ROOT_SYSTEM_CLASS);
                        }
                    }
                }
            }
        }
        // ── addSystemClassRootsIfMissing: boot-loader non-array classes not yet roots ─
        let mut synthetic_root_count = 0usize;
        // MAT materializes a synthetic <system class loader> object at 0x0 of
        // class java/lang/ClassLoader (no HPROF record). Capture that class's
        // instance shallow size so the report layer can inject the object.
        let mut system_classloader_shallow: Option<u32> = None;
        for (&caddr, ci) in &p1.class_map {
            if ci.loader_id != 0 {
                continue;
            }
            let name = p1.strings.get(&ci.name_id);
            if name.map(|n| n == "java/lang/ClassLoader").unwrap_or(false) {
                system_classloader_shallow = Some(instance_shallow_size(
                    caddr,
                    &p1.class_map,
                    ptr_size,
                    ref_size,
                    &mut size_cache,
                ));
            }
            let is_array = name.map(|n| n.starts_with('[')).unwrap_or(false);
            let is_prim_array = name
                .map(|n| is_primitive_array_class_name(n))
                .unwrap_or(false);
            if !should_add_system_class_root(is_array, is_prim_array, p1.has_sticky_class_roots) {
                continue;
            }
            if let Some(idx) = p1.id_map.index_of(caddr) {
                if !gc_root_set.contains(&(idx as u32)) {
                    gc_root_set.insert(idx as u32);
                    note_type(&mut root_type_of, idx as u32, heap::ROOT_SYSTEM_CLASS);
                    synthetic_root_count += 1;
                }
            }
        }

        // Resolve STACK_TRACE/STACK_FRAME into pre-rendered thread stacks while
        // pass1's string/class tables are still alive (they are freed just
        // below). Only traces that carry frames are kept. Small — one entry per
        // thread trace, off the per-object RSS budget.
        let thread_stacks = build_thread_stacks(&p1);

        // Pre-resolve every DISTINCT non-zero alloc stack-trace serial into its
        // frame lines while the STACK_FRAME/STACK_TRACE + string/class tables
        // are still alive (freed just below). Bounded by the number of distinct
        // traces (hundreds), so it stays off the per-object RSS budget.
        let alloc_frames_by_serial: Option<std::collections::HashMap<u32, Vec<String>>> =
            Some(resolve_alloc_frames(&p1));

        // Decode each thread's java.lang.Thread.name. Thread instance blobs were
        // captured during the 2a scan; remaining hops (String objects, backing
        // arrays) need 2 more targeted collect_blobs calls instead of 3.
        let thread_props = resolve_thread_names(&open, &p1, captured_thread_blobs)?;
        t_phase!("thread_names done");

        // Opt-in approximate duplicate-java.lang.String report. Runs two extra
        // full-file scans and keeps only hashes+lengths+counts (never the
        // decoded bytes), so RSS stays bounded. Must run while class_map/strings
        // are still alive (freed just below). `None` on the default path = zero
        // extra work, zero RSS.
        let dup_strings = if opts.find_duplicates {
            Some(resolve_duplicate_strings(&open, &p1)?)
        } else {
            None
        };

        // Opt-in duplicate-primitive-array waste scan. One extra full-file pass
        // alongside --find-duplicates; keeps only a hash→(count,type) map plus
        // an addr→hash map so we can later compute holder classes.
        let dup_prim_arrays: Option<DupPrimArrays> = if opts.find_duplicates {
            let (mut dpa, dup_addrs) = compute_dup_prim_arrays(&open, p1.id_size)?;
            // If --collections is also on we have the class_map/strings needed
            // to build FieldPlans and find which classes hold the most dup arrays.
            if opts.collections && !dup_addrs.is_empty() {
                dpa.top_array_holders =
                    compute_dup_array_holders(&open, &p1, &dup_addrs, p1.id_size)?;
            }
            Some(dpa)
        } else {
            None
        };

        // Opt-in boxed-number holder scan. Two extra full-file passes (collect
        // boxed-type addresses, then count references) when --collections is on.
        let boxed_number_holders: Vec<crate::report::BoxedNumberHolder> = if opts.collections {
            compute_boxed_holders(&open, &p1, p1.id_size)?
        } else {
            Vec::new()
        };

        // Capture java.lang.System's static `props` (a Properties/Hashtable of
        // String->String) via a bounded multi-pass worklist, while class_map/
        // strings/id_map are still alive. All captured sets are bounded (ONE
        // props object, capped at 4096 entries + their Strings/arrays), so this
        // stays off the per-object RSS budget on multi-GB dumps. Derives a JVM
        // version from the decoded properties. Falls back to empty/None (never
        // garbage) if the layout does not match the Hashtable form.
        let (system_properties, jvm_version) =
            resolve_system_properties(&open, &p1, captured_props_addr)?;
        t_phase!("system_props done");

        // Free class_ids now: build_field_decode_views was its last reader
        // (class_name_of_index uses it for referent class lookups). Releasing
        // here keeps peak RSS low before the edge-scan allocations.
        p1.class_ids = Vec::new();

        // class_map + strings are no longer needed; free before the large edge
        // arrays get allocated in Phase 3/4 to lower peak RSS. The STACK_FRAME/
        // STACK_TRACE maps were just consumed by build_thread_stacks and are
        // likewise dead — free them here too so they don't linger through the
        // peak-binding dominator/retained phases.
        p1.class_map = std::collections::HashMap::new();
        p1.strings = std::collections::HashMap::default();
        p1.stack_frames = std::collections::HashMap::default();
        p1.stack_traces = std::collections::HashMap::new();
        p1.stack_trace_thread = std::collections::HashMap::default();

        // ── Resolve thread→local synthetic edges ─────────────────────────
        let mut synthetic_edges: Vec<(u32, u32)> = Vec::new();
        // Per-thread count of local roots that resolve to a live object. Sized
        // by #threads only (bounded), so it stays off the per-object RSS budget.
        let mut thread_local_counts: std::collections::HashMap<u32, u64> =
            std::collections::HashMap::new();
        // Bounded per-thread sample of local object indices. Only populated when
        // the opt-in `--thread-locals` flag is set; empty otherwise (zero cost on
        // the default path).
        let mut thread_local_samples: std::collections::HashMap<u32, Vec<u32>> =
            std::collections::HashMap::new();
        // Gated frame→local map: per-thread (frame_number, local_idx) pairs, used
        // to build MAT's per-frame significant-locals interleave. Only populated
        // when `--thread-locals` is set; `u32::MAX` frame_number = no frame (JNI
        // local / native stack / thread block). Bounded per thread by the same
        // per-thread cap. Zero cost on the default path.
        let mut thread_local_frame_samples: std::collections::HashMap<u32, Vec<(u32, u32)>> =
            std::collections::HashMap::new();
        for &(thread_serial, frame_number, local_addr) in &p1.thread_local_pairs {
            let thread_obj_addr = match p1.thread_serial_to_obj_id.get(&thread_serial) {
                Some(&a) => a,
                None => continue,
            };
            let thread_idx = match p1.id_map.index_of(thread_obj_addr) {
                Some(i) => i as u32,
                None => continue,
            };
            let local_idx = match p1.id_map.index_of(local_addr) {
                Some(i) => i as u32,
                None => continue,
            };
            if thread_idx != local_idx {
                synthetic_edges.push((thread_idx, local_idx));
                *thread_local_counts.entry(thread_serial).or_insert(0) += 1;
                let sample = thread_local_samples.entry(thread_serial).or_default();
                if sample.len() < opts.thread_locals_per_thread {
                    sample.push(local_idx);
                }
                if opts.thread_locals_per_thread > 0 {
                    let fs = thread_local_frame_samples.entry(thread_serial).or_default();
                    if fs.len() < opts.thread_locals_per_thread {
                        fs.push((frame_number, local_idx));
                    }
                }
            }
        }
        // Dedup synthetic edges (same thread may reference same local multiple times)
        synthetic_edges.sort_unstable();
        synthetic_edges.dedup();

        // Add synthetic edge degrees to out_degree/in_degree
        for &(src, dst) in &synthetic_edges {
            out_degree[src as usize] += 1;
            in_degree[dst as usize] += 1;
        }

        // Sum in_degree by class NOW — after synthetic edges, before fwd_targets.
        // class_idx was compressed before the 2a scan; restoring it here costs
        // ~2 GB but fwd_targets doesn't exist yet, so this coexistence is
        // ~2 GB cheaper than the previous approach (restoring after fwd_targets).
        let n_hist_classes = class_names.len();
        let mut incoming_refs_per_class: Vec<u64> = vec![0u64; n_hist_classes];
        {
            let class_idx_restored = class_idx_c.restore()?;
            for (i, &d) in in_degree.iter().enumerate() {
                let ci = class_idx_restored[i] as usize;
                if ci < n_hist_classes {
                    incoming_refs_per_class[ci] += d as u64;
                }
            }
        } // class_idx_restored dropped here
        crate::trace::probe("pass2: incoming_refs_per_class computed (class_idx restore dropped)");

        let mut gc_root_indices: Vec<u32> = gc_root_set.into_iter().collect();
        gc_root_indices.sort_unstable();
        // Per-root type aligned 1:1 with the sorted indices. Every index in the
        // set came from a note_type call, so the lookup always hits; fall back
        // to ROOT_UNKNOWN defensively.
        let gc_root_types: Vec<u8> = gc_root_indices
            .iter()
            .map(|idx| root_type_of.get(idx).copied().unwrap_or(heap::ROOT_UNKNOWN))
            .collect();

        // ── Phase 3: Build forward-CSR offsets (prefix sum only) ────────
        // Done BEFORE compress-cold so out_degree (2 GB) is freed before the
        // compression transients, lowering the binding peak by ~2 GB.
        let mut fwd_offsets: Vec<u32> = Vec::with_capacity(n + 1);
        fwd_offsets.push(0u32);
        let mut edge_acc: u64 = 0;
        for i in 0..n {
            edge_acc += out_degree[i] as u64;
            if edge_acc > u32::MAX as u64 {
                return Err(io::Error::new(
                    ErrorKind::InvalidData,
                    format!(
                        "forward edge count exceeds u32::MAX ({edge_acc}); \
                         dump has too many references for the u32 CSR"
                    ),
                ));
            }
            fwd_offsets.push(edge_acc as u32);
        }
        crate::trace::drop_vec(out_degree); // dead after prefix sum

        // Compress shallow NOW, before fwd_targets (~6GB) is allocated.
        // class_idx and alloc_serial were already compressed before the 2a scan.
        let shallow_c = crate::cvec::CompressedU32::compress(&shallow, compress)?;
        if compress != crate::cvec::Codec::None {
            crate::trace::drop_vec(shallow);
            shallow = Vec::new();
        }
        // Compress in_degree (~2GB raw counts) before fwd_targets alloc.
        // in_degree is completely idle during the fwd fill; compressing it here
        // drops ~1.8 GB from the fwd-fill peak. Restored after the fwd fill for
        // the prefix-sum that builds in_cursors. Delta + zstd is very effective
        // on monotonically-constrained data (counts of edges per node).
        let in_degree_c = if compress != crate::cvec::Codec::None {
            let c = crate::cvec::CompressedU32::compress(&in_degree, compress)?;
            crate::trace::drop_vec(in_degree);
            in_degree = Vec::new();
            Some(c)
        } else {
            None
        };
        crate::trace::probe("pass2: after compress-cold shallow+in_degree (before fwd_targets)");
        t_phase!("compress-cold done");

        // ── Phase 3b: Build forward CSR ──────────────────────────────────
        // The forward fill runs FIRST (inside build); the inbound CSR is
        // deferred into InboundBuilder so its ~5.5GB does not coexist with
        // the rpo phase's arrays. The forward fill never touches inb_flat.
        let total_edges = *fwd_offsets.last().unwrap() as usize;
        let mut fwd_targets = crate::chunkvec::ChunkU32::zeroed(total_edges);
        // Optional per-edge field-name index, populated only under --ref-paths.
        // Parallel to fwd_targets (same indexing). 0 = "no name".
        // For --obj-graph: skip on large dumps (>100M edges) since Vec<u16> at that
        // scale costs ~15 GB on a 34G dump, negating the RSS savings.
        const FIELD_NAME_EDGE_CAP: usize = 100_000_000;
        let want_field_names =
            opts.ref_paths || (opts.obj_graph && total_edges <= FIELD_NAME_EDGE_CAP);
        let mut fwd_field_name_idx_opt: Option<Vec<u16>> = if want_field_names {
            Some(vec![0u16; total_edges])
        } else {
            None
        };
        // Interned field-name pool (pool[0] = ""). Built under --ref-paths or --obj-graph.
        let mut field_name_pool: Vec<String> = if want_field_names {
            let pool = vec![String::new()]; // index 0 = no name
            pool
        } else {
            Vec::new()
        };
        // Reverse map name -> pool index, for dedup during the fill.
        let mut field_name_pool_idx: std::collections::HashMap<String, u16> =
            std::collections::HashMap::new();
        crate::trace::probe("pass2: after fwd_targets alloc");
        // B3: no fwd_cursor clone. fwd_offsets is advanced in place as the
        // write cursor during the fill, then restored by right-shift below.
        {
            let mut r = source.open()?;
            let mut scratch: Vec<u8> = Vec::with_capacity(4096);
            let mut inb_flat_stub = crate::chunkvec::ChunkU32::zeroed(0);
            let mut in_degree_stub: Vec<u32> = Vec::new();
            loop {
                let (tag, length) = match r.next_record()? {
                    None => break,
                    Some(h) => h,
                };
                let result: io::Result<()> = (|| match tag {
                    tags::HEAP_DUMP | tags::HEAP_DUMP_SEGMENT => Self::fill_heap_2b(
                        &mut r,
                        id_size,
                        length,
                        &p1.id_map,
                        &class_addr_to_hist,
                        &field_plans_dense,
                        &field_plans_named_dense,
                        true,
                        false,
                        &mut fwd_targets,
                        &mut fwd_offsets,
                        &mut fwd_field_name_idx_opt,
                        &mut field_name_pool,
                        &mut field_name_pool_idx,
                        &mut inb_flat_stub,
                        &mut in_degree_stub,
                        &mut scratch,
                    ),
                    tags::HEAP_DUMP_END => Err(io::Error::new(HEAP_DUMP_END_KIND, "heap_dump_end")),
                    _ => r.skip(length),
                })();
                match result {
                    Ok(()) => {}
                    Err(e) if e.kind() == HEAP_DUMP_END_KIND => break,
                    Err(e)
                        if e.kind() == ErrorKind::UnexpectedEof
                            || e.kind() == ErrorKind::InvalidData =>
                    {
                        break;
                    }
                    Err(e) => return Err(e),
                }
            }
        }
        // Synthetic thread->local FORWARD edges. Their degrees were added to
        // out_degree above, so each fits within its node's slice.
        for &(src, dst) in &synthetic_edges {
            let pos = fwd_offsets[src as usize] as usize;
            fwd_targets.set(pos, dst);
            // Synthetic edges have no field name (index 0 = "no name").
            fwd_offsets[src as usize] += 1;
        }

        // B3 restore: each fwd_offsets[i] (i in 0..n) has advanced to node i's
        // END index; right-shift over (1..=n).rev() so fwd_offsets[node]..
        // fwd_offsets[node+1] again bounds node's slice. fwd_offsets[n]
        // (total_edges) was never a cursor and is preserved by starting at n.
        for i in (1..=n).rev() {
            fwd_offsets[i] = fwd_offsets[i - 1];
        }
        fwd_offsets[0] = 0;

        // Restore in_degree from compressed blob (if compressed above).
        if let Some(c) = in_degree_c {
            in_degree = c.restore()?;
        }

        // Prefix-sum in_degree counts → START cursors for the deferred inbound
        // build. in_degree[i] becomes node i's inbound slice START; total_inb
        // is the flat inbound length. inb_flat is NOT allocated here.
        let mut total_inb: u64 = 0;
        for d in in_degree.iter_mut() {
            let cnt = *d as u64;
            *d = total_inb as u32;
            total_inb += cnt;
        }
        // Inbound and forward edges are the same set counted from opposite ends,
        // so the forward-CSR guard above already bounds this. Assert the
        // invariant so a future divergence surfaces instead of silently
        // truncating a cursor.
        debug_assert!(
            total_inb <= u32::MAX as u64,
            "inbound edge total {total_inb} exceeds u32::MAX"
        );
        let in_cursors = in_degree; // renamed for clarity: prefix-summed START cursors

        // Precompute source_name before moving p1.id_map into InboundBuilder.
        let source_name = source.display_name().to_string();

        let alloc_stack_serial = std::mem::take(&mut p1.alloc_stack_serial);
        let mut gc_root_tag_counts: Vec<(u8, u64)> = p1
            .gc_root_tag_counts
            .iter()
            .map(|(&t, &c)| (t, c))
            .collect();
        gc_root_tag_counts.sort_by(|a, b| b.1.cmp(&a.1).then(a.0.cmp(&b.0)));
        let record_census = RecordCensus {
            utf8_records: p1.utf8_records,
            load_class_records: p1.load_class_records,
            unload_class_records: p1.unload_class_records,
            stack_frame_records: p1.stack_frame_records,
            stack_trace_records: p1.stack_trace_records,
            heap_dump_segments: p1.heap_dump_segments,
            instance_dumps: p1.instance_count,
            obj_array_dumps: p1.obj_array_count,
            prim_array_dumps: p1.prim_array_count,
            class_dumps: p1.class_dump_count,
            gc_root_tag_counts,
        };
        let mut graph = Graph {
            n,
            format: p1.format,
            file_size: p1.file_size,
            source_name,
            file_path: source.file_path().to_string(),
            id_size,
            ref_size: ref_size as u8,
            header_timestamp_ms: p1.header_timestamp_ms,
            gc_root_indices,
            gc_root_types,
            shallow,
            class_idx,
            class_names,
            class_loader_id,
            loader_labels,
            thread_stacks,
            thread_props,
            thread_local_counts,
            thread_local_samples,
            thread_local_frame_samples,
            system_properties,
            jvm_version,
            class_obj_class_idx,
            fwd_offsets,
            fwd_targets,
            synthetic_root_count,
            system_classloader_shallow,
            idom: Vec::new(),
            retained: Vec::new(),
            has_same_class_ancestor: crate::bitset::Bitset::default(),
            alloc_stack_serial,
            alloc_frames_by_serial,
            record_census,
            dup_strings,
            dup_prim_arrays,
            boxed_number_holders,
            arrays_by_size,
            incoming_refs_per_class,
            collections: fd_collections,
            references: fd_references,
            reference_referent_idx: fd_referent_idx,
            reference_null_referent_count: fd_null_referent_count,
            collection_attribution_raw: fd_attribution_raw,
            collection_attribution_truncated: fd_attribution_trunc,
            fields_by_size_raw: fd_fields_by_size_raw,
            coll_values_raw: fd_coll_values_raw,
            node_kv: fd_node_kv,
            direct_byte_buffer_capacity_sum: fd_dbb_capacity_sum,
            thread_local_null_key_count: fd_tl_null_key_count,
            tl_entry_records: fd_tl_entry_records,
            fwd_field_name_idx: fwd_field_name_idx_opt,
            field_name_pool: if opts.ref_paths || opts.obj_graph {
                Some(field_name_pool)
            } else {
                None
            },
            unreachable_retained: None,
            obj_graph_edges: None,
            type_ref_pairs: None,
            type_ref_pair_fields: None,
            class_ref_field_names: vec![],
        };

        // Populate class_ref_field_names when --field-stats is set (named plans were
        // built above because the condition now includes opts.field_stats).
        if opts.field_stats {
            let n_cls = graph.class_names.len();
            let mut schema: Vec<Vec<String>> = vec![Vec::new(); n_cls];
            for (ci, plan) in field_plans_named_dense.iter().enumerate() {
                if ci < n_cls {
                    schema[ci] = plan
                        .iter()
                        .map(|(_off, _excl, name)| name.clone())
                        .collect();
                }
            }
            graph.class_ref_field_names = schema;
        }

        // Package the deferred inbound-CSR construction. Moves id_map,
        // class_addrs, field_plans and synthetic_edges out of build (all
        // unused here after the forward fill).
        let inbound = InboundBuilder {
            source: source.clone(),
            id_size,
            n,
            id_map: Some(p1.id_map),
            id_map_c: None,
            id_map_codec: crate::cvec::Codec::None,
            // build_from_fwd drops these immediately; build() (file-scan path) is unused.
            class_addr_to_hist: HashMap::new(),
            field_plans_dense: Vec::new(),
            in_cursors,
            total_inb,
            synthetic_edges,
        };

        t_phase!("2b scan done");

        // Query results are tagged by slot inside `query_state`; the caller
        // reassembles them in input order after the late (retained) stage runs,
        // so no positional reorder happens here.
        Ok((
            graph,
            inbound,
            shallow_c,
            class_idx_c,
            alloc_serial_c,
            query_state,
            refwalk_csr,
            string_values,
            string_values_truncated,
        ))
    }

    /// First-scan heap walker that COUNTS out/in degrees per node and finalizes
    /// each object's authoritative shallow size (arrays/instances use their real
    /// element count / class blob). Produces the degree arrays that Phase 3
    /// prefix-sums into the CSR offsets; fills no edge targets itself.
    #[allow(clippy::too_many_arguments)]
    fn scan_heap_2a<'v>(
        r: &mut HprofReader,
        id_size: u8,
        mut remaining: u64,
        id_map: &crate::id_map::IdMap,
        class_addr_to_hist: &HashMap<u64, u32>,
        field_plans_dense: &[FieldPlan],
        out_degree: &mut Vec<u32>,
        in_degree: &mut Vec<u32>,
        scratch: &mut Vec<u8>,
        mut visitor: Option<&mut (dyn crate::query::ObjectVisitor + 'v)>,
        // Whether any active query targets an array class. When false, the scan
        // skips the per-array class-name construction (a String allocation per
        // array record) entirely — arrays still get their edges/degrees, they
        // just aren't delivered to the visitor.
        visit_arrays: bool,
        class_map: &HashMap<u64, crate::pass1::ClassInfo>,
        strings: &HashMap<u64, String>,
        capture_inst: &std::collections::HashSet<u64>,
        captured_inst: &mut HashMap<u64, (u64, Vec<u8>)>,
        capture_obj: &std::collections::HashSet<u64>,
        captured_obj: &mut HashMap<u64, Vec<u8>>,
        system_class_addr: u64,
        props_name_id: u64,
        captured_props_addr: &mut u64,
        mut fd: Option<&mut fielddecode::FieldDecodeState>,
        fd_p1: &Pass1,
        fd_shallow: &[u32],
        fd_descs: &[CollDesc],
    ) -> io::Result<()> {
        let ids = id_size as u64;
        let mut cache = crate::id_map::IndexCache::new();

        macro_rules! edge_if_valid {
            ($src:expr, $dst_addr:expr, $excl:expr) => {
                if $dst_addr != 0 {
                    if let Some(dst) = cache.index_of(id_map, $dst_addr) {
                        let src = $src as usize;
                        out_degree[src] += 1;
                        in_degree[dst] += 1;
                    }
                }
            };
        }

        // On truncated input, reads inside the heap segment may hit UnexpectedEof
        // mid-subrecord. Treat that as end-of-segment (break) rather than an error,
        // so we report everything successfully parsed before the cut-off.
        macro_rules! try_read {
            ($e:expr) => {
                match $e {
                    Err(e) if e.kind() == ErrorKind::UnexpectedEof => return Ok(()),
                    other => other?,
                }
            };
        }

        macro_rules! checked_sub {
            ($remaining:expr, $sz:expr) => {
                $remaining = $remaining
                    .checked_sub($sz)
                    .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "segment overrun"))?;
            };
        }

        while remaining > 0 {
            let sub_tag = try_read!(r.u1());
            checked_sub!(remaining, 1u64);

            match sub_tag {
                heap::ROOT_SYSTEM_CLASS
                | heap::ROOT_UNKNOWN
                | heap::ROOT_MONITOR_USED
                | heap::ROOT_INTERNED_STRING
                | heap::ROOT_DEBUGGER
                | heap::ROOT_VM_INTERNAL => {
                    try_read!(r.skip(ids));
                    checked_sub!(remaining, ids);
                }
                heap::ROOT_JNI_GLOBAL => {
                    try_read!(r.skip(2 * ids));
                    checked_sub!(remaining, 2 * ids);
                }
                heap::ROOT_JNI_LOCAL | heap::ROOT_JAVA_FRAME | heap::ROOT_JNI_MONITOR => {
                    try_read!(r.skip(ids + 8));
                    checked_sub!(remaining, ids + 8);
                }
                heap::ROOT_NATIVE_STACK | heap::ROOT_THREAD_BLOCK => {
                    try_read!(r.skip(ids + 4));
                    checked_sub!(remaining, ids + 4);
                }
                heap::ROOT_STICKY_CLASS => {
                    try_read!(r.skip(ids));
                    checked_sub!(remaining, ids);
                }
                heap::ROOT_THREAD_OBJ => {
                    try_read!(r.skip(ids + 8));
                    checked_sub!(remaining, ids + 8);
                }
                heap::CLASS_DUMP => {
                    let consumed = match Self::count_class_dump_edges(
                        r,
                        id_size,
                        id_map,
                        out_degree,
                        in_degree,
                        system_class_addr,
                        props_name_id,
                        captured_props_addr,
                    ) {
                        Err(e) if e.kind() == ErrorKind::UnexpectedEof => return Ok(()),
                        other => other?,
                    };
                    checked_sub!(remaining, consumed);
                }
                heap::INSTANCE_DUMP => {
                    let addr = try_read!(r.id());
                    try_read!(r.skip(4));
                    let class_id = try_read!(r.id());
                    let data_len = try_read!(r.u4()) as u64;
                    if data_len > remaining {
                        return Err(io::Error::new(
                            io::ErrorKind::InvalidData,
                            "array too large",
                        ));
                    }
                    try_read!(r.read_bytes_reuse(scratch, data_len as usize));
                    checked_sub!(remaining, ids + 4 + ids + 4 + data_len);

                    // Capture blob for wanted addresses (e.g. thread objects).
                    if !capture_inst.is_empty() && capture_inst.contains(&addr) {
                        captured_inst.insert(addr, (class_id, scratch.clone()));
                    }

                    // Field-decode hook must run on ALL instances (including
                    // those not in id_map), so it mirrors scan_all_records.
                    if let Some(ref mut fds) = fd {
                        fds.on_instance(addr, class_id, scratch, fd_p1, fd_shallow, fd_descs);
                    }

                    let src_idx = match id_map.index_of(addr) {
                        Some(i) => i,
                        None => continue,
                    };

                    if let Some(v) = visitor.as_deref_mut() {
                        v.visit_instance(src_idx, class_id, scratch);
                    }

                    // Edge: instance → class object
                    edge_if_valid!(src_idx, class_id, false);

                    // Edges from Object-type fields (dense Vec by class histogram idx,
                    // no HashMap lookup — Phase 0b already precomputed the per-class plan).
                    if let Some(&cidx) = class_addr_to_hist.get(&class_id) {
                        if (cidx as usize) < field_plans_dense.len() {
                            for &(off, _excluded) in &field_plans_dense[cidx as usize] {
                                let off = off as usize;
                                if off + id_size as usize <= scratch.len() {
                                    let ref_val = read_ref(&scratch[off..], id_size as usize);
                                    if ref_val != 0 {
                                        if let Some(dst) = cache.index_of(id_map, ref_val) {
                                            out_degree[src_idx] += 1;
                                            in_degree[dst] += 1;
                                        }
                                    }
                                }
                            }
                        }
                    }
                }
                heap::OBJ_ARRAY_DUMP => {
                    let addr = try_read!(r.id());
                    try_read!(r.skip(4));
                    let count = try_read!(r.u4()) as u64;
                    let elem_class_id = try_read!(r.id());
                    let byte_len = count.saturating_mul(ids);
                    if byte_len > remaining {
                        return Err(io::Error::new(
                            io::ErrorKind::InvalidData,
                            "array too large",
                        ));
                    }
                    try_read!(r.read_bytes_reuse(scratch, byte_len as usize));
                    checked_sub!(remaining, ids + 4 + 4 + ids + byte_len);

                    // Capture obj-array blob for wanted addresses (e.g. Hashtable table).
                    if !capture_obj.is_empty() && capture_obj.contains(&addr) {
                        captured_obj.insert(addr, scratch.clone());
                    }

                    // Field-decode hook must run on ALL arrays (including those not
                    // in id_map), so it mirrors scan_all_records.
                    if let Some(ref mut fds) = fd {
                        fds.on_obj_array(addr, elem_class_id, count, scratch, fd_p1, fd_shallow);
                    }

                    let src_idx = match id_map.index_of(addr) {
                        Some(i) => i,
                        None => continue,
                    };

                    // Shallow size already correct from Phase 0b; class_idx set by Phase 0c.

                    // OQL: deliver the array to any active query executor, resolving
                    // its own class name from the element-class string (HPROF stores
                    // object-array classes as `[L…;` descriptors).
                    if visit_arrays {
                        if let Some(v) = visitor.as_deref_mut() {
                            if let Some(raw) = class_map
                                .get(&elem_class_id)
                                .and_then(|ci| strings.get(&ci.name_id))
                            {
                                let name = crate::report::pretty_class_name(raw);
                                v.visit_array(src_idx, &name, count as u32);
                            }
                        }
                    }

                    // Edge: array → element class object
                    edge_if_valid!(src_idx, elem_class_id, false);

                    // Edges: array → non-null elements
                    for chunk in scratch.chunks(ids as usize) {
                        let ref_val = read_id(chunk, id_size);
                        if ref_val != 0 {
                            if let Some(dst) = cache.index_of(id_map, ref_val) {
                                out_degree[src_idx] += 1;
                                in_degree[dst] += 1;
                            }
                        }
                    }
                }
                heap::PRIM_ARRAY_NODATA_DUMP => {
                    // Android ART: same header as PRIM_ARRAY_DUMP but no element data.
                    try_read!(r.skip(ids + 4 + 4 + 1));
                    checked_sub!(remaining, ids + 4 + 4 + 1);
                }

                heap::PRIM_ARRAY_DUMP => {
                    let addr = try_read!(r.id());
                    try_read!(r.skip(4));
                    let count = try_read!(r.u4()) as u64;
                    let elem_type = try_read!(r.u1());
                    let esz = HprofType::from_code(elem_type)
                        .map(|t| t.byte_size() as u64)
                        .unwrap_or(1);
                    let byte_len = count.saturating_mul(esz);
                    if fd.is_some() {
                        try_read!(r.read_bytes_reuse(scratch, byte_len as usize));
                    } else {
                        try_read!(r.skip(byte_len));
                    }
                    checked_sub!(remaining, ids + 4 + 4 + 1 + byte_len);
                    // No object edges; shallow already set by Phase 0b.

                    if let Some(ref mut fds) = fd {
                        fds.on_prim_array(addr, elem_type, count, scratch, fd_p1, fd_shallow);
                    }

                    // OQL: deliver the primitive array to any active query executor.
                    // Primitive arrays carry no class-object address; synthesize the
                    // descriptor (`[C` → `char[]`) from the element type code.
                    if visit_arrays {
                        if let Some(v) = visitor.as_deref_mut() {
                            if let Some(src_idx) = id_map.index_of(addr) {
                                let raw = crate::pass2::sizing::prim_array_class_name(elem_type);
                                let name = crate::report::pretty_class_name(raw);
                                v.visit_array(src_idx, &name, count as u32);
                            }
                        }
                    }
                }
                heap::HEAP_DUMP_INFO => {
                    // Android ART: u4 heap_id + id heap_name_string_id — no edges, just skip.
                    try_read!(r.skip(4 + ids));
                    checked_sub!(remaining, 4 + ids);
                }
                other => {
                    return Err(io::Error::new(
                        ErrorKind::InvalidData,
                        format!("unknown heap sub-tag 0x{other:02x} in 2a"),
                    ));
                }
            }
        }
        Ok(())
    }

    /// Second-scan heap walker that FILLS the CSR edge arrays (degrees already
    /// counted by `scan_heap_2a`). `do_fwd`/`do_inb` select which side is being
    /// filled: the forward pass advances `fwd_offsets` in place as write
    /// cursors; the inbound pass writes into `inb_flat` at `in_degree` cursors,
    /// tagging excluded (weak/finalizer) referrers with the high bit.
    #[allow(clippy::too_many_arguments)]
    fn fill_heap_2b(
        r: &mut HprofReader,
        id_size: u8,
        mut remaining: u64,
        id_map: &crate::id_map::IdMap,
        class_addr_to_hist: &HashMap<u64, u32>,
        field_plans_dense: &[FieldPlan],
        field_plans_named_dense: &[FieldPlanNamed],
        do_fwd: bool,
        do_inb: bool,
        fwd_targets: &mut crate::chunkvec::ChunkU32,
        fwd_offsets: &mut Vec<u32>,
        fwd_field_name_idx: &mut Option<Vec<u16>>,
        field_name_pool: &mut Vec<String>,
        field_name_pool_idx: &mut std::collections::HashMap<String, u16>,
        inb_flat: &mut crate::chunkvec::ChunkU32,
        in_degree: &mut Vec<u32>,
        scratch: &mut Vec<u8>,
    ) -> io::Result<()> {
        let ids = id_size as u64;
        let mut cache = crate::id_map::IndexCache::new();
        let do_names = fwd_field_name_idx.is_some() && do_fwd;

        macro_rules! add_edge {
            ($src:expr, $dst_addr:expr, $excluded:expr, $name_idx:expr) => {
                if $dst_addr != 0 {
                    if let Some(dst) = cache.index_of(id_map, $dst_addr) {
                        let src = $src as usize;
                        if do_fwd {
                            // fwd_offsets[src] is the in-place write cursor.
                            let pos = fwd_offsets[src] as usize;
                            fwd_targets.set(pos, dst as u32);
                            if do_names {
                                if let Some(idx_vec) = fwd_field_name_idx.as_mut() {
                                    idx_vec[pos] = $name_idx;
                                }
                            }
                            fwd_offsets[src] += 1;
                        }
                        if do_inb {
                            // Inbound: store src with/without exclusion flag
                            let inb_val = if $excluded {
                                (src as u32) | 0x8000_0000u32
                            } else {
                                src as u32
                            };
                            inb_flat.set(in_degree[dst] as usize, inb_val);
                            in_degree[dst] += 1;
                        }
                    }
                }
            };
        }

        macro_rules! checked_sub {
            ($remaining:expr, $sz:expr) => {
                $remaining = $remaining
                    .checked_sub($sz)
                    .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "segment overrun"))?;
            };
        }

        macro_rules! try_read {
            ($e:expr) => {
                match $e {
                    Err(e) if e.kind() == ErrorKind::UnexpectedEof => return Ok(()),
                    other => other?,
                }
            };
        }

        while remaining > 0 {
            let sub_tag = try_read!(r.u1());
            checked_sub!(remaining, 1u64);

            match sub_tag {
                heap::ROOT_SYSTEM_CLASS
                | heap::ROOT_UNKNOWN
                | heap::ROOT_MONITOR_USED
                | heap::ROOT_INTERNED_STRING
                | heap::ROOT_DEBUGGER
                | heap::ROOT_VM_INTERNAL => {
                    try_read!(r.skip(ids));
                    checked_sub!(remaining, ids);
                }
                heap::ROOT_JNI_GLOBAL => {
                    try_read!(r.skip(2 * ids));
                    checked_sub!(remaining, 2 * ids);
                }
                heap::ROOT_JNI_LOCAL | heap::ROOT_JAVA_FRAME | heap::ROOT_JNI_MONITOR => {
                    try_read!(r.skip(ids + 8));
                    checked_sub!(remaining, ids + 8);
                }
                heap::ROOT_NATIVE_STACK | heap::ROOT_THREAD_BLOCK => {
                    try_read!(r.skip(ids + 4));
                    checked_sub!(remaining, ids + 4);
                }
                heap::ROOT_STICKY_CLASS => {
                    try_read!(r.skip(ids));
                    checked_sub!(remaining, ids);
                }
                heap::ROOT_THREAD_OBJ => {
                    try_read!(r.skip(ids + 8));
                    checked_sub!(remaining, ids + 8);
                }
                heap::CLASS_DUMP => {
                    let consumed = match Self::fill_class_dump_edges(
                        r,
                        id_size,
                        id_map,
                        do_fwd,
                        do_inb,
                        fwd_targets,
                        fwd_offsets,
                        fwd_field_name_idx,
                        inb_flat,
                        in_degree,
                    ) {
                        Err(e) if e.kind() == ErrorKind::UnexpectedEof => return Ok(()),
                        other => other?,
                    };
                    checked_sub!(remaining, consumed);
                }
                heap::INSTANCE_DUMP => {
                    let addr = try_read!(r.id());
                    try_read!(r.skip(4));
                    let class_id = try_read!(r.id());
                    let data_len = try_read!(r.u4()) as u64;
                    if data_len > remaining {
                        return Err(io::Error::new(
                            io::ErrorKind::InvalidData,
                            "array too large",
                        ));
                    }
                    try_read!(r.read_bytes_reuse(scratch, data_len as usize));
                    checked_sub!(remaining, ids + 4 + ids + 4 + data_len);

                    let src_idx = match id_map.index_of(addr) {
                        Some(i) => i,
                        None => continue,
                    };

                    // Edge: instance → class object
                    add_edge!(src_idx, class_id, false, 0u16);

                    // Edges from Object-type fields (dense Vec by class histogram idx)
                    if let Some(&cidx) = class_addr_to_hist.get(&class_id) {
                        let named_plan =
                            if do_names && (cidx as usize) < field_plans_named_dense.len() {
                                &field_plans_named_dense[cidx as usize]
                            } else {
                                &[][..]
                            };
                        if (cidx as usize) < field_plans_dense.len() {
                            for (fi, &(off, excluded)) in
                                field_plans_dense[cidx as usize].iter().enumerate()
                            {
                                let off = off as usize;
                                if off + id_size as usize <= scratch.len() {
                                    let ref_val = read_ref(&scratch[off..], id_size as usize);
                                    let name_idx = if do_names && fi < named_plan.len() {
                                        let fname = &named_plan[fi].2;
                                        if fname.is_empty() {
                                            0u16
                                        } else if let Some(&idx) = field_name_pool_idx.get(fname) {
                                            idx
                                        } else {
                                            let new_idx = field_name_pool.len() as u16;
                                            field_name_pool.push(fname.clone());
                                            field_name_pool_idx.insert(fname.clone(), new_idx);
                                            new_idx
                                        }
                                    } else {
                                        0u16
                                    };
                                    add_edge!(src_idx, ref_val, excluded, name_idx);
                                }
                            }
                        }
                    }
                }
                heap::OBJ_ARRAY_DUMP => {
                    let addr = try_read!(r.id());
                    try_read!(r.skip(4));
                    let count = try_read!(r.u4()) as u64;
                    let elem_class_id = try_read!(r.id());
                    let byte_len = count.saturating_mul(ids);
                    if byte_len > remaining {
                        return Err(io::Error::new(
                            io::ErrorKind::InvalidData,
                            "array too large",
                        ));
                    }
                    try_read!(r.read_bytes_reuse(scratch, byte_len as usize));
                    checked_sub!(remaining, ids + 4 + 4 + ids + byte_len);

                    let src_idx = match id_map.index_of(addr) {
                        Some(i) => i,
                        None => continue,
                    };

                    // Edge: array → element class
                    add_edge!(src_idx, elem_class_id, false, 0u16);

                    // Edges to elements
                    for chunk in scratch.chunks(ids as usize) {
                        let ref_val = read_id(chunk, id_size);
                        add_edge!(src_idx, ref_val, false, 0u16);
                    }
                }
                heap::PRIM_ARRAY_NODATA_DUMP => {
                    // Android ART: same header as PRIM_ARRAY_DUMP but no element data.
                    try_read!(r.skip(ids + 4 + 4 + 1));
                    checked_sub!(remaining, ids + 4 + 4 + 1);
                }

                heap::PRIM_ARRAY_DUMP => {
                    let _addr = try_read!(r.id());
                    try_read!(r.skip(4));
                    let count = try_read!(r.u4()) as u64;
                    let elem_type = try_read!(r.u1());
                    let esz = HprofType::from_code(elem_type)
                        .map(|t| t.byte_size() as u64)
                        .unwrap_or(1);
                    let byte_len = count.saturating_mul(esz);
                    try_read!(r.skip(byte_len));
                    checked_sub!(remaining, ids + 4 + 4 + 1 + byte_len);
                    // No object edges from prim arrays
                }
                other => {
                    return Err(io::Error::new(
                        ErrorKind::InvalidData,
                        format!("unknown heap sub-tag 0x{other:02x} in 2b"),
                    ));
                }
            }
        }
        Ok(())
    }

    /// FILL-phase counterpart to `count_class_dump_edges`: emits a class
    /// object's structural edges (→ superclass, → loader, → each Object-typed
    /// static field) into the forward and/or inbound CSR. Returns bytes consumed.
    #[allow(clippy::too_many_arguments)]
    pub(crate) fn fill_class_dump_edges(
        r: &mut HprofReader,
        id_size: u8,
        id_map: &crate::id_map::IdMap,
        do_fwd: bool,
        do_inb: bool,
        fwd_targets: &mut crate::chunkvec::ChunkU32,
        fwd_offsets: &mut Vec<u32>,
        fwd_field_name_idx: &mut Option<Vec<u16>>,
        inb_flat: &mut crate::chunkvec::ChunkU32,
        in_degree: &mut Vec<u32>,
    ) -> io::Result<u64> {
        let ids = id_size as u64;
        let mut consumed = 0u64;

        let class_addr = r.id()?;
        consumed += ids;
        r.skip(4)?;
        consumed += 4;
        let super_id = r.id()?;
        consumed += ids;
        let loader_id = r.id()?;
        consumed += ids;
        r.skip(ids * 4 + 4)?;
        consumed += ids * 4 + 4;

        let src_idx_opt = id_map.index_of(class_addr);
        let _ = fwd_field_name_idx;

        macro_rules! add_edge_inner {
            ($src:expr, $dst_addr:expr) => {
                if $dst_addr != 0 {
                    if let Some(dst) = id_map.index_of($dst_addr) {
                        let src = $src as usize;
                        if do_fwd {
                            // fwd_offsets[src] is the in-place write cursor.
                            let pos = fwd_offsets[src] as usize;
                            fwd_targets.set(pos, dst as u32);
                            fwd_offsets[src] += 1;
                        }
                        if do_inb {
                            inb_flat.set(in_degree[dst] as usize, src as u32);
                            in_degree[dst] += 1;
                        }
                    }
                }
            };
        }

        if let Some(src) = src_idx_opt {
            add_edge_inner!(src, super_id);
            add_edge_inner!(src, loader_id);
        }

        // Constant pool
        let cp = r.u2()? as u64;
        consumed += 2;
        for _ in 0..cp {
            r.skip(2)?;
            consumed += 2;
            let tp = r.u1()?;
            consumed += 1;
            let vs = value_size(tp, id_size);
            r.skip(vs)?;
            consumed += vs;
        }

        // Static fields
        let sc = r.u2()? as u64;
        consumed += 2;
        for _ in 0..sc {
            r.skip(ids)?;
            consumed += ids; // name_id
            let tp = r.u1()?;
            consumed += 1;
            let vs = value_size(tp, id_size);
            if tp == 2 {
                // Object static field
                let ref_val = read_id_from_reader(r, id_size)?;
                consumed += vs;
                if let Some(src) = src_idx_opt {
                    add_edge_inner!(src, ref_val);
                }
            } else {
                r.skip(vs)?;
                consumed += vs;
            }
        }

        // Instance fields (just skip)
        let ic = r.u2()? as u64;
        consumed += 2;
        let ic_skip = ic.saturating_mul(ids.saturating_add(1));
        r.skip(ic_skip)?;
        consumed += ic_skip;

        Ok(consumed)
    }
}

// ── Also need 2a version of CLASS_DUMP to count static obj edges ───────────
// We need a version that also counts degrees for CLASS_DUMP static fields.

impl Pass2 {
    /// COUNT-phase counterpart to `fill_class_dump_edges`: counts a class
    /// object's structural edges (→ superclass, → loader, → each Object-typed
    /// static field) into the degree arrays. Returns bytes consumed.
    /// If `system_class_addr != 0` and this CLASS_DUMP is for that class, also
    /// captures the static Object field with name_id `props_name_id` into
    /// `captured_props_addr` (used to locate java/lang/System.props).
    fn count_class_dump_edges(
        r: &mut HprofReader,
        id_size: u8,
        id_map: &crate::id_map::IdMap,
        out_degree: &mut Vec<u32>,
        in_degree: &mut Vec<u32>,
        system_class_addr: u64,
        props_name_id: u64,
        captured_props_addr: &mut u64,
    ) -> io::Result<u64> {
        let ids = id_size as u64;
        let mut consumed = 0u64;

        let class_addr = r.id()?;
        consumed += ids;
        r.skip(4)?;
        consumed += 4;
        let super_id = r.id()?;
        consumed += ids;
        let loader_id = r.id()?;
        consumed += ids;
        r.skip(ids * 4 + 4)?;
        consumed += ids * 4 + 4;

        let src_opt = id_map.index_of(class_addr);

        macro_rules! count_edge {
            ($dst_addr:expr) => {
                if $dst_addr != 0 {
                    if let Some(dst) = id_map.index_of($dst_addr) {
                        if let Some(src) = src_opt {
                            out_degree[src] += 1;
                            in_degree[dst] += 1;
                        }
                    }
                }
            };
        }

        if src_opt.is_some() {
            count_edge!(super_id);
            count_edge!(loader_id);
        }

        let cp = r.u2()? as u64;
        consumed += 2;
        for _ in 0..cp {
            r.skip(2)?;
            consumed += 2;
            let tp = r.u1()?;
            consumed += 1;
            let vs = value_size(tp, id_size);
            r.skip(vs)?;
            consumed += vs;
        }

        let sc = r.u2()? as u64;
        consumed += 2;
        let capture_props =
            system_class_addr != 0 && class_addr == system_class_addr && props_name_id != 0;
        for _ in 0..sc {
            let name_id = r.id()?;
            consumed += ids;
            let tp = r.u1()?;
            consumed += 1;
            let vs = value_size(tp, id_size);
            if tp == 2 {
                let ref_val = read_id_from_reader(r, id_size)?;
                consumed += vs;
                count_edge!(ref_val);
                if capture_props && name_id == props_name_id && ref_val != 0 {
                    *captured_props_addr = ref_val;
                }
            } else {
                r.skip(vs)?;
                consumed += vs;
            }
        }

        let ic = r.u2()? as u64;
        consumed += 2;
        let ic_skip = ic.saturating_mul(ids.saturating_add(1));
        r.skip(ic_skip)?;
        consumed += ic_skip;

        Ok(consumed)
    }
}

// ── Tests ──────────────────────────────────────────────────────────────────

#[cfg(test)]
mod tests {
    use super::*;
    use crate::pass1::{ClassInfo, Pass1};

    const DUMP: &str = "tests/fixtures/dump_0_fj-kmeans.hprof";

    #[test]
    fn pretty_binary_name_strips_l_and_semicolon_and_dots() {
        assert_eq!(pretty_binary_name("Lfoo/bar/Baz;"), "foo.bar.Baz");
        assert_eq!(pretty_binary_name("foo/bar/Baz"), "foo.bar.Baz");
        assert_eq!(pretty_binary_name("Baz"), "Baz");
    }

    #[test]
    fn render_frame_applies_hprof_line_conventions() {
        assert_eq!(
            render_frame(Some("foo.Bar"), Some("run"), Some("Bar.java"), 7, 42),
            "foo.Bar.run (Bar.java:42)"
        );
        assert_eq!(
            render_frame(Some("foo.Bar"), Some("run"), Some("Bar.java"), 7, -1),
            "foo.Bar.run (Bar.java)"
        );
        assert_eq!(
            render_frame(Some("foo.Bar"), Some("run"), Some("Bar.java"), 7, -2),
            "foo.Bar.run (Bar.java(Compiled Method))"
        );
        assert_eq!(
            render_frame(Some("foo.Bar"), Some("run"), Some("Bar.java"), 7, -3),
            "foo.Bar.run (Native Method)"
        );
    }

    #[test]
    fn render_frame_falls_back_when_strings_missing() {
        assert_eq!(
            render_frame(None, None, None, 99, -1),
            "<class#99>.<method> (Unknown Source)"
        );
    }

    #[test]
    fn pass2_graph_has_edges() {
        if !std::path::Path::new(DUMP).exists() {
            return;
        }
        let p1 = Pass1::run(&crate::source::HprofSource::from(DUMP), false).unwrap();
        let (g, inbound, _sc, _ci, _as, _q, _rw, _sv, _sv_trunc) = Pass2::build(
            &crate::source::HprofSource::from(DUMP),
            p1,
            crate::cvec::Codec::None,
            &crate::AnalyzeOptions::default(),
            &[],
            &mut std::collections::HashMap::new(),
            &mut std::collections::HashMap::new(),
        )
        .unwrap();
        assert!(!g.fwd_targets.is_empty(), "no forward edges");
        assert_eq!(g.fwd_offsets.len(), g.n + 1);
        // Identity dfn (node -> pre-order) suffices. build() now returns
        // blocked offsets: one sampled offset per INB_BLOCK nodes + a trailing
        // sentinel, so len == ceil(n / INB_BLOCK) + 1.
        let dfn: Vec<u32> = (0..g.n as u32).collect();
        let (inb_block_off, _inb_data) = inbound.build(&dfn).unwrap();
        assert_eq!(inb_block_off.len(), g.n.div_ceil(INB_BLOCK) + 1);
        for &r in &g.gc_root_indices {
            assert!((r as usize) < g.n, "gc_root idx {} out of range {}", r, g.n);
        }
        assert_eq!(g.class_idx.len(), g.n);
        assert!(!g.class_names.is_empty());
        // Only class objects (e.g. array classes with no static fields) may have
        // shallow 0 — MAT reports 0 for those. All other objects must be > 0.
        for i in 0..g.n {
            if g.shallow[i] == 0 {
                assert!(
                    g.class_obj_class_idx.contains_key(&(i as u32)),
                    "non-class object {i} has shallow 0"
                );
            }
        }
    }

    #[test]
    fn pass2_edge_counts_sane() {
        if !std::path::Path::new(DUMP).exists() {
            return;
        }
        let p1 = Pass1::run(&crate::source::HprofSource::from(DUMP), false).unwrap();
        let (g, _inbound, _sc, _ci, _as, _q, _rw, _sv, _sv_trunc) = Pass2::build(
            &crate::source::HprofSource::from(DUMP),
            p1,
            crate::cvec::Codec::None,
            &crate::AnalyzeOptions::default(),
            &[],
            &mut std::collections::HashMap::new(),
            &mut std::collections::HashMap::new(),
        )
        .unwrap();
        let fwd_edge_count: usize = g
            .fwd_offsets
            .windows(2)
            .map(|w| (w[1] - w[0]) as usize)
            .sum();
        assert!(
            fwd_edge_count > g.n / 2,
            "suspiciously few edges: {} for {} nodes",
            fwd_edge_count,
            g.n
        );
    }

    // Opt 1 invariant: `kind[i] == 3` (class_obj) is EXACTLY equivalent to the
    // object's address being present in the `class_addrs` set that pass2 builds
    // from `class_map.keys()`. scan_heap_2a relies on this to replace the
    // per-instance `class_addrs.contains(addr)` hash probe with `kind[src] == 3`
    // (src already computed).
    #[test]
    fn kind3_equals_class_addrs_membership() {
        if !std::path::Path::new(DUMP).exists() {
            return;
        }
        let p1 = Pass1::run(&crate::source::HprofSource::from(DUMP), false).unwrap();
        let class_addrs: std::collections::HashSet<u64> = p1.class_map.keys().cloned().collect();
        assert!(!class_addrs.is_empty(), "expected some class objects");
        let mut class_count = 0usize;
        for i in 0..p1.id_map.len() {
            let addr = p1.id_map.addr_at(i);
            let is_class_by_kind = p1.kind[i] == 3;
            let is_class_by_set = class_addrs.contains(&addr);
            assert_eq!(
                is_class_by_kind, is_class_by_set,
                "kind==3 vs class_addrs.contains disagree at index {i} (addr {addr:#x})"
            );
            if is_class_by_kind {
                class_count += 1;
            }
        }
        assert_eq!(
            class_count,
            class_addrs.len(),
            "every class address must appear exactly once as a kind==3 object"
        );
    }

    #[test]
    fn primitive_array_class_name_recognizes_all_prims() {
        for n in ["[Z", "[C", "[F", "[D", "[S", "[I", "[J", "[B"] {
            assert!(
                is_primitive_array_class_name(n),
                "{n} should be a primitive-array class name"
            );
        }
    }

    #[test]
    fn primitive_array_class_name_rejects_non_prims() {
        for n in [
            "[[I",                 // multi-dim int array
            "[Ljava/lang/String;", // object array
            "java/lang/String",    // ordinary class
            "[",                   // lone bracket
            "[ZZ",                 // too long
            "",                    // empty
            "Z",                   // no bracket
            "[X",                  // bracket + non-prim char
        ] {
            assert!(
                !is_primitive_array_class_name(n),
                "{n:?} must NOT be a primitive-array class name"
            );
        }
    }

    #[test]
    fn system_class_rooting_matches_mat_addsystemclassroots() {
        // MAT's addSystemClassRootsIfMissing (HprofParserHandlerImpl.fillIn):
        // the class-rooting loop runs ONLY when no sticky/SYSTEM_CLASS roots
        // exist in the dump, and roots only non-array boot-loader classes.

        // The normal HPROF case: sticky roots present -> MAT roots nothing.
        // We match that for ordinary (non-array) boot-loader classes: this is
        // the fix for the big-dump +4,645-object frontier over-marking.
        assert!(
            !should_add_system_class_root(false, false, true),
            "non-array boot class must NOT be synthetically rooted when sticky roots exist"
        );
        // No sticky roots -> MAT (and we) root non-array boot-loader classes.
        assert!(
            should_add_system_class_root(false, false, false),
            "non-array boot class must be rooted when the dump has no sticky roots"
        );

        // Object arrays / multi-dim arrays are never synthetically rooted,
        // regardless of sticky presence (MAT guard: !clazz.isArrayType()).
        assert!(!should_add_system_class_root(true, false, false));
        assert!(!should_add_system_class_root(true, false, true));

        // Primitive-array metadata classes ([Z etc.) are ALWAYS rooted
        // (Group B mirror of MAT's dominator root-attachment), independent of
        // sticky-root presence.
        assert!(should_add_system_class_root(true, true, true));
        assert!(should_add_system_class_root(true, true, false));
    }

    #[test]
    fn decode_latin1_string() {
        // coder 0 = LATIN1: each byte is a code point 0..=255.
        assert_eq!(decode_java_string(b"main", 0), "main");
        assert_eq!(decode_java_string(&[0xe9], 0), "é"); // 0xE9 = U+00E9
        assert_eq!(decode_java_string(&[], 0), "");
    }

    #[test]
    fn decode_utf16be_string() {
        // coder 1 = UTF-16BE: pair bytes big-endian.
        // "main" as UTF-16BE.
        let utf16: Vec<u8> = "main"
            .encode_utf16()
            .flat_map(|u| u.to_be_bytes())
            .collect();
        assert_eq!(decode_java_string(&utf16, 1), "main");
        // A non-Latin code point that needs UTF-16 (U+4E2D 中).
        let cjk: Vec<u8> = "中".encode_utf16().flat_map(|u| u.to_be_bytes()).collect();
        assert_eq!(decode_java_string(&cjk, 1), "中");
    }

    #[test]
    fn decode_java8_char_array_is_utf16() {
        // Java 8 Strings have a char[] value and NO coder field; the resolver
        // passes coder 1 (UTF16) for them. A char[] holds UTF-16BE code units.
        let chars: Vec<u8> = "hi".encode_utf16().flat_map(|u| u.to_be_bytes()).collect();
        assert_eq!(decode_java_string(&chars, 1), "hi");
    }

    #[test]
    fn field_offset_places_superclass_fields_after_subclass_fields() {
        // HPROF stores instance field VALUES subclass-first: the object's own
        // class fields precede the inherited superclass fields in the blob. Build
        // a synthetic two-class chain and confirm the inherited field's offset
        // lands *after* the subclass's own fields, and that the owner_class
        // filter skips a same-named field declared by the subclass.
        let mut strings: std::collections::HashMap<u64, String> =
            std::collections::HashMap::default();
        strings.insert(1, "java/lang/Thread".to_string());
        strings.insert(2, "Sub".to_string());
        strings.insert(10, "eetop".to_string()); // Thread field (Long)
        strings.insert(11, "name".to_string()); // Thread field (Object)
        strings.insert(20, "extra".to_string()); // Sub field (Int)
        strings.insert(21, "name".to_string()); // Sub's OWN shadowing "name"

        let obj_ref_width = 8usize;
        let thread = ClassInfo {
            name_id: 1,
            super_id: 0,
            fields: vec![(10, HprofType::Long), (11, HprofType::Object)],
            ..Default::default()
        };
        let sub = ClassInfo {
            name_id: 2,
            super_id: 100, // points at Thread
            fields: vec![(20, HprofType::Int), (21, HprofType::Object)],
            ..Default::default()
        };
        let mut class_map: HashMap<u64, ClassInfo> = HashMap::new();
        class_map.insert(100, thread);
        class_map.insert(200, sub);

        // Sub's own fields (int=4 + object=8) come first = 12 bytes, then Thread:
        // eetop(Long=8), then name(Object) at 12 + 8 = 20.
        let (off, t) = field_offset(
            200,
            "name",
            "java/lang/Thread",
            &class_map,
            &strings,
            obj_ref_width,
        )
        .expect("inherited Thread.name must resolve");
        assert_eq!(off, 20);
        assert_eq!(t, HprofType::Object);

        // For a pure java/lang/Thread instance, name is right after eetop = 8.
        let (off2, _) = field_offset(
            100,
            "name",
            "java/lang/Thread",
            &class_map,
            &strings,
            obj_ref_width,
        )
        .expect("Thread.name must resolve");
        assert_eq!(off2, 8);
    }
}

#[cfg(test)]
mod visitor_hook_tests {
    use crate::query::ObjectVisitor;

    // A minimal visitor that counts instance callbacks.
    struct Counter {
        n: usize,
    }
    impl crate::query::ObjectVisitor for Counter {
        fn visit_instance(&mut self, _src_idx: usize, _class_id: u64, _blob: &[u8]) {
            self.n += 1;
        }
    }

    #[test]
    fn counter_visitor_type_checks() {
        let mut c = Counter { n: 0 };
        let _dyn: &mut dyn crate::query::ObjectVisitor = &mut c;
        c.visit_instance(0, 0, &[]);
        assert_eq!(c.n, 1);
    }

    // A visitor that records exactly what it was called with, so we can assert
    // the trait-object dispatch forwards every argument unchanged.
    struct Recorder {
        calls: Vec<(usize, u64, usize)>,
    }
    impl crate::query::ObjectVisitor for Recorder {
        fn visit_instance(&mut self, src_idx: usize, class_id: u64, blob: &[u8]) {
            self.calls.push((src_idx, class_id, blob.len()));
        }
    }

    #[test]
    fn visitor_receives_all_args() {
        let mut rec = Recorder { calls: Vec::new() };
        {
            // Drive the visitor exclusively through a &mut dyn trait object,
            // mirroring how scan_heap_2a dispatches, to prove args pass intact.
            let v: &mut dyn crate::query::ObjectVisitor = &mut rec;
            v.visit_instance(0, 0x1000, &[]);
            v.visit_instance(7, 0x2000, &[0xAA, 0xBB, 0xCC]);
            v.visit_instance(42, 0xDEAD_BEEF, &[1, 2]);
        }
        assert_eq!(
            rec.calls,
            vec![(0, 0x1000, 0), (7, 0x2000, 3), (42, 0xDEAD_BEEF, 2)]
        );
    }
}