formualizer_eval/engine/

delta_edges.rs

use super::addr::VertexAddr;
use super::csr_edges::CsrEdges;
use super::vertex::VertexId;
use rustc_hash::{FxHashMap, FxHashSet};

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

    fn grid(row: u32, col: u32) -> VertexAddr {
        VertexAddr::grid(GridAddr::new(row, col))
    }

    use super::*;

    #[test]
    fn test_delta_slab_add_edge() {
        let csr = CsrEdges::from_adjacency(
            vec![(0u32, vec![1u32])],
            &[grid(0, 0), grid(0, 1), grid(0, 2)],
        );
        let mut delta = DeltaEdgeSlab::new();

        delta.add_edge(VertexId(0), VertexId(2));

        let merged = delta.merged_view(&csr, VertexId(0));
        assert_eq!(merged, vec![VertexId(1), VertexId(2)]);
    }

    #[test]
    fn test_delta_slab_remove_edge() {
        let csr = CsrEdges::from_adjacency(
            vec![(0u32, vec![1u32, 2u32, 3u32])],
            &[grid(0, 0), grid(0, 1), grid(0, 2), grid(0, 3)],
        );
        let mut delta = DeltaEdgeSlab::new();

        delta.remove_edge(VertexId(0), VertexId(2));

        let merged = delta.merged_view(&csr, VertexId(0));
        assert_eq!(merged, vec![VertexId(1), VertexId(3)]);
    }

    #[test]
    fn test_delta_slab_rebuild_threshold() {
        let mut edges = CsrMutableEdges::new();

        // Add 1000 edges through delta slab
        for i in 0..1000 {
            edges.add_edge(VertexId(i), VertexId(i + 1));
        }

        // Should trigger rebuild at threshold
        assert!(edges.delta_size() < 100); // Delta cleared after rebuild
    }

    #[test]
    fn test_delta_slab_multiple_operations() {
        let csr = CsrEdges::from_adjacency(
            vec![(0u32, vec![1u32, 2u32]), (1u32, vec![3u32])],
            &[grid(0, 0), grid(0, 1), grid(0, 2), grid(1, 0)],
        );
        let mut delta = DeltaEdgeSlab::new();

        // Multiple operations on same vertex
        delta.add_edge(VertexId(0), VertexId(3));
        delta.remove_edge(VertexId(0), VertexId(1));
        delta.add_edge(VertexId(0), VertexId(4));

        let merged = delta.merged_view(&csr, VertexId(0));
        assert_eq!(merged, vec![VertexId(2), VertexId(3), VertexId(4)]);
    }

    #[test]
    fn test_mutable_edges_exact_edge_count_includes_delta() {
        let mut edges = CsrMutableEdges::with_coords(vec![grid(0, 0), grid(0, 1), grid(0, 2)]);
        edges.add_edge(VertexId(0), VertexId(1));
        edges.add_edge(VertexId(0), VertexId(2));
        assert_eq!(edges.num_edges_exact(), 2);

        edges.remove_edge(VertexId(0), VertexId(1));
        assert_eq!(edges.num_edges_exact(), 1);
    }

    #[test]
    fn test_delta_slab_empty_base() {
        let csr = CsrEdges::empty();
        let mut delta = DeltaEdgeSlab::new();

        delta.add_edge(VertexId(0), VertexId(1));
        delta.add_edge(VertexId(0), VertexId(2));

        let merged = delta.merged_view(&csr, VertexId(0));
        assert_eq!(merged, vec![VertexId(1), VertexId(2)]);
    }

    #[test]
    fn test_delta_slab_remove_nonexistent() {
        let csr = CsrEdges::from_adjacency(vec![(0u32, vec![1u32])], &[grid(0, 0), grid(0, 1)]);
        let mut delta = DeltaEdgeSlab::new();

        // Remove edge that doesn't exist
        delta.remove_edge(VertexId(0), VertexId(2));

        let merged = delta.merged_view(&csr, VertexId(0));
        assert_eq!(merged, vec![VertexId(1)]); // No change
    }

    #[test]
    fn test_delta_slab_apply_to_csr() {
        let csr = CsrEdges::from_adjacency(
            vec![(0u32, vec![1u32]), (1u32, vec![2u32]), (2u32, vec![])],
            &[grid(0, 0), grid(0, 1), grid(1, 0)],
        );

        let mut delta = DeltaEdgeSlab::new();
        delta.add_edge(VertexId(0), VertexId(2));
        delta.remove_edge(VertexId(1), VertexId(2));
        delta.add_edge(VertexId(2), VertexId(0));

        // Apply delta and get new CSR
        let coords = vec![grid(0, 0), grid(0, 1), grid(1, 0)];
        let vertex_ids = vec![0u32, 1u32, 2u32];
        let new_csr = delta.apply_to_csr(&csr, &coords, &vertex_ids);

        assert_eq!(new_csr.out_edges(VertexId(0)), &[VertexId(1), VertexId(2)]);
        assert_eq!(new_csr.out_edges(VertexId(1)), &[]);
        assert_eq!(new_csr.out_edges(VertexId(2)), &[VertexId(0)]);
    }

    #[test]
    fn test_mutable_edges_auto_rebuild() {
        let mut edges = CsrMutableEdges::with_coords(vec![grid(0, 0), grid(0, 1), grid(1, 0)]);

        // Add initial edges
        edges.add_edge(VertexId(0), VertexId(1));
        edges.add_edge(VertexId(1), VertexId(2));

        // Perform many operations to trigger rebuild
        for _ in 0..500 {
            edges.add_edge(VertexId(2), VertexId(0));
            edges.remove_edge(VertexId(2), VertexId(0));
        }

        // Check that rebuild happened (delta is small)
        assert!(edges.delta_size() < 50);

        // Verify edges are still correct
        assert_eq!(edges.out_edges(VertexId(0)), vec![VertexId(1)]);
        assert_eq!(edges.out_edges(VertexId(1)), vec![VertexId(2)]);
    }

    #[test]
    fn test_mutable_edges_with_offset_vertex_ids() {
        use crate::engine::vertex_store::FIRST_NORMAL_VERTEX;

        let mut edges = CsrMutableEdges::new();

        // Add vertices with IDs starting at FIRST_NORMAL_VERTEX (1024)
        let base_id = FIRST_NORMAL_VERTEX;
        edges.add_vertex(grid(0, 0), base_id);
        edges.add_vertex(grid(0, 1), base_id + 1);
        edges.add_vertex(grid(1, 0), base_id + 2);

        // Add edges using offset IDs
        edges.add_edge(VertexId(base_id), VertexId(base_id + 1));
        edges.add_edge(VertexId(base_id + 1), VertexId(base_id + 2));
        edges.add_edge(VertexId(base_id + 2), VertexId(base_id));

        // Verify edges work correctly
        assert_eq!(
            edges.out_edges(VertexId(base_id)),
            vec![VertexId(base_id + 1)]
        );
        assert_eq!(
            edges.out_edges(VertexId(base_id + 1)),
            vec![VertexId(base_id + 2)]
        );
        assert_eq!(
            edges.out_edges(VertexId(base_id + 2)),
            vec![VertexId(base_id)]
        );

        // Force rebuild and verify again
        edges.rebuild();
        assert_eq!(
            edges.out_edges(VertexId(base_id)),
            vec![VertexId(base_id + 1)]
        );
        assert_eq!(
            edges.out_edges(VertexId(base_id + 1)),
            vec![VertexId(base_id + 2)]
        );
        assert_eq!(
            edges.out_edges(VertexId(base_id + 2)),
            vec![VertexId(base_id)]
        );
    }

    #[test]
    fn test_csr_coord_update() {
        let mut edges = CsrMutableEdges::new();

        edges.add_vertex(grid(1, 1), 1024);
        edges.add_vertex(grid(2, 2), 1025);
        edges.add_edge(VertexId(1024), VertexId(1025));

        // Update coordinate
        edges.update_addr(VertexId(1024), grid(5, 5));

        // Verify sorting remains correct after rebuild
        edges.rebuild();
        let out = edges.out_edges(VertexId(1024));
        assert_eq!(out, vec![VertexId(1025)]);
    }

    #[test]
    fn update_coord_uses_vertex_position_index() {
        let mut edges = CsrMutableEdges::new();
        let items: Vec<_> = (0..20_000u32).map(|id| (grid(id, id % 17), id)).collect();
        edges.add_vertices_batch(&items);

        let started = std::time::Instant::now();
        for id in 15_000..20_000u32 {
            edges.update_addr(VertexId(id), grid(id + 1, (id + 2) % 100));
        }
        let elapsed = started.elapsed();

        if !cfg!(debug_assertions) {
            assert!(
                elapsed < std::time::Duration::from_millis(50),
                "update_coord took {elapsed:?}"
            );
        }

        for id in 15_000..20_000u32 {
            let pos = edges.vertex_pos[&id];
            assert_eq!(edges.vertex_ids[pos], id);
            assert_eq!(edges.coords[pos], grid(id + 1, (id + 2) % 100));
        }
    }

    #[test]
    fn test_last_op_wins_add_then_remove() {
        let csr = CsrEdges::from_adjacency(vec![(0u32, vec![])], &[grid(0, 0)]);
        let mut delta = DeltaEdgeSlab::new();
        delta.add_edge(VertexId(0), VertexId(1));
        delta.remove_edge(VertexId(0), VertexId(1));
        let merged = delta.merged_view(&csr, VertexId(0));
        assert_eq!(merged, vec![]);
    }

    #[test]
    fn test_last_op_wins_remove_then_add() {
        let csr = CsrEdges::from_adjacency(vec![(0u32, vec![])], &[grid(0, 0)]);
        let mut delta = DeltaEdgeSlab::new();
        delta.remove_edge(VertexId(0), VertexId(1));
        delta.add_edge(VertexId(0), VertexId(1));
        let merged = delta.merged_view(&csr, VertexId(0));
        assert_eq!(merged, vec![VertexId(1)]);
    }

    #[test]
    fn test_dedup_additions_and_sorted() {
        let csr = CsrEdges::from_adjacency(
            vec![(0u32, vec![2u32])],
            &[grid(0, 0), grid(0, 1), grid(0, 2)],
        );
        let mut delta = DeltaEdgeSlab::new();
        // Add duplicates and out-of-order ids
        delta.add_edge(VertexId(0), VertexId(1));
        delta.add_edge(VertexId(0), VertexId(3));
        delta.add_edge(VertexId(0), VertexId(1)); // duplicate
        let merged = delta.merged_view(&csr, VertexId(0));
        // Should be deduped and sorted by VertexId
        assert_eq!(merged, vec![VertexId(1), VertexId(2), VertexId(3)]);
    }

    #[test]
    fn test_merged_in_view_add_and_remove() {
        let csr = CsrEdges::from_adjacency(
            vec![(0u32, vec![2u32]), (1u32, vec![2u32])],
            &[grid(0, 0), grid(0, 1), grid(0, 2), grid(0, 3)],
        );
        let mut delta = DeltaEdgeSlab::new();

        // New incoming edge 3 -> 2, removed incoming edge 0 -> 2.
        delta.add_edge(VertexId(3), VertexId(2));
        delta.remove_edge(VertexId(0), VertexId(2));

        let merged = delta.merged_in_view(&csr, VertexId(2));
        assert_eq!(merged, vec![VertexId(1), VertexId(3)]);
    }

    #[test]
    fn test_merged_in_view_last_op_wins() {
        let csr = CsrEdges::from_adjacency(vec![(0u32, vec![1u32])], &[grid(0, 0), grid(0, 1)]);
        let mut delta = DeltaEdgeSlab::new();

        delta.remove_edge(VertexId(0), VertexId(1));
        delta.add_edge(VertexId(0), VertexId(1));
        assert_eq!(delta.merged_in_view(&csr, VertexId(1)), vec![VertexId(0)]);

        delta.add_edge(VertexId(2), VertexId(1));
        delta.remove_edge(VertexId(2), VertexId(1));
        assert_eq!(delta.merged_in_view(&csr, VertexId(1)), vec![VertexId(0)]);
    }

    #[test]
    fn test_end_batch_below_threshold_defers_rebuild() {
        let mut edges = CsrMutableEdges::with_coords(vec![grid(0, 0), grid(0, 1), grid(0, 2)]);
        let before = edges.rebuild_count();

        edges.begin_batch();
        edges.add_edge(VertexId(0), VertexId(1));
        edges.add_edge(VertexId(0), VertexId(2));
        edges.end_batch();

        // Small edit bursts must not trigger a full rebuild (#125)...
        assert_eq!(edges.rebuild_count(), before);
        // ...while reads stay correct through the merged views.
        assert_eq!(edges.out_edges(VertexId(0)), vec![VertexId(1), VertexId(2)]);
        assert_eq!(edges.in_edges_merged(VertexId(1)), vec![VertexId(0)]);
        assert_eq!(edges.in_edges_merged(VertexId(2)), vec![VertexId(0)]);
    }

    #[test]
    fn test_end_batch_rebuilds_at_threshold() {
        let coords: Vec<VertexAddr> = (0..1200u32).map(|i| grid(i, 0)).collect();
        let mut edges = CsrMutableEdges::with_coords(coords);
        let before = edges.rebuild_count();

        edges.begin_batch();
        for i in 0..1100u32 {
            edges.add_edge(VertexId(i), VertexId(i + 1));
        }
        edges.end_batch();

        assert_eq!(edges.rebuild_count(), before + 1);
        assert_eq!(edges.delta_size(), 0);
        assert_eq!(edges.out_edges(VertexId(0)), vec![VertexId(1)]);
        assert_eq!(edges.in_edges(VertexId(1)), &[VertexId(0)]);
    }

    #[test]
    fn test_add_vertex_defers_rebuild() {
        let mut edges = CsrMutableEdges::new();
        edges.add_vertex(grid(0, 0), 1024);
        edges.add_vertex(grid(0, 1), 1025);
        assert_eq!(edges.rebuild_count(), 0);

        edges.add_edge(VertexId(1024), VertexId(1025));
        // Reads see the new vertex/edge before any rebuild.
        assert_eq!(edges.out_edges(VertexId(1024)), vec![VertexId(1025)]);
        assert_eq!(edges.in_edges_merged(VertexId(1025)), vec![VertexId(1024)]);

        // And the next rebuild folds the vertex into the CSR base.
        edges.rebuild();
        assert_eq!(edges.out_edges(VertexId(1024)), vec![VertexId(1025)]);
        assert_eq!(edges.in_edges(VertexId(1025)), &[VertexId(1024)]);
    }

    #[test]
    fn test_end_batch_rebuilds_on_coord_change_only() {
        let mut edges = CsrMutableEdges::with_coords(vec![grid(0, 0), grid(0, 1)]);
        edges.begin_batch();
        edges.update_addr(VertexId(0), grid(0, 2));
        // No ops, only coord changed; end_batch should rebuild due to coord_dirty
        edges.end_batch();
        // Smoke: out_edges call should not panic and reflect empty edges
        assert_eq!(edges.out_edges(VertexId(0)), Vec::<VertexId>::new());
    }
}

