sqry_core/graph/unified/storage/

metadata.rs

//! Sparse node metadata store for macro boundary analysis, classpath
//! provenance, and payload-less marker flags.
//!
//! This module provides [`NodeMetadataStore`], a sparse metadata store keyed by
//! full [`NodeId`] (index + generation) to prevent stale metadata when the
//! generational arena reuses a slot index with a new generation.
//!
//! # Two channels
//!
//! Each stored entry carries two independent channels:
//!
//! 1. **Typed payload** ([`TypedMetadata`]) — mutually-exclusive payload-bearing
//!    metadata (today: macro vs. classpath provenance). A node has at most one
//!    typed payload at a time.
//! 2. **Marker flags** ([`NodeFlags`]) — payload-less, independently composable
//!    boolean attributes (today: synthetic, address-taken, callsite-promiscuous).
//!    Any subset may be set simultaneously, AND may co-occur with a typed
//!    payload — e.g. a Rust function generated by a macro that is ALSO
//!    address-taken via `&foo` is `TypedMetadata::Macro(_)` PLUS
//!    `NodeFlags::ADDRESS_TAKEN`.
//!
//! Only nodes with metadata get entries, keeping memory overhead proportional
//! to the number of annotated symbols rather than total node count.

use std::collections::{BTreeMap, HashMap};

use serde::{Deserialize, Serialize};

use super::super::node::id::NodeId;
use super::dispatch_tables::DispatchTables;
use super::framework_routes::{FrameworkRouteMetadata, FrameworkRoutesMap};
use super::shape::ShapeDescriptor;

/// Optional metadata for nodes that participate in macro boundary analysis.
///
/// Stored separately from [`NodeEntry`] to avoid bloating the arena for the
/// majority of nodes that don't need macro metadata.
///
/// Each field is `Option` (or `Vec` with empty default) so only relevant
/// metadata consumes space in the serialized representation.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct MacroNodeMetadata {
    /// Whether this symbol was generated by macro expansion.
    pub macro_generated: Option<bool>,

    /// Qualified name of the macro that generated this symbol.
    pub macro_source: Option<String>,

    /// The cfg predicate string (e.g., `"test"`, `"feature = \"serde\""`).
    ///
    /// **Language-agnostic.** Despite the enclosing struct name
    /// (`MacroNodeMetadata`), `cfg_condition` is the canonical slot for
    /// any conditional-compilation guard the language has:
    ///
    /// - Rust: `#[cfg(...)]` predicate string from `cfg_analysis`
    ///   (e.g. `"target_os = \"linux\""`,
    ///   `"all(target_os = \"linux\", target_arch = \"amd64\")"`).
    /// - Go: file-level build constraint canonicalised by the
    ///   Go plugin's `build_constraints` module (e.g. `"linux"`,
    ///   `"linux && amd64"`, `"!windows"`, `"cgo"`). Per `01_SPEC` §3.3
    ///   and `02_DESIGN` §3.3 (T3.8), Go build tags are file-level —
    ///   every non-synthetic node staged from the same file shares the
    ///   same `cfg_condition` string.
    /// - Other languages: equivalent file-level / item-level
    ///   conditional-compilation guards may use the same slot. The
    ///   stored string is whatever canonical form the plugin chose;
    ///   cross-language structural comparison is handled by sqry-db's
    ///   `cfg_match` comparator (`02_DESIGN` §5.3.a).
    ///
    /// `None` means "no conditional-compilation guard recorded for this
    /// node" (the default for plain Rust items without `#[cfg]` and Go
    /// files without `//go:build`, `// +build`, recognised filename
    /// suffix, or `import "C"`).
    pub cfg_condition: Option<String>,

    /// Whether this cfg is active (`None` = unknown, requires external config).
    pub cfg_active: Option<bool>,

    /// Proc-macro kind for proc-macro function nodes.
    pub proc_macro_kind: Option<ProcMacroFunctionKind>,

    /// Whether expansion data came from cache vs live `cargo expand`.
    pub expansion_cached: Option<bool>,

    /// Unresolved attribute paths that could not be positively identified
    /// as proc-macro attributes. Stored for potential future resolution.
    pub unresolved_attributes: Vec<String>,
}

/// Classification of proc-macro function types.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ProcMacroFunctionKind {
    /// `#[proc_macro_derive(Name)]` — generates impls for structs/enums.
    Derive,
    /// `#[proc_macro_attribute]` — transforms annotated items.
    Attribute,
    /// `#[proc_macro]` — function-like `my_macro!(...)` invocation.
    FunctionLike,
}

/// Metadata for nodes originating from JVM classpath bytecode.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ClasspathNodeMetadata {
    /// Maven coordinates (e.g., `"com.google.guava:guava:33.0.0"`).
    pub coordinates: Option<String>,
    /// JAR file path this node was extracted from.
    pub jar_path: String,
    /// Fully qualified name in JVM format (e.g., `"java.util.HashMap"`).
    pub fqn: String,
    /// Whether this is a direct or transitive dependency.
    pub is_direct_dependency: bool,
}

/// Mutually-exclusive payload-bearing metadata variants.
///
/// A node has at most one `TypedMetadata` at a time (Macro is Rust-only,
/// Classpath is JVM-only — they cannot co-occur by language). Independent
/// boolean attributes are carried by [`NodeFlags`] instead.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum TypedMetadata {
    /// Rust macro-related metadata.
    Macro(MacroNodeMetadata),
    /// JVM classpath provenance metadata.
    Classpath(ClasspathNodeMetadata),
}

/// Payload-less marker flags, independently composable.
///
/// Each flag is a single bit in a `u8` bitset. Flags compose freely with each
/// other AND with the [`TypedMetadata`] channel — e.g. a node may carry
/// `TypedMetadata::Macro(_)` AND `NodeFlags::SYNTHETIC | NodeFlags::ADDRESS_TAKEN`
/// simultaneously.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[repr(transparent)]
pub struct NodeFlags(u8);

impl NodeFlags {
    /// Synthetic placeholder node marker (`C_SUPPRESS`).
    ///
    /// Identifies internal-use-only nodes that language plugins emit as
    /// shadows / scaffolding for binding-plane analysis (e.g. the Go
    /// plugin's `<field:operand.field>` field-access placeholders and the
    /// `name@<offset>` per-binding-site Variable nodes from the local-scope
    /// resolver). These nodes must be suppressed from user-facing search
    /// surfaces but remain reachable to internal callers via the explicit
    /// `include_synthetic` opt-in path.
    pub const SYNTHETIC: NodeFlags = NodeFlags(1 << 0);

    /// Function whose address has been taken at some site in the workspace
    /// (e.g. `&foo`, passed as an argument, stored in a function-pointer
    /// field). Populated by the C plugin's address-taken classifier; consumed
    /// by the C indirect-call resolver to scope type-signature matches.
    pub const ADDRESS_TAKEN: NodeFlags = NodeFlags(1 << 1);

    /// Call-site for which the indirect-call resolver exceeded the
    /// per-callsite cardinality cap. The original `Calls`-stub edge is
    /// preserved (no per-callee edges emitted) and the caller is flagged
    /// so planners and consumers can surface the over-cap fan-out.
    pub const CALLSITE_PROMISCUOUS: NodeFlags = NodeFlags(1 << 2);

    /// Import node that resolves to a Go standard-library package (e.g.
    /// `fmt`, `net/http`, `encoding/json`). Populated by the Go plugin's
    /// import classifier so callers can separate stdlib dependencies from
    /// third-party and relative imports.
    pub const IMPORT_STDLIB: NodeFlags = NodeFlags(1 << 3);

    /// Import node that resolves to a relative (intra-module) path (e.g. a
    /// `./sub` or `../pkg` style import). Populated by the Go plugin's
    /// import classifier; mutually distinct from [`Self::IMPORT_STDLIB`].
    pub const IMPORT_RELATIVE: NodeFlags = NodeFlags(1 << 4);

    /// Empty bitset.
    pub const EMPTY: NodeFlags = NodeFlags(0);

    /// Returns `true` iff every bit in `other` is set in `self`.
    #[must_use]
    pub const fn contains(self, other: NodeFlags) -> bool {
        (self.0 & other.0) == other.0
    }

    /// Set every bit in `other`.
    pub fn insert(&mut self, other: NodeFlags) {
        self.0 |= other.0;
    }

    /// Clear every bit in `other`.
    pub fn remove(&mut self, other: NodeFlags) {
        self.0 &= !other.0;
    }

    /// Returns `true` iff no bits are set.
    #[must_use]
    pub const fn is_empty(self) -> bool {
        self.0 == 0
    }

    /// Raw byte value (used by the wire format).
    #[must_use]
    pub const fn bits(self) -> u8 {
        self.0
    }

    /// Construct from a raw byte value (used by the wire format).
    #[must_use]
    pub const fn from_bits(bits: u8) -> NodeFlags {
        NodeFlags(bits)
    }
}

impl std::ops::BitOr for NodeFlags {
    type Output = NodeFlags;
    fn bitor(self, rhs: NodeFlags) -> NodeFlags {
        NodeFlags(self.0 | rhs.0)
    }
}

impl std::ops::BitOrAssign for NodeFlags {
    fn bitor_assign(&mut self, rhs: NodeFlags) {
        self.0 |= rhs.0;
    }
}

