//! Durable projection layered read source.
//!
//! Layered reads merge base CSR neighbors, durable delta segments, and the
//! current transaction-local delta into one deterministic source. Public SQL
//! read adoption is handled by a later phase; this module provides the pure
//! runtime surface and real segment loading boundary.

use roaring::RoaringBitmap;
use std::borrow::Cow;
use std::collections::{BTreeMap, HashMap, HashSet};
use std::fs;
use std::path::{Path, PathBuf};

use crate::edge_store::{EdgeStore, RelationshipId, NO_RELATIONSHIP_ID};
use crate::projection::chunk::SourceRange;
use crate::projection::manifest::{ManifestChunkRef, ManifestSegmentRef, ProjectionManifest};
use crate::projection::neighbors::{
    EdgeOverlay, Neighbor, NeighborIter, NeighborSource, OverlayDeletes, OverlayInsert,
    OverlayInserts, OwnedNeighborCursor, WeightedNeighbor, WeightedNeighborSource,
};
use crate::projection::segment::{DeltaSegment, SegmentKind};
use crate::projection::tx_delta;
use crate::safety::{GraphError, GraphResult};
use crate::types::{EdgeTypeId, TraversalDirection};

#[inline(always)]
#[allow(
    clippy::expect_used,
    reason = "owned and mmap edge stores reject reserved v6 type bytes before publication"
)]
fn logical_type_id_from_validated_v6(value: u8) -> EdgeTypeId {
    EdgeTypeId::from_v6_storage(value).expect("published topology contains validated v6 type IDs")
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
struct EdgeKey {
    source: u32,
    target: u32,
    type_id: EdgeTypeId,
    schema_reversed: bool,
    relationship_id: Option<RelationshipId>,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct LayeredEdge {
    target: u32,
    type_id: EdgeTypeId,
    schema_reversed: bool,
    weight: Option<u32>,
    relationship_id: Option<crate::edge_store::RelationshipId>,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum DurableEdgeState {
    Present {
        weight: Option<u32>,
        relationship_id: Option<crate::edge_store::RelationshipId>,
    },
    Deleted,
}

#[derive(Debug, Clone, Default)]
struct DurableEdges {
    inserts: Vec<LayeredEdge>,
    deletes: HashSet<(u32, EdgeTypeId, bool, Option<RelationshipId>)>,
}

type MergedEdgeKey = (u32, EdgeTypeId, bool, Option<RelationshipId>);

fn any_base_layer_winner(
    next_key: MergedEdgeKey,
    out_base: Option<LayeredEdge>,
    out_chunk: Option<LayeredEdge>,
    in_base: Option<LayeredEdge>,
    in_chunk: Option<LayeredEdge>,
) -> Option<LayeredEdge> {
    [out_base, out_chunk, in_base, in_chunk]
        .into_iter()
        .flatten()
        .rfind(|edge| {
            (
                edge.target,
                edge.type_id,
                edge.schema_reversed,
                edge.relationship_id,
            ) == next_key
        })
}

/// Fully decoded immutable durable state for one projection generation.
///
/// The engine builds this once while installing a validated manifest. Serving
/// queries borrow it, avoiding both segment I/O and repeated map construction.
pub(crate) struct LayeredSnapshot {
    base_chunk_ranges: Vec<SourceRange>,
    base_chunk_out: HashMap<u32, DurableEdges>,
    base_chunk_in: HashMap<u32, DurableEdges>,
    durable_out: HashMap<u32, DurableEdges>,
    durable_in: HashMap<u32, DurableEdges>,
    active_nodes: HashMap<u32, bool>,
    tenant_memberships: HashMap<u64, HashSet<u32>>,
    missing_relationship_identity_edge_types: RoaringBitmap,
}

/// Allocation preflight for constructing an immutable layered snapshot.
///
/// The estimate includes both direction indexes, their temporary merge maps,
/// output collections, per-segment weight indexes, and node/tenant metadata.
/// Callers can reserve [`Self::estimated_peak_bytes`] with their operation
/// governor before invoking [`LayeredSnapshot::try_build_from_refs_with_plan`].
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct LayeredSnapshotBuildPlan {
    base_chunk_ranges: usize,
    base_chunks: LayeredBuilderCapacity,
    segments: LayeredBuilderCapacity,
    estimated_peak_bytes: usize,
}

impl LayeredSnapshotBuildPlan {
    /// Return a conservative upper bound for private heap used while building.
    pub(crate) const fn estimated_peak_bytes(self) -> usize {
        self.estimated_peak_bytes
    }
}

#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
struct LayeredBuilderCapacity {
    segment_count: usize,
    edge_mutations: usize,
    max_edge_weights: usize,
    node_states: usize,
    tenant_rows: usize,
}

impl LayeredSnapshot {
    fn summarize_relationship_identity_completeness(
        &self,
        base: &EdgeStore,
        page: &mut Vec<Neighbor>,
    ) -> RoaringBitmap {
        let mut missing = RoaringBitmap::new();
        let layered = LayeredNeighbors::from_snapshot(base, self);
        for (direction, first, second) in [
            (
                TraversalDirection::Out,
                &self.base_chunk_out,
                &self.durable_out,
            ),
            (
                TraversalDirection::In,
                &self.base_chunk_in,
                &self.durable_in,
            ),
        ] {
            for source in first.keys().chain(second.keys()).copied() {
                let mut cursor = OwnedNeighborCursor::Start;
                loop {
                    page.clear();
                    let exhausted = layered.fill_directional_neighbors_internal(
                        direction,
                        source,
                        &mut cursor,
                        64,
                        page,
                        false,
                    );
                    for edge in &*page {
                        if edge.relationship_id.is_none() {
                            missing.insert(edge.type_id.get());
                        }
                    }
                    if exhausted {
                        break;
                    }
                }
            }
        }
        missing
    }

    fn refresh_relationship_identity_completeness_summary(&mut self, base: &EdgeStore) {
        let mut page = Vec::with_capacity(64);
        self.missing_relationship_identity_edge_types =
            self.summarize_relationship_identity_completeness(base, &mut page);
    }

    fn try_refresh_relationship_identity_completeness_summary(
        &mut self,
        base: &EdgeStore,
    ) -> GraphResult<()> {
        let mut page = Vec::new();
        page.try_reserve_exact(64)
            .map_err(layered_allocation_error)?;
        self.missing_relationship_identity_edge_types =
            self.summarize_relationship_identity_completeness(base, &mut page);
        Ok(())
    }

    /// Return whether an installed durable/base-chunk edge of any requested
    /// type lacks the relationship identity required by caller-scoped RLS.
    ///
    /// The snapshot computes this summary once while it is assembled. Serving
    /// queries only inspect the fixed-size summary; they never rescan segment
    /// adjacency.
    pub(crate) fn has_missing_relationship_identity_for_types(
        &self,
        active_edge_types: &RoaringBitmap,
    ) -> bool {
        active_edge_types.iter().any(|edge_type| {
            self.missing_relationship_identity_edge_types
                .contains(edge_type)
        })
    }

    pub(crate) fn has_any_missing_relationship_identity(&self) -> bool {
        !self.missing_relationship_identity_edge_types.is_empty()
    }

    /// Derive the immutable serving snapshot from validated decoded segments.
    pub(crate) fn build(
        base: &EdgeStore,
        base_chunks: &[DeltaSegment],
        segments: &[DeltaSegment],
    ) -> Self {
        let base_chunk_refs = base_chunks.iter().collect::<Vec<_>>();
        let segment_refs = segments.iter().collect::<Vec<_>>();
        Self::build_from_refs(base, &base_chunk_refs, &segment_refs)
    }

    /// Derive a snapshot without cloning already-decoded segment payloads.
    pub(crate) fn build_from_refs(
        base: &EdgeStore,
        base_chunks: &[&DeltaSegment],
        segments: &[&DeltaSegment],
    ) -> Self {
        let base_chunk_ranges = base_chunks
            .iter()
            .map(|chunk| SourceRange {
                start: chunk.header.source_start,
                end: chunk.header.source_end,
            })
            .collect::<Vec<_>>();
        let mut base_chunk_builder = LayeredBuilder::new(base);
        base_chunk_builder.apply_segment_refs(base_chunks);
        let base_chunk_output = base_chunk_builder.finish_replacement();
        let mut builder = LayeredBuilder::new(base);
        builder.apply_segment_refs(segments);
        let output = builder.finish();
        let mut snapshot = Self {
            base_chunk_ranges,
            base_chunk_out: base_chunk_output.durable_out,
            base_chunk_in: base_chunk_output.durable_in,
            durable_out: output.durable_out,
            durable_in: output.durable_in,
            active_nodes: output.active_nodes,
            tenant_memberships: output.tenant_memberships,
            missing_relationship_identity_edge_types: RoaringBitmap::new(),
        };
        snapshot.refresh_relationship_identity_completeness_summary(base);
        snapshot
    }

    /// Preflight all collection capacities required by a layered snapshot.
    ///
    /// # Errors
    ///
    /// Returns [`GraphError::Internal`] when input counts or the conservative
    /// heap estimate overflow `usize`.
    pub(crate) fn build_plan(
        base_chunks: &[&DeltaSegment],
        segments: &[&DeltaSegment],
    ) -> GraphResult<LayeredSnapshotBuildPlan> {
        let base_capacity = LayeredBuilderCapacity::try_from_segments(base_chunks)?;
        let segment_capacity = LayeredBuilderCapacity::try_from_segments(segments)?;
        let range_bytes = checked_bytes(base_chunks.len(), std::mem::size_of::<SourceRange>())?;
        let base_bytes = base_capacity.estimated_heap_bytes()?;
        let segment_bytes = segment_capacity.estimated_heap_bytes()?;
        let summary_page_bytes = checked_bytes(64, std::mem::size_of::<Neighbor>())?;
        let estimated_peak_bytes = range_bytes
            .checked_add(base_bytes)
            .and_then(|bytes| bytes.checked_add(segment_bytes))
            .and_then(|bytes| bytes.checked_add(summary_page_bytes))
            .ok_or_else(layered_size_overflow)?;
        Ok(LayeredSnapshotBuildPlan {
            base_chunk_ranges: base_chunks.len(),
            base_chunks: base_capacity,
            segments: segment_capacity,
            estimated_peak_bytes,
        })
    }

    /// Build a snapshot with fallible, pre-sized backing collections.
    ///
    /// # Errors
    ///
    /// Returns a typed allocation or size-overflow error.
    pub(crate) fn try_build_from_refs(
        base: &EdgeStore,
        base_chunks: &[&DeltaSegment],
        segments: &[&DeltaSegment],
    ) -> GraphResult<Self> {
        let plan = Self::build_plan(base_chunks, segments)?;
        Self::try_build_from_refs_with_plan(base, base_chunks, segments, plan)
    }

    /// Build from a previously validated allocation plan.
    ///
    /// # Errors
    ///
    /// Returns an internal error when the plan no longer matches the inputs,
    /// or a typed allocation error when a backing allocation fails.
    pub(crate) fn try_build_from_refs_with_plan(
        base: &EdgeStore,
        base_chunks: &[&DeltaSegment],
        segments: &[&DeltaSegment],
        plan: LayeredSnapshotBuildPlan,
    ) -> GraphResult<Self> {
        let actual = Self::build_plan(base_chunks, segments)?;
        if actual != plan {
            return Err(GraphError::Internal(
                "layered snapshot allocation plan does not match its segment inputs".to_string(),
            ));
        }
        let mut base_chunk_ranges = Vec::new();
        base_chunk_ranges
            .try_reserve_exact(plan.base_chunk_ranges)
            .map_err(layered_allocation_error)?;
        base_chunk_ranges.extend(base_chunks.iter().map(|chunk| SourceRange {
            start: chunk.header.source_start,
            end: chunk.header.source_end,
        }));
        let mut base_chunk_builder = LayeredBuilder::try_new(base, plan.base_chunks, base_chunks)?;
        base_chunk_builder.try_apply_segment_refs(base_chunks)?;
        let base_chunk_output = base_chunk_builder.try_finish_replacement()?;
        let mut builder = LayeredBuilder::try_new(base, plan.segments, segments)?;
        builder.try_apply_segment_refs(segments)?;
        let output = builder.try_finish()?;
        let mut snapshot = Self {
            base_chunk_ranges,
            base_chunk_out: base_chunk_output.durable_out,
            base_chunk_in: base_chunk_output.durable_in,
            durable_out: output.durable_out,
            durable_in: output.durable_in,
            active_nodes: output.active_nodes,
            tenant_memberships: output.tenant_memberships,
            missing_relationship_identity_edge_types: RoaringBitmap::new(),
        };
        snapshot.try_refresh_relationship_identity_completeness_summary(base)?;
        Ok(snapshot)
    }

    /// Conservatively estimate backend-private heap retained by this snapshot.
    pub(crate) fn estimated_heap_bytes(&self) -> usize {
        fn durable_map_bytes(map: &HashMap<u32, DurableEdges>) -> usize {
            let buckets = map
                .capacity()
                .saturating_mul(std::mem::size_of::<(u32, DurableEdges)>() * 2);
            let payloads = map.values().fold(0usize, |total, edges| {
                total
                    .saturating_add(
                        edges
                            .inserts
                            .capacity()
                            .saturating_mul(std::mem::size_of::<LayeredEdge>()),
                    )
                    .saturating_add(edges.deletes.capacity().saturating_mul(
                        std::mem::size_of::<(u32, EdgeTypeId, bool, Option<RelationshipId>)>() * 2,
                    ))
            });
            buckets.saturating_add(payloads)
        }

        self.base_chunk_ranges
            .capacity()
            .saturating_mul(std::mem::size_of::<SourceRange>())
            .saturating_add(durable_map_bytes(&self.base_chunk_out))
            .saturating_add(durable_map_bytes(&self.base_chunk_in))
            .saturating_add(durable_map_bytes(&self.durable_out))
            .saturating_add(durable_map_bytes(&self.durable_in))
            .saturating_add(
                self.active_nodes
                    .capacity()
                    .saturating_mul(std::mem::size_of::<(u32, bool)>() * 2),
            )
            .saturating_add(
                self.tenant_memberships
                    .capacity()
                    .saturating_mul(std::mem::size_of::<(u64, HashSet<u32>)>() * 2),
            )
            .saturating_add(
                self.tenant_memberships
                    .values()
                    .fold(0usize, |total, members| {
                        total.saturating_add(
                            members
                                .capacity()
                                .saturating_mul(std::mem::size_of::<u32>() * 2),
                        )
                    }),
            )
    }

    /// Conservative per-node upper bound for durable non-base neighbors.
    pub(crate) fn additional_neighbor_upper_bound(&self, direction: TraversalDirection) -> usize {
        fn inserted(map: &HashMap<u32, DurableEdges>) -> usize {
            map.values().fold(0usize, |total, edges| {
                total.saturating_add(edges.inserts.len())
            })
        }
        let outgoing = inserted(&self.base_chunk_out).saturating_add(inserted(&self.durable_out));
        let incoming = inserted(&self.base_chunk_in).saturating_add(inserted(&self.durable_in));
        match direction {
            TraversalDirection::Out => outgoing,
            TraversalDirection::In => incoming,
            TraversalDirection::Any => outgoing.saturating_add(incoming),
        }
    }
}

/// Source of decoded durable segments for a layered projection snapshot.
pub(crate) trait SegmentProvider {
    /// Load decoded segments that belong to one manifest snapshot.
    ///
    /// # Errors
    ///
    /// Returns segment loading or validation errors from the backing store.
    fn load_segments(&self) -> GraphResult<Vec<DeltaSegment>>;

    /// Load decoded base chunks that replace source-node ranges.
    ///
    /// # Errors
    ///
    /// Returns chunk loading or validation errors from the backing store.
    fn load_base_chunks(&self) -> GraphResult<Vec<DeltaSegment>> {
        Ok(Vec::new())
    }
}

/// Segment provider backed by a manifest and projection artifact directory.
#[allow(
    dead_code,
    reason = "Microphase 8 wires Engine read-path adoption to the manifest segment provider"
)]
pub(crate) struct ManifestSegmentProvider<'a> {
    root: &'a Path,
    manifest: &'a ProjectionManifest,
}

impl<'a> ManifestSegmentProvider<'a> {
    /// Borrow a manifest snapshot for real durable segment loading.
    #[allow(
        dead_code,
        reason = "Microphase 8 wires Engine read-path adoption to the manifest segment provider"
    )]
    pub(crate) fn new(root: &'a Path, manifest: &'a ProjectionManifest) -> Self {
        Self { root, manifest }
    }
}

impl SegmentProvider for ManifestSegmentProvider<'_> {
    fn load_segments(&self) -> GraphResult<Vec<DeltaSegment>> {
        self.manifest
            .segments
            .iter()
            .map(|segment| read_manifest_segment(self.root, segment))
            .collect()
    }

    fn load_base_chunks(&self) -> GraphResult<Vec<DeltaSegment>> {
        self.manifest
            .base_chunks
            .iter()
            .map(|chunk| read_manifest_chunk(self.root, chunk))
            .collect()
    }
}

#[allow(
    dead_code,
    reason = "Microphase 8 wires Engine read-path adoption to the manifest segment provider"
)]
fn read_manifest_segment(root: &Path, segment: &ManifestSegmentRef) -> GraphResult<DeltaSegment> {
    let path = root.join(PathBuf::from(&segment.path));
    let bytes = fs::read(&path)
        .map_err(|err| GraphError::Internal(format!("segment read failed: {err}")))?;
    let checksum = format!("crc32:{:08x}", crc32fast::hash(&bytes));
    if checksum != segment.checksum {
        return Err(GraphError::CorruptFile {
            reason: format!(
                "projection segment checksum mismatch for {}: expected {}, got {}",
                segment.path, segment.checksum, checksum
            ),
        });
    }
    DeltaSegment::from_bytes(&bytes)
}

fn read_manifest_chunk(root: &Path, chunk: &ManifestChunkRef) -> GraphResult<DeltaSegment> {
    let path = root.join(PathBuf::from(&chunk.path));
    let bytes = fs::read(&path)
        .map_err(|err| GraphError::Internal(format!("base chunk read failed: {err}")))?;
    let checksum = format!("crc32:{:08x}", crc32fast::hash(&bytes));
    if checksum != chunk.checksum {
        return Err(GraphError::CorruptFile {
            reason: format!(
                "projection base chunk checksum mismatch for {}: expected {}, got {}",
                chunk.path, chunk.checksum, checksum
            ),
        });
    }
    let segment = DeltaSegment::from_bytes(&bytes)?;
    if segment.header.source_start != chunk.source_start
        || segment.header.source_end != chunk.source_end
        || segment.header.kind != SegmentKind::Edge
        || segment.header.direction != TraversalDirection::Out
    {
        return Err(GraphError::CorruptFile {
            reason: format!(
                "projection base chunk {} metadata does not match manifest range {}..{}",
                chunk.path, chunk.source_start, chunk.source_end
            ),
        });
    }
    Ok(segment)
}