/// Delta slab for accumulating edge mutations between CSR rebuilds
///
/// Provides O(1) edge mutations by tracking additions and removals
/// separately, merging them with the base CSR on read.
#[derive(Debug)]
pub struct DeltaEdgeSlab {
    /// New edges to add, grouped by source vertex (set semantics to avoid duplicates)
    additions: FxHashMap<VertexId, FxHashSet<VertexId>>,

    /// Edges to remove, stored as sets for O(1) lookup
    removals: FxHashMap<VertexId, FxHashSet<VertexId>>,

    /// Reverse index of `additions`: target -> sources. Keeps incoming-edge
    /// reads delta-aware in O(degree) instead of O(V) scans (#125).
    additions_in: FxHashMap<VertexId, FxHashSet<VertexId>>,

    /// Reverse index of `removals`: target -> sources.
    removals_in: FxHashMap<VertexId, FxHashSet<VertexId>>,

    /// Total operation count for rebuild threshold
    op_count: usize,

    /// Flag indicating coordinates have changed and rebuild is needed
    coord_changed: bool,
}

impl DeltaEdgeSlab {
    /// Create a new empty delta slab
    pub fn new() -> Self {
        Self {
            additions: FxHashMap::default(),
            removals: FxHashMap::default(),
            additions_in: FxHashMap::default(),
            removals_in: FxHashMap::default(),
            op_count: 0,
            coord_changed: false,
        }
    }