/// Per-`NodeId` metadata entry: one typed payload slot + one flag bitset.
///
/// The two channels are independent — `mark_*` methods on
/// [`NodeMetadataStore`] update `flags` without disturbing `typed`, and
/// `insert_typed` updates `typed` without disturbing `flags`. This is what
/// enables co-occurrence of marker bits with typed payloads (e.g. a
/// macro-generated function whose address has been taken).
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct StoredEntry {
    /// Mutually-exclusive payload-bearing metadata. `None` when the node
    /// only carries marker flags (no macro / classpath provenance).
    pub typed: Option<TypedMetadata>,
    /// Independently-composable marker flags.
    pub flags: NodeFlags,
}

impl StoredEntry {
    /// Construct from a typed payload with empty flags.
    #[must_use]
    pub fn with_typed(typed: TypedMetadata) -> StoredEntry {
        StoredEntry {
            typed: Some(typed),
            flags: NodeFlags::EMPTY,
        }
    }

    /// Construct from flag bits with no typed payload.
    #[must_use]
    pub fn with_flags(flags: NodeFlags) -> StoredEntry {
        StoredEntry { typed: None, flags }
    }

    /// Returns `true` iff this entry has neither typed payload nor any flags set.
    #[must_use]
    pub fn is_vacant(&self) -> bool {
        self.typed.is_none() && self.flags.is_empty()
    }
}

/// Sparse metadata store keyed by full `NodeId` (index + generation).
///
/// Uses `(u32, u64)` tuple key to prevent stale metadata when the
/// generational arena reuses a slot index with a new generation.
/// A lookup with `NodeId { index: 5, generation: 3 }` will NOT match metadata
/// stored for `NodeId { index: 5, generation: 2 }`.
///
/// # Memory characteristics
///
/// For a typical large codebase (100K nodes), only ~5-10% of nodes have
/// metadata. A store with 10K entries at ~200 bytes each = ~2MB, which is
/// acceptable given snapshots are already 10-50MB.
///
/// # Serialization
///
/// The in-memory representation uses `HashMap` for O(1) lookups. For postcard
/// serialization (which doesn't support tuple keys natively), we serialize as
/// a `Vec` of [`NodeMetadataEntryV11`] structs with explicit `index`,
/// `generation`, `kind`, payload-slots, and `flags`, then reconstruct the
/// `HashMap` on deserialization.
///
/// The Phase β joint-stub fields ([`Self::framework_routes`] +
/// [`Self::dispatch_tables`]) are **not** carried by the in-store custom
/// serde impl — they ride the V12 snapshot envelope as separate slots and
/// are reattached via [`Self::set_framework_routes`] /
/// [`Self::set_dispatch_tables`] on load. Keeping them outside the entry
/// wire format preserves V11 metadata-store wire decoding when a V11
/// snapshot is upconverted in place.
#[derive(Debug, Clone, Default)]
pub struct NodeMetadataStore {
    /// Metadata entries keyed by `(NodeId::index(), NodeId::generation())`.
    entries: HashMap<(u32, u64), StoredEntry>,
    /// Plan A (V12, joint-stubs) — per-node framework-route metadata,
    /// populated by Phase 4f's framework extractors. Empty in the stub.
    framework_routes: FrameworkRoutesMap,
    /// Plan B (V12, joint-stubs) — per-snapshot dispatch-resolution side
    /// tables, populated by WS2 resolvers. Empty in the stub.
    dispatch_tables: DispatchTables,
    /// Identifier-blind per-function body-shape descriptors (V15), keyed by
    /// full `NodeId`.
    ///
    /// Unlike `framework_routes` / `dispatch_tables` (which are empty stubs
    /// reattached only at load time), this map IS populated during language
    /// staging and rides the existing take -> rekey -> merge metadata
    /// pipeline. Every `NodeId`-lifecycle and accounting hook on this type
    /// therefore has to carry it: the emptiness contract ([`Self::is_empty`] /
    /// [`Self::len`]), the staging->arena rekey
    /// (`build::parallel_commit::rekey_staging_metadata_to_arena`), the Phase
    /// 4c-prime loser drop (`build::unification::NodeRemapTable::apply_to_metadata_store`),
    /// the incremental-rebuild prune ([`Self::retain_entries`]), the
    /// staged->canonical [`Self::merge`], [`PartialEq`], `heap_bytes`, and the
    /// `NodeIdBearing::all_node_ids` union in `rebuild::coverage`. A miss in any
    /// one silently no-ops the feature or strands a descriptor on a tombstoned
    /// node.
    ///
    /// It is NOT carried by the custom `entries` serde (which writes only
    /// `entries`); the V15 snapshot envelope carries it in a dedicated slot,
    /// reattached via [`Self::set_shape_descriptors`] (the `framework_routes`
    /// envelope-slot precedent). `BTreeMap` (not `HashMap`) for deterministic
    /// serialization order (AC-1 / AC-8).
    shape_descriptors: BTreeMap<NodeId, ShapeDescriptor>,
}

/// Discriminant values for the on-wire `kind` byte.
const TYPED_KIND_NONE: u8 = 0;
const TYPED_KIND_MACRO: u8 = 1;
const TYPED_KIND_CLASSPATH: u8 = 2;

/// V11 wire-format entry for a single metadata record.
///
/// Adds `flags: u8` after the V7 layout. The `kind` byte now describes the
/// typed payload only (Macro / Classpath / None — synthetic moved to `flags`).
#[derive(Debug, Clone, Serialize, Deserialize)]
struct NodeMetadataEntryV11 {
    index: u32,
    generation: u64,
    /// Typed-payload discriminant: 0 = None, 1 = Macro, 2 = Classpath.
    kind: u8,
    /// Macro payload (present when `kind == TYPED_KIND_MACRO`).
    macro_data: Option<MacroNodeMetadata>,
    /// Classpath payload (present when `kind == TYPED_KIND_CLASSPATH`).
    classpath_data: Option<ClasspathNodeMetadata>,
    /// Marker-flag bitset (raw [`NodeFlags`] byte).
    flags: u8,
}

// Legacy V7 / V10 wire-format entries (`NodeMetadataEntryV7Legacy` +
// `LEGACY_V7_KIND_*` constants) moved into
// `sqry-core/src/graph/unified/persistence/legacy_v10.rs` in U03 codex
// iter-1, where they are owned by the versioned V10 wire-type module
// alongside the `EdgeKindV10` mirror. The codex review flagged the
// duplicate definition here as dead code; the canonical home is now the
// persistence/legacy_v10 module, which is where the V10 → V11 upconvert
// translates them into the live `StoredEntry { typed, flags }` shape.

impl Serialize for NodeMetadataStore {
    fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
        let entries: Vec<NodeMetadataEntryV11> = self
            .entries
            .iter()
            .map(|(&(index, generation), stored)| {
                let (kind, macro_data, classpath_data) = match &stored.typed {
                    None => (TYPED_KIND_NONE, None, None),
                    Some(TypedMetadata::Macro(m)) => (TYPED_KIND_MACRO, Some(m.clone()), None),
                    Some(TypedMetadata::Classpath(c)) => {
                        (TYPED_KIND_CLASSPATH, None, Some(c.clone()))
                    }
                };
                NodeMetadataEntryV11 {
                    index,
                    generation,
                    kind,
                    macro_data,
                    classpath_data,
                    flags: stored.flags.bits(),
                }
            })
            .collect();
        entries.serialize(serializer)
    }
}

impl<'de> Deserialize<'de> for NodeMetadataStore {
    fn deserialize<D: serde::Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
        let entries: Vec<NodeMetadataEntryV11> = Vec::deserialize(deserializer)?;
        let mut map = HashMap::with_capacity(entries.len());
        for e in entries {
            let typed = match e.kind {
                TYPED_KIND_NONE => None,
                TYPED_KIND_MACRO => Some(TypedMetadata::Macro(e.macro_data.unwrap_or_default())),
                TYPED_KIND_CLASSPATH => {
                    let data = e.classpath_data.ok_or_else(|| {
                        serde::de::Error::custom(
                            "missing classpath_data for Classpath typed metadata entry",
                        )
                    })?;
                    Some(TypedMetadata::Classpath(data))
                }
                other => {
                    return Err(serde::de::Error::custom(format!(
                        "unknown typed-metadata kind discriminant {other}"
                    )));
                }
            };
            let stored = StoredEntry {
                typed,
                flags: NodeFlags::from_bits(e.flags),
            };
            map.insert((e.index, e.generation), stored);
        }
        // Phase β joint-stubs: `framework_routes` and `dispatch_tables`
        // are NOT carried through the metadata-store custom serde wire
        // (V11 metadata-store payloads must continue to decode bit-for-bit
        // identically). The V12 snapshot envelope carries the new side
        // tables in dedicated slots and the loader reattaches them via
        // `Self::set_framework_routes` / `Self::set_dispatch_tables`
        // after this deserialization completes. Default to empty here.
        Ok(Self {
            entries: map,
            framework_routes: FrameworkRoutesMap::default(),
            dispatch_tables: DispatchTables::default(),
            // Envelope slot, same as the two stubs above: the V15 loader
            // reattaches the descriptors via `set_shape_descriptors` after
            // this entry-vec deserialization. Default to empty here.
            shape_descriptors: BTreeMap::new(),
        })
    }
}