/// Immutable layered neighbor source for one manifest snapshot.
pub(crate) struct LayeredNeighbors<'a> {
    base: &'a EdgeStore,
    base_in: Option<&'a EdgeStore>,
    base_chunk_ranges: Cow<'a, [SourceRange]>,
    base_chunk_out: Cow<'a, HashMap<u32, DurableEdges>>,
    base_chunk_in: Cow<'a, HashMap<u32, DurableEdges>>,
    durable_out: Cow<'a, HashMap<u32, DurableEdges>>,
    durable_in: Cow<'a, HashMap<u32, DurableEdges>>,
    committed_out_inserts: Cow<'a, OverlayInserts>,
    committed_out_deletes: Cow<'a, OverlayDeletes>,
    committed_in_inserts: Cow<'a, OverlayInserts>,
    committed_in_deletes: Cow<'a, OverlayDeletes>,
    active_nodes: Cow<'a, HashMap<u32, bool>>,
    tenant_memberships: Cow<'a, HashMap<u64, HashSet<u32>>>,
    tenant_filter: Option<u64>,
}

impl<'a> LayeredNeighbors<'a> {
    /// Build a layered source from already-decoded segments.
    pub(crate) fn new(base: &'a EdgeStore, segments: Vec<DeltaSegment>) -> Self {
        Self::new_with_tenant(base, segments, None)
    }

    /// Build a layered source with an optional tenant membership filter.
    pub(crate) fn new_with_tenant(
        base: &'a EdgeStore,
        segments: Vec<DeltaSegment>,
        tenant_filter: Option<u64>,
    ) -> Self {
        Self::new_with_options(base, None, segments, tenant_filter, None, None)
    }

    /// Build a layered source with reverse base CSR and committed overlays.
    pub(crate) fn new_with_options(
        base: &'a EdgeStore,
        base_in: Option<&'a EdgeStore>,
        segments: Vec<DeltaSegment>,
        tenant_filter: Option<u64>,
        committed_out: Option<EdgeOverlay>,
        committed_in: Option<EdgeOverlay>,
    ) -> Self {
        Self::new_with_base_chunks(
            base,
            base_in,
            Vec::new(),
            segments,
            tenant_filter,
            committed_out,
            committed_in,
        )
    }

    fn new_with_base_chunks(
        base: &'a EdgeStore,
        base_in: Option<&'a EdgeStore>,
        base_chunks: Vec<DeltaSegment>,
        segments: Vec<DeltaSegment>,
        tenant_filter: Option<u64>,
        committed_out: Option<EdgeOverlay>,
        committed_in: Option<EdgeOverlay>,
    ) -> Self {
        let snapshot = LayeredSnapshot::build(base, &base_chunks, &segments);
        let (committed_out_inserts, committed_out_deletes) = committed_out.unwrap_or_default();
        let (committed_in_inserts, committed_in_deletes) = committed_in.unwrap_or_default();
        Self {
            base,
            base_in,
            base_chunk_ranges: Cow::Owned(snapshot.base_chunk_ranges),
            base_chunk_out: Cow::Owned(snapshot.base_chunk_out),
            base_chunk_in: Cow::Owned(snapshot.base_chunk_in),
            durable_out: Cow::Owned(snapshot.durable_out),
            durable_in: Cow::Owned(snapshot.durable_in),
            committed_out_inserts: Cow::Owned(committed_out_inserts),
            committed_out_deletes: Cow::Owned(committed_out_deletes),
            committed_in_inserts: Cow::Owned(committed_in_inserts),
            committed_in_deletes: Cow::Owned(committed_in_deletes),
            active_nodes: Cow::Owned(snapshot.active_nodes),
            tenant_memberships: Cow::Owned(snapshot.tenant_memberships),
            tenant_filter,
        }
    }

    /// Build a layered source from a segment provider.
    ///
    /// # Errors
    ///
    /// Returns provider errors when segment loading fails.
    pub(crate) fn from_provider(
        base: &'a EdgeStore,
        provider: &impl SegmentProvider,
    ) -> GraphResult<Self> {
        Ok(Self::new_with_base_chunks(
            base,
            None,
            provider.load_base_chunks()?,
            provider.load_segments()?,
            None,
            None,
            None,
        ))
    }

    /// Build a layered source from a provider plus Engine-owned read overlays.
    ///
    /// # Errors
    ///
    /// Returns provider errors when segment loading fails.
    pub(crate) fn from_provider_with_overlays(
        base: &'a EdgeStore,
        base_in: &'a EdgeStore,
        provider: &impl SegmentProvider,
        committed_out: EdgeOverlay,
        committed_in: EdgeOverlay,
    ) -> GraphResult<Self> {
        Ok(Self::new_with_base_chunks(
            base,
            Some(base_in),
            provider.load_base_chunks()?,
            provider.load_segments()?,
            None,
            Some(committed_out),
            Some(committed_in),
        ))
    }

    /// Build a layered source from a generation-pinned derived snapshot.
    ///
    /// This is the serving-path constructor. Segment files are validated and
    /// decoded when the generation is installed; individual queries borrow
    /// that immutable snapshot and never reopen projection files.
    pub(crate) fn from_snapshot_with_overlays(
        base: &'a EdgeStore,
        base_in: &'a EdgeStore,
        snapshot: &'a LayeredSnapshot,
        committed_out: EdgeOverlay,
        committed_in: EdgeOverlay,
    ) -> Self {
        let (committed_out_inserts, committed_out_deletes) = committed_out;
        let (committed_in_inserts, committed_in_deletes) = committed_in;
        Self {
            base,
            base_in: Some(base_in),
            base_chunk_ranges: Cow::Borrowed(&snapshot.base_chunk_ranges),
            base_chunk_out: Cow::Borrowed(&snapshot.base_chunk_out),
            base_chunk_in: Cow::Borrowed(&snapshot.base_chunk_in),
            durable_out: Cow::Borrowed(&snapshot.durable_out),
            durable_in: Cow::Borrowed(&snapshot.durable_in),
            committed_out_inserts: Cow::Owned(committed_out_inserts),
            committed_out_deletes: Cow::Owned(committed_out_deletes),
            committed_in_inserts: Cow::Owned(committed_in_inserts),
            committed_in_deletes: Cow::Owned(committed_in_deletes),
            active_nodes: Cow::Borrowed(&snapshot.active_nodes),
            tenant_memberships: Cow::Borrowed(&snapshot.tenant_memberships),
            tenant_filter: None,
        }
    }

    /// Borrow a pinned derived snapshot without live overlays.
    pub(crate) fn from_snapshot(base: &'a EdgeStore, snapshot: &'a LayeredSnapshot) -> Self {
        Self {
            base,
            base_in: None,
            base_chunk_ranges: Cow::Borrowed(&snapshot.base_chunk_ranges),
            base_chunk_out: Cow::Borrowed(&snapshot.base_chunk_out),
            base_chunk_in: Cow::Borrowed(&snapshot.base_chunk_in),
            durable_out: Cow::Borrowed(&snapshot.durable_out),
            durable_in: Cow::Borrowed(&snapshot.durable_in),
            committed_out_inserts: Cow::Owned(HashMap::new()),
            committed_out_deletes: Cow::Owned(HashMap::new()),
            committed_in_inserts: Cow::Owned(HashMap::new()),
            committed_in_deletes: Cow::Owned(HashMap::new()),
            active_nodes: Cow::Borrowed(&snapshot.active_nodes),
            tenant_memberships: Cow::Borrowed(&snapshot.tenant_memberships),
            tenant_filter: None,
        }
    }

    /// Borrow a pinned snapshot plus one query-frozen directional overlay.
    pub(crate) fn from_snapshot_with_direction_overlay(
        base: &'a EdgeStore,
        base_in: &'a EdgeStore,
        snapshot: &'a LayeredSnapshot,
        direction: TraversalDirection,
        inserts: &'a OverlayInserts,
        deletes: &'a OverlayDeletes,
    ) -> Self {
        let empty_inserts = Cow::Owned(HashMap::new());
        let empty_deletes = Cow::Owned(HashMap::new());
        let (out_inserts, out_deletes, in_inserts, in_deletes) = match direction {
            TraversalDirection::Out => (
                Cow::Borrowed(inserts),
                Cow::Borrowed(deletes),
                empty_inserts,
                empty_deletes,
            ),
            TraversalDirection::In => (
                empty_inserts,
                empty_deletes,
                Cow::Borrowed(inserts),
                Cow::Borrowed(deletes),
            ),
            TraversalDirection::Any => (
                empty_inserts,
                empty_deletes,
                Cow::Owned(HashMap::new()),
                Cow::Owned(HashMap::new()),
            ),
        };
        Self {
            base,
            base_in: Some(base_in),
            base_chunk_ranges: Cow::Borrowed(&snapshot.base_chunk_ranges),
            base_chunk_out: Cow::Borrowed(&snapshot.base_chunk_out),
            base_chunk_in: Cow::Borrowed(&snapshot.base_chunk_in),
            durable_out: Cow::Borrowed(&snapshot.durable_out),
            durable_in: Cow::Borrowed(&snapshot.durable_in),
            committed_out_inserts: out_inserts,
            committed_out_deletes: out_deletes,
            committed_in_inserts: in_inserts,
            committed_in_deletes: in_deletes,
            active_nodes: Cow::Borrowed(&snapshot.active_nodes),
            tenant_memberships: Cow::Borrowed(&snapshot.tenant_memberships),
            tenant_filter: None,
        }
    }

    /// Borrow a pinned snapshot plus query-frozen overlays for both directions.
    pub(crate) fn from_snapshot_with_frozen_overlays(
        base: &'a EdgeStore,
        base_in: &'a EdgeStore,
        snapshot: &'a LayeredSnapshot,
        committed_out: &'a EdgeOverlay,
        committed_in: &'a EdgeOverlay,
    ) -> Self {
        Self {
            base,
            base_in: Some(base_in),
            base_chunk_ranges: Cow::Borrowed(&snapshot.base_chunk_ranges),
            base_chunk_out: Cow::Borrowed(&snapshot.base_chunk_out),
            base_chunk_in: Cow::Borrowed(&snapshot.base_chunk_in),
            durable_out: Cow::Borrowed(&snapshot.durable_out),
            durable_in: Cow::Borrowed(&snapshot.durable_in),
            committed_out_inserts: Cow::Borrowed(&committed_out.0),
            committed_out_deletes: Cow::Borrowed(&committed_out.1),
            committed_in_inserts: Cow::Borrowed(&committed_in.0),
            committed_in_deletes: Cow::Borrowed(&committed_in.1),
            active_nodes: Cow::Borrowed(&snapshot.active_nodes),
            tenant_memberships: Cow::Borrowed(&snapshot.tenant_memberships),
            tenant_filter: None,
        }
    }

    /// Borrow this layered snapshot as a neighbor source for `direction`.
    pub(crate) fn for_direction(
        &self,
        direction: TraversalDirection,
    ) -> DirectionalLayeredNeighbors<'_, 'a> {
        DirectionalLayeredNeighbors {
            layered: self,
            direction,
        }
    }

    /// Return weighted outgoing neighbors after durable deltas and transaction
    /// overlays have been applied.
    pub(crate) fn weighted_neighbors(&self, node_idx: u32) -> Vec<WeightedNeighbor> {
        self.merged_neighbors(TraversalDirection::Out, node_idx, false)
            .into_iter()
            .filter_map(|(target, edge)| {
                edge.weight.map(|weight| WeightedNeighbor {
                    target,
                    type_id: edge.type_id,
                    weight,
                    schema_reversed: edge.schema_reversed,
                    relationship_id: edge.relationship_id,
                })
            })
            .collect()
    }

    fn fill_weighted_out_neighbors(
        &self,
        node_idx: u32,
        cursor: &mut OwnedNeighborCursor,
        limit: usize,
        output: &mut Vec<WeightedNeighbor>,
    ) -> bool {
        if !self.node_visible(node_idx) || limit == 0 {
            return true;
        }
        let (targets, type_ids, schema_reversed, relationship_ids) = self
            .base
            .neighbors_with_schema_and_relationship_ids(node_idx);
        let weights = base_weight_slice(self.base, node_idx);
        let base_hidden = self.base_chunk_covers(node_idx);
        let chunk = self.base_chunk_out.get(&node_idx);
        let durable = self.durable_out.get(&node_idx);
        let overlay = self.committed_out_inserts.get(&node_idx).map(Vec::as_slice);
        let overlay_deleted = self.committed_out_deletes.get(&node_idx);
        let (mut base_pos, mut chunk_pos, mut durable_pos, mut overlay_pos, mut last_key) =
            match cursor {
                OwnedNeighborCursor::Layered {
                    base_pos,
                    chunk_pos,
                    durable_pos,
                    overlay_pos,
                    last_key,
                } => (*base_pos, *chunk_pos, *durable_pos, *overlay_pos, *last_key),
                _ => (0, 0, 0, 0, None),
            };
        let key = |edge: LayeredEdge| {
            (
                edge.target,
                edge.type_id,
                edge.schema_reversed,
                edge.relationship_id,
            )
        };
        let deleted = |set: Option<&HashSet<MergedEdgeKey>>, candidate: MergedEdgeKey| {
            set.is_some_and(|set| {
                set.contains(&candidate)
                    || set.contains(&(candidate.0, candidate.1, candidate.2, None))
            })
        };
        let mut examined = 0usize;
        while examined < limit {
            let base_edge = (!base_hidden && base_pos < targets.len()).then(|| LayeredEdge {
                target: targets[base_pos],
                type_id: type_ids.at(base_pos),
                schema_reversed: schema_reversed[base_pos] != 0,
                weight: weights.and_then(|weights| weights.get(base_pos).copied()),
                relationship_id: relationship_ids
                    .get(base_pos)
                    .copied()
                    .filter(|id| *id != NO_RELATIONSHIP_ID),
            });
            let chunk_edge = chunk
                .and_then(|edges| edges.inserts.get(chunk_pos))
                .copied();
            let durable_edge = durable
                .and_then(|edges| edges.inserts.get(durable_pos))
                .copied();
            let overlay_edge = overlay.and_then(|edges| edges.get(overlay_pos)).map(
                |&(target, type_id, schema_reversed, relationship_id)| LayeredEdge {
                    target,
                    type_id,
                    schema_reversed,
                    weight: None,
                    relationship_id,
                },
            );
            let Some(next_key) = [base_edge, chunk_edge, durable_edge, overlay_edge]
                .into_iter()
                .flatten()
                .map(key)
                .min()
            else {
                *cursor = OwnedNeighborCursor::Layered {
                    base_pos,
                    chunk_pos,
                    durable_pos,
                    overlay_pos,
                    last_key,
                };
                return true;
            };
            let equal_count = [base_edge, chunk_edge, durable_edge, overlay_edge]
                .into_iter()
                .flatten()
                .filter(|edge| key(*edge) == next_key)
                .count();
            if examined > 0 && examined.saturating_add(equal_count) > limit {
                break;
            }
            let durable_reinserts = durable_edge.is_some_and(|edge| key(edge) == next_key);
            let overlay_reinserts = overlay_edge.is_some_and(|edge| key(edge) == next_key);
            let mut selected = None;
            if base_edge.is_some_and(|edge| key(edge) == next_key) {
                selected = base_edge;
                base_pos += 1;
                examined += 1;
            }
            if chunk_edge.is_some_and(|edge| key(edge) == next_key) {
                selected = chunk_edge;
                chunk_pos += 1;
                examined += 1;
            }
            if durable_edge.is_some_and(|edge| key(edge) == next_key) {
                selected = durable_edge;
                durable_pos += 1;
                examined += 1;
            }
            if overlay_edge.is_some_and(|edge| key(edge) == next_key) {
                selected = overlay_edge;
                overlay_pos += 1;
                examined += 1;
            }
            let hidden_by_durable = deleted(durable.map(|edges| &edges.deletes), next_key);
            let hidden_by_overlay = deleted(overlay_deleted, next_key);
            let survives = (overlay_reinserts || !hidden_by_overlay)
                && (overlay_reinserts || durable_reinserts || !hidden_by_durable);
            if last_key != Some(next_key) && survives {
                if let Some(edge) = selected.filter(|edge| self.node_visible(edge.target)) {
                    if let Some(weight) = edge.weight {
                        output.push(WeightedNeighbor {
                            target: edge.target,
                            type_id: edge.type_id,
                            weight,
                            schema_reversed: edge.schema_reversed,
                            relationship_id: edge.relationship_id,
                        });
                    }
                }
            }
            last_key = Some(next_key);
        }
        *cursor = OwnedNeighborCursor::Layered {
            base_pos,
            chunk_pos,
            durable_pos,
            overlay_pos,
            last_key,
        };
        false
    }

    fn merged_neighbors(
        &self,
        direction: TraversalDirection,
        node_idx: u32,
        reversed: bool,
    ) -> Vec<(u32, LayeredEdge)> {
        if !self.node_visible(node_idx) {
            return Vec::new();
        }

        // Do not route an untouched base projection through the topology-keyed
        // merge map. A base CSR may contain parallel source relationship rows,
        // and the map is reserved for the legacy delta representation until
        // durable relationship keys reach segments and overlays.
        if direction == TraversalDirection::Out
            && self.base_chunk_out.is_empty()
            && self.durable_out.is_empty()
            && self.committed_out_inserts.is_empty()
            && self.committed_out_deletes.is_empty()
            && self.active_nodes.is_empty()
            && self.tenant_memberships.is_empty()
            && !tx_delta::edge_delta_dirty()
        {
            let (targets, type_ids, schema_reversed, relationship_ids) = self
                .base
                .neighbors_with_schema_and_relationship_ids(node_idx);
            let weights = base_weight_slice(self.base, node_idx);
            let mut neighbors = targets
                .iter()
                .zip(type_ids.iter())
                .zip(schema_reversed.iter())
                .enumerate()
                .filter_map(|(idx, ((&target, type_id), &schema_reversed))| {
                    self.node_visible(target).then_some((
                        target,
                        LayeredEdge {
                            target,
                            type_id,
                            schema_reversed: schema_reversed != 0,
                            weight: weights.and_then(|values| values.get(idx)).copied(),
                            relationship_id: relationship_ids
                                .get(idx)
                                .copied()
                                .filter(|&id| id != NO_RELATIONSHIP_ID),
                        },
                    ))
                })
                .collect::<Vec<_>>();
            if reversed {
                neighbors.reverse();
            }
            return neighbors;
        }

        let mut merged = BTreeMap::<MergedEdgeKey, LayeredEdge>::new();
        self.merge_base(direction, node_idx, &mut merged);
        self.merge_durable(direction, node_idx, &mut merged);
        self.merge_committed_overlay(direction, node_idx, &mut merged);
        self.merge_tx_delta(direction, node_idx, &mut merged);

        let mut neighbors = merged
            .into_iter()
            .filter(|((target, _, _, _), _)| self.node_visible(*target))
            .map(|((target, _, _, _), edge)| (target, edge))
            .collect::<Vec<_>>();
        if reversed {
            neighbors.reverse();
        }
        neighbors
    }