    fn reserve_additions(&mut self, additional: usize) {
        self.additions.reserve(additional);
        self.additions_in.reserve(additional);
    }

    /// Add an edge from source to target
    pub fn add_edge(&mut self, from: VertexId, to: VertexId) {
        // Last-op-wins: if previously removed, cancel the removal
        if let Some(rem) = self.removals.get_mut(&from) {
            rem.remove(&to);
        }
        if let Some(rem) = self.removals_in.get_mut(&to) {
            rem.remove(&from);
        }
        // Insert into additions set (forward and reverse)
        self.additions.entry(from).or_default().insert(to);
        self.additions_in.entry(to).or_default().insert(from);
        self.op_count += 1;
    }

    /// Remove an edge from source to target
    pub fn remove_edge(&mut self, from: VertexId, to: VertexId) {
        // Last-op-wins: if previously added in this slab, cancel the addition
        if let Some(adds) = self.additions.get_mut(&from) {
            adds.remove(&to);
        }
        if let Some(adds) = self.additions_in.get_mut(&to) {
            adds.remove(&from);
        }
        // Record removal (forward and reverse)
        self.removals.entry(from).or_default().insert(to);
        self.removals_in.entry(to).or_default().insert(from);
        self.op_count += 1;
    }

    /// Get a merged view of edges for a vertex, combining CSR and delta
    pub fn merged_view(&self, csr: &CsrEdges, v: VertexId) -> Vec<VertexId> {
        // Start from base CSR out-edges
        let mut result: Vec<_> = csr.out_edges(v).to_vec();
        // Remove edges marked for deletion
        if let Some(removes) = self.removals.get(&v) {
            result.retain(|e| !removes.contains(e));
        }
        // Add new edges (set semantics)
        if let Some(adds) = self.additions.get(&v) {
            result.extend(adds.iter().copied());
        }
        // Dedup deterministically and sort by VertexId for stable order
        let mut seen: FxHashSet<VertexId> = FxHashSet::default();
        result.retain(|e| seen.insert(*e));
        result.sort_by_key(|e| e.0);
        result
    }