impl NodeMetadataStore {
    /// Create a new empty metadata store.
    #[must_use]
    pub fn new() -> Self {
        Self::default()
    }

    // ---------------------------------------------------------------
    // Typed-payload accessors
    // ---------------------------------------------------------------

    /// Borrowed access to the typed payload for a node, if any.
    ///
    /// Returns `None` when the node has only marker flags or no entry at
    /// all. Stale (generation-mismatched) `NodeId`s also return `None`.
    #[must_use]
    pub fn get_typed(&self, node_id: NodeId) -> Option<&TypedMetadata> {
        self.entries
            .get(&(node_id.index(), node_id.generation()))?
            .typed
            .as_ref()
    }

    /// Mutable access to the typed payload for a node, if any.
    pub fn get_typed_mut(&mut self, node_id: NodeId) -> Option<&mut TypedMetadata> {
        self.entries
            .get_mut(&(node_id.index(), node_id.generation()))?
            .typed
            .as_mut()
    }

    /// Convenience accessor for the [`TypedMetadata::Macro`] variant only.
    ///
    /// Replaces the legacy `get` accessor. Returns `None` for classpath
    /// entries, marker-only entries, missing entries, and stale generations.
    #[must_use]
    pub fn get_macro(&self, node_id: NodeId) -> Option<&MacroNodeMetadata> {
        match self.get_typed(node_id)? {
            TypedMetadata::Macro(m) => Some(m),
            TypedMetadata::Classpath(_) => None,
        }
    }

    /// Mutable accessor for the [`TypedMetadata::Macro`] variant only.
    pub fn get_macro_mut(&mut self, node_id: NodeId) -> Option<&mut MacroNodeMetadata> {
        match self.get_typed_mut(node_id)? {
            TypedMetadata::Macro(m) => Some(m),
            TypedMetadata::Classpath(_) => None,
        }
    }

    // ---------------------------------------------------------------
    // Marker-flag accessors
    // ---------------------------------------------------------------

    /// Returns the marker-flag bitset (empty bitset when no entry exists).
    ///
    /// Cheap by-value `Copy` return — no borrow contention with
    /// [`Self::get_typed`].
    #[must_use]
    pub fn get_flags(&self, node_id: NodeId) -> NodeFlags {
        self.entries
            .get(&(node_id.index(), node_id.generation()))
            .map_or(NodeFlags::EMPTY, |e| e.flags)
    }

    /// Returns `true` iff the node has [`NodeFlags::SYNTHETIC`] set.
    #[must_use]
    pub fn is_synthetic(&self, node_id: NodeId) -> bool {
        self.get_flags(node_id).contains(NodeFlags::SYNTHETIC)
    }

    /// Returns `true` iff the node has [`NodeFlags::ADDRESS_TAKEN`] set.
    #[must_use]
    pub fn is_address_taken(&self, node_id: NodeId) -> bool {
        self.get_flags(node_id).contains(NodeFlags::ADDRESS_TAKEN)
    }

    /// Returns `true` iff the node has [`NodeFlags::CALLSITE_PROMISCUOUS`] set.
    #[must_use]
    pub fn is_callsite_promiscuous(&self, node_id: NodeId) -> bool {
        self.get_flags(node_id)
            .contains(NodeFlags::CALLSITE_PROMISCUOUS)
    }

    /// Returns `true` iff the node has [`NodeFlags::IMPORT_STDLIB`] set.
    #[must_use]
    pub fn is_import_stdlib(&self, node_id: NodeId) -> bool {
        self.get_flags(node_id).contains(NodeFlags::IMPORT_STDLIB)
    }

    /// Returns `true` iff the node has [`NodeFlags::IMPORT_RELATIVE`] set.
    #[must_use]
    pub fn is_import_relative(&self, node_id: NodeId) -> bool {
        self.get_flags(node_id).contains(NodeFlags::IMPORT_RELATIVE)
    }

    /// Set [`NodeFlags::SYNTHETIC`] on this node.
    ///
    /// Does NOT disturb any existing typed payload — a macro-generated node
    /// that is also marked synthetic retains both channels.
    pub fn mark_synthetic(&mut self, node_id: NodeId) {
        self.set_flag(node_id, NodeFlags::SYNTHETIC);
    }

    /// Set [`NodeFlags::ADDRESS_TAKEN`] on this node.
    ///
    /// Does NOT disturb any existing typed payload — preserves Macro /
    /// Classpath provenance for nodes that happen to be address-taken (e.g.
    /// a `DEFINE_HANDLER(my_irq)` C function that is also `&my_irq`'d).
    pub fn mark_address_taken(&mut self, node_id: NodeId) {
        self.set_flag(node_id, NodeFlags::ADDRESS_TAKEN);
    }

    /// Set [`NodeFlags::CALLSITE_PROMISCUOUS`] on this node.
    ///
    /// Does NOT disturb any existing typed payload.
    pub fn mark_callsite_promiscuous(&mut self, node_id: NodeId) {
        self.set_flag(node_id, NodeFlags::CALLSITE_PROMISCUOUS);
    }

    /// Set [`NodeFlags::IMPORT_STDLIB`] on this node.
    ///
    /// Does NOT disturb any existing typed payload (an import node keeps its
    /// provenance channel while gaining the stdlib classification).
    pub fn mark_import_stdlib(&mut self, node_id: NodeId) {
        self.set_flag(node_id, NodeFlags::IMPORT_STDLIB);
    }

    /// Set [`NodeFlags::IMPORT_RELATIVE`] on this node.
    ///
    /// Does NOT disturb any existing typed payload (an import node keeps its
    /// provenance channel while gaining the relative classification).
    pub fn mark_import_relative(&mut self, node_id: NodeId) {
        self.set_flag(node_id, NodeFlags::IMPORT_RELATIVE);
    }

    fn set_flag(&mut self, node_id: NodeId, flag: NodeFlags) {
        self.entries
            .entry((node_id.index(), node_id.generation()))
            .or_default()
            .flags
            .insert(flag);
    }

    // ---------------------------------------------------------------
    // Insertion / removal
    // ---------------------------------------------------------------

    /// Insert macro metadata for a node, replacing any existing typed
    /// payload at this `NodeId`. Marker flags on the existing entry are
    /// preserved.
    pub fn insert(&mut self, node_id: NodeId, metadata: MacroNodeMetadata) {
        self.insert_typed(node_id, TypedMetadata::Macro(metadata));
    }

    /// Insert a typed payload for a node, replacing any existing typed
    /// payload. Marker flags on the existing entry are preserved.
    pub fn insert_typed(&mut self, node_id: NodeId, typed: TypedMetadata) {
        let slot = self
            .entries
            .entry((node_id.index(), node_id.generation()))
            .or_default();
        slot.typed = Some(typed);
    }

    /// Insert a fully-formed [`StoredEntry`] for a node, replacing any
    /// existing entry. Used by bulk-remap paths (e.g. classpath emitter
    /// id remapping) and the snapshot upconvert path.
    pub fn insert_entry(&mut self, node_id: NodeId, entry: StoredEntry) {
        self.entries
            .insert((node_id.index(), node_id.generation()), entry);
    }

    /// Get or insert a default macro-metadata entry for a node.
    ///
    /// If the entry doesn't exist, it's created with an empty
    /// `MacroNodeMetadata` typed payload and empty flags.
    ///
    /// # Panics
    ///
    /// Panics if the existing entry has a non-Macro typed payload
    /// (i.e. [`TypedMetadata::Classpath`]). Callers must not mix typed
    /// payloads at the same key.
    pub fn get_or_insert_default(&mut self, node_id: NodeId) -> &mut MacroNodeMetadata {
        let slot = self
            .entries
            .entry((node_id.index(), node_id.generation()))
            .or_insert_with(|| {
                StoredEntry::with_typed(TypedMetadata::Macro(MacroNodeMetadata::default()))
            });
        if slot.typed.is_none() {
            slot.typed = Some(TypedMetadata::Macro(MacroNodeMetadata::default()));
        }
        match slot.typed.as_mut() {
            Some(TypedMetadata::Macro(m)) => m,
            Some(TypedMetadata::Classpath(_)) => {
                panic!("get_or_insert_default called on a Classpath typed metadata entry")
            }
            None => unreachable!("just populated above"),
        }
    }

    /// Remove the entry for a node and return its macro payload, if any.
    ///
    /// Returns `None` if no entry existed, or if the entry's typed payload
    /// was not [`TypedMetadata::Macro`]. The entry (including any flags) is
    /// removed in all cases when an entry was present.
    pub fn remove(&mut self, node_id: NodeId) -> Option<MacroNodeMetadata> {
        match self
            .entries
            .remove(&(node_id.index(), node_id.generation()))?
            .typed
        {
            Some(TypedMetadata::Macro(m)) => Some(m),
            Some(TypedMetadata::Classpath(_)) | None => None,
        }
    }

    /// Remove the entry for a node and return the full [`StoredEntry`].
    pub fn remove_entry(&mut self, node_id: NodeId) -> Option<StoredEntry> {
        self.entries
            .remove(&(node_id.index(), node_id.generation()))
    }