    fn merge_base(
        &self,
        direction: TraversalDirection,
        node_idx: u32,
        merged: &mut BTreeMap<MergedEdgeKey, LayeredEdge>,
    ) {
        match direction {
            TraversalDirection::Out => {
                if self.base_chunk_covers(node_idx) {
                    merge_durable_map(&self.base_chunk_out, node_idx, merged);
                } else {
                    let (targets, type_ids, schema_reversed, relationship_ids) = self
                        .base
                        .neighbors_with_schema_and_relationship_ids(node_idx);
                    let weights = base_weight_slice(self.base, node_idx);
                    for (idx, ((&target, type_id), &schema_reversed)) in targets
                        .iter()
                        .zip(type_ids.iter())
                        .zip(schema_reversed.iter())
                        .enumerate()
                    {
                        merged.insert(
                            (
                                target,
                                type_id,
                                schema_reversed != 0,
                                relationship_ids
                                    .get(idx)
                                    .copied()
                                    .filter(|&id| id != NO_RELATIONSHIP_ID),
                            ),
                            LayeredEdge {
                                target,
                                type_id,
                                schema_reversed: schema_reversed != 0,
                                weight: weights.and_then(|weights| weights.get(idx).copied()),
                                relationship_id: relationship_ids
                                    .get(idx)
                                    .copied()
                                    .filter(|&id| id != NO_RELATIONSHIP_ID),
                            },
                        );
                    }
                }
            }
            TraversalDirection::In => {
                if let Some(base_in) = self.base_in {
                    let (targets, type_ids, schema_reversed, relationship_ids) =
                        base_in.neighbors_with_schema_and_relationship_ids(node_idx);
                    let weights = base_weight_slice(base_in, node_idx);
                    for (idx, ((&target, type_id), &schema_reversed)) in targets
                        .iter()
                        .zip(type_ids.iter())
                        .zip(schema_reversed.iter())
                        .enumerate()
                    {
                        if self.base_chunk_covers(target) {
                            continue;
                        }
                        merged.insert(
                            (
                                target,
                                type_id,
                                schema_reversed != 0,
                                relationship_ids
                                    .get(idx)
                                    .copied()
                                    .filter(|&id| id != NO_RELATIONSHIP_ID),
                            ),
                            LayeredEdge {
                                target,
                                type_id,
                                schema_reversed: schema_reversed != 0,
                                weight: weights.and_then(|weights| weights.get(idx).copied()),
                                relationship_id: relationship_ids
                                    .get(idx)
                                    .copied()
                                    .filter(|&id| id != NO_RELATIONSHIP_ID),
                            },
                        );
                    }
                } else {
                    self.merge_base_in_by_scan(node_idx, merged);
                }
                merge_durable_map(&self.base_chunk_in, node_idx, merged);
            }
            TraversalDirection::Any => {
                self.merge_base(TraversalDirection::Out, node_idx, merged);
                self.merge_base(TraversalDirection::In, node_idx, merged);
            }
        }
    }

    fn merge_base_in_by_scan(
        &self,
        node_idx: u32,
        merged: &mut BTreeMap<MergedEdgeKey, LayeredEdge>,
    ) {
        for source in 0..self.base.node_count() {
            if self.base_chunk_covers(source) {
                continue;
            }
            let (targets, type_ids, schema_reversed, relationship_ids) =
                self.base.neighbors_with_schema_and_relationship_ids(source);
            let weights = base_weight_slice(self.base, source);
            for (idx, ((&target, type_id), &schema_reversed)) in targets
                .iter()
                .zip(type_ids.iter())
                .zip(schema_reversed.iter())
                .enumerate()
            {
                if target == node_idx {
                    merged.insert(
                        (
                            source,
                            type_id,
                            schema_reversed != 0,
                            relationship_ids
                                .get(idx)
                                .copied()
                                .filter(|&id| id != NO_RELATIONSHIP_ID),
                        ),
                        LayeredEdge {
                            target: source,
                            type_id,
                            schema_reversed: schema_reversed != 0,
                            weight: weights.and_then(|weights| weights.get(idx).copied()),
                            relationship_id: relationship_ids
                                .get(idx)
                                .copied()
                                .filter(|&id| id != NO_RELATIONSHIP_ID),
                        },
                    );
                }
            }
        }
        merge_durable_map(&self.base_chunk_in, node_idx, merged);
    }

    fn base_chunk_covers(&self, source: u32) -> bool {
        self.base_chunk_ranges
            .iter()
            .any(|range| range.contains(source))
    }

    fn merge_durable(
        &self,
        direction: TraversalDirection,
        node_idx: u32,
        merged: &mut BTreeMap<MergedEdgeKey, LayeredEdge>,
    ) {
        match direction {
            TraversalDirection::Out => merge_durable_map(&self.durable_out, node_idx, merged),
            TraversalDirection::In => merge_durable_map(&self.durable_in, node_idx, merged),
            TraversalDirection::Any => {
                merge_durable_map(&self.durable_out, node_idx, merged);
                merge_durable_map(&self.durable_in, node_idx, merged);
            }
        }
    }

    fn merge_committed_overlay(
        &self,
        direction: TraversalDirection,
        node_idx: u32,
        merged: &mut BTreeMap<MergedEdgeKey, LayeredEdge>,
    ) {
        match direction {
            TraversalDirection::Out => merge_committed_overlay_maps(
                &self.committed_out_inserts,
                &self.committed_out_deletes,
                node_idx,
                merged,
            ),
            TraversalDirection::In => merge_committed_overlay_maps(
                &self.committed_in_inserts,
                &self.committed_in_deletes,
                node_idx,
                merged,
            ),
            TraversalDirection::Any => {
                merge_committed_overlay_maps(
                    &self.committed_out_inserts,
                    &self.committed_out_deletes,
                    node_idx,
                    merged,
                );
                merge_committed_overlay_maps(
                    &self.committed_in_inserts,
                    &self.committed_in_deletes,
                    node_idx,
                    merged,
                );
            }
        }
    }

    fn merge_tx_delta(
        &self,
        direction: TraversalDirection,
        node_idx: u32,
        merged: &mut BTreeMap<MergedEdgeKey, LayeredEdge>,
    ) {
        let (inserts, deletes) = tx_delta::weighted_edge_overlay(direction);
        if let Some(deleted) = deletes.get(&node_idx) {
            for &(target, type_id, schema_reversed, relationship_id) in deleted {
                merged.retain(|&(edge_target, edge_type_id, edge_reversed, edge_id), _| {
                    edge_target != target
                        || edge_type_id != type_id
                        || edge_reversed != schema_reversed
                        || (relationship_id.is_some() && edge_id != relationship_id)
                });
            }
        }
        if let Some(inserted) = inserts.get(&node_idx) {
            for edge in inserted {
                merged.insert(
                    (
                        edge.target,
                        edge.type_id,
                        edge.schema_reversed,
                        edge.relationship_id,
                    ),
                    LayeredEdge {
                        type_id: edge.type_id,
                        target: edge.target,
                        schema_reversed: edge.schema_reversed,
                        weight: edge.weight,
                        relationship_id: edge.relationship_id,
                    },
                );
            }
        }
    }

    fn node_visible(&self, node_idx: u32) -> bool {
        if tx_delta::node_deleted(node_idx) {
            return false;
        }
        if !self.projection_node_visible(node_idx) {
            return false;
        }
        match self.tenant_filter {
            Some(tenant_hash) => self
                .tenant_memberships
                .get(&tenant_hash)
                .is_some_and(|members| members.contains(&node_idx)),
            None => true,
        }
    }

    fn projection_node_visible(&self, node_idx: u32) -> bool {
        self.active_nodes
            .get(&node_idx)
            .is_none_or(|active| *active)
    }

    fn fill_directional_neighbors(
        &self,
        direction: TraversalDirection,
        node_idx: u32,
        cursor: &mut OwnedNeighborCursor,
        limit: usize,
        output: &mut Vec<Neighbor>,
    ) -> bool {
        self.fill_directional_neighbors_internal(direction, node_idx, cursor, limit, output, true)
    }

    fn fill_directional_neighbors_internal(
        &self,
        direction: TraversalDirection,
        node_idx: u32,
        cursor: &mut OwnedNeighborCursor,
        limit: usize,
        output: &mut Vec<Neighbor>,
        apply_query_visibility: bool,
    ) -> bool {
        let source_visible = if apply_query_visibility {
            self.node_visible(node_idx)
        } else {
            self.projection_node_visible(node_idx)
        };
        if !source_visible || limit == 0 {
            return true;
        }
        if direction == TraversalDirection::Any {
            return self.fill_any_neighbors(node_idx, cursor, limit, output);
        }
        let (base, chunks, durable, overlay_inserts, overlay_deletes) = match direction {
            TraversalDirection::Out => (
                self.base,
                &*self.base_chunk_out,
                &*self.durable_out,
                &self.committed_out_inserts,
                &self.committed_out_deletes,
            ),
            TraversalDirection::In => (
                self.base_in.unwrap_or(self.base),
                &*self.base_chunk_in,
                &*self.durable_in,
                &self.committed_in_inserts,
                &self.committed_in_deletes,
            ),
            TraversalDirection::Any => unreachable!("Any is handled above"),
        };
        let (targets, type_ids, schema_reversed, relationship_ids) =
            base.neighbors_with_schema_and_relationship_ids(node_idx);
        let base_hidden = direction == TraversalDirection::Out && self.base_chunk_covers(node_idx);
        let chunk = chunks.get(&node_idx);
        let durable = durable.get(&node_idx);
        let overlay = overlay_inserts.get(&node_idx).map(Vec::as_slice);
        let overlay_deleted = overlay_deletes.get(&node_idx);
        let (mut base_pos, mut chunk_pos, mut durable_pos, mut overlay_pos, mut last_key) =
            match cursor {
                OwnedNeighborCursor::Layered {
                    base_pos,
                    chunk_pos,
                    durable_pos,
                    overlay_pos,
                    last_key,
                } => (*base_pos, *chunk_pos, *durable_pos, *overlay_pos, *last_key),
                _ => (0, 0, 0, 0, None),
            };
        let key = |edge: LayeredEdge| {
            (
                edge.target,
                edge.type_id,
                edge.schema_reversed,
                edge.relationship_id,
            )
        };
        let deleted = |set: Option<&HashSet<MergedEdgeKey>>, candidate: MergedEdgeKey| {
            set.is_some_and(|set| {
                set.contains(&candidate)
                    || set.contains(&(candidate.0, candidate.1, candidate.2, None))
            })
        };
        let mut examined = 0usize;
        while examined < limit {
            while !base_hidden
                && direction == TraversalDirection::In
                && base_pos < targets.len()
                && self.base_chunk_covers(targets[base_pos])
                && examined < limit
            {
                base_pos += 1;
                examined += 1;
            }
            if examined == limit {
                break;
            }
            let base_edge = (!base_hidden && base_pos < targets.len()).then(|| LayeredEdge {
                target: targets[base_pos],
                type_id: type_ids.at(base_pos),
                schema_reversed: schema_reversed[base_pos] != 0,
                weight: None,
                relationship_id: relationship_ids
                    .get(base_pos)
                    .copied()
                    .filter(|id| *id != NO_RELATIONSHIP_ID),
            });
            let chunk_edge = chunk
                .and_then(|edges| edges.inserts.get(chunk_pos))
                .copied();
            let durable_edge = durable
                .and_then(|edges| edges.inserts.get(durable_pos))
                .copied();
            let overlay_edge = overlay.and_then(|edges| edges.get(overlay_pos)).map(
                |&(target, type_id, schema_reversed, relationship_id)| LayeredEdge {
                    target,
                    type_id,
                    schema_reversed,
                    weight: None,
                    relationship_id,
                },
            );
            let Some(next_key) = [base_edge, chunk_edge, durable_edge, overlay_edge]
                .into_iter()
                .flatten()
                .map(key)
                .min()
            else {
                *cursor = OwnedNeighborCursor::Layered {
                    base_pos,
                    chunk_pos,
                    durable_pos,
                    overlay_pos,
                    last_key,
                };
                return true;
            };
            let equal_source_count = [base_edge, chunk_edge, durable_edge, overlay_edge]
                .into_iter()
                .flatten()
                .filter(|edge| key(*edge) == next_key)
                .count();
            // Equal keys must advance atomically so precedence remains exact.
            // A call may therefore consume one full four-source key group when
            // `limit` is smaller than that group, but it never starts another
            // group after the remaining raw-row budget is exhausted.
            if examined > 0 && examined.saturating_add(equal_source_count) > limit {
                break;
            }
            let durable_reinserts = durable_edge.is_some_and(|edge| key(edge) == next_key);
            let overlay_reinserts = overlay_edge.is_some_and(|edge| key(edge) == next_key);
            let mut selected = None;
            if base_edge.is_some_and(|edge| key(edge) == next_key) {
                selected = base_edge;
                base_pos += 1;
                examined += 1;
            }
            if chunk_edge.is_some_and(|edge| key(edge) == next_key) {
                selected = chunk_edge;
                chunk_pos += 1;
                examined += 1;
            }
            if durable_edge.is_some_and(|edge| key(edge) == next_key) {
                selected = durable_edge;
                durable_pos += 1;
                examined += 1;
            }
            if overlay_edge.is_some_and(|edge| key(edge) == next_key) {
                selected = overlay_edge;
                overlay_pos += 1;
                examined += 1;
            }
            let hidden_by_durable = deleted(durable.map(|edges| &edges.deletes), next_key);
            let hidden_by_overlay = deleted(overlay_deleted, next_key);
            let survives_overlay = overlay_reinserts || !hidden_by_overlay;
            let survives_durable = overlay_reinserts || durable_reinserts || !hidden_by_durable;
            let first_occurrence = last_key != Some(next_key);
            if first_occurrence && survives_overlay && survives_durable {
                if let Some(edge) = selected.filter(|edge| {
                    if apply_query_visibility {
                        self.node_visible(edge.target)
                    } else {
                        self.projection_node_visible(edge.target)
                    }
                }) {
                    output.push(Neighbor {
                        target: edge.target,
                        type_id: edge.type_id,
                        schema_reversed: edge.schema_reversed,
                        relationship_id: edge.relationship_id,
                    });
                }
            }
            last_key = Some(next_key);
        }
        *cursor = OwnedNeighborCursor::Layered {
            base_pos,
            chunk_pos,
            durable_pos,
            overlay_pos,
            last_key,
        };
        false
    }

    fn fill_any_neighbors(
        &self,
        node_idx: u32,
        cursor: &mut OwnedNeighborCursor,
        limit: usize,
        output: &mut Vec<Neighbor>,
    ) -> bool {
        let Some(base_in) = self.base_in else {
            return false;
        };
        let (out_targets, out_types, out_reversed, out_relationship_ids) = self
            .base
            .neighbors_with_schema_and_relationship_ids(node_idx);
        let (in_targets, in_types, in_reversed, in_relationship_ids) =
            base_in.neighbors_with_schema_and_relationship_ids(node_idx);
        let out_chunk = self.base_chunk_out.get(&node_idx);
        let in_chunk = self.base_chunk_in.get(&node_idx);
        let out_durable = self.durable_out.get(&node_idx);
        let in_durable = self.durable_in.get(&node_idx);
        let out_overlay = self.committed_out_inserts.get(&node_idx).map(Vec::as_slice);
        let in_overlay = self.committed_in_inserts.get(&node_idx).map(Vec::as_slice);
        let (
            mut out_base_pos,
            mut in_base_pos,
            mut out_chunk_pos,
            mut in_chunk_pos,
            mut out_durable_pos,
            mut in_durable_pos,
            mut out_overlay_pos,
            mut in_overlay_pos,
            mut last_key,
        ) = match cursor {
            OwnedNeighborCursor::LayeredAny {
                out_base_pos,
                in_base_pos,
                out_chunk_pos,
                in_chunk_pos,
                out_durable_pos,
                in_durable_pos,
                out_overlay_pos,
                in_overlay_pos,
                last_key,
            } => (
                *out_base_pos,
                *in_base_pos,
                *out_chunk_pos,
                *in_chunk_pos,
                *out_durable_pos,
                *in_durable_pos,
                *out_overlay_pos,
                *in_overlay_pos,
                *last_key,
            ),
            _ => (0, 0, 0, 0, 0, 0, 0, 0, None),
        };
        let key = |edge: LayeredEdge| {
            (
                edge.target,
                edge.type_id,
                edge.schema_reversed,
                edge.relationship_id,
            )
        };
        let deleted = |set: Option<&HashSet<MergedEdgeKey>>, candidate: MergedEdgeKey| {
            set.is_some_and(|set| {
                set.contains(&candidate)
                    || set.contains(&(candidate.0, candidate.1, candidate.2, None))
            })
        };
        let base_edge = |targets: &[u32],
                         types: crate::edge_store::EdgeTypeSlice<'_>,
                         reversed: &[u8],
                         identities: &[RelationshipId],
                         pos: usize| {
            (pos < targets.len()).then(|| LayeredEdge {
                target: targets[pos],
                type_id: types.at(pos),
                schema_reversed: reversed[pos] != 0,
                weight: None,
                relationship_id: identities
                    .get(pos)
                    .copied()
                    .filter(|id| *id != NO_RELATIONSHIP_ID),
            })
        };
        let overlay_edge = |overlay: Option<&[OverlayInsert]>, pos: usize| {
            overlay.and_then(|edges| edges.get(pos)).map(
                |&(target, type_id, schema_reversed, relationship_id)| LayeredEdge {
                    target,
                    type_id,
                    schema_reversed,
                    weight: None,
                    relationship_id,
                },
            )
        };
        let mut examined = 0usize;
        while examined < limit {
            while in_base_pos < in_targets.len()
                && self.base_chunk_covers(in_targets[in_base_pos])
                && examined < limit
            {
                in_base_pos += 1;
                examined += 1;
            }
            if examined == limit {
                break;
            }
            let out_base_edge = (!self.base_chunk_covers(node_idx))
                .then(|| {
                    base_edge(
                        out_targets,
                        out_types,
                        out_reversed,
                        out_relationship_ids,
                        out_base_pos,
                    )
                })
                .flatten();
            let in_base_edge = base_edge(
                in_targets,
                in_types,
                in_reversed,
                in_relationship_ids,
                in_base_pos,
            );
            let out_chunk_edge = out_chunk
                .and_then(|edges| edges.inserts.get(out_chunk_pos))
                .copied();
            let in_chunk_edge = in_chunk
                .and_then(|edges| edges.inserts.get(in_chunk_pos))
                .copied();
            let out_durable_edge = out_durable
                .and_then(|edges| edges.inserts.get(out_durable_pos))
                .copied();
            let in_durable_edge = in_durable
                .and_then(|edges| edges.inserts.get(in_durable_pos))
                .copied();
            let out_overlay_edge = overlay_edge(out_overlay, out_overlay_pos);
            let in_overlay_edge = overlay_edge(in_overlay, in_overlay_pos);
            let sources = [
                out_base_edge,
                in_base_edge,
                out_chunk_edge,
                in_chunk_edge,
                out_durable_edge,
                in_durable_edge,
                out_overlay_edge,
                in_overlay_edge,
            ];
            let Some(next_key) = sources.into_iter().flatten().map(key).min() else {
                *cursor = OwnedNeighborCursor::LayeredAny {
                    out_base_pos,
                    in_base_pos,
                    out_chunk_pos,
                    in_chunk_pos,
                    out_durable_pos,
                    in_durable_pos,
                    out_overlay_pos,
                    in_overlay_pos,
                    last_key,
                };
                return true;
            };
            let equal_source_count = sources
                .into_iter()
                .flatten()
                .filter(|edge| key(*edge) == next_key)
                .count();
            if examined > 0 && examined.saturating_add(equal_source_count) > limit {
                break;
            }
            let mut selected = any_base_layer_winner(
                next_key,
                out_base_edge,
                out_chunk_edge,
                in_base_edge,
                in_chunk_edge,
            );
            if out_base_edge.is_some_and(|edge| key(edge) == next_key) {
                out_base_pos += 1;
                examined += 1;
            }
            if out_chunk_edge.is_some_and(|edge| key(edge) == next_key) {
                out_chunk_pos += 1;
                examined += 1;
            }
            if in_base_edge.is_some_and(|edge| key(edge) == next_key) {
                in_base_pos += 1;
                examined += 1;
            }
            if in_chunk_edge.is_some_and(|edge| key(edge) == next_key) {
                in_chunk_pos += 1;
                examined += 1;
            }
            if deleted(out_durable.map(|edges| &edges.deletes), next_key) {
                selected = None;
            }
            if out_durable_edge.is_some_and(|edge| key(edge) == next_key) {
                selected = out_durable_edge;
                out_durable_pos += 1;
                examined += 1;
            }
            if deleted(in_durable.map(|edges| &edges.deletes), next_key) {
                selected = None;
            }
            if in_durable_edge.is_some_and(|edge| key(edge) == next_key) {
                selected = in_durable_edge;
                in_durable_pos += 1;
                examined += 1;
            }
            if deleted(self.committed_out_deletes.get(&node_idx), next_key) {
                selected = None;
            }
            if out_overlay_edge.is_some_and(|edge| key(edge) == next_key) {
                selected = out_overlay_edge;
                out_overlay_pos += 1;
                examined += 1;
            }
            if deleted(self.committed_in_deletes.get(&node_idx), next_key) {
                selected = None;
            }
            if in_overlay_edge.is_some_and(|edge| key(edge) == next_key) {
                selected = in_overlay_edge;
                in_overlay_pos += 1;
                examined += 1;
            }
            if last_key != Some(next_key) {
                if let Some(edge) = selected.filter(|edge| self.node_visible(edge.target)) {
                    output.push(Neighbor {
                        target: edge.target,
                        type_id: edge.type_id,
                        schema_reversed: edge.schema_reversed,
                        relationship_id: edge.relationship_id,
                    });
                }
            }
            last_key = Some(next_key);
        }
        *cursor = OwnedNeighborCursor::LayeredAny {
            out_base_pos,
            in_base_pos,
            out_chunk_pos,
            in_chunk_pos,
            out_durable_pos,
            in_durable_pos,
            out_overlay_pos,
            in_overlay_pos,
            last_key,
        };
        false
    }