    /// Get a merged view of *incoming* edges for a vertex, combining the CSR
    /// reverse edges with the delta's reverse index. O(in-degree), not O(V).
    pub fn merged_in_view(&self, csr: &CsrEdges, v: VertexId) -> Vec<VertexId> {
        let mut result: Vec<_> = csr.in_edges(v).to_vec();
        if let Some(removes) = self.removals_in.get(&v) {
            result.retain(|e| !removes.contains(e));
        }
        if let Some(adds) = self.additions_in.get(&v) {
            result.extend(adds.iter().copied());
        }
        let mut seen: FxHashSet<VertexId> = FxHashSet::default();
        result.retain(|e| seen.insert(*e));
        result.sort_by_key(|e| e.0);
        result
    }

    /// Check if the delta needs to be applied (rebuild threshold reached)
    pub fn needs_rebuild(&self) -> bool {
        self.op_count >= 1000 || self.coord_changed
    }

    /// Mark that coordinates have changed and rebuild is needed
    pub fn mark_dirty(&mut self) {
        self.coord_changed = true;
    }

    /// Get the current operation count
    pub fn op_count(&self) -> usize {
        self.op_count
    }

    /// Clear the delta slab
    pub fn clear(&mut self) {
        self.additions.clear();
        self.removals.clear();
        self.additions_in.clear();
        self.removals_in.clear();
        self.op_count = 0;
        self.coord_changed = false;
    }