    // ---------------------------------------------------------------
    // Iteration / bookkeeping
    // ---------------------------------------------------------------

    /// Returns the number of distinct nodes carrying any metadata: every
    /// entry, plus shape-only descriptors whose `NodeId` has no entry.
    ///
    /// Counts the UNION (not the sum) so a node that carries both an entry and
    /// a shape descriptor is counted once, keeping `len()` consistent with
    /// [`Self::is_empty`] (`len() == 0` iff `is_empty()`). Most functions carry
    /// a descriptor but no entry-metadata, so this is the common shape-only
    /// case that the build-pipeline drop gates must NOT discard.
    #[must_use]
    pub fn len(&self) -> usize {
        let shape_only = self
            .shape_descriptors
            .keys()
            .filter(|nid| !self.entries.contains_key(&(nid.index(), nid.generation())))
            .count();
        self.entries.len() + shape_only
    }

    /// Returns true if no nodes have metadata.
    ///
    /// Load-bearing: the chunked build pipeline uses `is_empty()` as a DROP
    /// gate (entrypoint, incremental, parallel-commit). A shape-only store
    /// (descriptors but no `entries`, the common case for ordinary functions)
    /// MUST report non-empty or it is silently discarded before the rekey +
    /// merge, no-op'ing the whole feature. Hence the `shape_descriptors` term.
    #[must_use]
    pub fn is_empty(&self) -> bool {
        self.entries.is_empty() && self.shape_descriptors.is_empty()
    }

    /// Iterate over entries whose typed payload is `Macro`, yielding the
    /// `MacroNodeMetadata`. Entries that are flag-only, classpath, or
    /// payload-less are skipped.
    pub fn iter(&self) -> impl Iterator<Item = ((u32, u64), &MacroNodeMetadata)> {
        self.entries.iter().filter_map(|(&k, v)| match &v.typed {
            Some(TypedMetadata::Macro(m)) => Some((k, m)),
            Some(TypedMetadata::Classpath(_)) | None => None,
        })
    }

    /// Iterate over every stored entry as `(key, &StoredEntry)`.
    ///
    /// Replaces the legacy `iter_all` accessor that returned `&NodeMetadata`.
    pub fn iter_entries(&self) -> impl Iterator<Item = ((u32, u64), &StoredEntry)> {
        self.entries.iter().map(|(&k, v)| (k, v))
    }

    /// Merge another metadata store into this one.
    ///
    /// On key collision the two [`StoredEntry`] channels merge independently,
    /// mirroring [`Self::insert_typed`]'s documented "marker flags on the
    /// existing entry are preserved" contract:
    ///
    /// - **flags** are OR-ed in (they are independently-composable markers, so
    ///   a later merge must never clear a bit an earlier one set). This keeps
    ///   the Go import-classification bits (`IMPORT_STDLIB` / `IMPORT_RELATIVE`)
    ///   alive when a build-tagged file later stamps a `cfg_condition` macro
    ///   payload (whose store carries empty flags) onto the same import node via
    ///   `stamp_cfg_condition_for_file`. A whole-entry overwrite here silently
    ///   dropped that classification (issue #467).
    /// - the **typed payload** takes `other`'s value when present, else keeps the
    ///   existing one, so a flags-only incoming entry never erases a typed
    ///   payload (and vice versa).
    ///
    /// Shape descriptors are whole-value per node and merge by overwrite on key
    /// collision: unlike the `framework_routes` / `dispatch_tables` stubs (which
    /// are reattached only at load time and so are deliberately NOT merged here),
    /// shape descriptors are produced per-file during staging and reach the
    /// authoritative store through exactly this merge, so they must ride it.
    pub fn merge(&mut self, other: &NodeMetadataStore) {
        for (&key, value) in &other.entries {
            let slot = self.entries.entry(key).or_default();
            slot.flags |= value.flags;
            if value.typed.is_some() {
                slot.typed = value.typed.clone();
            }
        }
        for (&node_id, descriptor) in &other.shape_descriptors {
            self.shape_descriptors.insert(node_id, descriptor.clone());
        }
    }

    /// Retain only entries whose `(index, generation)` key satisfies `keep`.
    ///
    /// Used by the Gate 0b [`NodeIdBearing`] impl
    /// (`sqry-core/src/graph/unified/rebuild/coverage.rs`) to drop
    /// metadata for tombstoned `NodeIds` during
    /// `RebuildGraph::finalize()`. Exposed at `pub(crate)` scope because
    /// only the rebuild pipeline needs predicate-based filtering;
    /// downstream callers use the targeted [`Self::remove`] /
    /// [`Self::remove_entry`] entry points.
    ///
    /// Live in the default build: the consumers are
    /// `RebuildGraph::finalize()` and the residue check (both via the
    /// [`NodeIdBearing::retain_nodes`] impl), reached from the ungated
    /// public `build::incremental::incremental_rebuild` -> `finalize`
    /// path (the `rebuild::coverage` unit tests exercise it too).
    ///
    /// [`NodeIdBearing`]: crate::graph::unified::rebuild::coverage::NodeIdBearing
    /// [`NodeIdBearing::retain_nodes`]: crate::graph::unified::rebuild::coverage::NodeIdBearing::retain_nodes
    pub(crate) fn retain_entries<F>(&mut self, mut keep: F)
    where
        F: FnMut(u32, u64) -> bool,
    {
        self.entries
            .retain(|&(index, generation), _entry| keep(index, generation));
        // Prune shape descriptors under the SAME predicate so a tombstoned
        // node cannot leave a stranded descriptor behind. Without this the
        // residue audit would flag a descriptor-only stale `NodeId`.
        self.shape_descriptors
            .retain(|node_id, _descriptor| keep(node_id.index(), node_id.generation()));
    }

    /// Test-only: clear the Phase A marker bits
    /// ([`NodeFlags::ADDRESS_TAKEN`] and [`NodeFlags::CALLSITE_PROMISCUOUS`])
    /// from every stored entry, leaving every other flag and the typed
    /// payload untouched.
    ///
    /// Used exclusively by `sqry-core/tests/snapshot_size_phase_a.rs`
    /// (U19) to materialize a "Phase-A-free" baseline snapshot for the
    /// +10% snapshot-size gate.
    ///
    /// Gated behind `cfg(any(test, feature = "test-support"))` so the
    /// helper is invisible to production builds and never accidentally
    /// invoked from non-test surfaces.
    #[cfg(any(test, feature = "test-support"))]
    pub fn clear_phase_a_flags_for_test(&mut self) {
        let mask = NodeFlags::ADDRESS_TAKEN | NodeFlags::CALLSITE_PROMISCUOUS;
        for slot in self.entries.values_mut() {
            slot.flags.remove(mask);
        }
    }

    // ---------------------------------------------------------------
    // Phase β joint-stubs: framework-routes + dispatch-tables
    // ---------------------------------------------------------------
    //
    // The accessors below are the public surface for Plan A's framework
    // route extractors and Plan B's dispatch resolvers. In this PR they
    // are read-only-empty by default — no resolver populates them yet.
    // The setters are used by the V12 snapshot load path to reattach the
    // envelope slots after the metadata-store entries Vec has been
    // deserialized.

    /// Read-only access to the framework-route map (Plan A).
    ///
    /// Empty in the stub. Populated by Plan A's Phase 4f extractor pass
    /// in the `feat/framework-route-extractors` downstream PR.
    #[must_use]
    pub fn framework_routes(&self) -> &FrameworkRoutesMap {
        &self.framework_routes
    }

    /// Mutable access to the framework-route map (Plan A).
    ///
    /// Reserved for Phase 4f extractors. The MCP filter / planner predicate
    /// that ship in this same PR read through [`Self::framework_routes`]
    /// only.
    pub fn framework_routes_mut(&mut self) -> &mut FrameworkRoutesMap {
        &mut self.framework_routes
    }

    /// Replace the framework-route map wholesale.
    ///
    /// Used by the V12 snapshot loader to reattach the envelope slot to
    /// the in-memory metadata store after entry-vec deserialization.
    pub fn set_framework_routes(&mut self, routes: FrameworkRoutesMap) {
        self.framework_routes = routes;
    }

    /// Lookup helper — returns the route metadata for a node if one was
    /// recorded by a framework extractor.
    #[must_use]
    pub fn framework_route(&self, node_id: NodeId) -> Option<&FrameworkRouteMetadata> {
        self.framework_routes.get(&node_id)
    }

    /// Read-only access to the dispatch-tables side store (Plan B).
    ///
    /// Empty in the stub. Populated by Plan B's WS2 resolver passes
    /// (JVM virtual / interface, Go interface, Python duck-typed,
    /// TypeScript structural, promiscuous-cap elision).
    #[must_use]
    pub fn dispatch_tables(&self) -> &DispatchTables {
        &self.dispatch_tables
    }

    /// Mutable access to the dispatch-tables side store (Plan B).
    pub fn dispatch_tables_mut(&mut self) -> &mut DispatchTables {
        &mut self.dispatch_tables
    }

    /// Replace the dispatch-tables wholesale.
    ///
    /// Used by the V12 snapshot loader to reattach the envelope slot to
    /// the in-memory metadata store after entry-vec deserialization.
    pub fn set_dispatch_tables(&mut self, tables: DispatchTables) {
        self.dispatch_tables = tables;
    }