    fn fill_directional_neighbors_reversed(
        &self,
        direction: TraversalDirection,
        node_idx: u32,
        cursor: &mut OwnedNeighborCursor,
        limit: usize,
        output: &mut Vec<Neighbor>,
    ) -> bool {
        if !self.node_visible(node_idx) || limit == 0 {
            return true;
        }
        if direction == TraversalDirection::Any {
            return self.fill_any_neighbors_reversed(node_idx, cursor, limit, output);
        }
        let (base, chunks, durable, overlay_inserts, overlay_deletes) = match direction {
            TraversalDirection::Out => (
                self.base,
                &*self.base_chunk_out,
                &*self.durable_out,
                &self.committed_out_inserts,
                &self.committed_out_deletes,
            ),
            TraversalDirection::In => (
                self.base_in.unwrap_or(self.base),
                &*self.base_chunk_in,
                &*self.durable_in,
                &self.committed_in_inserts,
                &self.committed_in_deletes,
            ),
            TraversalDirection::Any => unreachable!("Any is handled above"),
        };
        let (targets, type_ids, schema_reversed, relationship_ids) =
            base.neighbors_with_schema_and_relationship_ids(node_idx);
        let base_hidden = direction == TraversalDirection::Out && self.base_chunk_covers(node_idx);
        let chunk = chunks.get(&node_idx);
        let durable = durable.get(&node_idx);
        let overlay = overlay_inserts.get(&node_idx).map(Vec::as_slice);
        let overlay_deleted = overlay_deletes.get(&node_idx);
        let (mut bc, mut cc, mut dc, mut oc, mut last_key) = match cursor {
            OwnedNeighborCursor::LayeredReverse {
                base_consumed,
                chunk_consumed,
                durable_consumed,
                overlay_consumed,
                last_key,
            } => (
                *base_consumed,
                *chunk_consumed,
                *durable_consumed,
                *overlay_consumed,
                *last_key,
            ),
            _ => (0, 0, 0, 0, None),
        };
        let key = |edge: LayeredEdge| {
            (
                edge.target,
                edge.type_id,
                edge.schema_reversed,
                edge.relationship_id,
            )
        };
        let deleted = |set: Option<&HashSet<MergedEdgeKey>>, candidate: MergedEdgeKey| {
            set.is_some_and(|set| {
                set.contains(&candidate)
                    || set.contains(&(candidate.0, candidate.1, candidate.2, None))
            })
        };
        let from_end = |edges: &[LayeredEdge], consumed: usize| {
            consumed
                .checked_add(1)
                .and_then(|offset| edges.len().checked_sub(offset))
                .map(|pos| edges[pos])
        };
        let mut examined = 0usize;
        while examined < limit {
            while !base_hidden && direction == TraversalDirection::In && bc < targets.len() {
                let pos = targets.len() - 1 - bc;
                if !self.base_chunk_covers(targets[pos]) || examined == limit {
                    break;
                }
                bc += 1;
                examined += 1;
            }
            if examined == limit {
                break;
            }
            let base_edge = (!base_hidden && bc < targets.len()).then(|| {
                let pos = targets.len() - 1 - bc;
                LayeredEdge {
                    target: targets[pos],
                    type_id: type_ids.at(pos),
                    schema_reversed: schema_reversed[pos] != 0,
                    weight: None,
                    relationship_id: relationship_ids
                        .get(pos)
                        .copied()
                        .filter(|id| *id != NO_RELATIONSHIP_ID),
                }
            });
            let chunk_edge = chunk.and_then(|edges| from_end(&edges.inserts, cc));
            let durable_edge = durable.and_then(|edges| from_end(&edges.inserts, dc));
            let overlay_edge = overlay.and_then(|edges| {
                oc.checked_add(1)
                    .and_then(|offset| edges.len().checked_sub(offset))
                    .map(|pos| {
                        let (target, type_id, schema_reversed, relationship_id) = edges[pos];
                        LayeredEdge {
                            target,
                            type_id,
                            schema_reversed,
                            weight: None,
                            relationship_id,
                        }
                    })
            });
            let sources = [base_edge, chunk_edge, durable_edge, overlay_edge];
            let Some(next_key) = sources.into_iter().flatten().map(key).max() else {
                *cursor = OwnedNeighborCursor::LayeredReverse {
                    base_consumed: bc,
                    chunk_consumed: cc,
                    durable_consumed: dc,
                    overlay_consumed: oc,
                    last_key,
                };
                return true;
            };
            let equal = sources
                .into_iter()
                .flatten()
                .filter(|edge| key(*edge) == next_key)
                .count();
            if examined > 0 && examined.saturating_add(equal) > limit {
                break;
            }
            let durable_reinserts = durable_edge.is_some_and(|edge| key(edge) == next_key);
            let overlay_reinserts = overlay_edge.is_some_and(|edge| key(edge) == next_key);
            let mut selected = None;
            if base_edge.is_some_and(|edge| key(edge) == next_key) {
                selected = base_edge;
                bc += 1;
                examined += 1;
            }
            if chunk_edge.is_some_and(|edge| key(edge) == next_key) {
                selected = chunk_edge;
                cc += 1;
                examined += 1;
            }
            if durable_edge.is_some_and(|edge| key(edge) == next_key) {
                selected = durable_edge;
                dc += 1;
                examined += 1;
            }
            if overlay_edge.is_some_and(|edge| key(edge) == next_key) {
                selected = overlay_edge;
                oc += 1;
                examined += 1;
            }
            let survives_overlay = overlay_reinserts || !deleted(overlay_deleted, next_key);
            let survives_durable = overlay_reinserts
                || durable_reinserts
                || !deleted(durable.map(|edges| &edges.deletes), next_key);
            if last_key != Some(next_key) && survives_overlay && survives_durable {
                if let Some(edge) = selected.filter(|edge| self.node_visible(edge.target)) {
                    output.push(Neighbor {
                        target: edge.target,
                        type_id: edge.type_id,
                        schema_reversed: edge.schema_reversed,
                        relationship_id: edge.relationship_id,
                    });
                }
            }
            last_key = Some(next_key);
        }
        *cursor = OwnedNeighborCursor::LayeredReverse {
            base_consumed: bc,
            chunk_consumed: cc,
            durable_consumed: dc,
            overlay_consumed: oc,
            last_key,
        };
        false
    }

    fn fill_any_neighbors_reversed(
        &self,
        node_idx: u32,
        cursor: &mut OwnedNeighborCursor,
        limit: usize,
        output: &mut Vec<Neighbor>,
    ) -> bool {
        let Some(base_in) = self.base_in else {
            return false;
        };
        let (ot, oy, or, oi) = self
            .base
            .neighbors_with_schema_and_relationship_ids(node_idx);
        let (it, iy, ir, ii) = base_in.neighbors_with_schema_and_relationship_ids(node_idx);
        let oc = self.base_chunk_out.get(&node_idx);
        let ic = self.base_chunk_in.get(&node_idx);
        let od = self.durable_out.get(&node_idx);
        let id = self.durable_in.get(&node_idx);
        let oo = self.committed_out_inserts.get(&node_idx).map(Vec::as_slice);
        let io = self.committed_in_inserts.get(&node_idx).map(Vec::as_slice);
        let (mut ob, mut ib, mut occ, mut icc, mut odd, mut idd, mut ooo, mut ioo, mut last_key) =
            match cursor {
                OwnedNeighborCursor::LayeredAnyReverse {
                    out_base_consumed,
                    in_base_consumed,
                    out_chunk_consumed,
                    in_chunk_consumed,
                    out_durable_consumed,
                    in_durable_consumed,
                    out_overlay_consumed,
                    in_overlay_consumed,
                    last_key,
                } => (
                    *out_base_consumed,
                    *in_base_consumed,
                    *out_chunk_consumed,
                    *in_chunk_consumed,
                    *out_durable_consumed,
                    *in_durable_consumed,
                    *out_overlay_consumed,
                    *in_overlay_consumed,
                    *last_key,
                ),
                _ => (0, 0, 0, 0, 0, 0, 0, 0, None),
            };
        let key = |edge: LayeredEdge| {
            (
                edge.target,
                edge.type_id,
                edge.schema_reversed,
                edge.relationship_id,
            )
        };
        let deleted = |set: Option<&HashSet<MergedEdgeKey>>, candidate: MergedEdgeKey| {
            set.is_some_and(|set| {
                set.contains(&candidate)
                    || set.contains(&(candidate.0, candidate.1, candidate.2, None))
            })
        };
        let base_edge = |targets: &[u32],
                         types: crate::edge_store::EdgeTypeSlice<'_>,
                         reversed: &[u8],
                         ids: &[RelationshipId],
                         consumed: usize| {
            consumed
                .checked_add(1)
                .and_then(|offset| targets.len().checked_sub(offset))
                .map(|pos| LayeredEdge {
                    target: targets[pos],
                    type_id: types.at(pos),
                    schema_reversed: reversed[pos] != 0,
                    weight: None,
                    relationship_id: ids.get(pos).copied().filter(|id| *id != NO_RELATIONSHIP_ID),
                })
        };
        let stored = |edges: Option<&DurableEdges>, consumed: usize| {
            edges.and_then(|edges| {
                consumed
                    .checked_add(1)
                    .and_then(|offset| edges.inserts.len().checked_sub(offset))
                    .map(|pos| edges.inserts[pos])
            })
        };
        let overlay = |edges: Option<&[OverlayInsert]>, consumed: usize| {
            edges.and_then(|edges| {
                consumed
                    .checked_add(1)
                    .and_then(|offset| edges.len().checked_sub(offset))
                    .map(|pos| {
                        let (target, type_id, schema_reversed, relationship_id) = edges[pos];
                        LayeredEdge {
                            target,
                            type_id,
                            schema_reversed,
                            weight: None,
                            relationship_id,
                        }
                    })
            })
        };
        let mut examined = 0usize;
        while examined < limit {
            while ib < it.len() {
                let pos = it.len() - 1 - ib;
                if !self.base_chunk_covers(it[pos]) || examined == limit {
                    break;
                }
                ib += 1;
                examined += 1;
            }
            if examined == limit {
                break;
            }
            let obe = (!self.base_chunk_covers(node_idx))
                .then(|| base_edge(ot, oy, or, oi, ob))
                .flatten();
            let ibe = base_edge(it, iy, ir, ii, ib);
            let oce = stored(oc, occ);
            let ice = stored(ic, icc);
            let ode = stored(od, odd);
            let ide = stored(id, idd);
            let ooe = overlay(oo, ooo);
            let ioe = overlay(io, ioo);
            let sources = [obe, ibe, oce, ice, ode, ide, ooe, ioe];
            let Some(next_key) = sources.into_iter().flatten().map(key).max() else {
                *cursor = OwnedNeighborCursor::LayeredAnyReverse {
                    out_base_consumed: ob,
                    in_base_consumed: ib,
                    out_chunk_consumed: occ,
                    in_chunk_consumed: icc,
                    out_durable_consumed: odd,
                    in_durable_consumed: idd,
                    out_overlay_consumed: ooo,
                    in_overlay_consumed: ioo,
                    last_key,
                };
                return true;
            };
            let equal = sources
                .into_iter()
                .flatten()
                .filter(|edge| key(*edge) == next_key)
                .count();
            if examined > 0 && examined.saturating_add(equal) > limit {
                break;
            }
            let mut selected = any_base_layer_winner(next_key, obe, oce, ibe, ice);
            if obe.is_some_and(|edge| key(edge) == next_key) {
                ob += 1;
                examined += 1;
            }
            if oce.is_some_and(|edge| key(edge) == next_key) {
                occ += 1;
                examined += 1;
            }
            if ibe.is_some_and(|edge| key(edge) == next_key) {
                ib += 1;
                examined += 1;
            }
            if ice.is_some_and(|edge| key(edge) == next_key) {
                icc += 1;
                examined += 1;
            }
            if deleted(od.map(|edges| &edges.deletes), next_key) {
                selected = None;
            }
            if ode.is_some_and(|edge| key(edge) == next_key) {
                selected = ode;
                odd += 1;
                examined += 1;
            }
            if deleted(id.map(|edges| &edges.deletes), next_key) {
                selected = None;
            }
            if ide.is_some_and(|edge| key(edge) == next_key) {
                selected = ide;
                idd += 1;
                examined += 1;
            }
            if deleted(self.committed_out_deletes.get(&node_idx), next_key) {
                selected = None;
            }
            if ooe.is_some_and(|edge| key(edge) == next_key) {
                selected = ooe;
                ooo += 1;
                examined += 1;
            }
            if deleted(self.committed_in_deletes.get(&node_idx), next_key) {
                selected = None;
            }
            if ioe.is_some_and(|edge| key(edge) == next_key) {
                selected = ioe;
                ioo += 1;
                examined += 1;
            }
            if last_key != Some(next_key) {
                if let Some(edge) = selected.filter(|edge| self.node_visible(edge.target)) {
                    output.push(Neighbor {
                        target: edge.target,
                        type_id: edge.type_id,
                        schema_reversed: edge.schema_reversed,
                        relationship_id: edge.relationship_id,
                    });
                }
            }
            last_key = Some(next_key);
        }
        *cursor = OwnedNeighborCursor::LayeredAnyReverse {
            out_base_consumed: ob,
            in_base_consumed: ib,
            out_chunk_consumed: occ,
            in_chunk_consumed: icc,
            out_durable_consumed: odd,
            in_durable_consumed: idd,
            out_overlay_consumed: ooo,
            in_overlay_consumed: ioo,
            last_key,
        };
        false
    }
}

impl NeighborSource for LayeredNeighbors<'_> {
    fn neighbors(&self, node_idx: u32) -> NeighborIter<'_> {
        NeighborIter::Owned(
            self.merged_neighbors(TraversalDirection::Out, node_idx, false)
                .into_iter()
                .map(|(target, edge)| Neighbor {
                    target,
                    type_id: edge.type_id,
                    schema_reversed: edge.schema_reversed,
                    relationship_id: edge.relationship_id,
                })
                .collect::<Vec<_>>()
                .into_iter(),
        )
    }

    fn neighbors_reversed(&self, node_idx: u32) -> NeighborIter<'_> {
        NeighborIter::Owned(
            self.merged_neighbors(TraversalDirection::Out, node_idx, true)
                .into_iter()
                .map(|(target, edge)| Neighbor {
                    target,
                    type_id: edge.type_id,
                    schema_reversed: edge.schema_reversed,
                    relationship_id: edge.relationship_id,
                })
                .collect::<Vec<_>>()
                .into_iter(),
        )
    }

    fn fill_neighbors(
        &self,
        node_idx: u32,
        cursor: &mut OwnedNeighborCursor,
        limit: usize,
        output: &mut Vec<Neighbor>,
    ) -> bool {
        self.fill_directional_neighbors(TraversalDirection::Out, node_idx, cursor, limit, output)
    }

    fn fill_neighbors_reversed(
        &self,
        node_idx: u32,
        cursor: &mut OwnedNeighborCursor,
        limit: usize,
        output: &mut Vec<Neighbor>,
    ) -> bool {
        self.fill_directional_neighbors_reversed(
            TraversalDirection::Out,
            node_idx,
            cursor,
            limit,
            output,
        )
    }
}

/// Direction-specific neighbor source backed by one layered snapshot.
pub(crate) struct DirectionalLayeredNeighbors<'a, 'b> {
    layered: &'a LayeredNeighbors<'b>,
    direction: TraversalDirection,
}

impl NeighborSource for DirectionalLayeredNeighbors<'_, '_> {
    fn neighbors(&self, node_idx: u32) -> NeighborIter<'_> {
        NeighborIter::Owned(
            self.layered
                .merged_neighbors(self.direction, node_idx, false)
                .into_iter()
                .map(|(target, edge)| Neighbor {
                    target,
                    type_id: edge.type_id,
                    schema_reversed: edge.schema_reversed,
                    relationship_id: edge.relationship_id,
                })
                .collect::<Vec<_>>()
                .into_iter(),
        )
    }

    fn neighbors_reversed(&self, node_idx: u32) -> NeighborIter<'_> {
        NeighborIter::Owned(
            self.layered
                .merged_neighbors(self.direction, node_idx, true)
                .into_iter()
                .map(|(target, edge)| Neighbor {
                    target,
                    type_id: edge.type_id,
                    schema_reversed: edge.schema_reversed,
                    relationship_id: edge.relationship_id,
                })
                .collect::<Vec<_>>()
                .into_iter(),
        )
    }

    fn fill_neighbors(
        &self,
        node_idx: u32,
        cursor: &mut OwnedNeighborCursor,
        limit: usize,
        output: &mut Vec<Neighbor>,
    ) -> bool {
        self.layered
            .fill_directional_neighbors(self.direction, node_idx, cursor, limit, output)
    }

    fn fill_neighbors_reversed(
        &self,
        node_idx: u32,
        cursor: &mut OwnedNeighborCursor,
        limit: usize,
        output: &mut Vec<Neighbor>,
    ) -> bool {
        self.layered.fill_directional_neighbors_reversed(
            self.direction,
            node_idx,
            cursor,
            limit,
            output,
        )
    }
}