    /// Iterate over all (from, &additions) pairs in the delta. Used by
    /// build_from_adjacency to carry forward delta-only edges that the
    /// adjacency input does not cover.
    pub fn additions_iter(&self) -> impl Iterator<Item = (&VertexId, &FxHashSet<VertexId>)> {
        self.additions.iter()
    }

    /// Return removals scheduled for `from` (empty slice if none).
    pub fn removals_for(&self, from: VertexId) -> impl Iterator<Item = VertexId> + '_ {
        self.removals
            .get(&from)
            .into_iter()
            .flat_map(|set| set.iter().copied())
    }

    /// Apply delta to CSR, creating a new CSR structure
    pub fn apply_to_csr(
        &self,
        base: &CsrEdges,
        coords: &[VertexAddr],
        vertex_ids: &[u32],
    ) -> CsrEdges {
        let mut adjacency = Vec::with_capacity(vertex_ids.len());

        // Build new adjacency list by merging base and delta
        for &vid in vertex_ids {
            let v = VertexId(vid);
            let merged = self.merged_view(base, v);

            // Convert to u32 for adjacency format
            let targets: Vec<u32> = merged.into_iter().map(|id| id.0).collect();

            adjacency.push((vid, targets));
        }

        CsrEdges::from_adjacency(adjacency, coords)
    }
}

impl Default for DeltaEdgeSlab {
    fn default() -> Self {
        Self::new()
    }
}

/// Mutable edge storage combining CSR base with delta slab
///
/// Provides efficient edge mutations with automatic rebuild when
/// delta grows too large.
#[derive(Debug)]
pub struct CsrMutableEdges {
    /// Base CSR structure (immutable between rebuilds)
    base: CsrEdges,

    /// Delta slab for mutations
    delta: DeltaEdgeSlab,

    /// Vertex addresses (grid position or symbol identity) for deterministic ordering
    coords: Vec<VertexAddr>,

    /// Vertex IDs corresponding to coords array
    vertex_ids: Vec<u32>,

    /// Position of each vertex id in the coords and vertex_ids arrays.
    vertex_pos: FxHashMap<u32, usize>,