    // ---------------------------------------------------------------
    // Shape descriptors (V15 body-shape side table)
    // ---------------------------------------------------------------

    /// Insert (or replace) the shape descriptor for a node.
    ///
    /// Called from the build seam (`build::staging`) during staging, keyed by
    /// the staging-local `NodeId`, and again by
    /// `rekey_staging_metadata_to_arena` under the committed arena `NodeId`.
    pub fn insert_shape_descriptor(&mut self, node_id: NodeId, descriptor: ShapeDescriptor) {
        self.shape_descriptors.insert(node_id, descriptor);
    }

    /// Borrowed access to a node's shape descriptor, if one was computed.
    #[must_use]
    pub fn shape_descriptor(&self, node_id: NodeId) -> Option<&ShapeDescriptor> {
        self.shape_descriptors.get(&node_id)
    }

    /// Read-only access to the whole shape-descriptor map.
    ///
    /// Used by the V15 snapshot writer to extract the envelope payload, by the
    /// staging->arena rekey to iterate staging descriptors, and by the
    /// structural-index / query surfaces downstream.
    #[must_use]
    pub fn shape_descriptors(&self) -> &BTreeMap<NodeId, ShapeDescriptor> {
        &self.shape_descriptors
    }

    /// Remove a node's shape descriptor, returning it if present.
    ///
    /// Used by the Phase 4c-prime loser-drop
    /// (`NodeRemapTable::apply_to_metadata_store`) to evict a tombstoned
    /// loser's descriptor.
    pub fn remove_shape_descriptor(&mut self, node_id: NodeId) -> Option<ShapeDescriptor> {
        self.shape_descriptors.remove(&node_id)
    }

    /// Replace the shape-descriptor map wholesale.
    ///
    /// Used by the V15 snapshot loader to reattach the envelope slot after
    /// entry-vec deserialization (the `set_framework_routes` precedent).
    pub fn set_shape_descriptors(&mut self, descriptors: BTreeMap<NodeId, ShapeDescriptor>) {
        self.shape_descriptors = descriptors;
    }

    /// Iterate the `NodeId`s that carry a shape descriptor.
    ///
    /// The `NodeIdBearing` impl in `rebuild::coverage` chains this with the
    /// entry-derived `NodeId`s so the tombstone-residue audit sees descriptor-
    /// only nodes too.
    pub fn iter_shape_descriptor_node_ids(&self) -> impl Iterator<Item = NodeId> + '_ {
        self.shape_descriptors.keys().copied()
    }
}

impl PartialEq for NodeMetadataStore {
    fn eq(&self, other: &Self) -> bool {
        self.entries == other.entries
            && self.framework_routes == other.framework_routes
            && self.dispatch_tables == other.dispatch_tables
            // AC-8 round-trip equality must not be blind to descriptor loss.
            && self.shape_descriptors == other.shape_descriptors
    }
}

impl Eq for NodeMetadataStore {}