impl WeightedNeighborSource for LayeredNeighbors<'_> {
    fn has_weighted_edges(&self) -> bool {
        self.base.has_weights()
            || self
                .base_chunk_out
                .values()
                .any(|edges| edges.inserts.iter().any(|edge| edge.weight.is_some()))
            || self
                .durable_out
                .values()
                .any(|edges| edges.inserts.iter().any(|edge| edge.weight.is_some()))
            || tx_delta::weighted_edge_overlay(TraversalDirection::Out)
                .0
                .values()
                .any(|edges| edges.iter().any(|edge| edge.weight.is_some()))
    }

    fn weighted_neighbors(&self, node_idx: u32) -> Vec<WeightedNeighbor> {
        self.weighted_neighbors(node_idx)
    }

    fn fill_weighted_neighbors(
        &self,
        node_idx: u32,
        cursor: &mut OwnedNeighborCursor,
        limit: usize,
        output: &mut Vec<WeightedNeighbor>,
    ) -> bool {
        self.fill_weighted_out_neighbors(node_idx, cursor, limit, output)
    }
}

fn base_weight_slice(base: &EdgeStore, node_idx: u32) -> Option<&[u32]> {
    let (_, _, weights) = base.neighbors_weighted(node_idx);
    (!weights.is_empty()).then_some(weights)
}

fn merge_durable_map(
    durable: &HashMap<u32, DurableEdges>,
    node_idx: u32,
    merged: &mut BTreeMap<MergedEdgeKey, LayeredEdge>,
) {
    if let Some(edges) = durable.get(&node_idx) {
        for &(target, type_id, schema_reversed, relationship_id) in &edges.deletes {
            if let Some(relationship_id) = relationship_id {
                merged.remove(&(target, type_id, schema_reversed, Some(relationship_id)));
            } else {
                merged.retain(|&(edge_target, edge_type_id, edge_reversed, _), _| {
                    edge_target != target
                        || edge_type_id != type_id
                        || edge_reversed != schema_reversed
                });
            }
        }
        for edge in &edges.inserts {
            merged.insert(
                (
                    edge.target,
                    edge.type_id,
                    edge.schema_reversed,
                    edge.relationship_id,
                ),
                *edge,
            );
        }
    }
}

fn merge_committed_overlay_maps(
    inserts: &OverlayInserts,
    deletes: &OverlayDeletes,
    node_idx: u32,
    merged: &mut BTreeMap<MergedEdgeKey, LayeredEdge>,
) {
    if let Some(deleted) = deletes.get(&node_idx) {
        for &(target, type_id, schema_reversed, relationship_id) in deleted {
            merged.retain(|&(edge_target, edge_type_id, edge_reversed, edge_id), _| {
                edge_target != target
                    || edge_type_id != type_id
                    || edge_reversed != schema_reversed
                    || (relationship_id.is_some() && edge_id != relationship_id)
            });
        }
    }
    if let Some(inserted) = inserts.get(&node_idx) {
        for &(target, type_id, schema_reversed, relationship_id) in inserted {
            merged.insert(
                (target, type_id, schema_reversed, relationship_id),
                LayeredEdge {
                    target,
                    type_id,
                    schema_reversed,
                    weight: None,
                    relationship_id,
                },
            );
        }
    }
}

struct LayeredBuilder<'a> {
    base: &'a EdgeStore,
    out_edges: HashMap<EdgeKey, DurableEdgeState>,
    in_edges: HashMap<EdgeKey, DurableEdgeState>,
    active_nodes: HashMap<u32, bool>,
    tenant_memberships: HashMap<u64, HashSet<u32>>,
}

struct LayeredBuildOutput {
    durable_out: HashMap<u32, DurableEdges>,
    durable_in: HashMap<u32, DurableEdges>,
    active_nodes: HashMap<u32, bool>,
    tenant_memberships: HashMap<u64, HashSet<u32>>,
}

impl LayeredBuilderCapacity {
    fn try_from_segments(segments: &[&DeltaSegment]) -> GraphResult<Self> {
        let mut capacity = Self {
            segment_count: segments.len(),
            ..Self::default()
        };
        for segment in segments {
            let mutations = segment
                .edge_inserts
                .len()
                .checked_add(segment.edge_deletes.len())
                .ok_or_else(layered_size_overflow)?;
            capacity.edge_mutations = capacity
                .edge_mutations
                .checked_add(mutations)
                .ok_or_else(layered_size_overflow)?;
            capacity.max_edge_weights = capacity.max_edge_weights.max(segment.edge_weights.len());
            capacity.node_states = capacity
                .node_states
                .checked_add(segment.node_states.len())
                .ok_or_else(layered_size_overflow)?;
            capacity.tenant_rows = capacity
                .tenant_rows
                .checked_add(segment.tenants.len())
                .ok_or_else(layered_size_overflow)?;
        }
        Ok(capacity)
    }

    fn estimated_heap_bytes(self) -> GraphResult<usize> {
        let ordered = checked_bytes(self.segment_count, std::mem::size_of::<&DeltaSegment>())?;
        let direction_entry = std::mem::size_of::<(EdgeKey, DurableEdgeState)>();
        let intermediate_edges = checked_bytes(self.edge_mutations, direction_entry)?
            .checked_mul(4)
            .ok_or_else(layered_size_overflow)?;
        let output_map_entry = std::mem::size_of::<(u32, DurableEdges)>();
        let output_maps = checked_bytes(self.edge_mutations, output_map_entry)?
            .checked_mul(4)
            .ok_or_else(layered_size_overflow)?;
        let insert_payloads =
            checked_bytes(self.edge_mutations, std::mem::size_of::<LayeredEdge>())?
                .checked_mul(2)
                .ok_or_else(layered_size_overflow)?;
        let delete_payloads = checked_bytes(
            self.edge_mutations,
            std::mem::size_of::<(u32, EdgeTypeId, bool, Option<RelationshipId>)>(),
        )?
        .checked_mul(4)
        .ok_or_else(layered_size_overflow)?;
        let weight_index =
            checked_bytes(self.max_edge_weights, std::mem::size_of::<(EdgeKey, u32)>())?
                .checked_mul(2)
                .ok_or_else(layered_size_overflow)?;
        let node_index = checked_bytes(self.node_states, std::mem::size_of::<(u32, bool)>())?
            .checked_mul(2)
            .ok_or_else(layered_size_overflow)?;
        let tenant_index =
            checked_bytes(self.tenant_rows, std::mem::size_of::<(u64, HashSet<u32>)>())?
                .checked_mul(2)
                .and_then(|bytes| {
                    checked_bytes(self.tenant_rows, std::mem::size_of::<u32>())
                        .ok()?
                        .checked_mul(2)
                        .and_then(|members| bytes.checked_add(members))
                })
                .ok_or_else(layered_size_overflow)?;
        [
            ordered,
            intermediate_edges,
            output_maps,
            insert_payloads,
            delete_payloads,
            weight_index,
            node_index,
            tenant_index,
        ]
        .into_iter()
        .try_fold(0usize, |total, bytes| {
            total.checked_add(bytes).ok_or_else(layered_size_overflow)
        })
    }
}

fn checked_bytes(count: usize, item_size: usize) -> GraphResult<usize> {
    count
        .checked_mul(item_size)
        .ok_or_else(layered_size_overflow)
}

fn layered_size_overflow() -> GraphError {
    GraphError::Internal("layered snapshot allocation size overflowed usize".to_string())
}

fn layered_allocation_error(_error: std::collections::TryReserveError) -> GraphError {
    GraphError::Oom {
        used_mb: 0,
        need_mb: 1,
        limit_mb: crate::config::MEMORY_LIMIT_MB.get().max(1) as u64,
    }
}

fn try_tenant_memberships(
    segments: &[&DeltaSegment],
    tenant_rows: usize,
) -> GraphResult<HashMap<u64, HashSet<u32>>> {
    let mut counts = HashMap::<u64, usize>::new();
    counts
        .try_reserve(tenant_rows)
        .map_err(layered_allocation_error)?;
    for tenant in segments.iter().flat_map(|segment| &segment.tenants) {
        let count = counts.entry(tenant.tenant_hash).or_default();
        *count = count.checked_add(1).ok_or_else(layered_size_overflow)?;
    }
    let mut memberships = HashMap::new();
    memberships
        .try_reserve(counts.len())
        .map_err(layered_allocation_error)?;
    for (tenant_hash, count) in counts {
        let mut members = HashSet::new();
        members
            .try_reserve(count)
            .map_err(layered_allocation_error)?;
        memberships.insert(tenant_hash, members);
    }
    Ok(memberships)
}

impl<'a> LayeredBuilder<'a> {
    fn new(base: &'a EdgeStore) -> Self {
        Self {
            base,
            out_edges: HashMap::new(),
            in_edges: HashMap::new(),
            active_nodes: HashMap::new(),
            tenant_memberships: HashMap::new(),
        }
    }

    fn try_new(
        base: &'a EdgeStore,
        capacity: LayeredBuilderCapacity,
        segments: &[&DeltaSegment],
    ) -> GraphResult<Self> {
        let mut out_edges = HashMap::new();
        out_edges
            .try_reserve(capacity.edge_mutations)
            .map_err(layered_allocation_error)?;
        let mut in_edges = HashMap::new();
        in_edges
            .try_reserve(capacity.edge_mutations)
            .map_err(layered_allocation_error)?;
        let mut active_nodes = HashMap::new();
        active_nodes
            .try_reserve(capacity.node_states)
            .map_err(layered_allocation_error)?;
        let tenant_memberships = try_tenant_memberships(segments, capacity.tenant_rows)?;
        Ok(Self {
            base,
            out_edges,
            in_edges,
            active_nodes,
            tenant_memberships,
        })
    }

    fn apply_segments(&mut self, segments: &[DeltaSegment]) {
        let refs = segments.iter().collect::<Vec<_>>();
        self.apply_segment_refs(&refs);
    }

    fn apply_segment_refs(&mut self, segments: &[&DeltaSegment]) {
        let mut ordered = segments.to_vec();
        ordered.sort_by_key(|segment| {
            (
                segment.header.sync_watermark,
                segment.header.level,
                segment.header.source_start,
                segment.header.source_end,
            )
        });
        for segment in ordered {
            match segment.header.kind {
                SegmentKind::Edge => self.apply_edge_segment(segment),
                SegmentKind::Node => self.apply_node_segment(segment),
            }
        }
    }

    fn try_apply_segment_refs(&mut self, segments: &[&DeltaSegment]) -> GraphResult<()> {
        let mut ordered = Vec::new();
        ordered
            .try_reserve_exact(segments.len())
            .map_err(layered_allocation_error)?;
        ordered.extend_from_slice(segments);
        ordered.sort_by_key(|segment| {
            (
                segment.header.sync_watermark,
                segment.header.level,
                segment.header.source_start,
                segment.header.source_end,
            )
        });
        for segment in ordered {
            match segment.header.kind {
                SegmentKind::Edge => self.try_apply_edge_segment(segment)?,
                SegmentKind::Node => self.apply_node_segment(segment),
            }
        }
        Ok(())
    }

    fn apply_edge_segment(&mut self, segment: &DeltaSegment) {
        let weights = segment
            .edge_weights
            .iter()
            .map(|row| {
                (
                    EdgeKey {
                        source: row.source,
                        target: row.target,
                        type_id: row.type_id,
                        schema_reversed: row.schema_reversed,
                        relationship_id: row.relationship_id,
                    },
                    row.weight,
                )
            })
            .collect::<HashMap<_, _>>();
        for edge in &segment.edge_deletes {
            self.remove_edge(
                segment.header.direction,
                edge.source,
                edge.target,
                edge.type_id,
                edge.schema_reversed,
                edge.relationship_id,
            );
        }
        for edge in &segment.edge_inserts {
            let layered_edge = LayeredEdge {
                target: edge.target,
                type_id: edge.type_id,
                schema_reversed: edge.schema_reversed,
                weight: weights
                    .get(&EdgeKey {
                        source: edge.source,
                        target: edge.target,
                        type_id: edge.type_id,
                        schema_reversed: edge.schema_reversed,
                        relationship_id: edge.relationship_id,
                    })
                    .copied(),
                relationship_id: edge.relationship_id,
            };
            self.insert_edge(segment.header.direction, edge.source, layered_edge);
        }
    }

    fn try_apply_edge_segment(&mut self, segment: &DeltaSegment) -> GraphResult<()> {
        let mut weights = HashMap::new();
        weights
            .try_reserve(segment.edge_weights.len())
            .map_err(layered_allocation_error)?;
        weights.extend(segment.edge_weights.iter().map(|row| {
            (
                EdgeKey {
                    source: row.source,
                    target: row.target,
                    type_id: row.type_id,
                    schema_reversed: row.schema_reversed,
                    relationship_id: row.relationship_id,
                },
                row.weight,
            )
        }));
        for edge in &segment.edge_deletes {
            self.remove_edge(
                segment.header.direction,
                edge.source,
                edge.target,
                edge.type_id,
                edge.schema_reversed,
                edge.relationship_id,
            );
        }
        for edge in &segment.edge_inserts {
            let layered_edge = LayeredEdge {
                target: edge.target,
                type_id: edge.type_id,
                schema_reversed: edge.schema_reversed,
                weight: weights
                    .get(&EdgeKey {
                        source: edge.source,
                        target: edge.target,
                        type_id: edge.type_id,
                        schema_reversed: edge.schema_reversed,
                        relationship_id: edge.relationship_id,
                    })
                    .copied(),
                relationship_id: edge.relationship_id,
            };
            self.insert_edge(segment.header.direction, edge.source, layered_edge);
        }
        Ok(())
    }

    fn apply_node_segment(&mut self, segment: &DeltaSegment) {
        for node in &segment.node_states {
            self.active_nodes.insert(node.node_idx, node.active);
        }
        for tenant in &segment.tenants {
            let members = self
                .tenant_memberships
                .entry(tenant.tenant_hash)
                .or_default();
            if tenant.tombstone {
                members.remove(&tenant.node_idx);
            } else {
                members.insert(tenant.node_idx);
            }
        }
    }

    fn insert_edge(&mut self, direction: TraversalDirection, source: u32, edge: LayeredEdge) {
        match direction {
            TraversalDirection::Out => {
                self.out_edges.insert(
                    EdgeKey {
                        source,
                        target: edge.target,
                        type_id: edge.type_id,
                        schema_reversed: edge.schema_reversed,
                        relationship_id: edge.relationship_id,
                    },
                    DurableEdgeState::Present {
                        weight: edge.weight,
                        relationship_id: edge.relationship_id,
                    },
                );
                self.in_edges.insert(
                    EdgeKey {
                        source: edge.target,
                        target: source,
                        type_id: edge.type_id,
                        schema_reversed: edge.schema_reversed,
                        relationship_id: edge.relationship_id,
                    },
                    DurableEdgeState::Present {
                        weight: edge.weight,
                        relationship_id: edge.relationship_id,
                    },
                );
            }
            TraversalDirection::In => {
                self.in_edges.insert(
                    EdgeKey {
                        source,
                        target: edge.target,
                        type_id: edge.type_id,
                        schema_reversed: edge.schema_reversed,
                        relationship_id: edge.relationship_id,
                    },
                    DurableEdgeState::Present {
                        weight: edge.weight,
                        relationship_id: edge.relationship_id,
                    },
                );
                self.out_edges.insert(
                    EdgeKey {
                        source: edge.target,
                        target: source,
                        type_id: edge.type_id,
                        schema_reversed: edge.schema_reversed,
                        relationship_id: edge.relationship_id,
                    },
                    DurableEdgeState::Present {
                        weight: edge.weight,
                        relationship_id: edge.relationship_id,
                    },
                );
            }
            TraversalDirection::Any => {
                self.insert_edge(TraversalDirection::Out, source, edge);
                self.insert_edge(
                    TraversalDirection::In,
                    edge.target,
                    LayeredEdge {
                        target: source,
                        ..edge
                    },
                );
            }
        }
    }

    fn remove_edge(
        &mut self,
        direction: TraversalDirection,
        source: u32,
        target: u32,
        type_id: EdgeTypeId,
        schema_reversed: bool,
        relationship_id: Option<RelationshipId>,
    ) {
        match direction {
            TraversalDirection::Out => {
                record_edge_delete(
                    &mut self.out_edges,
                    EdgeKey {
                        source,
                        target,
                        type_id,
                        schema_reversed,
                        relationship_id,
                    },
                );
                record_edge_delete(
                    &mut self.in_edges,
                    EdgeKey {
                        source: target,
                        target: source,
                        type_id,
                        schema_reversed,
                        relationship_id,
                    },
                );
            }
            TraversalDirection::In => {
                record_edge_delete(
                    &mut self.in_edges,
                    EdgeKey {
                        source,
                        target,
                        type_id,
                        schema_reversed,
                        relationship_id,
                    },
                );
                record_edge_delete(
                    &mut self.out_edges,
                    EdgeKey {
                        source: target,
                        target: source,
                        type_id,
                        schema_reversed,
                        relationship_id,
                    },
                );
            }
            TraversalDirection::Any => {
                self.remove_edge(
                    TraversalDirection::Out,
                    source,
                    target,
                    type_id,
                    schema_reversed,
                    relationship_id,
                );
                self.remove_edge(
                    TraversalDirection::In,
                    target,
                    source,
                    type_id,
                    schema_reversed,
                    relationship_id,
                );
            }
        }
    }

    fn finish(self) -> LayeredBuildOutput {
        self.finish_with_base_duplicate_filter(true)
    }

    fn finish_replacement(self) -> LayeredBuildOutput {
        self.finish_with_base_duplicate_filter(false)
    }

    fn try_finish(self) -> GraphResult<LayeredBuildOutput> {
        self.try_finish_with_base_duplicate_filter(true)
    }

    fn try_finish_replacement(self) -> GraphResult<LayeredBuildOutput> {
        self.try_finish_with_base_duplicate_filter(false)
    }

    fn try_finish_with_base_duplicate_filter(
        self,
        suppress_base_duplicates: bool,
    ) -> GraphResult<LayeredBuildOutput> {
        let Self {
            base,
            out_edges,
            in_edges,
            active_nodes,
            tenant_memberships,
        } = self;
        Ok(LayeredBuildOutput {
            durable_out: try_finish_direction(base, out_edges, suppress_base_duplicates)?,
            durable_in: try_finish_direction(base, in_edges, suppress_base_duplicates)?,
            active_nodes,
            tenant_memberships,
        })
    }