    /// Nested batch depth; non-zero defers automatic rebuilds.
    batch_depth: usize,

    /// Number of full CSR rebuilds performed (observability / regression tests).
    rebuild_count: u64,
}

impl CsrMutableEdges {
    /// Create new mutable edges with empty base
    pub fn new() -> Self {
        Self {
            base: CsrEdges::empty(),
            delta: DeltaEdgeSlab::new(),
            coords: Vec::new(),
            vertex_ids: Vec::new(),
            vertex_pos: FxHashMap::default(),
            batch_depth: 0,
            rebuild_count: 0,
        }
    }

    /// Create with initial vertex coordinates
    pub fn with_coords(coords: Vec<VertexAddr>) -> Self {
        let num_vertices = coords.len();
        let vertex_ids: Vec<u32> = (0..num_vertices as u32).collect();
        let adjacency: Vec<_> = vertex_ids.iter().map(|&id| (id, Vec::new())).collect();
        let vertex_pos = vertex_ids
            .iter()
            .enumerate()
            .map(|(idx, &id)| (id, idx))
            .collect();

        Self {
            base: CsrEdges::from_adjacency(adjacency, &coords),
            delta: DeltaEdgeSlab::new(),
            coords,
            vertex_ids,
            vertex_pos,
            batch_depth: 0,
            rebuild_count: 0,
        }
    }

    pub(crate) fn reserve_prepared_additions(&mut self, vertices: usize, edges: usize) {
        self.coords.reserve(vertices);
        self.vertex_ids.reserve(vertices);
        self.vertex_pos.reserve(vertices);
        self.delta.reserve_additions(edges);
    }

    /// Add an edge, rebuilding if threshold reached
    pub fn add_edge(&mut self, from: VertexId, to: VertexId) {
        self.delta.add_edge(from, to);
        self.maybe_rebuild();
    }

    /// Remove an edge, rebuilding if threshold reached
    pub fn remove_edge(&mut self, from: VertexId, to: VertexId) {
        self.delta.remove_edge(from, to);
        self.maybe_rebuild();
    }

    /// Get outgoing edges for a vertex (merged view)
    pub fn out_edges(&self, v: VertexId) -> Vec<VertexId> {
        if self.delta.op_count() == 0 {
            self.base.out_edges(v).to_vec()
        } else {
            self.delta.merged_view(&self.base, v)
        }
    }

    /// Borrow outgoing edges when no delta mutations are pending.
    ///
    /// This is a zero-allocation hot path for read-heavy evaluation/scheduling phases.
    #[inline]
    pub fn out_edges_ref(&self, v: VertexId) -> Option<&[VertexId]> {
        if self.delta.op_count() == 0 {
            Some(self.base.out_edges(v))
        } else {
            None
        }
    }

    /// Get incoming edges from base CSR (delta not applied for performance)
    /// After rebuild, this will include all changes
    pub fn in_edges(&self, v: VertexId) -> &[VertexId] {
        self.base.in_edges(v)
    }

    /// Get incoming edges for a vertex with pending delta mutations applied.
    ///
    /// O(in-degree + pending delta entries for `v`); never scans all vertices.
    pub fn in_edges_merged(&self, v: VertexId) -> Vec<VertexId> {
        if self.delta.op_count() == 0 {
            self.base.in_edges(v).to_vec()
        } else {
            self.delta.merged_in_view(&self.base, v)
        }
    }

    /// Borrow incoming edges when no delta mutations are pending.
    ///
    /// This is a zero-allocation hot path for read-heavy evaluation/scheduling phases.
    #[inline]
    pub fn in_edges_ref(&self, v: VertexId) -> Option<&[VertexId]> {
        if self.delta.op_count() == 0 {
            Some(self.base.in_edges(v))
        } else {
            None
        }
    }