impl crate::graph::unified::memory::GraphMemorySize for NodeMetadataStore {
    fn heap_bytes(&self) -> usize {
        use crate::graph::unified::memory::HASHMAP_ENTRY_OVERHEAD;

        let base = self.entries.capacity()
            * (std::mem::size_of::<(u32, u64)>()
                + std::mem::size_of::<StoredEntry>()
                + HASHMAP_ENTRY_OVERHEAD);
        // Account for heap Strings inside each typed payload. Marker-flag
        // entries are payload-less — only the `flags` byte itself, which is
        // counted via mem::size_of::<StoredEntry>() in `base`.
        let inner: usize = self
            .entries
            .values()
            .map(|entry| match &entry.typed {
                None => 0,
                Some(TypedMetadata::Macro(m)) => {
                    m.macro_source.as_ref().map_or(0, String::capacity)
                        + m.cfg_condition.as_ref().map_or(0, String::capacity)
                        + m.unresolved_attributes
                            .iter()
                            .map(String::capacity)
                            .sum::<usize>()
                        + m.unresolved_attributes.capacity() * std::mem::size_of::<String>()
                }
                Some(TypedMetadata::Classpath(c)) => {
                    c.coordinates.as_ref().map_or(0, String::capacity)
                        + c.jar_path.capacity()
                        + c.fqn.capacity()
                }
            })
            .sum();
        // Phase β joint-stubs: account for the framework-routes BTreeMap
        // and the dispatch-tables side store. The stubs are empty by
        // construction; this code accounts for whatever Plan A / Plan B
        // populate downstream without needing a second size-impl edit.
        let framework_routes_bytes = self.framework_routes.len()
            * (std::mem::size_of::<NodeId>() + std::mem::size_of::<FrameworkRouteMetadata>());
        // Phase β joint-stubs (V12 DispatchTables shape — Plan B DESIGN
        // §3.7): five per-plane collections, each contributing
        // `len * (NodeId + entry-type)` heap bytes. Empty until Plan B's
        // resolver PRs (`U_WS2_2_*` ...) populate the planes.
        let dt = &self.dispatch_tables;
        let dispatch_tables_bytes = dt.jvm_virtual.len()
            * (std::mem::size_of::<NodeId>()
                + std::mem::size_of::<super::dispatch_tables::JvmDispatchEntry>())
            + dt.go_interface.len()
                * (std::mem::size_of::<NodeId>()
                    + std::mem::size_of::<super::dispatch_tables::GoDispatchEntry>())
            + dt.python_duck.len()
                * (std::mem::size_of::<NodeId>()
                    + std::mem::size_of::<super::dispatch_tables::PythonDispatchEntry>())
            + dt.ts_structural.len()
                * (std::mem::size_of::<NodeId>()
                    + std::mem::size_of::<super::dispatch_tables::TsDispatchEntry>())
            + dt.cap_hits.len() * std::mem::size_of::<super::dispatch_tables::CapHit>();
        // V15 body-shape side table. `ShapeDescriptor` is fixed-size POD (no
        // heap-allocated fields: the cf histogram and the MinHash lanes are
        // inline arrays), so `size_of` captures the whole payload; charge the
        // key alongside it, mirroring the framework-routes accounting above.
        // This delta feeds `CodeGraph::heap_bytes` and the daemon
        // admission/LRU memory budget, so a shape-only graph is accounted, not
        // admitted past `memory_limit_mb` undercounted.
        let shape_descriptors_bytes = self.shape_descriptors.len()
            * (std::mem::size_of::<NodeId>() + std::mem::size_of::<ShapeDescriptor>());
        base + inner + framework_routes_bytes + dispatch_tables_bytes + shape_descriptors_bytes
    }
}

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

    #[test]
    fn test_metadata_store_basic_operations() {
        let mut store = NodeMetadataStore::new();
        assert!(store.is_empty());
        assert_eq!(store.len(), 0);

        let node = NodeId::new(5, 1);
        let metadata = MacroNodeMetadata {
            macro_generated: Some(true),
            macro_source: Some("derive_Debug".to_string()),
            ..Default::default()
        };

        store.insert(node, metadata.clone());
        assert_eq!(store.len(), 1);
        assert!(!store.is_empty());

        let retrieved = store.get_macro(node).unwrap();
        assert_eq!(retrieved.macro_generated, Some(true));
        assert_eq!(retrieved.macro_source.as_deref(), Some("derive_Debug"));
    }

    #[test]
    fn test_metadata_full_nodeid_key() {
        let mut store = NodeMetadataStore::new();

        let node_gen1 = NodeId::new(5, 1);
        let node_gen2 = NodeId::new(5, 2);

        store.insert(
            node_gen1,
            MacroNodeMetadata {
                macro_generated: Some(true),
                ..Default::default()
            },
        );

        // Same index, different generation → should NOT match
        assert!(store.get_macro(node_gen2).is_none());

        // Same index, same generation → should match
        assert!(store.get_macro(node_gen1).is_some());
    }

    #[test]
    fn test_metadata_slot_reuse_no_stale_data() {
        let mut store = NodeMetadataStore::new();

        // Simulate: node at index 5 gen 1 has metadata
        let old_node = NodeId::new(5, 1);
        store.insert(
            old_node,
            MacroNodeMetadata {
                cfg_condition: Some("test".to_string()),
                ..Default::default()
            },
        );

        // Simulate: slot 5 is reused with generation 2 (new node)
        let new_node = NodeId::new(5, 2);

        // New node should NOT see old metadata
        assert!(store.get_macro(new_node).is_none());

        // Old node still accessible
        assert_eq!(
            store.get_macro(old_node).unwrap().cfg_condition.as_deref(),
            Some("test")
        );
    }

    #[test]
    fn test_metadata_store_postcard_roundtrip() {
        let mut store = NodeMetadataStore::new();

        store.insert(
            NodeId::new(1, 0),
            MacroNodeMetadata {
                macro_generated: Some(true),
                macro_source: Some("derive_Debug".to_string()),
                cfg_condition: Some("test".to_string()),
                cfg_active: Some(true),
                proc_macro_kind: Some(ProcMacroFunctionKind::Derive),
                expansion_cached: Some(false),
                unresolved_attributes: vec!["my_attr".to_string()],
            },
        );

        store.insert(
            NodeId::new(42, 3),
            MacroNodeMetadata {
                cfg_condition: Some("feature = \"serde\"".to_string()),
                ..Default::default()
            },
        );

        let bytes = postcard::to_allocvec(&store).expect("serialize");
        let deserialized: NodeMetadataStore = postcard::from_bytes(&bytes).expect("deserialize");

        assert_eq!(store, deserialized);
    }

    #[test]
    fn test_empty_metadata_store_zero_overhead() {
        let store = NodeMetadataStore::new();
        let bytes = postcard::to_allocvec(&store).expect("serialize");

        // Empty HashMap serializes to a single varint length of 0
        assert!(
            bytes.len() <= 2,
            "Empty store should serialize to minimal bytes, got {} bytes",
            bytes.len()
        );
    }

    #[test]
    fn test_metadata_store_merge() {
        let mut store1 = NodeMetadataStore::new();
        let mut store2 = NodeMetadataStore::new();

        store1.insert(
            NodeId::new(1, 0),
            MacroNodeMetadata {
                macro_generated: Some(true),
                ..Default::default()
            },
        );

        store2.insert(
            NodeId::new(2, 0),
            MacroNodeMetadata {
                cfg_condition: Some("test".to_string()),
                ..Default::default()
            },
        );

        store1.merge(&store2);
        assert_eq!(store1.len(), 2);
        assert!(store1.get_macro(NodeId::new(1, 0)).is_some());
        assert!(store1.get_macro(NodeId::new(2, 0)).is_some());
    }

    #[test]
    fn test_proc_macro_function_kind_serde() {
        let kinds = [
            ProcMacroFunctionKind::Derive,
            ProcMacroFunctionKind::Attribute,
            ProcMacroFunctionKind::FunctionLike,
        ];

        for kind in kinds {
            let bytes = postcard::to_allocvec(&kind).expect("serialize");
            let deserialized: ProcMacroFunctionKind =
                postcard::from_bytes(&bytes).expect("deserialize");
            assert_eq!(kind, deserialized);
        }
    }

    #[test]
    fn test_metadata_get_or_insert_default() {
        let mut store = NodeMetadataStore::new();
        let node = NodeId::new(10, 0);

        // First access creates default
        let meta = store.get_or_insert_default(node);
        meta.cfg_condition = Some("test".to_string());

        // Second access returns existing
        let meta = store.get_macro(node).unwrap();
        assert_eq!(meta.cfg_condition.as_deref(), Some("test"));
    }

    #[test]
    fn test_metadata_remove() {
        let mut store = NodeMetadataStore::new();
        let node = NodeId::new(1, 0);

        store.insert(
            node,
            MacroNodeMetadata {
                macro_generated: Some(true),
                ..Default::default()
            },
        );

        assert!(store.get_macro(node).is_some());
        let removed = store.remove(node);
        assert!(removed.is_some());
        assert!(store.get_macro(node).is_none());
        assert!(store.is_empty());
    }

    #[test]
    fn test_metadata_store_large_scale() {
        let mut store = NodeMetadataStore::new();

        // Insert 10K entries (simulating ~10% of a 100K-node codebase)
        for i in 0..10_000u32 {
            store.insert(
                NodeId::new(i, 0),
                MacroNodeMetadata {
                    cfg_condition: Some(format!("feature_{i}")),
                    ..Default::default()
                },
            );
        }

        assert_eq!(store.len(), 10_000);

        // Verify O(1) lookups
        assert!(store.get_macro(NodeId::new(0, 0)).is_some());
        assert!(store.get_macro(NodeId::new(5_000, 0)).is_some());
        assert!(store.get_macro(NodeId::new(9_999, 0)).is_some());
        assert!(store.get_macro(NodeId::new(10_000, 0)).is_none());

        // Verify round-trip
        let bytes = postcard::to_allocvec(&store).expect("serialize");
        let deserialized: NodeMetadataStore = postcard::from_bytes(&bytes).expect("deserialize");
        assert_eq!(store, deserialized);
    }

    #[test]
    fn test_classpath_metadata_insert_and_get() {
        let mut store = NodeMetadataStore::new();
        let node = NodeId::new(100, 0);

        let cp_meta = ClasspathNodeMetadata {
            coordinates: Some("com.google.guava:guava:33.0.0".to_string()),
            jar_path: "/home/user/.m2/repository/guava-33.0.0.jar".to_string(),
            fqn: "com.google.common.collect.ImmutableList".to_string(),
            is_direct_dependency: true,
        };

        store.insert_typed(node, TypedMetadata::Classpath(cp_meta.clone()));
        assert_eq!(store.len(), 1);

        // get_macro should return None (only returns Macro variant)
        assert!(store.get_macro(node).is_none());

        // get_typed should return the classpath metadata
        let retrieved = store.get_typed(node).unwrap();
        match retrieved {
            TypedMetadata::Classpath(cp) => {
                assert_eq!(cp.fqn, "com.google.common.collect.ImmutableList");
                assert_eq!(
                    cp.coordinates.as_deref(),
                    Some("com.google.guava:guava:33.0.0")
                );
                assert!(cp.is_direct_dependency);
            }
            TypedMetadata::Macro(_) => panic!("expected Classpath variant"),
        }
    }

    #[test]
    fn test_classpath_metadata_postcard_roundtrip() {
        let mut store = NodeMetadataStore::new();

        // Mix of macro and classpath metadata
        store.insert(
            NodeId::new(1, 0),
            MacroNodeMetadata {
                macro_generated: Some(true),
                ..Default::default()
            },
        );

        store.insert_typed(
            NodeId::new(2, 0),
            TypedMetadata::Classpath(ClasspathNodeMetadata {
                coordinates: Some("org.slf4j:slf4j-api:2.0.0".to_string()),
                jar_path: "slf4j-api-2.0.0.jar".to_string(),
                fqn: "org.slf4j.Logger".to_string(),
                is_direct_dependency: false,
            }),
        );

        let bytes = postcard::to_allocvec(&store).expect("serialize");
        let deserialized: NodeMetadataStore = postcard::from_bytes(&bytes).expect("deserialize");
        assert_eq!(store, deserialized);
        assert_eq!(deserialized.len(), 2);

        // Verify macro entry
        assert!(deserialized.get_macro(NodeId::new(1, 0)).is_some());

        // Verify classpath entry via get_typed
        let cp = deserialized.get_typed(NodeId::new(2, 0)).unwrap();
        assert!(matches!(cp, TypedMetadata::Classpath(_)));
    }

    #[test]
    fn test_node_metadata_store_json_roundtrip() {
        let mut store = NodeMetadataStore::new();

        store.insert(
            NodeId::new(1, 0),
            MacroNodeMetadata {
                macro_generated: Some(true),
                macro_source: Some("serde_derive".to_string()),
                ..Default::default()
            },
        );

        store.insert_typed(
            NodeId::new(2, 0),
            TypedMetadata::Classpath(ClasspathNodeMetadata {
                coordinates: None,
                jar_path: "rt.jar".to_string(),
                fqn: "java.lang.String".to_string(),
                is_direct_dependency: true,
            }),
        );

        let json = serde_json::to_string(&store).unwrap();
        let deserialized: NodeMetadataStore = serde_json::from_str(&json).unwrap();
        assert_eq!(store, deserialized);
    }

    #[test]
    fn test_iter_entries_includes_both_types() {
        let mut store = NodeMetadataStore::new();

        store.insert(
            NodeId::new(1, 0),
            MacroNodeMetadata {
                macro_generated: Some(true),
                ..Default::default()
            },
        );

        store.insert_typed(
            NodeId::new(2, 0),
            TypedMetadata::Classpath(ClasspathNodeMetadata {
                coordinates: None,
                jar_path: "test.jar".to_string(),
                fqn: "com.example.Test".to_string(),
                is_direct_dependency: true,
            }),
        );

        // iter() only yields macro entries
        let macro_entries: Vec<_> = store.iter().collect();
        assert_eq!(macro_entries.len(), 1);

        // iter_entries() yields all entries
        let all_entries: Vec<_> = store.iter_entries().collect();
        assert_eq!(all_entries.len(), 2);
    }

    #[test]
    fn test_remove_entry_classpath() {
        let mut store = NodeMetadataStore::new();
        let node = NodeId::new(50, 0);

        store.insert_typed(
            node,
            TypedMetadata::Classpath(ClasspathNodeMetadata {
                coordinates: None,
                jar_path: "test.jar".to_string(),
                fqn: "Test".to_string(),
                is_direct_dependency: true,
            }),
        );

        assert_eq!(store.len(), 1);

        // remove() returns None for non-macro entries, but still removes
        let removed = store.remove(node);
        assert!(removed.is_none());
        assert!(store.is_empty());
    }

    #[test]
    fn test_remove_entry_typed() {
        let mut store = NodeMetadataStore::new();
        let node = NodeId::new(50, 0);

        store.insert_typed(
            node,
            TypedMetadata::Classpath(ClasspathNodeMetadata {
                coordinates: None,
                jar_path: "test.jar".to_string(),
                fqn: "Test".to_string(),
                is_direct_dependency: true,
            }),
        );

        // remove_entry() returns the full StoredEntry
        let removed = store.remove_entry(node);
        assert!(matches!(
            removed.as_ref().and_then(|e| e.typed.as_ref()),
            Some(TypedMetadata::Classpath(_))
        ));
        assert!(store.is_empty());
    }

    // ------------------------------------------------------------------
    // V15 shape-descriptor side table: emptiness contract + lifecycle hooks
    // ------------------------------------------------------------------

    mod shape_descriptor_tests {
        use super::*;
        use crate::graph::unified::build::shape::CfBucket;
        use crate::graph::unified::memory::GraphMemorySize;
        use crate::graph::unified::storage::shape::ShapeDescriptor;

        fn descriptor_with_branches(n: u16) -> ShapeDescriptor {
            let mut d = ShapeDescriptor::default();
            d.cf_histogram[CfBucket::Branch.index()] = n;
            d
        }

        #[test]
        fn shape_only_store_is_non_empty_and_counts() {
            // The single most dangerous miss: a store with descriptors but no
            // entries (the common case for ordinary functions) MUST report
            // non-empty so the build-pipeline drop gates do not discard it.
            let mut store = NodeMetadataStore::new();
            assert!(store.is_empty());
            assert_eq!(store.len(), 0);

            store.insert_shape_descriptor(NodeId::new(7, 1), ShapeDescriptor::default());
            assert!(!store.is_empty(), "shape-only store must be non-empty");
            assert_eq!(store.len(), 1);
        }

        #[test]
        fn len_counts_union_not_sum() {
            // A node carrying BOTH an entry and a descriptor is counted once,
            // keeping len() consistent with is_empty().
            let mut store = NodeMetadataStore::new();
            let both = NodeId::new(1, 1);
            store.insert(both, MacroNodeMetadata::default());
            store.insert_shape_descriptor(both, ShapeDescriptor::default());
            store.insert_shape_descriptor(NodeId::new(2, 1), ShapeDescriptor::default());
            // 1 node with an entry (also a descriptor) + 1 shape-only node = 2.
            assert_eq!(store.len(), 2);
        }

        #[test]
        fn merge_carries_shape_descriptors() {
            let mut dst = NodeMetadataStore::new();
            let mut src = NodeMetadataStore::new();
            src.insert_shape_descriptor(NodeId::new(3, 1), descriptor_with_branches(5));
            dst.merge(&src);
            assert_eq!(
                dst.shape_descriptor(NodeId::new(3, 1))
                    .map(|d| d.cf_histogram[CfBucket::Branch.index()]),
                Some(5),
                "merge must carry shape descriptors into the authoritative store"
            );
        }

        #[test]
        fn retain_entries_prunes_shape_descriptors() {
            let mut store = NodeMetadataStore::new();
            let keep = NodeId::new(10, 1);
            let drop = NodeId::new(11, 1);
            store.insert_shape_descriptor(keep, ShapeDescriptor::default());
            store.insert_shape_descriptor(drop, ShapeDescriptor::default());
            store.retain_entries(|index, _generation| index == 10);
            assert!(store.shape_descriptor(keep).is_some());
            assert!(
                store.shape_descriptor(drop).is_none(),
                "retain must prune descriptors under the same predicate"
            );
        }

        #[test]
        fn partial_eq_is_sensitive_to_descriptor_loss() {
            let mut a = NodeMetadataStore::new();
            let mut b = NodeMetadataStore::new();
            a.insert_shape_descriptor(NodeId::new(4, 1), descriptor_with_branches(2));
            b.insert_shape_descriptor(NodeId::new(4, 1), descriptor_with_branches(9));
            assert_ne!(a, b, "equality must not be blind to descriptor differences");
            b.insert_shape_descriptor(NodeId::new(4, 1), descriptor_with_branches(2));
            assert_eq!(a, b);
        }

        #[test]
        fn heap_bytes_grows_with_descriptors() {
            let store_empty = NodeMetadataStore::new();
            let base = store_empty.heap_bytes();
            let mut store = NodeMetadataStore::new();
            store.insert_shape_descriptor(NodeId::new(5, 1), ShapeDescriptor::default());
            assert!(
                store.heap_bytes() > base,
                "a descriptor must increase the accounted heap size"
            );
        }

        #[test]
        fn set_and_read_shape_descriptors_envelope_slot() {
            let mut map: BTreeMap<NodeId, ShapeDescriptor> = BTreeMap::new();
            map.insert(NodeId::new(6, 1), descriptor_with_branches(3));
            let mut store = NodeMetadataStore::new();
            store.set_shape_descriptors(map);
            assert_eq!(store.shape_descriptors().len(), 1);
            assert_eq!(
                store
                    .shape_descriptor(NodeId::new(6, 1))
                    .map(|d| d.cf_histogram[CfBucket::Branch.index()]),
                Some(3)
            );
        }
    }

    // ------------------------------------------------------------------
    // U02_NODE_FLAGS: NodeFlags / StoredEntry / co-occurrence semantics
    // ------------------------------------------------------------------

    mod node_flags_tests {
        use super::*;

        #[test]
        fn node_flags_bit_composition() {
            // SYNTHETIC | ADDRESS_TAKEN coexist; setting one doesn't clear the other.
            let mut f = NodeFlags::EMPTY;
            assert!(f.is_empty());
            f.insert(NodeFlags::SYNTHETIC);
            assert!(f.contains(NodeFlags::SYNTHETIC));
            assert!(!f.contains(NodeFlags::ADDRESS_TAKEN));

            f.insert(NodeFlags::ADDRESS_TAKEN);
            assert!(
                f.contains(NodeFlags::SYNTHETIC),
                "ADDRESS_TAKEN insert must not clear SYNTHETIC"
            );
            assert!(f.contains(NodeFlags::ADDRESS_TAKEN));

            // contains(EMPTY) is trivially true.
            assert!(f.contains(NodeFlags::EMPTY));

            // remove preserves the other bit.
            f.remove(NodeFlags::SYNTHETIC);
            assert!(!f.contains(NodeFlags::SYNTHETIC));
            assert!(f.contains(NodeFlags::ADDRESS_TAKEN));

            // BitOr / BitOrAssign work for ergonomic construction.
            let combined = NodeFlags::SYNTHETIC | NodeFlags::CALLSITE_PROMISCUOUS;
            assert!(combined.contains(NodeFlags::SYNTHETIC));
            assert!(combined.contains(NodeFlags::CALLSITE_PROMISCUOUS));
            assert!(!combined.contains(NodeFlags::ADDRESS_TAKEN));

            let mut acc = NodeFlags::EMPTY;
            acc |= NodeFlags::ADDRESS_TAKEN;
            acc |= NodeFlags::CALLSITE_PROMISCUOUS;
            assert!(acc.contains(NodeFlags::ADDRESS_TAKEN | NodeFlags::CALLSITE_PROMISCUOUS));
        }

        #[test]
        fn stored_entry_flags_only_no_typed_payload() {
            // typed = None + flags = ADDRESS_TAKEN ONLY: pure marker entry.
            let entry = StoredEntry::with_flags(NodeFlags::ADDRESS_TAKEN);
            assert!(entry.typed.is_none());
            assert!(entry.flags.contains(NodeFlags::ADDRESS_TAKEN));
            assert!(!entry.flags.contains(NodeFlags::SYNTHETIC));
            assert!(!entry.is_vacant());

            // A truly vacant entry is detectable.
            assert!(StoredEntry::default().is_vacant());
        }

        #[test]
        fn co_occurrence_macro_and_address_taken() {
            // The design's whole point: TypedMetadata::Macro AND
            // NodeFlags::ADDRESS_TAKEN coexist in one entry. Both channels
            // return positively.
            let mut store = NodeMetadataStore::new();
            let node = NodeId::new(42, 1);

            store.insert(
                node,
                MacroNodeMetadata {
                    macro_generated: Some(true),
                    macro_source: Some("DEFINE_HANDLER".to_string()),
                    ..Default::default()
                },
            );
            store.mark_address_taken(node);

            // Typed channel: Macro payload preserved.
            let typed = store.get_typed(node).expect("typed entry present");
            assert!(matches!(typed, TypedMetadata::Macro(_)));
            let macro_meta = store.get_macro(node).expect("macro payload present");
            assert_eq!(macro_meta.macro_source.as_deref(), Some("DEFINE_HANDLER"));

            // Flag channel: ADDRESS_TAKEN set.
            assert!(store.is_address_taken(node));
            assert!(!store.is_synthetic(node));
            assert!(!store.is_callsite_promiscuous(node));

            // Adding another flag must NOT disturb the typed payload.
            store.mark_synthetic(node);
            assert!(store.is_synthetic(node));
            assert!(store.is_address_taken(node));
            assert!(
                store.get_macro(node).is_some(),
                "mark_synthetic must not clobber Macro payload"
            );
        }

        #[test]
        fn synthetic_via_flag_not_typed() {
            // After mark_synthetic, get_typed == None AND is_synthetic == true
            // (no typed payload was created — the marker lives in flags alone).
            let mut store = NodeMetadataStore::new();
            let node = NodeId::new(7, 1);

            assert!(!store.is_synthetic(node), "missing entry must report false");

            store.mark_synthetic(node);
            assert!(store.is_synthetic(node));
            assert!(
                store.get_typed(node).is_none(),
                "mark_synthetic must not populate the typed slot"
            );
            assert!(store.get_macro(node).is_none());

            // Stale generation must NOT see the synthetic flag.
            let stale = NodeId::new(7, 2);
            assert!(!store.is_synthetic(stale));
        }

        #[test]
        fn mark_address_taken_preserves_typed_payload() {
            // Co-occurrence regression test: mark_address_taken on a node with
            // existing Macro payload must NOT replace or drop the payload.
            let mut store = NodeMetadataStore::new();
            let node = NodeId::new(101, 3);
            let macro_meta = MacroNodeMetadata {
                macro_generated: Some(true),
                macro_source: Some("foo_macro".to_string()),
                cfg_condition: Some("feature = \"x\"".to_string()),
                ..Default::default()
            };
            store.insert(node, macro_meta.clone());

            store.mark_address_taken(node);

            assert!(store.is_address_taken(node));
            let retrieved = store.get_macro(node).expect("Macro payload preserved");
            assert_eq!(retrieved, &macro_meta);
        }

        #[test]
        fn mark_callsite_promiscuous_independent_of_typed() {
            // A callsite-promiscuous node may have no typed payload AND may
            // have any other flag set independently.
            let mut store = NodeMetadataStore::new();
            let node = NodeId::new(202, 0);

            store.mark_callsite_promiscuous(node);
            assert!(store.is_callsite_promiscuous(node));
            assert!(!store.is_synthetic(node));
            assert!(!store.is_address_taken(node));
            assert!(store.get_typed(node).is_none());

            // Add SYNTHETIC: must compose, not clobber.
            store.mark_synthetic(node);
            assert!(store.is_callsite_promiscuous(node));
            assert!(store.is_synthetic(node));
        }

        #[test]
        fn import_classification_flags_distinct_bits_no_collision() {
            // IMPORT_STDLIB (1<<3) and IMPORT_RELATIVE (1<<4) occupy fresh bits
            // that do not overlap each other or the pre-existing trio.
            assert_eq!(NodeFlags::IMPORT_STDLIB.bits(), 1 << 3);
            assert_eq!(NodeFlags::IMPORT_RELATIVE.bits(), 1 << 4);

            let existing =
                NodeFlags::SYNTHETIC | NodeFlags::ADDRESS_TAKEN | NodeFlags::CALLSITE_PROMISCUOUS;
            assert!(!existing.contains(NodeFlags::IMPORT_STDLIB));
            assert!(!existing.contains(NodeFlags::IMPORT_RELATIVE));
            assert!(!NodeFlags::IMPORT_STDLIB.contains(NodeFlags::IMPORT_RELATIVE));
            assert!(!NodeFlags::IMPORT_RELATIVE.contains(NodeFlags::IMPORT_STDLIB));
        }

        #[test]
        fn mark_import_flags_roundtrip_via_set_flag() {
            let mut store = NodeMetadataStore::new();
            let stdlib_node = NodeId::new(303, 0);
            let relative_node = NodeId::new(404, 0);

            // Missing entries report false before any marking.
            assert!(!store.is_import_stdlib(stdlib_node));
            assert!(!store.is_import_relative(stdlib_node));

            store.mark_import_stdlib(stdlib_node);
            store.mark_import_relative(relative_node);

            // Each helper sets exactly its own bit, nothing else.
            assert!(store.is_import_stdlib(stdlib_node));
            assert!(!store.is_import_relative(stdlib_node));
            assert!(!store.is_synthetic(stdlib_node));
            assert!(!store.is_address_taken(stdlib_node));
            assert!(!store.is_callsite_promiscuous(stdlib_node));

            assert!(store.is_import_relative(relative_node));
            assert!(!store.is_import_stdlib(relative_node));

            // Marker-only entries carry no typed payload.
            assert!(store.get_typed(stdlib_node).is_none());
            assert!(store.get_typed(relative_node).is_none());

            // Both classifications may co-occur on one node without clobber.
            store.mark_import_relative(stdlib_node);
            assert!(store.is_import_stdlib(stdlib_node));
            assert!(store.is_import_relative(stdlib_node));

            // Stale generation must NOT observe the flags.
            let stale = NodeId::new(303, 1);
            assert!(!store.is_import_stdlib(stale));
            assert!(!store.is_import_relative(stale));
        }

        #[test]
        fn get_flags_returns_empty_for_missing_entry() {
            let store = NodeMetadataStore::new();
            let missing = NodeId::new(999, 0);
            assert!(store.get_flags(missing).is_empty());
            assert!(!store.is_synthetic(missing));
            assert!(!store.is_address_taken(missing));
            assert!(!store.is_callsite_promiscuous(missing));
        }

        #[test]
        fn get_macro_roundtrips_typed_macro_storage() {
            // get_macro returns Some for a TypedMetadata::Macro entry created
            // via either `insert` (convenience) or `insert_typed` (raw).
            let mut store = NodeMetadataStore::new();
            let node_a = NodeId::new(1, 0);
            let node_b = NodeId::new(2, 0);
            let payload_a = MacroNodeMetadata {
                cfg_condition: Some("a".to_string()),
                ..Default::default()
            };
            let payload_b = MacroNodeMetadata {
                cfg_condition: Some("b".to_string()),
                ..Default::default()
            };

            store.insert(node_a, payload_a.clone());
            store.insert_typed(node_b, TypedMetadata::Macro(payload_b.clone()));

            assert_eq!(store.get_macro(node_a), Some(&payload_a));
            assert_eq!(store.get_macro(node_b), Some(&payload_b));
        }

        #[test]
        fn serialize_deserialize_preserves_typed_and_flags() {
            // Build a store mixing every shape: Macro+flags, Classpath alone,
            // flags-only (single + multi), and a stale-generation key.
            let mut store = NodeMetadataStore::new();

            store.insert(
                NodeId::new(1, 0),
                MacroNodeMetadata {
                    macro_generated: Some(true),
                    ..Default::default()
                },
            );
            store.mark_address_taken(NodeId::new(1, 0));

            store.insert_typed(
                NodeId::new(2, 0),
                TypedMetadata::Classpath(ClasspathNodeMetadata {
                    coordinates: None,
                    jar_path: "x.jar".to_string(),
                    fqn: "com.example.X".to_string(),
                    is_direct_dependency: true,
                }),
            );

            store.mark_synthetic(NodeId::new(3, 0));
            store.mark_address_taken(NodeId::new(4, 9));
            store.mark_callsite_promiscuous(NodeId::new(4, 9));

            let bytes = postcard::to_allocvec(&store).expect("serialize");
            let decoded: NodeMetadataStore = postcard::from_bytes(&bytes).expect("deserialize");

            assert_eq!(store, decoded);

            // Spot-check that both channels round-tripped, not just equality.
            assert!(decoded.get_macro(NodeId::new(1, 0)).is_some());
            assert!(decoded.is_address_taken(NodeId::new(1, 0)));
            assert!(matches!(
                decoded.get_typed(NodeId::new(2, 0)),
                Some(TypedMetadata::Classpath(_))
            ));
            assert!(decoded.is_synthetic(NodeId::new(3, 0)));
            assert!(decoded.is_address_taken(NodeId::new(4, 9)));
            assert!(decoded.is_callsite_promiscuous(NodeId::new(4, 9)));
        }

        #[test]
        fn json_serialize_deserialize_preserves_typed_and_flags() {
            // serde_json path used by MCP export. Same shape as the postcard
            // round-trip — the Serialize/Deserialize impls are wire-format
            // agnostic.
            let mut store = NodeMetadataStore::new();
            store.insert(
                NodeId::new(5, 5),
                MacroNodeMetadata {
                    macro_generated: Some(true),
                    ..Default::default()
                },
            );
            store.mark_address_taken(NodeId::new(5, 5));
            store.mark_synthetic(NodeId::new(9, 0));

            let json = serde_json::to_string(&store).expect("json serialize");
            let decoded: NodeMetadataStore = serde_json::from_str(&json).expect("json deserialize");
            assert_eq!(store, decoded);
        }

        #[test]
        fn insert_entry_bulk_remap_path() {
            // The classpath emitter rebuilds the store from `iter_entries()`
            // + `insert_entry()` during id remapping; that path must preserve
            // both channels.
            let mut original = NodeMetadataStore::new();
            original.insert_typed(
                NodeId::new(10, 0),
                TypedMetadata::Classpath(ClasspathNodeMetadata {
                    coordinates: Some("g:a:1".to_string()),
                    jar_path: "j.jar".to_string(),
                    fqn: "F".to_string(),
                    is_direct_dependency: true,
                }),
            );
            original.mark_address_taken(NodeId::new(10, 0));
            original.mark_synthetic(NodeId::new(11, 0));

            // Simulate the emitter's remap: copy via iter_entries.
            let mut remapped = NodeMetadataStore::new();
            for (key, entry) in original.iter_entries() {
                let nid = NodeId::new(key.0, key.1);
                remapped.insert_entry(nid, entry.clone());
            }

            assert_eq!(original, remapped);
            assert!(remapped.is_address_taken(NodeId::new(10, 0)));
            assert!(matches!(
                remapped.get_typed(NodeId::new(10, 0)),
                Some(TypedMetadata::Classpath(_))
            ));
            assert!(remapped.is_synthetic(NodeId::new(11, 0)));
        }
    }
}