    fn finish_with_base_duplicate_filter(
        self,
        suppress_base_duplicates: bool,
    ) -> LayeredBuildOutput {
        let Self {
            base,
            out_edges,
            in_edges,
            active_nodes,
            tenant_memberships,
        } = self;
        LayeredBuildOutput {
            durable_out: finish_direction(base, out_edges, suppress_base_duplicates),
            durable_in: finish_direction(base, in_edges, suppress_base_duplicates),
            active_nodes,
            tenant_memberships,
        }
    }
}

fn finish_direction(
    base: &EdgeStore,
    edges: HashMap<EdgeKey, DurableEdgeState>,
    suppress_base_duplicates: bool,
) -> HashMap<u32, DurableEdges> {
    let mut ordered = edges.into_iter().collect::<Vec<_>>();
    ordered.sort_by_key(|(key, _)| *key);
    let mut out = HashMap::<u32, DurableEdges>::new();
    for (key, state) in ordered {
        match state {
            DurableEdgeState::Present {
                weight,
                relationship_id,
            } => {
                if suppress_base_duplicates && base_edge_exists(base, key) && weight.is_none() {
                    continue;
                }
                out.entry(key.source)
                    .or_default()
                    .inserts
                    .push(LayeredEdge {
                        target: key.target,
                        type_id: key.type_id,
                        schema_reversed: key.schema_reversed,
                        weight,
                        relationship_id,
                    });
            }
            DurableEdgeState::Deleted => {
                out.entry(key.source).or_default().deletes.insert((
                    key.target,
                    key.type_id,
                    key.schema_reversed,
                    key.relationship_id,
                ));
            }
        }
    }
    out
}

fn try_finish_direction(
    base: &EdgeStore,
    edges: HashMap<EdgeKey, DurableEdgeState>,
    suppress_base_duplicates: bool,
) -> GraphResult<HashMap<u32, DurableEdges>> {
    let mut ordered = Vec::new();
    ordered
        .try_reserve_exact(edges.len())
        .map_err(layered_allocation_error)?;
    ordered.extend(edges);
    ordered.sort_by_key(|(key, _)| *key);

    let mut counts = HashMap::<u32, (usize, usize)>::new();
    counts
        .try_reserve(ordered.len())
        .map_err(layered_allocation_error)?;
    for (key, state) in &ordered {
        if suppress_base_duplicates
            && matches!(state, DurableEdgeState::Present { weight: None, .. })
            && base_edge_exists(base, *key)
        {
            continue;
        }
        let count = counts.entry(key.source).or_default();
        match state {
            DurableEdgeState::Present { .. } => count.0 += 1,
            DurableEdgeState::Deleted => count.1 += 1,
        }
    }
    let mut out = HashMap::<u32, DurableEdges>::new();
    out.try_reserve(counts.len())
        .map_err(layered_allocation_error)?;
    for (source, (insert_count, delete_count)) in counts {
        let mut inserts = Vec::new();
        inserts
            .try_reserve_exact(insert_count)
            .map_err(layered_allocation_error)?;
        let mut deletes = HashSet::new();
        deletes
            .try_reserve(delete_count)
            .map_err(layered_allocation_error)?;
        out.insert(source, DurableEdges { inserts, deletes });
    }
    for (key, state) in ordered {
        match state {
            DurableEdgeState::Present {
                weight,
                relationship_id,
            } => {
                if suppress_base_duplicates && base_edge_exists(base, key) && weight.is_none() {
                    continue;
                }
                let Some(edges) = out.get_mut(&key.source) else {
                    return Err(GraphError::Internal(
                        "layered output preflight omitted an insert source".to_string(),
                    ));
                };
                edges.inserts.push(LayeredEdge {
                    target: key.target,
                    type_id: key.type_id,
                    schema_reversed: key.schema_reversed,
                    weight,
                    relationship_id,
                });
            }
            DurableEdgeState::Deleted => {
                let Some(edges) = out.get_mut(&key.source) else {
                    return Err(GraphError::Internal(
                        "layered output preflight omitted a delete source".to_string(),
                    ));
                };
                edges.deletes.insert((
                    key.target,
                    key.type_id,
                    key.schema_reversed,
                    key.relationship_id,
                ));
            }
        }
    }
    Ok(out)
}

fn record_edge_delete(edges: &mut HashMap<EdgeKey, DurableEdgeState>, key: EdgeKey) {
    if key.relationship_id.is_none() {
        edges.retain(|existing, _| {
            existing.source != key.source
                || existing.target != key.target
                || existing.type_id != key.type_id
                || existing.schema_reversed != key.schema_reversed
        });
    }
    edges.insert(key, DurableEdgeState::Deleted);
}