    /// Visit incoming edges without materializing the complete in-degree.
    ///
    /// The caller-owned budget is charged once for every base or delta entry
    /// examined. `false` means an entry remained when the budget was
    /// exhausted. Pending removals are filtered and pending additions are read
    /// through the reverse delta index, so this has the same delta-aware
    /// semantics as [`Self::in_edges_merged`]. A duplicate base/addition pair
    /// may be presented twice; bounded semantic callers already deduplicate by
    /// address and charging the duplicate is the conservative accounting rule.
    pub(crate) fn visit_in_edges_bounded(
        &self,
        v: VertexId,
        remaining_work: &mut u64,
        visitor: &mut dyn FnMut(VertexId) -> bool,
    ) -> bool {
        let removals = self.delta.removals_in.get(&v);
        for &source in self.base.in_edges(v) {
            if *remaining_work == 0 {
                return false;
            }
            *remaining_work -= 1;
            if removals.is_some_and(|set| set.contains(&source)) {
                continue;
            }
            if !visitor(source) {
                return false;
            }
        }
        if let Some(additions) = self.delta.additions_in.get(&v) {
            for &source in additions {
                if *remaining_work == 0 {
                    return false;
                }
                *remaining_work -= 1;
                if !visitor(source) {
                    return false;
                }
            }
        }
        true
    }

    /// Get the current delta size
    pub fn delta_size(&self) -> usize {
        self.delta.op_count()
    }

    /// Return the exact number of logical outgoing dependency edges, including pending delta
    /// mutations.
    ///
    /// This is intended for read-only observability. When the delta slab is non-empty, the
    /// implementation walks the known vertex ids and merges each outgoing edge list, so callers
    /// should avoid putting it on hot evaluation paths.
    pub fn num_edges_exact(&self) -> usize {
        if self.delta.op_count() == 0 {
            return self.base.num_edges();
        }

        self.vertex_ids
            .iter()
            .map(|&id| self.out_edges(VertexId(id)).len())
            .sum()
    }

    /// Force a rebuild of the CSR structure
    pub fn rebuild(&mut self) {
        if self.delta.op_count() > 0 || self.delta.needs_rebuild() {
            self.base = self
                .delta
                .apply_to_csr(&self.base, &self.coords, &self.vertex_ids);
            self.delta.clear();
            self.rebuild_count += 1;
        }
    }

    /// Number of full CSR rebuilds performed so far.
    ///
    /// Per-edit dependency updates must amortize rebuilds (#125); regression
    /// tests assert on this counter instead of wall-clock time.
    pub fn rebuild_count(&self) -> u64 {
        self.rebuild_count
    }

    /// Check and perform rebuild if threshold reached
    fn maybe_rebuild(&mut self) {
        if self.batch_depth == 0 && self.delta.needs_rebuild() {
            self.rebuild();
        }
    }

    /// Enter batch mode - defer rebuilds until the outer end_batch() call.
    pub fn begin_batch(&mut self) {
        self.batch_depth = self.batch_depth.saturating_add(1);
    }

    /// Exit batch mode and rebuild only when the amortization threshold (or a
    /// coordinate change) demands it.
    ///
    /// Rebuilding unconditionally here made every single-formula edit O(V)
    /// and cell-by-cell edit loops O(N^2) (#125). Pending delta mutations are
    /// visible to readers through the merged out/in views, so deferring the
    /// rebuild is safe.
    pub fn end_batch(&mut self) {
        self.batch_depth = self.batch_depth.saturating_sub(1);
        if self.batch_depth == 0 && self.delta.needs_rebuild() {
            self.rebuild();
        }
    }

    /// End a batch without compacting the delta slab into the full CSR.
    /// Prepared graph transactions use this to keep application work local.
    pub(crate) fn end_batch_deferred(&mut self) {
        self.batch_depth = self.batch_depth.saturating_sub(1);
    }

    /// Add a new vertex with its coordinate and ID
    ///
    /// Does NOT rebuild the CSR base: reads of a vertex that is not in the
    /// base yet gracefully resolve to "no edges" plus any pending delta
    /// mutations, and the next rebuild picks the vertex up from
    /// `coords`/`vertex_ids` (#125).
    pub fn add_vertex(&mut self, addr: VertexAddr, vertex_id: u32) -> usize {
        let idx = self.coords.len();
        self.coords.push(addr);
        self.vertex_ids.push(vertex_id);
        self.vertex_pos.insert(vertex_id, idx);
        idx
    }

    /// Add many vertices at once; single rebuild at end.
    pub fn add_vertices_batch(&mut self, items: &[(VertexAddr, u32)]) {
        if items.is_empty() {
            return;
        }
        let start_len = self.coords.len();
        self.coords.reserve(items.len());
        self.vertex_ids.reserve(items.len());
        for (addr, vid) in items {
            let idx = self.coords.len();
            self.coords.push(*addr);
            self.vertex_ids.push(*vid);
            self.vertex_pos.insert(*vid, idx);
        }
        // Single rebuild to incorporate all new vertices.
        self.rebuild();
        debug_assert_eq!(self.coords.len(), start_len + items.len());
    }