fn base_edge_exists(base: &EdgeStore, key: EdgeKey) -> bool {
    let (targets, type_ids, schema_reversed, relationship_ids) =
        base.neighbors_with_schema_and_relationship_ids(key.source);
    targets
        .iter()
        .zip(type_ids.iter())
        .zip(schema_reversed.iter())
        .enumerate()
        .any(|(idx, ((&target, type_id), &schema_reversed))| {
            target == key.target
                && type_id == key.type_id
                && (schema_reversed != 0) == key.schema_reversed
                && relationship_ids
                    .get(idx)
                    .copied()
                    .filter(|&id| id != NO_RELATIONSHIP_ID)
                    == key.relationship_id
        })
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::edge_store::{IdentifiedRawEdge, RawEdge, SortedEdgeStoreBuilder};
    use crate::projection::manifest::{ManifestSegmentRef, ProjectionManifest};
    use crate::projection::neighbors::{
        CsrNeighbors, Neighbor, OwnedNeighborCursor, WeightedNeighbor,
    };
    use crate::projection::segment::{
        SegmentEdge, SegmentEdgeWeight, SegmentNodeState, SegmentTenant,
    };
    use crate::projection::test_fixtures::{
        assert_full_csr_equivalence, edge_store_from_tuples, weighted_edge_store_from_tuples,
        ProjectionArtifactDir,
    };
    use proptest::prelude::*;

    fn collect_layered_pages(
        neighbors: &impl NeighborSource,
        node_idx: u32,
        page_size: usize,
    ) -> Vec<Neighbor> {
        let mut cursor = OwnedNeighborCursor::default();
        let mut output = Vec::new();
        let mut pages = 0usize;
        loop {
            pages += 1;
            assert!(pages <= 1_000, "layered pager must make progress");
            let mut page = Vec::new();
            let exhausted = neighbors.fill_neighbors(node_idx, &mut cursor, page_size, &mut page);
            assert!(
                page.len() <= page_size,
                "one layered page exceeded its row bound"
            );
            output.extend(page);
            if exhausted {
                return output;
            }
        }
    }

    fn collect_layered_pages_requiring_progress(
        neighbors: &impl NeighborSource,
        node_idx: u32,
        page_size: usize,
    ) -> Vec<Neighbor> {
        let mut cursor = OwnedNeighborCursor::default();
        let mut output = Vec::new();
        for _ in 0..1_000 {
            let before = cursor.clone();
            let mut page = Vec::new();
            let exhausted = neighbors.fill_neighbors(node_idx, &mut cursor, page_size, &mut page);
            assert!(
                exhausted || cursor != before,
                "a non-exhausted layered page must advance bounded owned state"
            );
            assert!(
                page.len() <= page_size,
                "one layered page exceeded its output-row bound"
            );
            output.extend(page);
            if exhausted {
                return output;
            }
        }
        panic!("layered pager did not exhaust within its bounded fixture")
    }

    fn collect_layered_reverse_pages(
        neighbors: &impl NeighborSource,
        node_idx: u32,
        page_size: usize,
    ) -> Vec<Neighbor> {
        let mut cursor = OwnedNeighborCursor::default();
        let mut output = Vec::new();
        for _ in 0..1_000 {
            let before = cursor.clone();
            let mut page = Vec::new();
            let exhausted =
                neighbors.fill_neighbors_reversed(node_idx, &mut cursor, page_size, &mut page);
            assert!(exhausted || cursor != before, "reverse pager must advance");
            assert!(page.len() <= page_size, "reverse page exceeded row bound");
            output.extend(page);
            if exhausted {
                return output;
            }
        }
        panic!("reverse layered pager failed to finish")
    }

    #[test]
    fn owned_layered_cursor_preserves_tombstone_precedence_and_parallel_relationships() {
        let base = edge_store_from_tuples(5, &[(0, 1, 1), (1, 0, 1)]);
        let mut older_out =
            DeltaSegment::new(SegmentKind::Edge, 0, TraversalDirection::Out, 0, 5, 1)
                .expect("older outbound segment");
        older_out.edge_inserts.extend([
            SegmentEdge {
                source: 0,
                target: 2,
                type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
                schema_reversed: false,
                relationship_id: Some(10),
            },
            SegmentEdge {
                source: 0,
                target: 2,
                type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
                schema_reversed: false,
                relationship_id: Some(11),
            },
        ]);
        let mut newer_out =
            DeltaSegment::new(SegmentKind::Edge, 0, TraversalDirection::Out, 0, 5, 2)
                .expect("newer outbound segment");
        newer_out.edge_deletes.extend([
            SegmentEdge {
                source: 0,
                target: 1,
                type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
                schema_reversed: false,
                relationship_id: None,
            },
            SegmentEdge {
                source: 0,
                target: 2,
                type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
                schema_reversed: false,
                relationship_id: Some(10),
            },
        ]);
        newer_out.edge_inserts.push(SegmentEdge {
            source: 0,
            target: 3,
            type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
            schema_reversed: false,
            relationship_id: Some(12),
        });

        let mut older_in = DeltaSegment::new(SegmentKind::Edge, 0, TraversalDirection::In, 0, 5, 1)
            .expect("older inbound segment");
        older_in.edge_inserts.push(SegmentEdge {
            source: 0,
            target: 2,
            type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
            schema_reversed: false,
            relationship_id: Some(20),
        });
        let mut newer_in = DeltaSegment::new(SegmentKind::Edge, 0, TraversalDirection::In, 0, 5, 2)
            .expect("newer inbound segment");
        newer_in.edge_deletes.push(SegmentEdge {
            source: 0,
            target: 2,
            type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
            schema_reversed: false,
            relationship_id: Some(20),
        });
        newer_in.edge_inserts.push(SegmentEdge {
            source: 0,
            target: 4,
            type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
            schema_reversed: false,
            relationship_id: Some(21),
        });

        let layered = LayeredNeighbors::new(&base, vec![older_out, older_in, newer_out, newer_in]);
        for direction in [TraversalDirection::Out, TraversalDirection::In] {
            let directional = layered.for_direction(direction);
            let expected = directional.neighbors(0).collect::<Vec<_>>();
            let fixture_expected = match direction {
                TraversalDirection::Out => vec![
                    Neighbor {
                        target: 2,
                        type_id: EdgeTypeId::from_v6_storage(1)
                            .expect("fixture type ID is valid v6"),
                        schema_reversed: false,
                        relationship_id: Some(11),
                    },
                    Neighbor {
                        target: 3,
                        type_id: EdgeTypeId::from_v6_storage(1)
                            .expect("fixture type ID is valid v6"),
                        schema_reversed: false,
                        relationship_id: Some(12),
                    },
                ],
                TraversalDirection::In => vec![
                    Neighbor {
                        target: 1,
                        type_id: EdgeTypeId::from_v6_storage(1)
                            .expect("fixture type ID is valid v6"),
                        schema_reversed: false,
                        relationship_id: None,
                    },
                    Neighbor {
                        target: 4,
                        type_id: EdgeTypeId::from_v6_storage(1)
                            .expect("fixture type ID is valid v6"),
                        schema_reversed: false,
                        relationship_id: Some(21),
                    },
                ],
                TraversalDirection::Any => unreachable!("fixture checks one direction at a time"),
            };
            assert_eq!(
                &expected, &fixture_expected,
                "fixture must exercise wildcard tombstones, identified deletes, parallel identities, and later-segment inserts"
            );
            for page_size in [1, 2, 3] {
                assert_eq!(
                    collect_layered_pages(&directional, 0, page_size),
                    expected,
                    "layered paging changed current {direction:?} byte/order semantics"
                );
                assert_eq!(
                    collect_layered_reverse_pages(&directional, 0, page_size),
                    directional.neighbors_reversed(0).collect::<Vec<_>>(),
                    "reverse layered paging changed {direction:?} precedence/order"
                );
            }
        }
    }

    #[test]
    fn owned_layered_cursor_matches_current_order_in_both_directions() {
        let base = edge_store_from_tuples(4, &[(0, 1, 1), (0, 2, 1), (3, 0, 1)]);
        let base_in = edge_store_from_tuples(4, &[(1, 0, 1), (2, 0, 1), (0, 3, 1)]);
        let layered =
            LayeredNeighbors::new_with_options(&base, Some(&base_in), Vec::new(), None, None, None);
        for direction in [TraversalDirection::Out, TraversalDirection::In] {
            let directional = layered.for_direction(direction);
            assert_eq!(
                collect_layered_pages(&directional, 0, 1),
                directional.neighbors(0).collect::<Vec<_>>()
            );
        }
    }

    #[test]
    fn owned_layered_cursor_yields_progress_after_zero_output_raw_page() {
        let base = edge_store_from_tuples(4, &[(0, 1, 1), (0, 2, 1), (0, 3, 1)]);
        let mut segment = DeltaSegment::new(SegmentKind::Edge, 0, TraversalDirection::Out, 0, 4, 1)
            .expect("delete segment");
        segment
            .edge_deletes
            .extend((1..=3).map(|target| SegmentEdge {
                source: 0,
                target,
                type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
                schema_reversed: false,
                relationship_id: None,
            }));
        let layered = LayeredNeighbors::new(&base, vec![segment]);
        let mut cursor = OwnedNeighborCursor::default();
        let mut page = Vec::new();

        let exhausted = layered.fill_neighbors(0, &mut cursor, 1, &mut page);

        assert!(!exhausted);
        assert!(page.is_empty());
        assert!(matches!(
            cursor,
            OwnedNeighborCursor::Layered { base_pos: 1, .. }
        ));

        let mut reverse_cursor = OwnedNeighborCursor::default();
        let mut reverse_page = Vec::new();
        let reverse_exhausted =
            layered.fill_neighbors_reversed(0, &mut reverse_cursor, 1, &mut reverse_page);
        assert!(!reverse_exhausted);
        assert!(reverse_page.is_empty());
        assert!(matches!(
            reverse_cursor,
            OwnedNeighborCursor::LayeredReverse {
                base_consumed: 1,
                ..
            }
        ));
    }

    #[test]
    fn owned_layered_cursor_bounds_examined_raw_rows() {
        let base = edge_store_from_tuples(
            18,
            &(1..18).map(|target| (0, target, 1)).collect::<Vec<_>>(),
        );
        let layered = LayeredNeighbors::new(&base, Vec::new());
        let mut cursor = OwnedNeighborCursor::default();
        let mut page = Vec::new();

        let exhausted = layered.fill_neighbors(0, &mut cursor, 3, &mut page);

        assert!(!exhausted);
        assert_eq!(page.len(), 3);
        assert!(matches!(
            cursor,
            OwnedNeighborCursor::Layered { base_pos: 3, .. }
        ));
    }

    #[test]
    fn owned_layered_cursor_preserves_parallel_base_relationship_identity_order() {
        let mut builder = SortedEdgeStoreBuilder::new(5, false);
        for relationship_id in [10, 20] {
            builder
                .try_push_identified(IdentifiedRawEdge {
                    edge: RawEdge {
                        source: 0,
                        target: 1,
                        type_id: EdgeTypeId::from_v6_storage(1)
                            .expect("fixture type ID is valid v6"),
                        weight: None,
                        schema_reversed: false,
                    },
                    relationship_id,
                })
                .expect("production-ordered identified base edge");
        }
        let base = builder.finish();
        let mut durable = DeltaSegment::new(SegmentKind::Edge, 0, TraversalDirection::Out, 0, 5, 1)
            .expect("durable segment");
        durable.edge_inserts.push(SegmentEdge {
            source: 0,
            target: 2,
            type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
            schema_reversed: false,
            relationship_id: Some(30),
        });
        let layered = LayeredNeighbors::new(&base, vec![durable]);

        assert_eq!(
            collect_layered_pages(&layered, 0, 1),
            layered.neighbors(0).collect::<Vec<_>>()
        );
    }

    #[test]
    fn owned_layered_cursor_collapses_duplicate_full_base_keys_across_pages() {
        let mut builder = SortedEdgeStoreBuilder::new(4, false);
        for _ in 0..3 {
            builder
                .try_push(RawEdge {
                    source: 0,
                    target: 1,
                    type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
                    weight: None,
                    schema_reversed: false,
                })
                .expect("duplicate identityless base edge");
        }
        let base = builder.finish();
        let mut durable = DeltaSegment::new(SegmentKind::Edge, 0, TraversalDirection::Out, 0, 4, 1)
            .expect("durable segment activates layered oracle");
        durable.edge_inserts.push(SegmentEdge {
            source: 0,
            target: 2,
            type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
            schema_reversed: false,
            relationship_id: Some(40),
        });
        let layered = LayeredNeighbors::new(&base, vec![durable]);

        assert_eq!(
            collect_layered_pages(&layered, 0, 1),
            layered.neighbors(0).collect::<Vec<_>>()
        );
    }

    #[test]
    fn owned_layered_cursor_pages_base_chunk_replacements_without_base_leakage() {
        let base = edge_store_from_tuples(6, &[(0, 1, 1), (0, 2, 1), (5, 0, 1)]);
        let base_in = edge_store_from_tuples(6, &[(1, 0, 1), (2, 0, 1), (0, 5, 1)]);
        let mut chunk = DeltaSegment::new(SegmentKind::Edge, 1, TraversalDirection::Out, 0, 1, 1)
            .expect("base chunk");
        chunk.edge_inserts.extend([
            SegmentEdge {
                source: 0,
                target: 3,
                type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
                schema_reversed: false,
                relationship_id: Some(31),
            },
            SegmentEdge {
                source: 0,
                target: 4,
                type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
                schema_reversed: false,
                relationship_id: Some(32),
            },
        ]);
        let layered = LayeredNeighbors::new_with_base_chunks(
            &base,
            Some(&base_in),
            vec![chunk],
            Vec::new(),
            None,
            None,
            None,
        );

        for direction in [TraversalDirection::Out, TraversalDirection::In] {
            let directional = layered.for_direction(direction);
            assert_eq!(
                collect_layered_pages(&directional, 0, 1),
                directional.neighbors(0).collect::<Vec<_>>()
            );
        }
        assert_eq!(
            collect_layered_pages(&layered.for_direction(TraversalDirection::Out), 0, 1)
                .into_iter()
                .map(|neighbor| neighbor.target)
                .collect::<Vec<_>>(),
            vec![3, 4]
        );
        let inbound = layered.for_direction(TraversalDirection::In);
        for target in [1, 2, 3, 4] {
            assert_eq!(
                collect_layered_pages(&inbound, target, 1),
                inbound.neighbors(target).collect::<Vec<_>>()
            );
        }
        assert!(collect_layered_pages(&inbound, 1, 1).is_empty());
        assert!(collect_layered_pages(&inbound, 2, 1).is_empty());
        for target in [3, 4] {
            assert_eq!(
                collect_layered_pages(&inbound, target, 1)
                    .into_iter()
                    .map(|neighbor| neighbor.target)
                    .collect::<Vec<_>>(),
                vec![0]
            );
        }
    }

    #[test]
    fn owned_layered_cursor_preserves_durable_and_frozen_overlay_precedence() {
        let base = edge_store_from_tuples(5, &[(0, 1, 1)]);
        let mut durable = DeltaSegment::new(SegmentKind::Edge, 0, TraversalDirection::Out, 0, 5, 1)
            .expect("durable segment");
        durable.edge_deletes.push(SegmentEdge {
            source: 0,
            target: 1,
            type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
            schema_reversed: false,
            relationship_id: None,
        });
        durable.edge_inserts.push(SegmentEdge {
            source: 0,
            target: 2,
            type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
            schema_reversed: false,
            relationship_id: Some(50),
        });
        let mut overlay_inserts = OverlayInserts::new();
        overlay_inserts.entry(0).or_default().push((
            1,
            crate::types::EdgeTypeId::test_v6(1),
            false,
            None,
        ));
        let mut overlay_deletes = OverlayDeletes::new();
        overlay_deletes.entry(0).or_default().insert((
            2,
            crate::types::EdgeTypeId::test_v6(1),
            false,
            None,
        ));
        let layered = LayeredNeighbors::new_with_options(
            &base,
            None,
            vec![durable],
            None,
            Some((overlay_inserts, overlay_deletes)),
            None,
        );

        for page_size in [1, 2, 3] {
            assert_eq!(
                collect_layered_pages(&layered, 0, page_size),
                layered.neighbors(0).collect::<Vec<_>>()
            );
        }
        assert_eq!(
            layered
                .neighbors(0)
                .map(|neighbor| neighbor.target)
                .collect::<Vec<_>>(),
            vec![1]
        );
    }

    #[test]
    fn owned_layered_cursor_preserves_inbound_durable_and_frozen_overlay_precedence() {
        let base = edge_store_from_tuples(5, &[]);
        let base_in = edge_store_from_tuples(5, &[(0, 1, 1)]);
        let mut durable = DeltaSegment::new(SegmentKind::Edge, 0, TraversalDirection::In, 0, 5, 1)
            .expect("inbound durable segment");
        durable.edge_deletes.push(SegmentEdge {
            source: 0,
            target: 1,
            type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
            schema_reversed: false,
            relationship_id: None,
        });
        durable.edge_inserts.push(SegmentEdge {
            source: 0,
            target: 2,
            type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
            schema_reversed: false,
            relationship_id: Some(60),
        });
        let mut overlay_inserts = OverlayInserts::new();
        overlay_inserts.entry(0).or_default().push((
            1,
            crate::types::EdgeTypeId::test_v6(1),
            false,
            None,
        ));
        let mut overlay_deletes = OverlayDeletes::new();
        overlay_deletes.entry(0).or_default().insert((
            2,
            crate::types::EdgeTypeId::test_v6(1),
            false,
            None,
        ));
        let layered = LayeredNeighbors::new_with_options(
            &base,
            Some(&base_in),
            vec![durable],
            None,
            None,
            Some((overlay_inserts, overlay_deletes)),
        );
        let inbound = layered.for_direction(TraversalDirection::In);

        for page_size in [1, 2, 3] {
            assert_eq!(
                collect_layered_pages(&inbound, 0, page_size),
                inbound.neighbors(0).collect::<Vec<_>>()
            );
        }
        assert_eq!(
            inbound
                .neighbors(0)
                .map(|neighbor| neighbor.target)
                .collect::<Vec<_>>(),
            vec![1]
        );
    }

    #[test]
    fn owned_layered_any_cursor_preserves_shared_map_precedence_and_order() {
        let base = edge_store_from_tuples(8, &[(0, 1, 1), (0, 2, 1), (5, 0, 1)]);
        let base_in = edge_store_from_tuples(8, &[(1, 0, 1), (2, 0, 1), (0, 5, 1)]);

        let mut chunk_out =
            DeltaSegment::new(SegmentKind::Edge, 1, TraversalDirection::Out, 0, 1, 1)
                .expect("outbound base chunk");
        chunk_out.edge_inserts.extend([
            SegmentEdge {
                source: 0,
                target: 3,
                type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
                schema_reversed: false,
                relationship_id: Some(30),
            },
            SegmentEdge {
                source: 0,
                target: 4,
                type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
                schema_reversed: false,
                relationship_id: Some(31),
            },
        ]);
        let mut chunk_in =
            DeltaSegment::new(SegmentKind::Edge, 1, TraversalDirection::Out, 5, 6, 2)
                .expect("inbound base chunk");
        chunk_in.edge_inserts.push(SegmentEdge {
            source: 5,
            target: 0,
            type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
            schema_reversed: false,
            relationship_id: Some(32),
        });

        let mut durable_out =
            DeltaSegment::new(SegmentKind::Edge, 0, TraversalDirection::Out, 0, 8, 3)
                .expect("outbound durable segment");
        durable_out.edge_deletes.push(SegmentEdge {
            source: 0,
            target: 3,
            type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
            schema_reversed: false,
            relationship_id: None,
        });
        for (target, relationship_id) in [(6, 60), (7, 70)] {
            durable_out.edge_inserts.push(SegmentEdge {
                source: 0,
                target,
                type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
                schema_reversed: false,
                relationship_id: Some(relationship_id),
            });
            durable_out.edge_weights.push(SegmentEdgeWeight {
                source: 0,
                target,
                type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
                schema_reversed: false,
                relationship_id: Some(relationship_id),
                weight: 7,
            });
        }
        let mut durable_in =
            DeltaSegment::new(SegmentKind::Edge, 0, TraversalDirection::In, 0, 8, 4)
                .expect("inbound durable segment");
        for relationship_id in [63, 64] {
            durable_in.edge_inserts.push(SegmentEdge {
                source: 0,
                target: 2,
                type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
                schema_reversed: false,
                relationship_id: Some(relationship_id),
            });
        }
        durable_in.edge_inserts.push(SegmentEdge {
            source: 0,
            target: 7,
            type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
            schema_reversed: false,
            relationship_id: Some(70),
        });
        durable_in.edge_weights.push(SegmentEdgeWeight {
            source: 0,
            target: 7,
            type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
            schema_reversed: false,
            relationship_id: Some(70),
            weight: 9,
        });

        let mut overlay_out_inserts = OverlayInserts::new();
        overlay_out_inserts.entry(0).or_default().push((
            3,
            crate::types::EdgeTypeId::test_v6(1),
            false,
            None,
        ));
        let mut overlay_out_deletes = OverlayDeletes::new();
        overlay_out_deletes.entry(0).or_default().insert((
            6,
            crate::types::EdgeTypeId::test_v6(1),
            false,
            None,
        ));
        let mut overlay_in_inserts = OverlayInserts::new();
        overlay_in_inserts.entry(0).or_default().push((
            6,
            crate::types::EdgeTypeId::test_v6(1),
            false,
            Some(60),
        ));
        let mut overlay_in_deletes = OverlayDeletes::new();
        overlay_in_deletes.entry(0).or_default().insert((
            2,
            crate::types::EdgeTypeId::test_v6(1),
            false,
            Some(62),
        ));

        let layered = LayeredNeighbors::new_with_base_chunks(
            &base,
            Some(&base_in),
            vec![chunk_out, chunk_in],
            vec![durable_out, durable_in],
            None,
            Some((overlay_out_inserts, overlay_out_deletes)),
            Some((overlay_in_inserts, overlay_in_deletes)),
        );
        let any = layered.for_direction(TraversalDirection::Any);
        let expected = any.neighbors(0).collect::<Vec<_>>();
        assert_eq!(
            expected,
            vec![
                Neighbor {
                    target: 2,
                    type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
                    schema_reversed: false,
                    relationship_id: Some(63)
                },
                Neighbor {
                    target: 2,
                    type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
                    schema_reversed: false,
                    relationship_id: Some(64)
                },
                Neighbor {
                    target: 3,
                    type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
                    schema_reversed: false,
                    relationship_id: None
                },
                Neighbor {
                    target: 4,
                    type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
                    schema_reversed: false,
                    relationship_id: Some(31)
                },
                Neighbor {
                    target: 5,
                    type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
                    schema_reversed: false,
                    relationship_id: Some(32)
                },
                Neighbor {
                    target: 6,
                    type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
                    schema_reversed: false,
                    relationship_id: Some(60)
                },
                Neighbor {
                    target: 7,
                    type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
                    schema_reversed: false,
                    relationship_id: Some(70)
                },
            ]
        );
        let (_, winning_payload) = layered
            .merged_neighbors(TraversalDirection::Any, 0, false)
            .into_iter()
            .find(|(target, edge)| *target == 7 && edge.relationship_id == Some(70))
            .expect("shared-key durable edge must survive");
        assert_eq!(
            winning_payload.weight,
            Some(9),
            "within one precedence layer In must overwrite an identical Out full key"
        );
        for page_size in 1..=expected.len() {
            assert_eq!(
                collect_layered_pages_requiring_progress(&any, 0, page_size),
                expected,
                "owned Any cursor changed shared-map order at page size {page_size}"
            );
            assert_eq!(
                collect_layered_reverse_pages(&any, 0, page_size),
                any.neighbors_reversed(0).collect::<Vec<_>>(),
                "reverse Any cursor changed shared-map order at page size {page_size}"
            );
        }
    }

    #[test]
    fn owned_layered_any_base_layer_applies_complete_out_before_in() {
        let key = (3, crate::types::EdgeTypeId::test_v6(1), false, Some(30));
        let out_chunk = LayeredEdge {
            target: 3,
            type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
            schema_reversed: false,
            weight: Some(7),
            relationship_id: Some(30),
        };
        let in_base = LayeredEdge {
            weight: Some(9),
            ..out_chunk
        };

        assert_eq!(
            any_base_layer_winner(key, None, Some(out_chunk), Some(in_base), None),
            Some(in_base),
            "eager Any applies the complete outbound base/chunk layer before inbound base"
        );
    }

    #[test]
    fn owned_layered_any_cursor_yields_after_bounded_zero_output_work() {
        let base = edge_store_from_tuples(6, &[(0, 1, 1), (0, 2, 1), (3, 0, 1), (4, 0, 1)]);
        let base_in = edge_store_from_tuples(6, &[(1, 0, 1), (2, 0, 1), (0, 3, 1), (0, 4, 1)]);
        let mut durable = DeltaSegment::new(SegmentKind::Edge, 0, TraversalDirection::Any, 0, 6, 1)
            .expect("Any tombstone segment");
        durable.edge_deletes.extend([
            SegmentEdge {
                source: 0,
                target: 1,
                type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
                schema_reversed: false,
                relationship_id: None,
            },
            SegmentEdge {
                source: 0,
                target: 2,
                type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
                schema_reversed: false,
                relationship_id: None,
            },
            SegmentEdge {
                source: 0,
                target: 3,
                type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
                schema_reversed: false,
                relationship_id: None,
            },
            SegmentEdge {
                source: 0,
                target: 4,
                type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
                schema_reversed: false,
                relationship_id: None,
            },
        ]);
        let layered = LayeredNeighbors::new_with_options(
            &base,
            Some(&base_in),
            vec![durable],
            None,
            None,
            None,
        );
        let any = layered.for_direction(TraversalDirection::Any);
        let mut cursor = OwnedNeighborCursor::default();
        let before = cursor.clone();
        let mut page = Vec::new();

        let exhausted = any.fill_neighbors(0, &mut cursor, 2, &mut page);

        assert!(
            !exhausted,
            "one bounded raw page must yield before exhaustion"
        );
        assert!(page.is_empty(), "tombstoned raw work must not leak output");
        assert_ne!(cursor, before, "a Progress page must advance owned state");

        let mut reverse_cursor = OwnedNeighborCursor::default();
        let reverse_before = reverse_cursor.clone();
        let mut reverse_page = Vec::new();
        let reverse_exhausted =
            any.fill_neighbors_reversed(0, &mut reverse_cursor, 2, &mut reverse_page);
        assert!(!reverse_exhausted);
        assert!(reverse_page.is_empty());
        assert_ne!(reverse_cursor, reverse_before);
    }

    proptest! {
        #[test]
        fn owned_layered_any_cursor_matches_eager_for_generated_direction_pairs(
            out_mask in 0u8..64,
            in_mask in 0u8..64,
            page_size in 1usize..5,
        ) {
            let out_edges = (1u32..7)
                .filter(|target| out_mask & (1 << (target - 1)) != 0)
                .map(|target| (0, target, 1))
                .collect::<Vec<_>>();
            let in_edges = (1u32..7)
                .filter(|target| in_mask & (1 << (target - 1)) != 0)
                .map(|target| (0, target, 1))
                .collect::<Vec<_>>();
            let base = edge_store_from_tuples(7, &out_edges);
            let base_in = edge_store_from_tuples(7, &in_edges);
            let mut durable =
                DeltaSegment::new(SegmentKind::Edge, 0, TraversalDirection::Out, 0, 7, 1)
                    .expect("durable segment activates layered paging");
            durable.edge_inserts.push(SegmentEdge {
                source: 0,
                target: 6,
                type_id: EdgeTypeId::from_v6_storage(2).expect("fixture type ID is valid v6"),
                schema_reversed: false,
                relationship_id: Some(90),
            });
            let layered = LayeredNeighbors::new_with_options(
                &base,
                Some(&base_in),
                vec![durable],
                None,
                None,
                None,
            );
            let any = layered.for_direction(TraversalDirection::Any);

            prop_assert_eq!(
                collect_layered_pages_requiring_progress(&any, 0, page_size),
                any.neighbors(0).collect::<Vec<_>>()
            );
            prop_assert_eq!(
                collect_layered_reverse_pages(&any, 0, page_size),
                any.neighbors_reversed(0).collect::<Vec<_>>()
            );
        }
    }

    proptest! {
        #[test]
        fn owned_layered_cursor_matches_eager_for_generated_mutation_sequences(
            inbound in any::<bool>(),
            base_mask in 0u8..64,
            operations in prop::collection::vec((1u32..7, any::<bool>(), 0u8..3), 0..24),
            page_size in 1usize..5,
        ) {
            let direction = if inbound {
                TraversalDirection::In
            } else {
                TraversalDirection::Out
            };
            let base_edges = (1u32..7)
                .filter(|target| base_mask & (1 << (target - 1)) != 0)
                .map(|target| (0, target, 1))
                .collect::<Vec<_>>();
            let base = edge_store_from_tuples(7, &base_edges);
            let mut segments = Vec::with_capacity(operations.len());
            for (offset, (target, insert, identity_variant)) in
                operations.into_iter().enumerate()
            {
                let mut segment = DeltaSegment::new(
                    SegmentKind::Edge,
                    0,
                    direction,
                    0,
                    7,
                    i64::try_from(offset + 1).expect("bounded generated watermark"),
                )
                .expect("generated segment");
                let edge = SegmentEdge {
                    source: 0,
                    target,
                    type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
                    schema_reversed: false,
                    relationship_id: (identity_variant != 0)
                        .then_some(u32::from(identity_variant)),
                };
                if insert {
                    segment.edge_inserts.push(edge);
                } else {
                    segment.edge_deletes.push(edge);
                }
                segments.push(segment);
            }
            let layered = LayeredNeighbors::new_with_options(
                &base,
                Some(&base),
                segments,
                None,
                None,
                None,
            );
            let directional = layered.for_direction(direction);

            prop_assert_eq!(
                collect_layered_pages(&directional, 0, page_size),
                directional.neighbors(0).collect::<Vec<_>>()
            );
            prop_assert_eq!(
                collect_layered_reverse_pages(&directional, 0, page_size),
                directional.neighbors_reversed(0).collect::<Vec<_>>()
            );
        }
    }

    #[test]
    fn fallible_snapshot_plan_preserves_ordered_layered_results() {
        let base = edge_store_from_tuples(4, &[(0, 1, 1)]);
        let mut older = DeltaSegment::new(SegmentKind::Edge, 0, TraversalDirection::Out, 0, 4, 1)
            .expect("older segment");
        older.edge_inserts.push(SegmentEdge {
            source: 0,
            target: 2,
            type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
            schema_reversed: false,
            relationship_id: Some(10),
        });
        let mut newer = DeltaSegment::new(SegmentKind::Edge, 0, TraversalDirection::Out, 0, 4, 2)
            .expect("newer segment");
        newer.edge_deletes.push(SegmentEdge {
            source: 0,
            target: 1,
            type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
            schema_reversed: false,
            relationship_id: None,
        });
        newer.edge_inserts.push(SegmentEdge {
            source: 0,
            target: 3,
            type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
            schema_reversed: false,
            relationship_id: Some(11),
        });
        let refs = [&newer, &older];
        let plan = LayeredSnapshot::build_plan(&[], &refs).expect("snapshot plan");
        assert!(plan.estimated_peak_bytes() > 0);
        let snapshot = LayeredSnapshot::try_build_from_refs_with_plan(&base, &[], &refs, plan)
            .expect("bounded snapshot");
        let layered = LayeredNeighbors::from_snapshot(&base, &snapshot);

        assert_eq!(
            layered.neighbors(0).collect::<Vec<_>>(),
            vec![
                Neighbor {
                    target: 2,
                    type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
                    schema_reversed: false,
                    relationship_id: Some(10),
                },
                Neighbor {
                    target: 3,
                    type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
                    schema_reversed: false,
                    relationship_id: Some(11),
                },
            ]
        );
    }

    #[test]
    fn relationship_identity_summary_uses_effective_out_in_and_base_chunk_precedence() {
        let base = edge_store_from_tuples(6, &[]);
        let mut chunk = DeltaSegment::new(SegmentKind::Edge, 0, TraversalDirection::Out, 0, 6, 1)
            .expect("base chunk");
        chunk.edge_inserts.push(SegmentEdge {
            source: 0,
            target: 1,
            type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
            schema_reversed: false,
            relationship_id: None,
        });
        let mut delete_chunk =
            DeltaSegment::new(SegmentKind::Edge, 0, TraversalDirection::Out, 0, 6, 2)
                .expect("durable delete");
        delete_chunk.edge_deletes.push(SegmentEdge {
            source: 0,
            target: 1,
            type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
            schema_reversed: false,
            relationship_id: None,
        });
        let mut inbound = DeltaSegment::new(SegmentKind::Edge, 0, TraversalDirection::In, 0, 6, 3)
            .expect("inbound insert");
        inbound.edge_inserts.push(SegmentEdge {
            source: 4,
            target: 5,
            type_id: EdgeTypeId::from_v6_storage(2).expect("fixture type ID is valid v6"),
            schema_reversed: true,
            relationship_id: None,
        });
        let chunk_refs = [&chunk];
        let segment_refs = [&delete_chunk, &inbound];
        let snapshot = LayeredSnapshot::try_build_from_refs(&base, &chunk_refs, &segment_refs)
            .expect("summarized snapshot");
        let type_one = roaring::RoaringBitmap::from_iter([1]);
        let type_two = roaring::RoaringBitmap::from_iter([2]);

        assert!(!snapshot.has_missing_relationship_identity_for_types(&type_one));
        assert!(snapshot.has_missing_relationship_identity_for_types(&type_two));
    }

    #[test]
    fn relationship_identity_summary_ignores_ambient_transaction_visibility() {
        tx_delta::clear_for_test();
        let base = edge_store_from_tuples(3, &[]);
        let mut segment = DeltaSegment::new(SegmentKind::Edge, 0, TraversalDirection::Out, 0, 3, 1)
            .expect("durable segment");
        segment.edge_inserts.push(SegmentEdge {
            source: 0,
            target: 1,
            type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
            schema_reversed: false,
            relationship_id: None,
        });
        tx_delta::record_deleted_node(1).expect("ambient transaction tombstone");
        let snapshot = LayeredSnapshot::try_build_from_refs(&base, &[], &[&segment])
            .expect("snapshot summary must ignore tx visibility");
        tx_delta::clear_for_test();
        let active = roaring::RoaringBitmap::from_iter([1]);
        assert!(snapshot.has_missing_relationship_identity_for_types(&active));
    }

    #[test]
    fn relationship_identity_summary_honors_durable_node_visibility() {
        let base = edge_store_from_tuples(3, &[]);
        let mut edge_segment =
            DeltaSegment::new(SegmentKind::Edge, 0, TraversalDirection::Out, 0, 3, 1)
                .expect("durable edge");
        edge_segment.edge_inserts.push(SegmentEdge {
            source: 0,
            target: 1,
            type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
            schema_reversed: false,
            relationship_id: None,
        });
        let mut node_segment =
            DeltaSegment::new(SegmentKind::Node, 0, TraversalDirection::Any, 0, 3, 2)
                .expect("durable node state");
        node_segment.node_states.push(SegmentNodeState {
            node_idx: 1,
            active: false,
        });
        let snapshot =
            LayeredSnapshot::try_build_from_refs(&base, &[], &[&edge_segment, &node_segment])
                .expect("snapshot with tombstoned target");
        let active = roaring::RoaringBitmap::from_iter([1]);
        assert!(!snapshot.has_missing_relationship_identity_for_types(&active));
    }

    #[test]
    fn layered_neighbors_equal_full_rebuild_for_insert_delete_sequence() {
        let base = edge_store_from_tuples(4, &[(0, 1, 1), (0, 2, 1)]);
        let mut insert = DeltaSegment::new(SegmentKind::Edge, 0, TraversalDirection::Out, 0, 4, 1)
            .expect("insert segment");
        insert.edge_inserts.push(SegmentEdge {
            source: 0,
            target: 3,
            type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
            schema_reversed: false,
            relationship_id: None,
        });
        let mut delete = DeltaSegment::new(SegmentKind::Edge, 0, TraversalDirection::Out, 0, 4, 2)
            .expect("delete segment");
        delete.edge_deletes.push(SegmentEdge {
            source: 0,
            target: 1,
            type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
            schema_reversed: false,
            relationship_id: None,
        });
        let layered = LayeredNeighbors::new(&base, vec![insert, delete]);
        let full_rebuild = edge_store_from_tuples(4, &[(0, 2, 1), (0, 3, 1)]);
        let expected = CsrNeighbors::new(&full_rebuild);

        assert_full_csr_equivalence(4, &expected, &layered);
    }

    #[test]
    fn layered_neighbors_preserve_segment_relationship_id() {
        let base = edge_store_from_tuples(4, &[(0, 1, 1)]);
        let mut segment = DeltaSegment::new(SegmentKind::Edge, 0, TraversalDirection::Out, 0, 4, 1)
            .expect("insert segment");
        segment.edge_inserts.push(SegmentEdge {
            source: 0,
            target: 3,
            type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
            schema_reversed: false,
            relationship_id: Some(77),
        });
        let layered = LayeredNeighbors::new(&base, vec![segment]);

        assert_eq!(
            layered.neighbors(0).collect::<Vec<_>>(),
            vec![
                Neighbor {
                    target: 1,
                    type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
                    schema_reversed: false,
                    relationship_id: None,
                },
                Neighbor {
                    target: 3,
                    type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
                    schema_reversed: false,
                    relationship_id: Some(77),
                },
            ]
        );
    }

    #[test]
    fn layered_neighbors_tx_delta_wins_over_durable_segments() {
        tx_delta::clear_for_test();
        let base = edge_store_from_tuples(4, &[(0, 1, 1)]);
        let mut durable = DeltaSegment::new(SegmentKind::Edge, 0, TraversalDirection::Out, 0, 4, 1)
            .expect("durable segment");
        durable.edge_inserts.push(SegmentEdge {
            source: 0,
            target: 2,
            type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
            schema_reversed: false,
            relationship_id: None,
        });
        tx_delta::record_deleted_edge(0, 2, crate::types::EdgeTypeId::test_v6(1))
            .expect("record tx delete");
        tx_delta::record_added_edge(
            0,
            tx_delta::DeltaEdge {
                target: 3,
                type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
                weight: None,
                schema_reversed: false,
                relationship_id: None,
            },
        )
        .expect("record tx insert");
        let layered = LayeredNeighbors::new(&base, vec![durable]);

        assert_eq!(
            layered.neighbors(0).collect::<Vec<_>>(),
            vec![
                Neighbor {
                    target: 1,
                    type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
                    schema_reversed: false,
                    relationship_id: None,
                },
                Neighbor {
                    target: 3,
                    type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
                    schema_reversed: false,
                    relationship_id: None,
                },
            ]
        );
        tx_delta::clear_for_test();
    }

    #[test]
    fn layered_neighbors_inbound_direction_matches_full_rebuild() {
        let base = edge_store_from_tuples(4, &[(1, 0, 1)]);
        let mut inbound = DeltaSegment::new(SegmentKind::Edge, 0, TraversalDirection::In, 0, 4, 1)
            .expect("inbound segment");
        inbound.edge_inserts.push(SegmentEdge {
            source: 0,
            target: 2,
            type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
            schema_reversed: false,
            relationship_id: None,
        });
        let layered = LayeredNeighbors::new(&base, vec![inbound]);
        let actual = layered.merged_neighbors(TraversalDirection::In, 0, false);

        assert_eq!(
            actual
                .into_iter()
                .map(|(target, edge)| Neighbor {
                    target,
                    type_id: edge.type_id,
                    schema_reversed: edge.schema_reversed,
                    relationship_id: None,
                })
                .collect::<Vec<_>>(),
            vec![
                Neighbor {
                    target: 1,
                    type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
                    schema_reversed: false,
                    relationship_id: None,
                },
                Neighbor {
                    target: 2,
                    type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
                    schema_reversed: false,
                    relationship_id: None,
                },
            ]
        );
    }

    #[test]
    fn layered_neighbors_suppresses_duplicates_across_layers() {
        let base = edge_store_from_tuples(3, &[(0, 1, 1)]);
        let mut segment = DeltaSegment::new(SegmentKind::Edge, 0, TraversalDirection::Out, 0, 3, 1)
            .expect("segment");
        segment.edge_inserts.push(SegmentEdge {
            source: 0,
            target: 1,
            type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
            schema_reversed: false,
            relationship_id: None,
        });
        segment.edge_inserts.push(SegmentEdge {
            source: 0,
            target: 2,
            type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
            schema_reversed: false,
            relationship_id: None,
        });
        segment.edge_inserts.push(SegmentEdge {
            source: 0,
            target: 2,
            type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
            schema_reversed: false,
            relationship_id: None,
        });
        let layered = LayeredNeighbors::new(&base, vec![segment]);

        assert_eq!(
            layered.neighbors(0).collect::<Vec<_>>(),
            vec![
                Neighbor {
                    target: 1,
                    type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
                    schema_reversed: false,
                    relationship_id: None,
                },
                Neighbor {
                    target: 2,
                    type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
                    schema_reversed: false,
                    relationship_id: None,
                },
            ]
        );
    }

    #[test]
    fn weighted_shortest_path_uses_durable_weight_segments() {
        let base = weighted_edge_store_from_tuples(4, &[(0, 1, 1, 10)]);
        let mut segment = DeltaSegment::new(SegmentKind::Edge, 0, TraversalDirection::Out, 0, 4, 1)
            .expect("segment");
        segment.edge_inserts.push(SegmentEdge {
            source: 0,
            target: 2,
            type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
            schema_reversed: false,
            relationship_id: None,
        });
        segment.edge_weights.push(SegmentEdgeWeight {
            source: 0,
            target: 2,
            type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
            relationship_id: None,
            weight: 3,
            schema_reversed: false,
        });
        let layered = LayeredNeighbors::new(&base, vec![segment]);

        assert_eq!(
            layered.weighted_neighbors(0),
            vec![
                WeightedNeighbor {
                    target: 1,
                    type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
                    relationship_id: None,
                    weight: 10,
                    schema_reversed: false,
                },
                WeightedNeighbor {
                    target: 2,
                    type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
                    relationship_id: None,
                    weight: 3,
                    schema_reversed: false,
                },
            ]
        );
    }

    #[test]
    fn weighted_layered_cursor_matches_eager_across_pages_and_precedence() {
        let base =
            weighted_edge_store_from_tuples(5, &[(0, 1, 1, 10), (0, 2, 1, 20), (0, 4, 1, 40)]);
        let mut segment = DeltaSegment::new(SegmentKind::Edge, 0, TraversalDirection::Out, 0, 5, 1)
            .expect("segment");
        segment.edge_deletes.push(SegmentEdge {
            source: 0,
            target: 2,
            type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
            schema_reversed: false,
            relationship_id: None,
        });
        segment.edge_inserts.push(SegmentEdge {
            source: 0,
            target: 3,
            type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
            schema_reversed: false,
            relationship_id: None,
        });
        segment.edge_weights.push(SegmentEdgeWeight {
            source: 0,
            target: 3,
            type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
            relationship_id: None,
            weight: 3,
            schema_reversed: false,
        });
        // The durable snapshot has already combined edge and weight rows, so
        // the cursor must preserve the same resolved winner as the eager map.
        segment.edge_inserts.push(SegmentEdge {
            source: 0,
            target: 4,
            type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
            schema_reversed: false,
            relationship_id: None,
        });
        let layered = LayeredNeighbors::new(&base, vec![segment]);
        let eager = layered.weighted_neighbors(0);
        assert_eq!(
            eager.iter().map(|edge| edge.target).collect::<Vec<_>>(),
            [1, 3, 4]
        );
        for page_size in [1, 2, 3] {
            let mut cursor = OwnedNeighborCursor::default();
            let mut actual = Vec::new();
            loop {
                let mut page = Vec::new();
                let exhausted =
                    layered.fill_weighted_neighbors(0, &mut cursor, page_size, &mut page);
                actual.extend(page);
                if exhausted {
                    break;
                }
            }
            assert_eq!(actual, eager, "weighted page size {page_size} diverged");
        }
    }

    #[test]
    fn layered_reads_hide_transaction_deleted_nodes() {
        tx_delta::clear_for_test();
        let base = edge_store_from_tuples(4, &[(0, 1, 1), (0, 2, 1), (3, 0, 1)]);
        tx_delta::record_deleted_node(0).expect("record source node delete");
        let source_deleted = LayeredNeighbors::new(&base, Vec::new());

        assert_eq!(source_deleted.neighbors(0).collect::<Vec<_>>(), Vec::new());

        tx_delta::clear_for_test();
        tx_delta::record_deleted_node(2).expect("record target node delete");
        let target_deleted = LayeredNeighbors::new(&base, Vec::new());

        assert_eq!(
            target_deleted.neighbors(0).collect::<Vec<_>>(),
            vec![Neighbor {
                target: 1,
                type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
                schema_reversed: false,
                relationship_id: None,
            }]
        );
        tx_delta::clear_for_test();
    }

    #[test]
    fn weighted_neighbors_preserve_transaction_local_insert_weights() {
        tx_delta::clear_for_test();
        let base = weighted_edge_store_from_tuples(4, &[(0, 1, 1, 10)]);
        tx_delta::record_added_edge(
            0,
            tx_delta::DeltaEdge {
                target: 2,
                type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
                weight: Some(4),
                schema_reversed: false,
                relationship_id: None,
            },
        )
        .expect("record weighted tx insert");
        let layered = LayeredNeighbors::new(&base, Vec::new());

        assert_eq!(
            layered.weighted_neighbors(0),
            vec![
                WeightedNeighbor {
                    target: 1,
                    type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
                    relationship_id: None,
                    weight: 10,
                    schema_reversed: false,
                },
                WeightedNeighbor {
                    target: 2,
                    type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
                    relationship_id: None,
                    weight: 4,
                    schema_reversed: false,
                },
            ]
        );
        tx_delta::clear_for_test();
    }

    #[test]
    fn layered_reads_apply_tenant_filter_and_node_visibility_segments() {
        let base = edge_store_from_tuples(4, &[(0, 1, 1), (0, 2, 1), (0, 3, 1)]);
        let tenant_hash = xxhash_rust::xxh3::xxh3_64(b"tenant-42");
        let mut node_segment =
            DeltaSegment::new(SegmentKind::Node, 0, TraversalDirection::Any, 0, 4, 1)
                .expect("node segment");
        node_segment.tenants.push(SegmentTenant {
            node_idx: 0,
            tenant_hash,
            tenant: Some("tenant-42".to_string()),
            tombstone: false,
        });
        node_segment.node_states.push(SegmentNodeState {
            node_idx: 2,
            active: false,
        });
        node_segment.tenants.push(SegmentTenant {
            node_idx: 1,
            tenant_hash,
            tenant: Some("tenant-42".to_string()),
            tombstone: false,
        });
        node_segment.tenants.push(SegmentTenant {
            node_idx: 2,
            tenant_hash,
            tenant: Some("tenant-42".to_string()),
            tombstone: false,
        });
        let layered =
            LayeredNeighbors::new_with_tenant(&base, vec![node_segment], Some(tenant_hash));

        assert_eq!(
            layered.neighbors(0).collect::<Vec<_>>(),
            vec![Neighbor {
                target: 1,
                type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
                schema_reversed: false,
                relationship_id: None,
            }]
        );
    }

    #[test]
    fn manifest_segment_provider_loads_real_segments() {
        let dir = ProjectionArtifactDir::new("manifest_segment_provider_loads_real_segments");
        let base = edge_store_from_tuples(2, &[]);
        let mut segment = DeltaSegment::new(SegmentKind::Edge, 0, TraversalDirection::Out, 0, 2, 1)
            .expect("segment");
        segment.edge_inserts.push(SegmentEdge {
            source: 0,
            target: 1,
            type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
            schema_reversed: false,
            relationship_id: None,
        });
        let segment_path = dir.segment_path(1, 0);
        segment
            .write_to_path(&segment_path)
            .expect("segment writes");
        let manifest = manifest_with_segment(&dir, 1, &segment_path, &segment);
        let provider = ManifestSegmentProvider::new(dir.path(), &manifest);
        let layered =
            LayeredNeighbors::from_provider(&base, &provider).expect("provider loads segments");

        assert_eq!(
            layered.neighbors(0).collect::<Vec<_>>(),
            vec![Neighbor {
                target: 1,
                type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
                schema_reversed: false,
                relationship_id: None,
            }]
        );
    }

    #[test]
    fn manifest_segment_provider_rejects_checksum_mismatch() {
        let dir = ProjectionArtifactDir::new("manifest_segment_provider_rejects_checksum_mismatch");
        let base = edge_store_from_tuples(2, &[]);
        let mut segment = DeltaSegment::new(SegmentKind::Edge, 0, TraversalDirection::Out, 0, 2, 1)
            .expect("segment");
        segment.edge_inserts.push(SegmentEdge {
            source: 0,
            target: 1,
            type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
            schema_reversed: false,
            relationship_id: None,
        });
        let segment_path = dir.segment_path(1, 0);
        segment
            .write_to_path(&segment_path)
            .expect("segment writes");
        let mut manifest = manifest_with_segment(&dir, 1, &segment_path, &segment);
        manifest.segments[0].checksum = "crc32:00000000".to_string();
        let provider = ManifestSegmentProvider::new(dir.path(), &manifest);

        let err = match LayeredNeighbors::from_provider(&base, &provider) {
            Ok(_) => panic!("checksum mismatch should reject segment"),
            Err(err) => err,
        };

        assert!(matches!(err, GraphError::CorruptFile { .. }));
    }

    #[test]
    fn durable_delete_then_later_insert_reactivates_edge() {
        let base = edge_store_from_tuples(2, &[(0, 1, 1)]);
        let mut delete = DeltaSegment::new(SegmentKind::Edge, 0, TraversalDirection::Out, 0, 2, 1)
            .expect("delete segment");
        delete.edge_deletes.push(SegmentEdge {
            source: 0,
            target: 1,
            type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
            schema_reversed: false,
            relationship_id: None,
        });
        let mut insert = DeltaSegment::new(SegmentKind::Edge, 0, TraversalDirection::Out, 0, 2, 2)
            .expect("insert segment");
        insert.edge_inserts.push(SegmentEdge {
            source: 0,
            target: 1,
            type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
            schema_reversed: false,
            relationship_id: None,
        });
        let layered = LayeredNeighbors::new(&base, vec![insert, delete]);

        assert_eq!(
            layered.neighbors(0).collect::<Vec<_>>(),
            vec![Neighbor {
                target: 1,
                type_id: EdgeTypeId::from_v6_storage(1).expect("fixture type ID is valid v6"),
                schema_reversed: false,
                relationship_id: None,
            }]
        );
    }

    proptest! {
        #[test]
        fn layered_without_segments_matches_base_csr(
            node_count in 1u32..16,
            raw_edges in prop::collection::vec((0u32..16, 0u32..16, 0u8..4), 0..96),
            query_node in 0u32..16,
        ) {
            tx_delta::clear_for_test();
            let edges = raw_edges
                .into_iter()
                .filter(|(source, target, _)| *source < node_count && *target < node_count)
                .collect::<Vec<_>>();
            let base = edge_store_from_tuples(node_count, &edges);
            let csr = CsrNeighbors::new(&base);
            let layered = LayeredNeighbors::new(&base, Vec::new());

            prop_assert_eq!(
                csr.neighbors(query_node).collect::<Vec<_>>(),
                layered.neighbors(query_node).collect::<Vec<_>>()
            );
            prop_assert_eq!(
                csr.neighbors_reversed(query_node).collect::<Vec<_>>(),
                layered.neighbors_reversed(query_node).collect::<Vec<_>>()
            );
        }
    }

    fn manifest_with_segment(
        dir: &ProjectionArtifactDir,
        generation_id: u64,
        segment_path: &Path,
        segment: &DeltaSegment,
    ) -> ProjectionManifest {
        let relative = segment_path
            .strip_prefix(dir.path())
            .expect("segment is inside artifact dir")
            .to_string_lossy()
            .to_string();
        let bytes = fs::read(segment_path).expect("segment bytes read");
        let mut manifest = ProjectionManifest::base_only(
            generation_id,
            "base.pggraph",
            "crc32:base",
            1,
            segment.header.sync_watermark,
            1,
        );
        manifest.segments.push(ManifestSegmentRef {
            path: relative,
            checksum: format!("crc32:{:08x}", crc32fast::hash(&bytes)),
            level: segment.header.level,
            source_start: segment.header.source_start,
            source_end: segment.header.source_end,
            sync_watermark: segment.header.sync_watermark,
        });
        manifest
    }
}