    /// Update the stored address for a vertex in the cache
    /// Marks for rebuild to maintain sort order
    pub fn update_addr(&mut self, vertex_id: VertexId, new_addr: VertexAddr) {
        if let Some(&pos) = self.vertex_pos.get(&vertex_id.0) {
            debug_assert_eq!(
                self.vertex_ids[pos], vertex_id.0,
                "vertex_pos out of sync with vertex_ids at position {pos}"
            );
            self.coords[pos] = new_addr;
            // Force rebuild on next access to maintain sort invariants
            self.delta.mark_dirty();
        }
    }

    /// Return a copy of `adjacency` extended with the current base+delta
    /// out-edges of every existing vertex that the input does not cover.
    ///
    /// Named-range pass-through vertices (NamedScalar/NamedArray) emit edges
    /// to their underlying cells via `add_edge` during load; those edges live
    /// in `base`/`delta` but are not part of the formula-target adjacency that
    /// bulk-ingest's finalize hands to [`build_from_adjacency`]. Feeding the
    /// raw adjacency straight to that (pure) builder would therefore silently
    /// drop the pass-through vertices' out-edges, and `build_demand_subgraph`
    /// could never reach the underlying cells. Callers run this first to merge
    /// those edges back in, then pass the result to `build_from_adjacency`.
    ///
    /// Must be called BEFORE `build_from_adjacency`, which overwrites
    /// `base`/`delta`/`vertex_ids`.
    pub fn adjacency_with_carried_forward_edges(
        &self,
        mut adjacency: Vec<(u32, Vec<u32>)>,
    ) -> Vec<(u32, Vec<u32>)> {
        let covered: FxHashSet<u32> = adjacency.iter().map(|(vid, _)| *vid).collect();
        // Carry forward base+delta out-edges for ANY existing vertex not in
        // the new adjacency input. `vertex_ids` already tracks every vertex
        // allocated so far, including named-range pass-through vertices.
        for &vid in &self.vertex_ids {
            if covered.contains(&vid) {
                continue;
            }
            let v = VertexId(vid);
            // Merge with any pending delta edges for this vertex.
            let merged = if self.delta.op_count() == 0 {
                self.base.out_edges(v).to_vec()
            } else {
                self.delta.merged_view(&self.base, v)
            };
            if !merged.is_empty() {
                adjacency.push((vid, merged.into_iter().map(|v| v.0).collect()));
            }
        }
        // Also carry forward delta-only additions (additions to vertices that
        // weren't in vertex_ids yet — e.g., freshly allocated names whose
        // add_vertex didn't trigger a rebuild). Pure deltas should be rare
        // here, but include them for completeness.
        for (&from, adds) in self.delta.additions_iter() {
            if covered.contains(&from.0) {
                continue;
            }
            if adjacency.iter().any(|(v, _)| *v == from.0) {
                continue;
            }
            let removals: FxHashSet<u32> = self.delta.removals_for(from).map(|v| v.0).collect();
            let mut base_set: FxHashSet<u32> =
                self.base.out_edges(from).iter().map(|v| v.0).collect();
            for r in &removals {
                base_set.remove(r);
            }
            for &add in adds {
                base_set.insert(add.0);
            }
            if !base_set.is_empty() {
                let mut targets: Vec<u32> = base_set.into_iter().collect();
                targets.sort_unstable();
                adjacency.push((from.0, targets));
            }
        }
        adjacency
    }

    /// Build underlying CSR directly from adjacency and provided coords/ids.
    /// This replaces the current base and clears the delta slab.
    ///
    /// Pure builder: it uses exactly the edges in `adjacency` and does not
    /// consult the existing `base`/`delta`. To preserve edges for vertices
    /// absent from `adjacency` (e.g. named-range pass-through vertices), run
    /// [`adjacency_with_carried_forward_edges`] first and pass its result in.
    pub fn build_from_adjacency(
        &mut self,
        adjacency: Vec<(u32, Vec<u32>)>,
        coords: Vec<VertexAddr>,
        vertex_ids: Vec<u32>,
    ) {
        self.base = CsrEdges::from_adjacency(adjacency, &coords);
        self.coords = coords;
        self.vertex_ids = vertex_ids;
        self.vertex_pos = self
            .vertex_ids
            .iter()
            .enumerate()
            .map(|(idx, &id)| (id, idx))
            .collect();
        self.delta.clear();
    }
}

impl Default for CsrMutableEdges {
    fn default() -> Self {
        Self::new()
    }
}