fnx_classes/

digraph.rs

//! Directed graph (DiGraph) storage.
//!
//! Mirrors the undirected [`Graph`] API with directed semantics:
//! - Edge `(u, v)` is distinct from `(v, u)`.
//! - Adjacency is split into **successors** (outgoing) and **predecessors** (incoming).
//! - `neighbors(n)` returns successors (matching NetworkX convention).

use crate::{AttrMap, EdgeSnapshot, FxIndexMap, GraphError};
use fnx_runtime::{
    CgseValue, CompatibilityMode, DecisionAction, DecisionRecord, EvidenceLedger, EvidenceTerm,
    RuntimePolicy, bool_evidence, count_evidence,
};
use indexmap::{IndexMap, IndexSet};
use serde::{Deserialize, Serialize};

// ---------------------------------------------------------------------------
// DirectedEdgeKey — order-preserving (NOT canonicalized)
// ---------------------------------------------------------------------------

#[derive(Debug, Clone, PartialEq, Eq, Hash)]
struct DirectedEdgeKey {
    source: String,
    target: String,
}

impl DirectedEdgeKey {
    fn new(source: &str, target: &str) -> Self {
        Self {
            source: source.to_owned(),
            target: target.to_owned(),
        }
    }
}

#[derive(Hash, PartialEq, Eq, Clone, Copy)]
struct DirectedEdgeKeyRef<'a> {
    source: &'a str,
    target: &'a str,
}

impl<'a> DirectedEdgeKeyRef<'a> {
    fn new(source: &'a str, target: &'a str) -> Self {
        Self { source, target }
    }
}

impl<'a> indexmap::Equivalent<DirectedEdgeKey> for DirectedEdgeKeyRef<'a> {
    fn equivalent(&self, key: &DirectedEdgeKey) -> bool {
        self.source == key.source && self.target == key.target
    }
}

// ---------------------------------------------------------------------------
// DiGraphSnapshot
// ---------------------------------------------------------------------------

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct DiGraphSnapshot {
    pub mode: CompatibilityMode,
    pub nodes: Vec<String>,
    /// br-snapnodeattrs: per-node attributes preserved across snapshot
    /// round-trip. Defaults to empty for backward compatibility.
    #[serde(default)]
    pub node_attrs: std::collections::BTreeMap<String, AttrMap>,
    /// Edges in source→target order. `left` = source, `right` = target.
    pub edges: Vec<EdgeSnapshot>,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct MultiDiEdgeSnapshot {
    pub source: String,
    pub target: String,
    pub key: usize,
    pub attrs: AttrMap,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct MultiDiGraphSnapshot {
    pub mode: CompatibilityMode,
    pub nodes: Vec<String>,
    /// See DiGraphSnapshot::node_attrs.
    #[serde(default)]
    pub node_attrs: std::collections::BTreeMap<String, AttrMap>,
    pub edges: Vec<MultiDiEdgeSnapshot>,
}

// ---------------------------------------------------------------------------
// DiGraph
// ---------------------------------------------------------------------------

/// br-r37-c1-d58s8 P1: integer CSR view of a DiGraph's adjacency —
/// succ/pred targets are node INDICES (IndexMap insertion order) with
/// per-row slices in ROW order, so index walks reproduce the String
/// walks exactly. Built once per revision via `DiGraph::csr()`.
#[derive(Debug)]
pub struct DiCsr {
    pub succ_offsets: Vec<usize>,
    pub succ_targets: Vec<u32>,
    pub pred_offsets: Vec<usize>,
    pub pred_targets: Vec<u32>,
}

impl DiCsr {
    #[must_use]
    pub fn successors(&self, idx: usize) -> &[u32] {
        &self.succ_targets[self.succ_offsets[idx]..self.succ_offsets[idx + 1]]
    }

    #[must_use]
    pub fn predecessors(&self, idx: usize) -> &[u32] {
        &self.pred_targets[self.pred_offsets[idx]..self.pred_offsets[idx + 1]]
    }
}

type DiCsrCache = std::sync::Arc<std::sync::RwLock<Option<(u64, std::sync::Arc<DiCsr>)>>>;
type DiAllIntCache = std::sync::Arc<std::sync::RwLock<Option<(u64, String, bool)>>>;

/// Revision-keyed integer CSR view of a MultiDiGraph's distinct successor and
/// predecessor rows. Parallel-edge keys are intentionally collapsed because this
/// view is for multiplicity-insensitive structural algorithms.
#[derive(Debug)]
pub struct MultiDiCsr {
    pub succ_offsets: Vec<usize>,
    pub succ_targets: Vec<u32>,
    pub pred_offsets: Vec<usize>,
    pub pred_targets: Vec<u32>,
}

impl MultiDiCsr {
    #[must_use]
    pub fn successors(&self, idx: usize) -> &[u32] {
        &self.succ_targets[self.succ_offsets[idx]..self.succ_offsets[idx + 1]]
    }

    #[must_use]
    pub fn predecessors(&self, idx: usize) -> &[u32] {
        &self.pred_targets[self.pred_offsets[idx]..self.pred_offsets[idx + 1]]
    }
}

type MultiDiCsrCache = std::sync::Arc<std::sync::RwLock<Option<(u64, std::sync::Arc<MultiDiCsr>)>>>;

/// br-r37-c1-s8dj1: node POSITION pair -> position in `edges`.
type EdgePairMap = std::collections::HashMap<(usize, usize), usize, rustc_hash::FxBuildHasher>;

/// br-r37-c1-s8dj1: the pair lookaside, stamped with the graph revision.
///
/// CLONES COLD, AND IS NEVER SHARED. `MultiDiGraph` derives `Clone`, which would
/// otherwise hand both copies the same cache. The stamp does not rescue an
/// INCREMENTALLY maintained map from that: the two graphs' revisions advance
/// independently and can COINCIDE, so the clone updates the shared map and
/// stamps its own revision, and the original then reads a map describing a
/// different graph. A cold clone costs one rebuild and removes the class.
#[derive(Debug, Default)]
pub(crate) struct EdgePairIndex {
    map: std::sync::RwLock<Option<(u64, EdgePairMap)>>,
    /// br-r37-c1-s8dj1: the revision the warm-up counter belongs to.
    probed_revision: std::sync::atomic::AtomicU64,
    /// Probes seen at that revision without an index. See `PAIR_INDEX_WARMUP`.
    probes_since_build: std::sync::atomic::AtomicUsize,
}

impl Clone for EdgePairIndex {
    fn clone(&self) -> Self {
        Self::default()
    }
}

/// br-r37-c1-s8dj1: probes at one revision before the O(E) map is built.
///
/// THE GUARD EXISTS BECAUSE THE BUILD IS NOT FREE. Building costs a hash per
/// endpoint of every edge, so a caller that probes a large graph a HANDFUL of
/// times would pay O(E) to save O(node key length) — measured at E=8000 as
/// 6308 ns/probe against 592 ns for the undirected class, a loss on exactly the
/// sparse-probing shape that never benefits.
///
/// SO THE THRESHOLD SCALES WITH THE BUILD IT AMORTISES, and a constant will not
/// do: 64 probes justify a build at E=100 and are nowhere near justifying one at
/// E=8000. Waiting until the caller has made roughly half as many probes as the
/// build costs entries bounds the waste at a small constant factor of the work
/// already spent, which is the ordinary amortisation rule. Below the threshold
/// the string path answers, exactly as the class did before this index existed.
const PAIR_INDEX_WARMUP_FLOOR: usize = 64;

#[inline]
fn pair_index_warmup(edge_pairs: usize) -> usize {
    PAIR_INDEX_WARMUP_FLOOR.max(edge_pairs / 2)
}

#[derive(Debug, Clone)]
pub struct DiGraph {
    mode: CompatibilityMode,
    revision: u64,
    /// name <-> slot and the insertion order (see `crate::NodeOrder`).
    node_order: crate::NodeOrder,
    node_attrs: Vec<AttrMap>,
    /// slot -> successor slots / predecessor slots (see `crate::SlotRows`).
    succ_indices: crate::SlotRows,
    pred_indices: crate::SlotRows,
    edges: crate::FxIndexMap<(usize, usize), AttrMap>,
    runtime_policy: RuntimePolicy,
    csr_cache: DiCsrCache,
    all_int_cache: DiAllIntCache,
}

impl DiGraph {
    /// br-r37-c1-clearedgesinplace (cc): drop all edges in place, keeping nodes (index,
    /// order, attrs). O(E)+O(V): clear the edges map + every succ/pred index row; the
    /// revision bump invalidates the CSR + all-int caches. Sibling of Graph::clear_edges
    /// and MultiDiGraph::clear_edges. The binding's prior path collected the mirror keys
    /// and called remove_edge per edge (each an O(degree) row retain -> O(E*degree),
    /// measured ~0.003-0.011x vs nx); this is the in-place fix.
    pub fn clear_edges(&mut self) {
        if self.edges.is_empty() {
            return;
        }
        self.edges.clear();
        for row in self.succ_indices.rows_mut() {
            row.clear();
        }
        for row in self.pred_indices.rows_mut() {
            row.clear();
        }
        self.revision = self.revision.saturating_add(1);
    }

    /// br-r37-c1-7dpyg: structural clone with a FRESH RuntimePolicy —
    /// see Graph::clone_with_fresh_policy.
    #[must_use]
    pub fn clone_with_fresh_policy(&self) -> Self {
        Self {
            mode: self.mode,
            revision: self.revision,
            node_order: self.node_order.clone(),
            node_attrs: self.node_attrs.clone(),
            succ_indices: self.succ_indices.clone(),
            pred_indices: self.pred_indices.clone(),
            edges: self.edges.clone(),
            runtime_policy: RuntimePolicy::new(self.mode),
            csr_cache: self.csr_cache.clone(),
            all_int_cache: self.all_int_cache.clone(),
        }
    }

    // -----------------------------------------------------------------------
    // Constructors
    // -----------------------------------------------------------------------

    #[must_use]
    pub fn new(mode: CompatibilityMode) -> Self {
        Self {
            mode,
            revision: 0,
            node_order: crate::NodeOrder::default(),
            node_attrs: Vec::new(),
            succ_indices: crate::SlotRows::default(),
            pred_indices: crate::SlotRows::default(),
            edges: crate::FxIndexMap::default(),
            runtime_policy: RuntimePolicy::new(mode),
            csr_cache: std::sync::Arc::default(),
            all_int_cache: std::sync::Arc::default(),
        }
    }

    #[must_use]
    pub fn with_runtime_policy(runtime_policy: RuntimePolicy) -> Self {
        let mode = runtime_policy.mode();
        Self {
            mode,
            revision: 0,
            node_order: crate::NodeOrder::default(),
            node_attrs: Vec::new(),
            succ_indices: crate::SlotRows::default(),
            pred_indices: crate::SlotRows::default(),
            edges: crate::FxIndexMap::default(),
            runtime_policy,
            csr_cache: std::sync::Arc::default(),
            all_int_cache: std::sync::Arc::default(),
        }
    }

    /// br-r37-c1-indsub: directed twin of `Graph::induced_subgraph_ordered` —
    /// build the induced subgraph over `order` (parent node indices, already in
    /// the caller's output order) with no node-name hashing.
    ///
    /// `add_edge_with_attrs` costs two `String` allocations, four name hashes
    /// and TWO `record_decision` ledger entries per arc (the second carrying
    /// four `EvidenceTerm`s, eight more String allocations). Everything it
    /// re-derives is already known here. Invariants reproduced directly:
    /// `edges` keyed by the ORIENTED new-index pair (direction is identity for
    /// a directed edge, so no canonicalization), and a succ/pred row records
    /// its endpoint when the arc is emitted, so both row orders are the walk
    /// order — matching nx's `add_edges_from` over the induced walk.
    ///
    /// `order` must contain distinct, in-range parent indices.
    #[must_use]
    pub fn induced_subgraph_ordered(&self, order: &[usize], runtime_policy: RuntimePolicy) -> Self {
        let mut position = vec![usize::MAX; self.node_order.slot_count()];
        for (slot, &parent_idx) in order.iter().enumerate() {
            let parent_slot = self.node_order.get_index(parent_idx).unwrap().1;
            position[parent_slot] = slot;
        }

        let mut node_order = crate::NodeOrder::default();
        node_order.reserve(order.len());
        let mut node_attrs = Vec::with_capacity(order.len());

        for &parent_idx in order {
            let (name, parent_slot) = self
                .node_order
                .get_index(parent_idx)
                .expect("induced subgraph order must hold valid parent node indices");
            node_order.push(name.clone());
            node_attrs.push(self.node_attrs[parent_slot].clone());
        }

        let mut succ_indices = vec![Vec::new(); order.len()];
        let mut pred_indices = vec![Vec::new(); order.len()];
        let mut edges = crate::FxIndexMap::default();
        for (new_slot, &parent_idx) in order.iter().enumerate() {
            let parent_slot = self.node_order.get_index(parent_idx).unwrap().1;
            for &parent_target in &self.succ_indices[parent_slot] {
                let other = position[parent_target];
                if other == usize::MAX {
                    continue;
                }
                let attrs = self
                    .edges
                    .get(&(parent_slot, parent_target))
                    .cloned()
                    .unwrap_or_default();
                edges.insert((new_slot, other), attrs);
                succ_indices[new_slot].push(other);
                pred_indices[other].push(new_slot);
            }
        }

        Self {
            mode: runtime_policy.mode(),
            revision: 0,
            node_order,
            node_attrs,
            succ_indices: crate::SlotRows::new(succ_indices),
            pred_indices: crate::SlotRows::new(pred_indices),
            edges,
            runtime_policy,
            csr_cache: std::sync::Arc::default(),
            all_int_cache: std::sync::Arc::default(),
        }
    }

    #[must_use]
    pub fn strict() -> Self {
        Self::new(CompatibilityMode::Strict)
    }

    #[must_use]
    pub fn hardened() -> Self {
        Self::new(CompatibilityMode::Hardened)
    }

    // -----------------------------------------------------------------------
    // Read-only queries
    // -----------------------------------------------------------------------

    #[must_use]
    pub fn mode(&self) -> CompatibilityMode {
        self.mode
    }

    #[must_use]
    pub fn node_count(&self) -> usize {
        self.node_order.len()
    }

    #[must_use]
    pub fn edge_count(&self) -> usize {
        self.edges.len()
    }

    #[must_use]
    pub fn revision(&self) -> u64 {
        self.revision
    }

    /// br-r37-c1-d58s8 P1: revision-keyed integer CSR of both row
    /// families. O(V+E) rebuild only when the graph changed since the
    /// cached build; kernels get index walks identical to the String
    /// walks (per-row order preserved).
    #[must_use]
    pub fn csr(&self) -> std::sync::Arc<DiCsr> {
        if let Ok(guard) = self.csr_cache.read()
            && let Some((rev, csr)) = guard.as_ref()
            && *rev == self.revision
        {
            return csr.clone();
        }
        let built = std::sync::Arc::new(self.build_csr());
        if let Ok(mut guard) = self.csr_cache.write() {
            *guard = Some((self.revision, built.clone()));
        }
        built
    }

    /// br-r37-c1-d58s8: revision-keyed memo of "every edge's `attr`
    /// value is an integer (missing = default 1 = int)".
    #[must_use]
    pub fn edge_weights_all_int(&self, attr: &str) -> bool {
        if let Ok(guard) = self.all_int_cache.read()
            && let Some((rev, a, v)) = guard.as_ref()
            && *rev == self.revision
            && a == attr
        {
            return *v;
        }
        let v = self
            .edges
            .values()
            .all(|attrs| attrs.get(attr).is_none_or(fnx_runtime::CgseValue::is_int));
        if let Ok(mut guard) = self.all_int_cache.write() {
            *guard = Some((self.revision, attr.to_owned(), v));
        }
        v
    }

    fn build_csr(&self) -> DiCsr {
        // br-r37-c1-d58s8 DiGraph flip P1: build from the EAGER index rows
        // — zero String hashing (was a get_index_of per neighbor).
        let n = self.node_order.len();
        let mut succ_offsets = Vec::with_capacity(n + 1);
        let mut pred_offsets = Vec::with_capacity(n + 1);
        let mut succ_targets = Vec::new();
        let mut pred_targets = Vec::new();
        succ_offsets.push(0);
        pred_offsets.push(0);

        let mut slot_to_pos = vec![0; self.node_order.slot_count()];
        for (pos, (_, slot)) in self.node_order.iter().enumerate() {
            slot_to_pos[slot] = pos;
        }

        for slot in self.node_order.values() {
            succ_targets.extend(
                self.succ_indices[slot]
                    .iter()
                    .map(|&v| u32::try_from(slot_to_pos[v]).unwrap_or(u32::MAX)),
            );
            succ_offsets.push(succ_targets.len());
            pred_targets.extend(
                self.pred_indices[slot]
                    .iter()
                    .map(|&u| u32::try_from(slot_to_pos[u]).unwrap_or(u32::MAX)),
            );
            pred_offsets.push(pred_targets.len());
        }
        DiCsr {
            succ_offsets,
            succ_targets,
            pred_offsets,
            pred_targets,
        }
    }

    /// br-r37-c1-d58s8: consistency oracle — eager index rows must mirror
    /// the String rows exactly (order included). Test-only.
    #[doc(hidden)]
    #[must_use]
    pub fn debug_index_rows_consistent(&self) -> bool {
        // br-r37-c1-d58s8 P3: the String rows this oracle compared
        // against are GONE; the index rows are the single store. Length
        // sanity only (parity batteries cover content).
        self.succ_indices.len() == self.node_order.slot_count()
            && self.pred_indices.len() == self.node_order.slot_count()
    }

    #[must_use]
    pub fn has_node(&self, node: &str) -> bool {
        self.node_order.contains_key(node)
    }

    /// Check for directed edge source→target.
    #[must_use]
    /// br-r37-c1-d58s8: resolve a String pair to the ORIENTED index key.
    #[inline]
    fn edge_pair_key(&self, source: &str, target: &str) -> Option<(usize, usize)> {
        Some((*self.node_order.get(source)?, *self.node_order.get(target)?))
    }

    /// br-r37-c1-1l8s0: direct slice access to the eager succ index row, by
    /// node position (through the position view while slots are not dense).
    #[must_use]
    #[inline]
    pub fn successors_indices(&self, idx: usize) -> Option<&[usize]> {
        if self.node_order.is_dense() {
            self.succ_indices.get(idx).map(Vec::as_slice)
        } else {
            self.succ_indices
                .positions(&self.node_order)
                .get(idx)
                .map(Vec::as_slice)
        }
    }

    /// Direct slice access to the eager pred index row, by node position.
    #[must_use]
    #[inline]
    pub fn predecessors_indices(&self, idx: usize) -> Option<&[usize]> {
        if self.node_order.is_dense() {
            self.pred_indices.get(idx).map(Vec::as_slice)
        } else {
            self.pred_indices
                .positions(&self.node_order)
                .get(idx)
                .map(Vec::as_slice)
        }
    }

    pub fn has_edge(&self, source: &str, target: &str) -> bool {
        self.edge_pair_key(source, target)
            .is_some_and(|k| self.edges.contains_key(&k))
    }

    /// br-r37-c1-04z53 (cc): directed `has_edge` straight from insertion-order
    /// indices (source-major), skipping the `i.to_string()` heap alloc for int
    /// nodes. Names come from the node table by index (borrowed); the caller
    /// guards each index with `node_index_matches_int`. Mirror of
    /// `Graph::has_edge_by_indices`.
    ///
    /// br-r37-c1-p1tvg: this used to resolve both indices back to NAMES and
    /// call `has_edge`, which hashes two Strings to rediscover the very
    /// indices it was handed — exactly the cost a caller pays an index route
    /// to avoid. The edges map has been keyed by the ORIENTED
    /// `(source_idx, target_idx)` pair since the d58s8 flip, so the pair IS
    /// the key. An out-of-range index still answers `false`: every key in the
    /// map names two live nodes, so no key can carry one.
    #[must_use]
    pub fn has_edge_by_indices(&self, source_idx: usize, target_idx: usize) -> bool {
        let Some(s_slot) = self.node_order.slot_at(source_idx) else {
            return false;
        };
        let Some(t_slot) = self.node_order.slot_at(target_idx) else {
            return false;
        };
        self.edges.contains_key(&(s_slot, t_slot))
    }

    #[must_use]
    pub fn nodes_ordered(&self) -> Vec<&str> {
        self.node_order.keys().map(String::as_str).collect()
    }

    /// br-cc-nbunchbulk: does the node at contiguous index `idx` have the exact int
    /// name `idx`? O(1) index probe + no-alloc parse — the int membership fast path
    /// for `_nbunch_present` (sibling of the undirected Graph accessor).
    #[must_use]
    pub fn node_index_matches_int(&self, idx: usize) -> bool {
        self.node_order
            .get_index(idx)
            .is_some_and(|(k, _)| k.parse::<usize>() == Ok(idx))
    }

    /// sfq4w.3: see `Graph::node_names_are_positions`.
    #[must_use]
    pub fn node_names_are_positions(&self) -> bool {
        self.node_order.names_are_positions()
    }

    #[must_use]
    pub fn get_node_index(&self, node: &str) -> Option<usize> {
        self.node_order.get_index_of(node)
    }

    #[must_use]
    pub fn get_node_name(&self, index: usize) -> Option<&str> {
        self.node_order.get_index(index).map(|(k, _)| k.as_str())
    }

    // -- Directed adjacency queries ----------------------------------------

    /// Successors of `node` (outgoing neighbors). Returns `None` if node absent.
    #[must_use]
    pub fn successors(&self, node: &str) -> Option<Vec<&str>> {
        // br-r37-c1-d58s8 DiGraph flip P2: serve from the eager index
        // rows + name table (order-faithful; oracle-validated in P1).
        let &idx = self.node_order.get(node)?;
        Some(
            self.succ_indices[idx]
                .iter()
                .map(|&i| {
                    self.node_order.names[i]
                        .as_deref()
                        .expect("index rows hold valid node slots")
                })
                .collect(),
        )
    }

    #[must_use]
    pub fn successors_iter(&self, node: &str) -> Option<impl Iterator<Item = &str> + '_> {
        // br-r37-c1-d58s8 DiGraph flip P2: index-backed.
        let &idx = self.node_order.get(node)?;
        Some(self.succ_indices[idx].iter().map(move |&i| {
            self.node_order.names[i]
                .as_deref()
                .expect("index rows hold valid node slots")
        }))
    }

    /// yr2oc.4: the attribute maps of `node`'s out-edges in successor-row
    /// order. Read by slot: no position translation after a removal.
    #[must_use]
    pub fn out_edge_attrs(&self, node: &str) -> Option<impl Iterator<Item = &AttrMap> + '_> {
        let &slot = self.node_order.get(node)?;
        Some(self.succ_indices[slot].iter().map(move |&target| {
            self.edges
                .get(&(slot, target))
                .expect("successor rows have matching edge storage")
        }))
    }

    /// yr2oc.4: the attribute maps of `node`'s in-edges in predecessor-row
    /// order. Read by slot: no position translation after a removal.
    #[must_use]
    pub fn in_edge_attrs(&self, node: &str) -> Option<impl Iterator<Item = &AttrMap> + '_> {
        let &slot = self.node_order.get(node)?;
        Some(self.pred_indices[slot].iter().map(move |&source| {
            self.edges
                .get(&(source, slot))
                .expect("predecessor rows have matching edge storage")
        }))
    }

    /// Predecessors of `node` (incoming neighbors). Returns `None` if node absent.
    #[must_use]
    pub fn predecessors(&self, node: &str) -> Option<Vec<&str>> {
        // br-r37-c1-d58s8 DiGraph flip P2: index-backed.
        let &idx = self.node_order.get(node)?;
        Some(
            self.pred_indices[idx]
                .iter()
                .map(|&i| {
                    self.node_order.names[i]
                        .as_deref()
                        .expect("index rows hold valid node slots")
                })
                .collect(),
        )
    }

    /// br-r37-c1-u3qyn: restore explicit succ/pred row orders (pickle
    /// round-trip) — see Graph::apply_row_orders. `which` selects the
    /// adjacency side.
    pub fn apply_row_orders(&mut self, orders: &[(String, Vec<String>)], pred: bool) {
        // br-r37-c1-d58s8 P3: pure integer-row reorder per family.
        for (node, order) in orders {
            let Some(&idx) = self.node_order.get(node.as_str()) else {
                continue;
            };
            let rows = if pred {
                &mut self.pred_indices
            } else {
                &mut self.succ_indices
            };
            let row = &rows[idx];
            let row_set: std::collections::HashSet<usize> = row.iter().copied().collect();
            let mut placed: std::collections::HashSet<usize> = std::collections::HashSet::new();
            let mut new_row: Vec<usize> = Vec::with_capacity(row.len());
            for v in order {
                if let Some(&v_idx) = self.node_order.get(v.as_str())
                    && row_set.contains(&v_idx)
                    && placed.insert(v_idx)
                {
                    new_row.push(v_idx);
                }
            }
            for &v_idx in row {
                if placed.insert(v_idx) {
                    new_row.push(v_idx);
                }
            }
            rows.rows_mut()[idx] = new_row;
        }
    }

    /// br-r37-c1-0ek49: reorder every PRED row into NetworkX's
    /// `DiGraph.copy()` walk order. nx copy rebuilds via the u-major
    /// succ walk (`(u, v, d) for u in _adj for v in _adj[u]`), which
    /// recreates succ rows in their original order but fills each pred
    /// row in walk order: `(pos(u), index of v within succ[u])`. Call on
    /// a fresh clone inside copy-shaped constructors; graph content is
    /// unchanged, only pred row order moves.
    pub fn reorder_pred_rows_for_nx_copy_walk(&mut self) {
        // br-r37-c1-predrebuild: rebuild pred rows in nx's u-major copy-walk order
        // DIRECTLY from succ rows in O(E), replacing the prior per-pred-entry
        // linear search `succ_indices[u].position(|w| w == v)` (O(E * degree) —
        // quadratic on dense graphs). Iterating succ in (u, succ-index) order and
        // appending u to pred[v] reproduces exactly the
        // (pos(u), index-of-v-within-succ[u]) ordering the old sort produced,
        // including parallel-edge multiplicity — byte-identical pred row order.
        //
        // Walking the rows in slot order is position order only once slots are
        // positions.
        self.ensure_compact();
        let mut new_rows: Vec<Vec<usize>> = self
            .pred_indices
            .iter()
            .map(|row| Vec::with_capacity(row.len()))
            .collect();
        for (u, succ_row) in self.succ_indices.iter().enumerate() {
            for &v in succ_row {
                new_rows[v].push(u);
            }
        }
        *self.pred_indices.rows_mut() = new_rows;
    }

    /// br-r37-c1-36v6r: order each successor and predecessor row as
    /// `source`'s row for the node of the same name - its succ / pred rows,
    /// or its pred / succ rows when `swap` (a reverse view's). A graph built
    /// from a filtered or reverse view of `source` by a copy-shaped walk
    /// fills pred rows in succ-walk order (networkx's `view.copy()` does the
    /// same), where the view's pred row is `source`'s row filtered. Entries
    /// `source`'s row lacks keep their relative order after the matched
    /// ones. Graph content is unchanged; `revision` moves if any row does.
    pub fn reorder_rows_like(&mut self, source: &Self, swap: bool) {
        let mut to_source = vec![usize::MAX; self.node_order.slot_count()];
        let mut from_source = vec![usize::MAX; source.node_order.slot_count()];
        for (name, slot) in self.node_order.iter() {
            if let Some(&source_slot) = source.node_order.get(name) {
                to_source[slot] = source_slot;
                from_source[source_slot] = slot;
            }
        }
        let (succ_source, pred_source) = if swap {
            (&source.pred_indices, &source.succ_indices)
        } else {
            (&source.succ_indices, &source.pred_indices)
        };
        let slots: Vec<usize> = self.node_order.values().collect();
        let succ_changed = crate::reorder_slot_rows_like(
            &mut self.succ_indices,
            succ_source,
            &slots,
            &to_source,
            &from_source,
        );
        let pred_changed = crate::reorder_slot_rows_like(
            &mut self.pred_indices,
            pred_source,
            &slots,
            &to_source,
            &from_source,
        );
        if succ_changed || pred_changed {
            self.revision = self.revision.saturating_add(1);
            // Clones share the CSR memo, keyed by revision alone.
            self.csr_cache = std::sync::Arc::default();
        }
    }

    /// br-r37-c1-36v6r: order every successor and predecessor row as given -
    /// `succ_rows[i]` / `pred_rows[i]` are the rows of the node at position
    /// `i`, neighbours as positions - for a row order no graph holds, such
    /// as a `to_directed(as_view=True)` view's (both rows the undirected
    /// graph's row). All or nothing: false, and nothing changes, unless every
    /// given row is a permutation of the current one.
    pub fn set_row_orders(&mut self, succ_rows: &[Vec<usize>], pred_rows: &[Vec<usize>]) -> bool {
        let slots: Vec<usize> = self.node_order.values().collect();
        let mut succ = self.succ_indices.clone();
        let mut pred = self.pred_indices.clone();
        let (Some(succ_changed), Some(pred_changed)) = (
            crate::set_slot_row_orders(&mut succ, &slots, succ_rows),
            crate::set_slot_row_orders(&mut pred, &slots, pred_rows),
        ) else {
            return false;
        };
        if succ_changed || pred_changed {
            self.succ_indices = succ;
            self.pred_indices = pred;
            self.revision = self.revision.saturating_add(1);
            self.csr_cache = std::sync::Arc::default();
        }
        true
    }

    #[must_use]
    pub fn predecessors_iter(&self, node: &str) -> Option<impl Iterator<Item = &str> + '_> {
        // br-r37-c1-d58s8 DiGraph flip P2: index-backed.
        let &idx = self.node_order.get(node)?;
        Some(self.pred_indices[idx].iter().map(move |&i| {
            self.node_order.names[i]
                .as_deref()
                .expect("index rows hold valid node slots")
        }))
    }

    /// Neighbors = successors (matches NetworkX `DiGraph.neighbors()` convention).
    #[must_use]
    pub fn neighbors(&self, node: &str) -> Option<Vec<&str>> {
        self.successors(node)
    }

    #[must_use]
    pub fn neighbors_iter(&self, node: &str) -> Option<impl Iterator<Item = &str> + '_> {
        self.successors_iter(node)
    }

    #[must_use]
    pub fn neighbor_count(&self, node: &str) -> usize {
        // br-r37-c1-d58s8 DiGraph flip P2: index-backed.
        self.node_order
            .get(node)
            .map_or(0, |&i| self.succ_indices[i].len())
    }

    /// Out-degree: number of successors.
    #[must_use]
    pub fn out_degree(&self, node: &str) -> usize {
        self.node_order
            .get(node)
            .map_or(0, |&i| self.succ_indices[i].len())
    }

    /// In-degree: number of predecessors.
    #[must_use]
    pub fn in_degree(&self, node: &str) -> usize {
        self.node_order
            .get(node)
            .map_or(0, |&i| self.pred_indices[i].len())
    }

    /// Total degree: in_degree + out_degree.
    #[must_use]
    pub fn degree(&self, node: &str) -> usize {
        // br-r37-c1-3gt4y: both rows share the same contiguous node index.
        // Resolve the name once instead of paying the IndexMap lookup in both
        // in_degree() and out_degree(). Missing nodes still report degree zero.
        self.node_order.get(node).map_or(0, |&idx| {
            self.pred_indices[idx].len() + self.succ_indices[idx].len()
        })
    }

    /// br-r37-c1-degidx: index-based degree accessors — O(1), zero
    /// String hashing (the &str path costs get_index_of per call;
    /// degree pays it twice). Used by the directed DegreeView iterator
    /// which walks nodes in index order.
    #[must_use]
    pub fn out_degree_by_index(&self, idx: usize) -> usize {
        self.node_order
            .slot_at(idx)
            .map_or(0, |slot| self.succ_indices[slot].len())
    }

    #[must_use]
    pub fn in_degree_by_index(&self, idx: usize) -> usize {
        self.node_order
            .slot_at(idx)
            .map_or(0, |slot| self.pred_indices[slot].len())
    }

    #[must_use]
    pub fn degree_by_index(&self, idx: usize) -> usize {
        self.out_degree_by_index(idx) + self.in_degree_by_index(idx)
    }

    /// Outgoing edges from `node` as (source, target) pairs.
    #[must_use]
    pub fn out_edges<'a>(&'a self, node: &'a str) -> Vec<(&'a str, &'a str)> {
        // br-r37-c1-d58s8 DiGraph flip P2: index-backed.
        self.node_order.get(node).map_or_else(Vec::new, |&idx| {
            self.succ_indices[idx]
                .iter()
                .map(|&i| {
                    (
                        node,
                        self.node_order.names[i]
                            .as_deref()
                            .expect("index rows hold valid node slots"),
                    )
                })
                .collect()
        })
    }

    /// Incoming edges to `node` as (source, target) pairs.
    #[must_use]
    pub fn in_edges<'a>(&'a self, node: &'a str) -> Vec<(&'a str, &'a str)> {
        // br-r37-c1-d58s8 DiGraph flip P2: index-backed.
        self.node_order.get(node).map_or_else(Vec::new, |&idx| {
            self.pred_indices[idx]
                .iter()
                .map(|&i| {
                    (
                        self.node_order.names[i]
                            .as_deref()
                            .expect("index rows hold valid node slots"),
                        node,
                    )
                })
                .collect()
        })
    }

    // -- Attribute queries -------------------------------------------------

    #[must_use]
    pub fn node_attrs(&self, node: &str) -> Option<&AttrMap> {
        let &slot = self.node_order.get(node)?;
        Some(&self.node_attrs[slot])
    }

    #[must_use]
    #[inline]
    pub fn node_attrs_by_index(&self, idx: usize) -> Option<&AttrMap> {
        self.node_order
            .slot_at(idx)
            .map(|slot| &self.node_attrs[slot])
    }

    /// Attributes of directed edge source→target.
    #[must_use]
    pub fn edge_attrs(&self, source: &str, target: &str) -> Option<&AttrMap> {
        self.edges.get(&self.edge_pair_key(source, target)?)
    }

    /// br-r37-c1-d58s8: index-keyed attr access (oriented). Zero
    /// node-map probes for callers already holding indices.
    #[must_use]
    #[inline]
    pub fn edge_attrs_by_indices(&self, source_idx: usize, target_idx: usize) -> Option<&AttrMap> {
        let s_slot = self.node_order.slot_at(source_idx)?;
        let t_slot = self.node_order.slot_at(target_idx)?;
        self.edges.get(&(s_slot, t_slot))
    }

    /// br-r37-c1-hasattrlazyfix: does ANY directed edge carry the attribute `key` in the
    /// authoritative Rust storage (the inner `edges` AttrMaps), regardless of whether the
    /// lazy `edge_py_attrs` Python mirror has been materialized? ~O(E) key checks.
    #[must_use]
    pub fn any_edge_has_attr(&self, key: &str) -> bool {
        self.edges.values().any(|attrs| attrs.contains_key(key))
    }

    /// Directed twin of `Graph::all_edge_attr_values_scalar`: true if EVERY inner
    /// edge attr value is a faithful scalar (Int/Float/Bool). No-alloc scan over
    /// the edge store values — contrast `edges_ordered_borrowed`, which allocates
    /// an O(E) Vec and resolves node names per edge. Used by the reverse gate.
    #[must_use]
    pub fn all_edge_attr_values_scalar(&self) -> bool {
        self.edges.values().all(|attrs| {
            attrs.values().all(|v| {
                matches!(
                    v,
                    fnx_runtime::CgseValue::Int(_)
                        | fnx_runtime::CgseValue::Float(_)
                        | fnx_runtime::CgseValue::Bool(_)
                )
            })
        })
    }

    /// br-r37-c1-hasanyattrlazyfix: any node / any edge carries a Python-visible attr per
    /// the authoritative inner storage (not the lazy `*_py_attrs` mirrors).
    #[must_use]
    pub fn any_node_has_attrs(&self) -> bool {
        self.node_attrs.iter().any(|attrs| !attrs.is_empty())
    }

    #[must_use]
    pub fn any_edge_has_attrs(&self) -> bool {
        self.edges.values().any(|attrs| !attrs.is_empty())
    }

    /// br-r37-c1-prdir: iterate every directed edge as
    /// `((source_idx, target_idx), &AttrMap)` straight from the index-keyed edge
    /// store, in insertion order. Lets COO/CSR exporters read each edge's weight
    /// with ONE attr lookup instead of the per-edge `edges.get(&(u,v))` hash that
    /// the successors-then-`edge_attrs_by_indices` path pays.
    pub fn edges_indexed(&self) -> impl Iterator<Item = ((usize, usize), &AttrMap)> + '_ {
        let slot_positions = self.node_order.slot_positions();
        self.edges
            .iter()
            .map(move |(&(source, target), attrs)| match &slot_positions {
                Some(positions) => ((positions[source], positions[target]), attrs),
                None => ((source, target), attrs),
            })
    }

    #[must_use]
    pub fn evidence_ledger(&self) -> &EvidenceLedger {
        self.runtime_policy.decision_log()
    }

    pub fn drain_decision_records(&mut self) -> Vec<DecisionRecord> {
        self.runtime_policy.drain_decision_records()
    }

    /// Record a bounded recovery performed outside the graph (e.g. a reader
    /// skipping a malformed line in hardened mode) in this graph's ledger.
    pub fn record_external_recovery(
        &mut self,
        operation: impl Into<std::borrow::Cow<'static, str>>,
        rationale: impl Into<std::borrow::Cow<'static, str>>,
    ) {
        self.runtime_policy.record(
            operation,
            DecisionAction::FullValidate,
            0.7,
            rationale,
            Vec::new(),
        );
    }

    #[must_use]
    pub fn runtime_policy(&self) -> &RuntimePolicy {
        &self.runtime_policy
    }

    pub fn set_runtime_policy(&mut self, runtime_policy: RuntimePolicy) {
        self.mode = runtime_policy.mode();
        self.runtime_policy = runtime_policy;
    }

    /// Type identity: always `true` for DiGraph.
    #[must_use]
    pub fn is_directed(&self) -> bool {
        true
    }

    /// Type identity: always `false` for DiGraph (not a multigraph).
    #[must_use]
    pub fn is_multigraph(&self) -> bool {
        false
    }

    // -----------------------------------------------------------------------
    // Mutations
    // -----------------------------------------------------------------------

    pub fn add_node(&mut self, node: impl AsRef<str> + Into<String>) -> bool {
        self.add_node_with_attrs(node, AttrMap::new())
    }

    /// br-r37-c1-wa1b9: the storage half of [`Self::add_node_with_attrs`],
    /// without its ledger record. See the undirected `Graph` sibling.
    /// br-r37-c1-jc9e4: the key is materialised only on the VACANT path, the
    /// directed mirror of the `Graph` sibling. This used to open with `let node
    /// = node.into();`, so a `&str` caller allocated an owned `String` on every
    /// call — including the calls that insert nothing, where it was built purely
    /// to probe the map and then dropped by `entry`. The allocation census
    /// measured the directed pair at the undirected pair's pre-lever numbers,
    /// so the fix did not transfer on its own and had to be applied here too.
    ///
    /// Occupied-branch semantics are unchanged: `IndexMap::entry(..).or_default()`
    /// on a present key returns the existing value without reordering, which is
    /// exactly what `get_mut` returns.
    fn add_node_with_attrs_unrecorded(
        &mut self,
        node: impl AsRef<str> + Into<String>,
        attrs: AttrMap,
    ) -> (bool, bool, usize, Option<usize>) {
        let node_ref = node.as_ref();
        if let Some(&slot) = self.node_order.get(node_ref) {
            let bucket = &mut self.node_attrs[slot];
            let changed = !attrs.is_empty()
                && attrs
                    .iter()
                    .any(|(key, value)| bucket.get(key) != Some(value));
            bucket.extend(attrs);
            if changed {
                self.revision = self.revision.saturating_add(1);
            }
            return (changed, true, bucket.len(), None);
        }

        let slot = self.node_order.push(node.into());
        self.node_attrs.push(attrs);
        self.succ_indices.rows_mut().push(Vec::new());
        self.pred_indices.rows_mut().push(Vec::new());
        let attr_count = self.node_attrs[slot].len();
        self.revision = self.revision.saturating_add(1);
        (true, false, attr_count, Some(slot))
    }

    fn resolve_empty_node_for_edge_unrecorded(&mut self, node: &str) -> (usize, bool) {
        if let Some(&slot) = self.node_order.get(node) {
            return (slot, false);
        }
        let slot = self.node_order.push(node.to_owned());
        self.node_attrs.push(AttrMap::new());
        self.succ_indices.rows_mut().push(Vec::new());
        self.pred_indices.rows_mut().push(Vec::new());
        self.revision = self.revision.saturating_add(1);
        (slot, true)
    }

    pub fn add_node_with_attrs(
        &mut self,
        node: impl AsRef<str> + Into<String>,
        attrs: AttrMap,
    ) -> bool {
        self.compact_before_growth();
        let (changed, existed, attrs_count, _) = self.add_node_with_attrs_unrecorded(node, attrs);
        self.record_decision(
            "add_node",
            0.0,
            false,
            vec![
                EvidenceTerm {
                    signal: "node_preexisting".into(),
                    observed_value: bool_evidence(existed),
                    log_likelihood_ratio: -3.0,
                },
                EvidenceTerm {
                    signal: "attrs_count".into(),
                    observed_value: count_evidence(attrs_count),
                    log_likelihood_ratio: -1.0,
                },
            ],
        );
        changed
    }

    pub fn apply_node_defaults(&mut self, defaults: &AttrMap) -> bool {
        if defaults.is_empty() {
            return false;
        }
        let mut inserted = 0usize;
        for slot in self.node_order.values() {
            let attrs = &mut self.node_attrs[slot];
            for (key, value) in defaults {
                if !attrs.contains_key(key) {
                    attrs.insert(key.clone(), value.clone());
                    inserted += 1;
                }
            }
        }
        if inserted > 0 {
            self.revision = self.revision.saturating_add(1);
            self.record_decision(
                "apply_node_defaults",
                0.0,
                false,
                vec![EvidenceTerm {
                    signal: "defaults_applied".into(),
                    observed_value: count_evidence(inserted),
                    log_likelihood_ratio: 0.0,
                }],
            );
            return true;
        }
        false
    }

    pub fn add_edge(
        &mut self,
        source: impl AsRef<str>,
        target: impl AsRef<str>,
    ) -> Result<(), GraphError> {
        self.add_edge_with_attrs(source, target, AttrMap::new())
    }

    /// br-r37-c1-jc9e4: the directed mirror of the `Graph` sibling — the
    /// endpoint keys are never owned here, and each is resolved with ONE map
    /// probe. This used to own both keys, clone each again for autocreation,
    /// and then hash each up to three times (`contains_key`, `get_index_of`
    /// for the edge key, `get_index_of` again for the successor/predecessor
    /// rows). The census measured the directed pair at the undirected pair's
    /// pre-lever numbers, so none of that transferred on its own.
    pub fn add_edge_with_attrs(
        &mut self,
        source: impl AsRef<str>,
        target: impl AsRef<str>,
        attrs: AttrMap,
    ) -> Result<(), GraphError> {
        self.compact_before_growth();
        let source = source.as_ref();
        let target = target.as_ref();

        let unknown_feature = attrs
            .keys()
            .any(|key| key.starts_with("__fnx_incompatible"));
        let self_loop = source == target;
        let incompatibility_probability = if unknown_feature {
            1.0
        } else if self_loop {
            0.22
        } else {
            0.08
        };

        // br-r37-c1-eo88t: the gate decision earns a ledger record only when it
        // REFUSES. A refusal returns from here, so the outcome record at the end
        // of this function never runs and the gate record is the whole audit
        // trail for the rejected mutation. On the allow path the outcome record
        // carries the same operation/mode/action/incompatibility_probability
        // plus strictly richer evidence, so recording the gate as well wrote two
        // near-identical records for every edge on the hottest mutation path.
        let action = self
            .runtime_policy
            .action_for(incompatibility_probability, unknown_feature);

        if action == DecisionAction::FailClosed || action == DecisionAction::FullValidate {
            self.record_decision(
                "add_edge",
                incompatibility_probability,
                unknown_feature,
                vec![EvidenceTerm {
                    signal: "unknown_incompatible_feature".into(),
                    observed_value: bool_evidence(unknown_feature),
                    log_likelihood_ratio: 12.0,
                }],
            );
            return Err(GraphError::FailClosed {
                operation: "add_edge",
                reason: "incompatible edge metadata".to_owned(),
            });
        }

        // Auto-create nodes.
        let (source_slot, source_autocreated) = self.resolve_empty_node_for_edge_unrecorded(source);
        let (target_slot, target_autocreated) = if self_loop {
            (source_slot, source_autocreated)
        } else {
            self.resolve_empty_node_for_edge_unrecorded(target)
        };

        let edge_key = (source_slot, target_slot);
        let (new_edge, changed, edge_attr_count) = match self.edges.entry(edge_key) {
            indexmap::map::Entry::Vacant(entry) => {
                let count = attrs.len();
                entry.insert(attrs);
                (true, true, count)
            }
            indexmap::map::Entry::Occupied(mut entry) => {
                let edge_attrs = entry.get_mut();
                let changed = !attrs.is_empty()
                    && attrs
                        .iter()
                        .any(|(key, value)| edge_attrs.get(key) != Some(value));
                edge_attrs.extend(attrs);
                (false, changed, edge_attrs.len())
            }
        };

        if new_edge {
            self.succ_indices.rows_mut()[source_slot].push(target_slot);
            self.pred_indices.rows_mut()[target_slot].push(source_slot);
        }

        if changed {
            self.revision = self.revision.saturating_add(1);
        }

        self.record_decision(
            "add_edge",
            incompatibility_probability,
            unknown_feature,
            vec![
                EvidenceTerm {
                    signal: "self_loop".into(),
                    observed_value: bool_evidence(self_loop),
                    log_likelihood_ratio: -0.5,
                },
                EvidenceTerm {
                    signal: "edge_attr_count".into(),
                    observed_value: count_evidence(edge_attr_count),
                    log_likelihood_ratio: -2.0,
                },
                EvidenceTerm {
                    signal: "source_autocreated".into(),
                    observed_value: bool_evidence(source_autocreated),
                    log_likelihood_ratio: -1.25,
                },
                EvidenceTerm {
                    signal: "target_autocreated".into(),
                    observed_value: bool_evidence(target_autocreated),
                    log_likelihood_ratio: -1.25,
                },
            ],
        );

        Ok(())
    }

    /// Bulk-add a sequence of attribute-free directed edges. Bypasses
    /// the per-edge `runtime_policy.record_decision` call that
    /// [`add_edge_with_attrs`] makes. Instead, emit one summary record
    /// for the whole batch.
    ///
    /// Intended for fnx-internal callers that build a fresh graph
    /// from a known-good edge list where per-edge compatibility
    /// accounting would add constant-factor overhead without adding
    /// useful policy evidence. Nodes referenced by edges are
    /// auto-created if absent, matching `add_edge` semantics.
    #[must_use]
    pub fn extend_edges_unrecorded<I, S, T>(&mut self, edges: I) -> usize
    where
        I: IntoIterator<Item = (S, T)>,
        S: Into<String>,
        T: Into<String>,
    {
        self.compact_before_growth();
        let iterator = edges.into_iter();
        let (lower_bound, _) = iterator.size_hint();
        self.node_order.reserve(lower_bound);
        self.edges.reserve(lower_bound);

        let mut inserted = 0usize;
        for (source, target) in iterator {
            let source = source.into();
            let target = target.into();
            let (s_slot, _) = match self.node_order.get(&source) {
                Some(&slot) => (slot, false),
                None => {
                    let slot = self.node_order.push(source.clone());
                    self.node_attrs.push(AttrMap::new());
                    self.succ_indices.rows_mut().push(Vec::new());
                    self.pred_indices.rows_mut().push(Vec::new());
                    (slot, true)
                }
            };
            let (t_slot, _) = if source == target {
                (s_slot, false)
            } else {
                match self.node_order.get(&target) {
                    Some(&slot) => (slot, false),
                    None => {
                        let slot = self.node_order.push(target.clone());
                        self.node_attrs.push(AttrMap::new());
                        self.succ_indices.rows_mut().push(Vec::new());
                        self.pred_indices.rows_mut().push(Vec::new());
                        (slot, true)
                    }
                }
            };
            let edge_key = (s_slot, t_slot);
            if self.edges.contains_key(&edge_key) {
                continue;
            }
            self.edges.insert(edge_key, AttrMap::new());
            self.succ_indices.rows_mut()[s_slot].push(t_slot);
            self.pred_indices.rows_mut()[t_slot].push(s_slot);
            inserted += 1;
        }

        if inserted > 0 {
            self.revision = self
                .revision
                .saturating_add(u64::try_from(inserted).unwrap_or(u64::MAX));
            self.record_decision(
                "extend_edges_unrecorded",
                0.0,
                false,
                vec![EvidenceTerm {
                    signal: "batch_edge_count".into(),
                    observed_value: count_evidence(inserted),
                    log_likelihood_ratio: -1.0,
                }],
            );
        }

        inserted
    }

    /// br-r37-c1-tcidx: bulk-add directed edges by EXISTING node index — the directed sibling of
    /// `Graph::extend_existing_index_edges_unrecorded`. All endpoints MUST already exist (callers
    /// gate on all nodes being pre-added). A new edge takes an empty AttrMap; a duplicate
    /// `(source_idx, target_idx)` pair is skipped. Emits the IDENTICAL succ/pred adjacency,
    /// revision bump, and batch ledger summary as `extend_edges_unrecorded` given the same
    /// resolved index sequence — but with ZERO String hashing (endpoints arrive as integer
    /// indices). Unblocks the O(|V|²)-closure `transitive_closure` from paying a per-edge
    /// `get_index_of` on both endpoints.
    pub fn extend_existing_index_edges_unrecorded<I>(&mut self, edges: I) -> usize
    where
        I: IntoIterator<Item = (usize, usize)>,
    {
        self.compact_before_growth();
        let iterator = edges.into_iter();
        self.edges.reserve(iterator.size_hint().0);

        let mut inserted = 0usize;
        for (s_idx, t_idx) in iterator {
            debug_assert!(s_idx < self.node_order.len());
            debug_assert!(t_idx < self.node_order.len());
            let s_slot = self
                .node_order
                .slot_at(s_idx)
                .expect("validated source index");
            let t_slot = self
                .node_order
                .slot_at(t_idx)
                .expect("validated target index");
            let edge_key = (s_slot, t_slot);
            if self.edges.contains_key(&edge_key) {
                continue;
            }
            self.edges.insert(edge_key, AttrMap::new());
            self.succ_indices.rows_mut()[s_slot].push(t_slot);
            self.pred_indices.rows_mut()[t_slot].push(s_slot);
            inserted += 1;
        }

        if inserted > 0 {
            self.revision = self
                .revision
                .saturating_add(u64::try_from(inserted).unwrap_or(u64::MAX));
            self.record_decision(
                "extend_edges_unrecorded",
                0.0,
                false,
                vec![EvidenceTerm {
                    signal: "batch_edge_count".into(),
                    observed_value: count_evidence(inserted),
                    log_likelihood_ratio: -1.0,
                }],
            );
        }

        inserted
    }

    /// br-r37-c1-digbatch: bulk-add plain nodes (no attrs), one summary ledger
    /// record — the directed sibling of `Graph::extend_nodes_unrecorded`, backing
    /// PyDiGraph's fast integer-node path (`add_nodes_from(range)` / int list).
    /// First insertion wins order; existing nodes are skipped.
    pub fn extend_nodes_unrecorded<I, N>(&mut self, nodes: I) -> usize
    where
        I: IntoIterator<Item = N>,
        N: Into<String>,
    {
        self.compact_before_growth();
        let iterator = nodes.into_iter();
        let (lower_bound, _) = iterator.size_hint();
        self.node_order.reserve(lower_bound);

        let mut inserted = 0usize;
        for node in iterator {
            let node = node.into();
            if self.node_order.contains_key(&node) {
                continue;
            }
            self.node_order.push(node);
            self.node_attrs.push(AttrMap::new());
            self.succ_indices.rows_mut().push(Vec::new());
            self.pred_indices.rows_mut().push(Vec::new());
            inserted += 1;
        }
        if inserted > 0 {
            self.revision = self
                .revision
                .saturating_add(u64::try_from(inserted).unwrap_or(u64::MAX));
            self.record_decision(
                "extend_nodes_unrecorded",
                0.0,
                false,
                vec![EvidenceTerm {
                    signal: "batch_node_count".into(),
                    observed_value: count_evidence(inserted),
                    log_likelihood_ratio: -1.0,
                }],
            );
        }
        inserted
    }

    /// br-r37-c1-dgctor: bulk-add nodes WITH attrs, one summary ledger
    /// record — the directed sibling of `Graph::extend_nodes_unrecorded`
    /// extended with per-node AttrMaps. First insertion wins the order;
    /// re-inserting an existing node merges attrs (matching
    /// `add_node_with_attrs`).
    #[must_use]
    pub fn extend_nodes_with_attrs_unrecorded<I>(&mut self, nodes: I) -> usize
    where
        I: IntoIterator<Item = (String, AttrMap)>,
    {
        self.compact_before_growth();
        let iterator = nodes.into_iter();
        let (lower_bound, _) = iterator.size_hint();
        self.node_order.reserve(lower_bound);

        let mut inserted = 0usize;
        for (node, attrs) in iterator {
            if let Some(&slot) = self.node_order.get(&node) {
                self.node_attrs[slot].extend(attrs);
                continue;
            }
            self.node_order.push(node);
            self.node_attrs.push(attrs);
            self.succ_indices.rows_mut().push(Vec::new());
            self.pred_indices.rows_mut().push(Vec::new());
            inserted += 1;
        }
        if inserted > 0 {
            self.revision = self
                .revision
                .saturating_add(u64::try_from(inserted).unwrap_or(u64::MAX));
            self.record_decision(
                "extend_nodes_unrecorded",
                0.0,
                false,
                vec![EvidenceTerm {
                    signal: "batch_node_count".into(),
                    observed_value: count_evidence(inserted),
                    log_likelihood_ratio: -1.0,
                }],
            );
        }
        inserted
    }

    /// br-r37-c1-dgctor: bulk-add ATTRIBUTED directed edges without
    /// per-edge ledger records — the directed sibling of
    /// `Graph::extend_edges_with_attrs_unrecorded`. Insert-or-MERGE
    /// (duplicate (source, target) extends the existing AttrMap, matching
    /// `add_edge_with_attrs`); nodes auto-created in first-appearance
    /// order. Callers must pre-screen "__fnx_incompatible" attr keys.
    #[must_use]
    pub fn extend_edges_with_attrs_unrecorded<I>(&mut self, edges: I) -> usize
    where
        I: IntoIterator<Item = (String, String, AttrMap)>,
    {
        self.compact_before_growth();
        let iterator = edges.into_iter();
        let (lower_bound, _) = iterator.size_hint();
        self.node_order.reserve(lower_bound);
        self.edges.reserve(lower_bound);

        let mut inserted = 0usize;
        let mut merged_changed = false;
        for (source, target, attrs) in iterator {
            let (s_slot, _) = match self.node_order.get(&source) {
                Some(&slot) => (slot, false),
                None => {
                    let slot = self.node_order.push(source.clone());
                    self.node_attrs.push(AttrMap::new());
                    self.succ_indices.rows_mut().push(Vec::new());
                    self.pred_indices.rows_mut().push(Vec::new());
                    (slot, true)
                }
            };
            let (t_slot, _) = if source == target {
                (s_slot, false)
            } else {
                match self.node_order.get(&target) {
                    Some(&slot) => (slot, false),
                    None => {
                        let slot = self.node_order.push(target.clone());
                        self.node_attrs.push(AttrMap::new());
                        self.succ_indices.rows_mut().push(Vec::new());
                        self.pred_indices.rows_mut().push(Vec::new());
                        (slot, true)
                    }
                }
            };
            let edge_key = (s_slot, t_slot);
            if let Some(existing) = self.edges.get_mut(&edge_key) {
                if attrs.is_empty() {
                    continue;
                }
                if attrs
                    .iter()
                    .any(|(key, value)| existing.get(key) != Some(value))
                {
                    merged_changed = true;
                }
                existing.extend(attrs);
                continue;
            }
            self.edges.insert(edge_key, attrs);
            self.succ_indices.rows_mut()[s_slot].push(t_slot);
            self.pred_indices.rows_mut()[t_slot].push(s_slot);
            inserted += 1;
        }
        if inserted > 0 || merged_changed {
            self.revision = self
                .revision
                .saturating_add(u64::try_from(inserted.max(1)).unwrap_or(u64::MAX));
            self.record_decision(
                "extend_edges_unrecorded",
                0.0,
                false,
                vec![EvidenceTerm {
                    signal: "batch_edge_count".into(),
                    observed_value: count_evidence(inserted),
                    log_likelihood_ratio: -1.0,
                }],
            );
        }
        inserted
    }

    /// Bulk-add ATTRIBUTED directed edges when the caller has already
    /// identified newly-created nodes. This keeps edge-map insertion in global
    /// input order, then stages successor and predecessor row commits by row so
    /// adjacency maps are probed once per touched row rather than once per edge.
    #[must_use]
    pub fn extend_prepared_edges_with_attrs_row_staged_unrecorded<I, N>(
        &mut self,
        new_nodes: N,
        edges: I,
    ) -> usize
    where
        I: IntoIterator<Item = (String, String, AttrMap)>,
        N: IntoIterator<Item = String>,
    {
        self.compact_before_growth();
        for node in new_nodes {
            if self.node_order.contains_key(&node) {
                continue;
            }
            self.node_order.push(node);
            self.node_attrs.push(AttrMap::new());
            self.succ_indices.rows_mut().push(Vec::new());
            self.pred_indices.rows_mut().push(Vec::new());
        }

        let mut inserted = 0usize;
        let mut merged_changed = false;

        for (source, target, attrs) in edges {
            let s_slot = *self.node_order.get(&source).expect("source created above");
            let t_slot = *self.node_order.get(&target).expect("target created above");
            let edge_key = (s_slot, t_slot);
            if let Some(existing) = self.edges.get_mut(&edge_key) {
                if attrs.is_empty() {
                    continue;
                }
                if attrs
                    .iter()
                    .any(|(key, value)| existing.get(key) != Some(value))
                {
                    merged_changed = true;
                }
                existing.extend(attrs);
                continue;
            }

            self.succ_indices.rows_mut()[s_slot].push(t_slot);
            self.pred_indices.rows_mut()[t_slot].push(s_slot);
            self.edges.insert(edge_key, attrs);
            inserted += 1;
        }

        if inserted > 0 || merged_changed {
            self.revision = self
                .revision
                .saturating_add(u64::try_from(inserted.max(1)).unwrap_or(u64::MAX));
            self.record_decision(
                "extend_edges_unrecorded",
                0.0,
                false,
                vec![EvidenceTerm {
                    signal: "batch_edge_count".into(),
                    observed_value: count_evidence(inserted),
                    log_likelihood_ratio: -1.0,
                }],
            );
        }

        inserted
    }

    /// Bulk-add attributed directed edges into a fresh graph when the caller has
    /// already assigned NetworkX-compatible node indices in first-seen order.
    ///
    /// This bypasses the string-key lookup path in
    /// [`extend_prepared_edges_with_attrs_row_staged_unrecorded`]: node labels
    /// are installed once, then edges are committed directly by `(source_idx,
    /// target_idx)`. Duplicate directed edges merge attributes without appending
    /// another successor/predecessor entry, matching `DiGraph.add_edges_from`.
    #[must_use]
    /// br-r37-c1-dodattrbatch: bulk-add ATTRIBUTED directed edges by EXISTING
    /// node index. All endpoints MUST already exist (callers gate on int-labeled
    /// nodes). A new edge takes the AttrMap; a duplicate (source, target) MERGES
    /// last-wins like add_edge_with_attrs. Sibling of the fresh-index attr extend
    /// minus node creation; integer index rows only (no String hashing).
    pub fn extend_existing_index_edges_with_attrs_unrecorded<I>(&mut self, edges: I) -> usize
    where
        I: IntoIterator<Item = (usize, usize, AttrMap)>,
    {
        self.compact_before_growth();
        let node_count = self.node_order.len();
        let edges = edges.into_iter();
        self.edges.reserve(edges.size_hint().0);
        let mut inserted = 0usize;
        let mut merged_changed = false;
        for (source_idx, target_idx, attrs) in edges {
            debug_assert!(source_idx < node_count && target_idx < node_count);
            let s_slot = self
                .node_order
                .slot_at(source_idx)
                .expect("validated source index");
            let t_slot = self
                .node_order
                .slot_at(target_idx)
                .expect("validated target index");
            let edge_key = (s_slot, t_slot);
            if let Some(existing) = self.edges.get_mut(&edge_key) {
                if attrs.is_empty() {
                    continue;
                }
                if attrs
                    .iter()
                    .any(|(key, value)| existing.get(key) != Some(value))
                {
                    merged_changed = true;
                }
                existing.extend(attrs);
                continue;
            }
            self.edges.insert(edge_key, attrs);
            self.succ_indices.rows_mut()[s_slot].push(t_slot);
            self.pred_indices.rows_mut()[t_slot].push(s_slot);
            inserted += 1;
        }
        if inserted > 0 || merged_changed {
            self.revision = self
                .revision
                .saturating_add(u64::try_from(inserted).unwrap_or(u64::MAX));
        }
        inserted
    }

    pub fn extend_fresh_index_edges_with_attrs_unrecorded<I, N>(
        &mut self,
        nodes: N,
        edges: I,
    ) -> usize
    where
        I: IntoIterator<Item = (usize, usize, AttrMap)>,
        N: IntoIterator<Item = String>,
    {
        if !self.node_order.is_empty()
            || !self.succ_indices.is_empty()
            || !self.pred_indices.is_empty()
            || !self.edges.is_empty()
        {
            return 0;
        }

        let node_labels: Vec<String> = nodes.into_iter().collect();
        let node_count = node_labels.len();
        self.node_order = crate::NodeOrder::from_names(node_labels);
        self.node_attrs = vec![AttrMap::new(); node_count];
        *self.succ_indices.rows_mut() = vec![Vec::new(); node_count];
        *self.pred_indices.rows_mut() = vec![Vec::new(); node_count];

        let mut inserted = 0usize;
        let mut merged_changed = false;
        let edges = edges.into_iter();
        self.edges.reserve(edges.size_hint().0);
        for (source_idx, target_idx, attrs) in edges {
            if source_idx >= node_count || target_idx >= node_count {
                continue;
            }
            let edge_key = (source_idx, target_idx);
            if let Some(existing) = self.edges.get_mut(&edge_key) {
                if attrs.is_empty() {
                    continue;
                }
                if attrs
                    .iter()
                    .any(|(key, value)| existing.get(key) != Some(value))
                {
                    merged_changed = true;
                }
                existing.extend(attrs);
                continue;
            }
            self.edges.insert(edge_key, attrs);
            self.succ_indices.rows_mut()[source_idx].push(target_idx);
            self.pred_indices.rows_mut()[target_idx].push(source_idx);
            inserted += 1;
        }

        if node_count > 0 || inserted > 0 || merged_changed {
            self.revision = self.revision.saturating_add(
                u64::try_from(node_count.saturating_add(inserted)).unwrap_or(u64::MAX),
            );
            self.record_decision(
                "extend_fresh_index_edges_unrecorded",
                0.0,
                false,
                vec![
                    EvidenceTerm {
                        signal: "batch_node_count".into(),
                        observed_value: count_evidence(node_count),
                        log_likelihood_ratio: -1.0,
                    },
                    EvidenceTerm {
                        signal: "batch_edge_count".into(),
                        observed_value: count_evidence(inserted),
                        log_likelihood_ratio: -1.0,
                    },
                ],
            );
        }

        inserted
    }

    pub fn apply_edge_defaults(&mut self, defaults: &AttrMap) -> bool {
        if defaults.is_empty() {
            return false;
        }
        let mut inserted = 0usize;
        for attrs in self.edges.values_mut() {
            for (key, value) in defaults {
                if !attrs.contains_key(key) {
                    attrs.insert(key.clone(), value.clone());
                    inserted += 1;
                }
            }
        }
        if inserted > 0 {
            self.revision = self.revision.saturating_add(1);
            self.record_decision(
                "apply_edge_defaults",
                0.0,
                false,
                vec![EvidenceTerm {
                    signal: "defaults_applied".into(),
                    observed_value: count_evidence(inserted),
                    log_likelihood_ratio: 0.0,
                }],
            );
            return true;
        }
        false
    }

    /// br-r37-c1-sjf4t: overwrite the attribute map for an existing edge.
    pub fn replace_edge_attrs(&mut self, source: &str, target: &str, attrs: AttrMap) -> bool {
        let Some(edge_key) = self.edge_pair_key(source, target) else {
            return false;
        };
        if let Some(slot) = self.edges.get_mut(&edge_key) {
            if *slot != attrs {
                *slot = attrs;
                self.revision = self.revision.saturating_add(1);
            }
            true
        } else {
            false
        }
    }

    /// br-r37-c1-sjf4t: overwrite the attribute map for an existing node.
    pub fn replace_node_attrs(&mut self, node: &str, attrs: AttrMap) -> bool {
        if let Some(&slot) = self.node_order.get(node) {
            if self.node_attrs[slot] != attrs {
                self.node_attrs[slot] = attrs;
                self.revision = self.revision.saturating_add(1);
            }
            true
        } else {
            false
        }
    }

    /// Remove directed edge source→target. Returns `true` if it existed.
    pub fn remove_edge(&mut self, source: &str, target: &str) -> bool {
        let (Some(&s_slot), Some(&t_slot)) =
            (self.node_order.get(source), self.node_order.get(target))
        else {
            return false;
        };
        let pair = (s_slot, t_slot);
        let removed = self.edges.swap_remove(&pair).is_some();
        if removed {
            self.succ_indices.rows_mut()[s_slot].retain(|&i| i != t_slot);
            self.pred_indices.rows_mut()[t_slot].retain(|&i| i != s_slot);
            self.revision = self.revision.saturating_add(1);
        }
        removed
    }

    /// Remove node and all incident edges (both incoming and outgoing).
    pub fn remove_node(&mut self, node: &str) -> bool {
        let removed = self.remove_node_deferred_compaction(node);
        if removed {
            self.compact_when_sparse();
        }
        removed
    }

    fn remove_node_deferred_compaction(&mut self, node: &str) -> bool {
        // O(1): the node's entry is tombstoned (or popped), never shifted.
        let Some((slot, popped)) = self.node_order.remove(node) else {
            return false;
        };

        // Drop incident edges and remove this slot from neighbors' succ/pred rows.
        let succ_rows = self.succ_indices.rows_mut();
        let pred_rows = self.pred_indices.rows_mut();
        let succs = std::mem::take(&mut succ_rows[slot]);
        for &t in &succs {
            self.edges.swap_remove(&(slot, t));
            if t != slot {
                pred_rows[t].retain(|&i| i != slot);
            }
        }
        let preds = std::mem::take(&mut pred_rows[slot]);
        for &s in &preds {
            self.edges.swap_remove(&(s, slot));
            if s != slot {
                succ_rows[s].retain(|&i| i != slot);
            }
        }

        if popped {
            self.node_attrs.pop();
            succ_rows.pop();
            pred_rows.pop();
        } else {
            self.node_attrs[slot].clear();
        }

        self.revision = self.revision.saturating_add(1);
        true
    }

    /// Compact once the freed slots outnumber the live nodes (see
    /// `Graph::compact_when_sparse`).
    fn compact_when_sparse(&mut self) {
        let removed = self.node_order.slot_count() - self.node_order.len();
        if removed > self.node_order.len() {
            self.compact_internal();
        }
    }

    /// Compact at the first insertion after a run of removals (see
    /// `Graph::compact_before_growth`).
    fn compact_before_growth(&mut self) {
        if !self.node_order.is_dense() {
            self.compact_internal();
        }
    }

    pub fn remove_nodes_from<'a, I>(&mut self, nodes: I) -> (usize, usize)
    where
        I: IntoIterator<Item = &'a str>,
    {
        let mut removed_nodes = 0usize;
        let old_edge_count = self.edges.len();
        for node in nodes {
            if self.remove_node_deferred_compaction(node) {
                removed_nodes += 1;
            }
        }
        let removed_edges = old_edge_count.saturating_sub(self.edges.len());
        if removed_nodes > 0 {
            self.compact_when_sparse();
        }
        (removed_nodes, removed_edges)
    }

    /// Nothing removed since the last compaction: every node's slot is its
    /// position (O(1)).
    #[inline]
    #[must_use]
    pub fn is_compact(&self) -> bool {
        self.node_order.is_dense()
    }

    pub fn ensure_compact(&mut self) {
        if self.is_compact() {
            return;
        }
        self.compact_internal();
    }

    fn compact_internal(&mut self) {
        let n = self.node_order.len();
        let old_slots: Vec<usize> = self.node_order.values().collect();
        let old_to_new = self.node_order.compact();

        let mut new_node_attrs = Vec::with_capacity(n);
        let mut new_succ_indices = Vec::with_capacity(n);
        let mut new_pred_indices = Vec::with_capacity(n);
        let succ_rows = self.succ_indices.rows_mut();
        let pred_rows = self.pred_indices.rows_mut();
        for old_slot in old_slots {
            new_node_attrs.push(std::mem::take(&mut self.node_attrs[old_slot]));
            new_succ_indices.push(std::mem::take(&mut succ_rows[old_slot]));
            new_pred_indices.push(std::mem::take(&mut pred_rows[old_slot]));
        }

        // Remap neighbor slots in succ_indices and pred_indices
        for row in &mut new_succ_indices {
            for target in row {
                *target = old_to_new[*target];
            }
        }
        for row in &mut new_pred_indices {
            for source in row {
                *source = old_to_new[*source];
            }
        }

        // Remap edge keys
        let old_edges = std::mem::take(&mut self.edges);
        let mut new_edges =
            FxIndexMap::with_capacity_and_hasher(old_edges.len(), rustc_hash::FxBuildHasher);

        for ((old_s, old_t), attrs) in old_edges {
            new_edges.insert((old_to_new[old_s], old_to_new[old_t]), attrs);
        }

        self.node_attrs = new_node_attrs;
        *self.succ_indices.rows_mut() = new_succ_indices;
        *self.pred_indices.rows_mut() = new_pred_indices;
        self.edges = new_edges;
        self.revision = self.revision.saturating_add(1);
    }

    // -----------------------------------------------------------------------
    // Snapshot / ordered iteration
    // -----------------------------------------------------------------------

    /// Edges in deterministic order: iterate nodes in insertion order, then
    /// each node's successors in insertion order.
    #[must_use]
    pub fn edges_ordered(&self) -> Vec<EdgeSnapshot> {
        let mut ordered = Vec::with_capacity(self.edges.len());

        for (node, u) in self.node_order.iter() {
            for &t in &self.succ_indices[u] {
                let target = self.node_order.names[t]
                    .as_deref()
                    .expect("valid target slot");
                if let Some(attrs) = self.edges.get(&(u, t)) {
                    ordered.push(EdgeSnapshot {
                        left: node.clone(),
                        right: target.to_owned(),
                        attrs: attrs.clone(),
                    });
                }
            }
        }

        ordered
    }

    #[must_use]
    pub fn edges_ordered_indices_borrowed(&self) -> Vec<(usize, usize, &AttrMap)> {
        let mut ordered = Vec::with_capacity(self.edges.len());
        let mut slot_to_pos = vec![0; self.node_order.slot_count()];
        for (pos, (_, slot)) in self.node_order.iter().enumerate() {
            slot_to_pos[slot] = pos;
        }

        for (u_pos, (_, u)) in self.node_order.iter().enumerate() {
            for &t in &self.succ_indices[u] {
                if let Some(attrs) = self.edges.get(&(u, t)) {
                    let t_pos = slot_to_pos[t];
                    ordered.push((u_pos, t_pos, attrs));
                }
            }
        }

        ordered
    }

    #[must_use]
    pub fn edges_ordered_borrowed(&self) -> Vec<(&str, &str, &AttrMap)> {
        let mut ordered = Vec::with_capacity(self.edges.len());

        for (u_name, u) in self.node_order.iter() {
            for &t in &self.succ_indices[u] {
                if let Some(attrs) = self.edges.get(&(u, t)) {
                    let t_name = self.node_order.names[t]
                        .as_deref()
                        .expect("valid target slot");
                    ordered.push((u_name.as_str(), t_name, attrs));
                }
            }
        }

        ordered
    }

    /// br-r37-c1-wsize (cc): integer `size(weight)` from the store — directed
    /// twin of `Graph::weighted_size_int`. A directed edge `(u,v,w)` adds `w` to
    /// `out_degree(u)` and `in_degree(v)`, so `sum(degree)/2` halves back to each
    /// edge's weight once; a self-loop adds `w` to both in and out of one node →
    /// `2w` → halves to `w`, again one stored edge counted once. Returns `None` on
    /// any non-integer weight (caller uses the exact degree path); missing weight
    /// defaults to nx's int `1`.
    #[must_use]
    pub fn weighted_size_int(&self, weight: &str) -> Option<i128> {
        let mut total: i128 = 0;
        for attrs in self.edges.values() {
            let value = match attrs.get(weight) {
                Some(CgseValue::Int(v)) => i128::from(*v),
                Some(_) => return None,
                None => 1,
            };
            total = total.checked_add(value)?;
        }
        Some(total)
    }

    /// br-r37-c1-revborrow: O(V+E) directed transpose. The reverse of a
    /// `DiGraph` keeps the SAME node table (identical insertion order, identical
    /// indices), so reversing is a pure topology flip — no String hashing, no
    /// per-edge node-index lookups, no re-insertion through the name table (the
    /// edge-iterate + `edges_ordered()` clone + `extend_edges_with_attrs_*`
    /// re-hash path is O(E) String/hash work this avoids entirely).
    ///
    /// Successor rows are rebuilt by walking the ORIGINAL successor rows in
    /// node-major (ascending source index) order and appending the reversed
    /// endpoint, so the reversed graph's `edges_ordered()` stream is
    /// byte-identical to NetworkX's `reverse()`, which walks `self.edges()`
    /// (u-major successor order) and adds each `(v, u)`. `pred_indices[u]` ends
    /// up equal to the original `succ_indices[u]`, matching NetworkX's reversed
    /// `pred` order as well. Node + edge attribute maps are cloned verbatim.
    #[must_use]
    pub fn reversed(&self) -> Self {
        let slot_count = self.node_order.slot_count();
        let mut succ_indices: Vec<Vec<usize>> = vec![Vec::new(); slot_count];
        let mut pred_indices: Vec<Vec<usize>> = vec![Vec::new(); slot_count];
        for (i, (succ_row, pred_row)) in succ_indices
            .iter_mut()
            .zip(pred_indices.iter_mut())
            .enumerate()
        {
            // reversed succ row = original pred row; reversed pred row = original
            // succ row — pre-size so neither row reallocates.
            succ_row.reserve(self.pred_indices[i].len());
            pred_row.reserve(self.succ_indices[i].len());
        }
        let mut edges: crate::FxIndexMap<(usize, usize), AttrMap> =
            crate::FxIndexMap::with_capacity_and_hasher(
                self.edges.len(),
                rustc_hash::FxBuildHasher,
            );
        for u in self.node_order.values() {
            for &t in &self.succ_indices[u] {
                // original edge u -> t becomes t -> u in the reverse
                succ_indices[t].push(u);
                pred_indices[u].push(t);
                if let Some(attrs) = self.edges.get(&(u, t)) {
                    edges.insert((t, u), attrs.clone());
                }
            }
        }
        Self {
            mode: self.mode,
            revision: 0,
            node_order: self.node_order.clone(),
            node_attrs: self.node_attrs.clone(),
            succ_indices: crate::SlotRows::new(succ_indices),
            pred_indices: crate::SlotRows::new(pred_indices),
            edges,
            runtime_policy: self.runtime_policy.clone(),
            csr_cache: std::sync::Arc::default(),
            all_int_cache: std::sync::Arc::default(),
        }
    }

    /// Directed edge endpoints in the same order as `edges_ordered_borrowed`.
    #[must_use]
    pub fn edges_ordered_indices(&self) -> Vec<(usize, usize)> {
        let mut ordered = Vec::with_capacity(self.edges.len());
        let mut slot_to_pos = vec![0; self.node_order.slot_count()];
        for (pos, (_, slot)) in self.node_order.iter().enumerate() {
            slot_to_pos[slot] = pos;
        }

        for (u_pos, (_, u)) in self.node_order.iter().enumerate() {
            for &t in &self.succ_indices[u] {
                if self.edges.contains_key(&(u, t)) {
                    let t_pos = slot_to_pos[t];
                    ordered.push((u_pos, t_pos));
                }
            }
        }

        ordered
    }

    #[must_use]
    pub fn snapshot(&self) -> DiGraphSnapshot {
        let mut node_attrs: std::collections::BTreeMap<String, AttrMap> =
            std::collections::BTreeMap::new();
        for (name, slot) in self.node_order.iter() {
            if !self.node_attrs[slot].is_empty() {
                node_attrs.insert(name.clone(), self.node_attrs[slot].clone());
            }
        }
        DiGraphSnapshot {
            mode: self.mode,
            nodes: self.node_order.keys().cloned().collect(),
            node_attrs,
            edges: self.edges_ordered(),
        }
    }

    /// Convert to an undirected Graph by dropping directionality.
    /// Both (u→v) and (v→u) merge into a single undirected edge.
    /// When both exist, the latter's attributes overwrite the former's.
    #[must_use]
    pub fn to_undirected(&self) -> crate::Graph {
        let mut g = crate::Graph::with_runtime_policy(self.runtime_policy.clone());
        for (node, slot) in self.node_order.iter() {
            g.add_node_with_attrs(node.clone(), self.node_attrs[slot].clone());
        }
        // Iterate in canonical node->successor order (`edges_ordered`) rather
        // than the private `edges` IndexMap order: this matches networkx's
        // reciprocal-edge merge order (for a<->b the later-processed direction's
        // attrs win) AND keeps the `edges` map order unobservable, so
        // `remove_node` can drop incident edges with O(1) swap_remove.
        //
        // br-r37-c1-toundirbatch (cc): collect the directed edges (same
        // edges_ordered order) + one extend_edges_with_attrs_unrecorded instead
        // of per-edge add_edge_with_attrs (a policy record EACH — and for a<->b
        // reciprocal digraphs HALF the calls are merge-dups). The batch helper
        // dedups on the canonical endpoint pair and merges attrs via the same
        // `extend` (later direction wins), in the identical order; all nodes are
        // pre-added so endpoints resolve to existing indices → byte-identical.
        let edges: Vec<(String, String, AttrMap)> = self
            .edges_ordered()
            .into_iter()
            .map(|snap| (snap.left, snap.right, snap.attrs))
            .collect();
        let _ = g.extend_edges_with_attrs_unrecorded(edges);
        g
    }

    // -----------------------------------------------------------------------
    // Internal
    // -----------------------------------------------------------------------

    fn record_decision(
        &mut self,
        operation: &'static str,
        incompatibility_probability: f64,
        unknown_incompatible_feature: bool,
        evidence: Vec<EvidenceTerm>,
    ) -> DecisionAction {
        let action = self
            .runtime_policy
            .action_for(incompatibility_probability, unknown_incompatible_feature);
        self.runtime_policy.record(
            operation,
            action,
            incompatibility_probability,
            "argmin expected loss over {allow,full_validate,fail_closed}",
            evidence,
        );
        action
    }
}

/// yr2oc.1: the MultiDiGraph node table: [`crate::NodeOrder`] (name <-> slot
/// and the insertion order, with O(1) removal) plus slot-indexed attributes.
///
/// A removal tombstones its entry, and the table compacts once tombstones
/// outnumber live nodes. Nothing outside the table holds a slot (the rows and
/// the edge map are keyed by name), so compaction moves only the attributes.
#[derive(Debug, Clone, Default)]
struct MultiDiNodes {
    order: crate::NodeOrder,
    attrs: Vec<AttrMap>,
}

impl MultiDiNodes {
    fn len(&self) -> usize {
        self.order.len()
    }

    fn is_empty(&self) -> bool {
        self.order.is_empty()
    }

    fn contains_key(&self, name: &str) -> bool {
        self.order.contains_key(name)
    }

    fn get(&self, name: &str) -> Option<&AttrMap> {
        Some(&self.attrs[*self.order.get(name)?])
    }

    fn get_mut(&mut self, name: &str) -> Option<&mut AttrMap> {
        let slot = *self.order.get(name)?;
        Some(&mut self.attrs[slot])
    }

    /// The position of a node: O(1) until a removal, O(log V) after.
    fn get_index_of(&self, name: &str) -> Option<usize> {
        self.order.get_index_of(name)
    }

    fn get_index(&self, position: usize) -> Option<(&String, &AttrMap)> {
        let (name, slot) = self.order.get_index(position)?;
        Some((name, &self.attrs[slot]))
    }

    fn keys(&self) -> impl Iterator<Item = &String> + '_ {
        self.order.keys()
    }

    fn iter(&self) -> impl Iterator<Item = (&String, &AttrMap)> + '_ {
        self.order
            .iter()
            .map(|(name, slot)| (name, &self.attrs[slot]))
    }

    fn reserve(&mut self, additional: usize) {
        self.order.reserve(additional);
        self.attrs.reserve(additional);
    }

    /// Append a node that is not in the table.
    fn push(&mut self, name: String, attrs: AttrMap) -> &mut AttrMap {
        let slot = self.order.push(name);
        debug_assert_eq!(slot, self.attrs.len());
        self.attrs.push(attrs);
        &mut self.attrs[slot]
    }

    /// `IndexMap::insert`: a present node keeps its place and takes `attrs`.
    fn insert(&mut self, name: String, attrs: AttrMap) -> Option<AttrMap> {
        if let Some(existing) = self.get_mut(&name) {
            return Some(std::mem::replace(existing, attrs));
        }
        self.push(name, attrs);
        None
    }

    /// Remove a node, keeping the order of the rest: O(1), plus an amortised
    /// O(1) share of compaction.
    fn remove(&mut self, name: &str) -> Option<AttrMap> {
        let (slot, popped) = self.order.remove(name)?;
        let attrs = if popped {
            self.attrs.pop().expect("attributes are slot-parallel")
        } else {
            std::mem::take(&mut self.attrs[slot])
        };
        if self.order.slot_count() - self.order.len() > self.order.len() {
            let old_to_new = self.order.compact();
            self.attrs = std::mem::take(&mut self.attrs)
                .into_iter()
                .zip(old_to_new)
                .filter_map(|(attrs, new)| (new != usize::MAX).then_some(attrs))
                .collect();
        }
        Some(attrs)
    }

    /// Name -> position for a whole-graph walk: one O(V) table instead of a
    /// rank query per lookup while removals have separated slots and positions.
    fn positions(&self) -> NodePositions<'_> {
        NodePositions {
            order: &self.order,
            table: self.order.slot_positions(),
        }
    }
}

/// See [`MultiDiNodes::positions`].
struct NodePositions<'a> {
    order: &'a crate::NodeOrder,
    table: Option<Vec<usize>>,
}

impl NodePositions<'_> {
    fn of(&self, name: &str) -> Option<usize> {
        let slot = *self.order.get(name)?;
        Some(self.table.as_ref().map_or(slot, |table| table[slot]))
    }
}

/// A MultiDiGraph's successor or predecessor rows: node -> neighbour -> keys.
/// The outer map is unordered (yr2oc.1): every ordered walk goes through the
/// node table, so a removal drops a row in O(1) instead of shifting the map.
type MultiDiRows =
    rustc_hash::FxHashMap<String, crate::FxIndexMap<String, crate::FxIndexSet<usize>>>;

#[derive(Debug, Clone)]
pub struct MultiDiGraph {
    mode: CompatibilityMode,
    revision: u64,
    nodes: MultiDiNodes,
    successors: MultiDiRows,
    predecessors: MultiDiRows,
    /// br-r37-c1-19ngg: the key -> attrs bucket is Fx-hashed too (it was the
    /// one map here left on SipHash); insertion-ordered, so order is unchanged.
    edges: crate::FxIndexMap<DirectedEdgeKey, crate::FxIndexMap<usize, AttrMap>>,
    runtime_policy: RuntimePolicy,
    edge_count: usize,
    csr_cache: MultiDiCsrCache,
    /// br-r37-c1-s8dj1: (source position, target position) -> position in
    /// `edges`, so an edge can be reached without hashing the node keys.
    ///
    /// Every level of this graph's storage is keyed by `String`, so resolving an
    /// edge cost a hash of the whole node key however the caller arrived — which
    /// is why `has_edge_by_indices` could only convert its indices BACK to names,
    /// and the directed class kept a 3.7x key-length slope on `has_edge(u, v)`
    /// and 4.4x on the keyed form while the slot-addressed undirected twin sat at
    /// networkx's own 1.9x.
    ///
    /// BUILT LAZILY AND MAINTAINED INCREMENTALLY. A first attempt rebuilt it on
    /// every revision change, which made an add-then-check loop O(E) per probe
    /// and measured 137x slower than networkx at E=8000 — the rebuild was worse
    /// than the hashing it removed. Adds now patch the map in O(1), because
    /// inserting into `edges` APPENDS and adding a node APPENDS, so no existing
    /// position moves.
    ///
    /// THE STAMP IS THE SAFETY NET, not merely an optimisation. Removals do move
    /// both index spaces (`remove_node` moves every later node position down,
    /// `remove_edge` swap-removes from `edges`) and are deliberately NOT patched
    /// here: they bump `revision` without re-stamping, so the next read finds a
    /// stale stamp and rebuilds. Any mutation path not handled explicitly
    /// therefore degrades to a correct rebuild rather than a wrong answer.
    edge_pair_index: EdgePairIndex,
}

impl MultiDiGraph {
    /// br-r37-c1-7qqr8: positional index of a node, matching
    /// `DiGraph::get_node_index`. MultiDiGraph carried the
    /// `NodeIndexLookupCache` (br-r37-c1-ic4cv) but only ever used its
    /// present-set half, so there was no accessor to populate the index half
    /// from — which is what the edge-attr index lookaside needs.
    #[must_use]
    pub fn get_node_index(&self, node: &str) -> Option<usize> {
        self.nodes.get_index_of(node)
    }

    /// br-r37-c1-7dpyg: structural clone with a FRESH RuntimePolicy —
    /// see Graph::clone_with_fresh_policy.
    #[must_use]
    pub fn clone_with_fresh_policy(&self) -> Self {
        Self {
            mode: self.mode,
            revision: self.revision,
            nodes: self.nodes.clone(),
            successors: self.successors.clone(),
            predecessors: self.predecessors.clone(),
            edges: self.edges.clone(),
            runtime_policy: RuntimePolicy::new(self.mode),
            edge_count: self.edge_count,
            csr_cache: self.csr_cache.clone(),
            edge_pair_index: EdgePairIndex::default(),
        }
    }

    /// br-r37-c1-s0d4x: reorder every PRED row into NetworkX's
    /// `MultiDiGraph.copy()` walk order — the multigraph counterpart of
    /// DiGraph::reorder_pred_rows_for_nx_copy_walk: each pred row's cells
    /// sort by `(pos(u), index of v within succ[u])`; succ rows are
    /// recreated in their original order by the u-major walk.
    pub fn reorder_pred_rows_for_nx_copy_walk(&mut self) {
        // br-r37-c1-predrebuild: build each pred row's u-major copy-walk order
        // DIRECTLY by walking succ in pos(u) order, replacing the prior
        // per-pred-entry double `get_index_of` lookup + per-row sort. Each pred
        // row holds DISTINCT predecessors u, so pos(u) is the sole ordering key
        // (the succ-index tiebreak never fires), and appending u to pred[v] while
        // iterating succ in insertion (= position) order yields the exact same
        // ordering — byte-identical pred row order, no sort, no index lookups.
        let mut order_map: IndexMap<String, Vec<String>> =
            IndexMap::with_capacity(self.predecessors.len());
        for u in self.nodes.keys() {
            let Some(succ_row) = self.successors.get(u.as_str()) else {
                continue;
            };
            for v in succ_row.keys() {
                order_map.entry(v.clone()).or_default().push(u.clone());
            }
        }
        let orders: Vec<(String, Vec<String>)> = order_map.into_iter().collect();
        self.apply_row_orders(&orders, true);
    }

    /// br-r37-c1-36v6r: order the successor and predecessor rows as given -
    /// a materialised view's rows, which no insertion history reproduces -
    /// and move the revision, so revision-keyed memos (the CSR) rebuild from
    /// the new rows. Keyed cells move wholesale.
    pub fn set_row_orders(
        &mut self,
        succ: &[(String, Vec<String>)],
        pred: &[(String, Vec<String>)],
    ) {
        self.apply_row_orders(succ, false);
        self.apply_row_orders(pred, true);
        self.revision = self.revision.saturating_add(1);
        // Clones share the CSR memo, keyed by revision alone.
        self.csr_cache = std::sync::Arc::default();
    }

    /// br-r37-c1-36v6r: order each successor and predecessor row as
    /// `source`'s row for the node of the same name - its succ / pred rows,
    /// or its pred / succ rows when `swap` (a reverse view's); see
    /// DiGraph::reorder_rows_like. A stable sort on the neighbour's position
    /// in the source row, so entries the source row lacks keep their
    /// relative order after the matched ones. Keyed cells move wholesale;
    /// `revision` moves if any row does.
    pub fn reorder_rows_like(&mut self, source: &Self, swap: bool) {
        fn reorder(rows: &mut MultiDiRows, source_rows: &MultiDiRows) -> bool {
            let mut changed = false;
            for (node, row) in rows.iter_mut() {
                let Some(source_row) = source_rows.get(node.as_str()) else {
                    continue;
                };
                if row.len() < 2 {
                    continue;
                }
                let rank = |w: &String| source_row.get_index_of(w.as_str()).unwrap_or(usize::MAX);
                if !row.keys().map(rank).is_sorted() {
                    row.sort_by_cached_key(|w, _| rank(w));
                    changed = true;
                }
            }
            changed
        }
        let (succ_source, pred_source) = if swap {
            (&source.predecessors, &source.successors)
        } else {
            (&source.successors, &source.predecessors)
        };
        let succ_changed = reorder(&mut self.successors, succ_source);
        let pred_changed = reorder(&mut self.predecessors, pred_source);
        if succ_changed || pred_changed {
            self.revision = self.revision.saturating_add(1);
            self.csr_cache = std::sync::Arc::default();
        }
    }

    /// br-r37-c1-u3qyn: restore explicit succ/pred row orders (pickle
    /// round-trip) — see Graph::apply_row_orders. Keyed cells move
    /// wholesale; `pred` selects the adjacency side.
    pub fn apply_row_orders(&mut self, orders: &[(String, Vec<String>)], pred: bool) {
        let map = if pred {
            &mut self.predecessors
        } else {
            &mut self.successors
        };
        for (node, order) in orders {
            let Some(row) = map.get_mut(node.as_str()) else {
                continue;
            };
            let mut old = std::mem::take(row);
            let mut new_row =
                crate::FxIndexMap::with_capacity_and_hasher(old.len(), rustc_hash::FxBuildHasher);
            for v in order {
                if let Some((k, val)) = old.shift_remove_entry(v.as_str()) {
                    new_row.insert(k, val);
                }
            }
            for (k, val) in old {
                new_row.insert(k, val);
            }
            *row = new_row;
        }
    }

    #[must_use]
    pub fn new(mode: CompatibilityMode) -> Self {
        Self {
            mode,
            revision: 0,
            nodes: MultiDiNodes::default(),
            successors: MultiDiRows::default(),
            predecessors: MultiDiRows::default(),
            edges: crate::FxIndexMap::default(),
            runtime_policy: RuntimePolicy::new(mode),
            edge_count: 0,
            csr_cache: std::sync::Arc::default(),
            edge_pair_index: EdgePairIndex::default(),
        }
    }

    #[must_use]
    pub fn with_runtime_policy(runtime_policy: RuntimePolicy) -> Self {
        let mode = runtime_policy.mode();
        Self {
            mode,
            revision: 0,
            nodes: MultiDiNodes::default(),
            successors: MultiDiRows::default(),
            predecessors: MultiDiRows::default(),
            edges: crate::FxIndexMap::default(),
            runtime_policy,
            edge_count: 0,
            csr_cache: std::sync::Arc::default(),
            edge_pair_index: EdgePairIndex::default(),
        }
    }

    #[must_use]
    pub fn strict() -> Self {
        Self::new(CompatibilityMode::Strict)
    }

    #[must_use]
    pub fn hardened() -> Self {
        Self::new(CompatibilityMode::Hardened)
    }

    #[must_use]
    pub fn mode(&self) -> CompatibilityMode {
        self.mode
    }

    #[must_use]
    pub fn node_count(&self) -> usize {
        self.nodes.len()
    }

    #[must_use]
    pub fn edge_count(&self) -> usize {
        self.edge_count
    }

    #[must_use]
    pub fn number_of_selfloops(&self) -> usize {
        self.edges
            .iter()
            .filter(|(edge_key, _)| edge_key.source == edge_key.target)
            .map(|(_, edge_bucket)| edge_bucket.len())
            .sum()
    }

    /// Return an iterator over keys for edges from source to target.
    /// Edge keys as Vec.
    #[must_use]
    pub fn edge_keys(&self, source: &str, target: &str) -> Option<Vec<usize>> {
        self.successors
            .get(source)?
            .get(target)
            .map(|keys| keys.iter().copied().collect())
    }

    /// Edge keys as iterator.
    pub fn edge_keys_iter(
        &self,
        source: &str,
        target: &str,
    ) -> Option<impl Iterator<Item = &usize>> {
        self.successors
            .get(source)?
            .get(target)
            .map(|keys| keys.iter())
    }

    /// Every (source, target) pair joined by more than one edge, with its
    /// edge count, in edge insertion order.
    pub fn parallel_edge_counts(&self) -> impl Iterator<Item = (&str, &str, usize)> {
        self.edges
            .iter()
            .filter(|(_, keys)| keys.len() > 1)
            .map(|(edge, keys)| (edge.source.as_str(), edge.target.as_str(), keys.len()))
    }

    #[must_use]
    pub fn revision(&self) -> u64 {
        self.revision
    }

    /// Revision-keyed integer CSR of distinct successor and predecessor rows.
    /// Row order follows the existing map-backed rows exactly, so consumers can
    /// replace string-key walks without changing observable traversal order.
    #[must_use]
    pub fn csr(&self) -> std::sync::Arc<MultiDiCsr> {
        if let Ok(guard) = self.csr_cache.read()
            && let Some((rev, csr)) = guard.as_ref()
            && *rev == self.revision
        {
            return csr.clone();
        }
        let built = std::sync::Arc::new(self.build_csr());
        if let Ok(mut guard) = self.csr_cache.write() {
            *guard = Some((self.revision, built.clone()));
        }
        built
    }

    fn build_csr(&self) -> MultiDiCsr {
        let n = self.nodes.len();
        let distinct_succ_edges: usize = self.successors.values().map(IndexMap::len).sum();
        let distinct_pred_edges: usize = self.predecessors.values().map(IndexMap::len).sum();
        let mut succ_offsets = Vec::with_capacity(n + 1);
        let mut pred_offsets = Vec::with_capacity(n + 1);
        let mut succ_targets = Vec::with_capacity(distinct_succ_edges);
        let mut pred_targets = Vec::with_capacity(distinct_pred_edges);
        succ_offsets.push(0);
        pred_offsets.push(0);
        let positions = self.nodes.positions();
        for node in self.nodes.keys() {
            if let Some(row) = self.successors.get(node.as_str()) {
                succ_targets.extend(row.keys().filter_map(|target| {
                    positions
                        .of(target.as_str())
                        .map(|idx| u32::try_from(idx).unwrap_or(u32::MAX))
                }));
            }
            succ_offsets.push(succ_targets.len());

            if let Some(row) = self.predecessors.get(node.as_str()) {
                pred_targets.extend(row.keys().filter_map(|source| {
                    positions
                        .of(source.as_str())
                        .map(|idx| u32::try_from(idx).unwrap_or(u32::MAX))
                }));
            }
            pred_offsets.push(pred_targets.len());
        }
        MultiDiCsr {
            succ_offsets,
            succ_targets,
            pred_offsets,
            pred_targets,
        }
    }

    #[must_use]
    pub fn has_node(&self, node: &str) -> bool {
        self.nodes.contains_key(node)
    }

    #[must_use]
    pub fn has_edge(&self, source: &str, target: &str) -> bool {
        self.edges
            .get(&DirectedEdgeKeyRef::new(source, target))
            .is_some_and(|edge_bucket| !edge_bucket.is_empty())
    }

    /// br-r37-c1-04z53 (cc): directed `has_edge` straight from insertion-order
    /// indices (source-major), skipping the `i.to_string()` heap alloc for int
    /// nodes. Names come from the node table by index (borrowed); the caller
    /// guards each index with `node_index_matches_int`. Mirror of
    /// `Graph::has_edge_by_indices`.
    /// br-r37-c1-s8dj1: O(E) rebuild of the pair lookaside. Paid on the first
    /// probe, and again after any mutation that did not patch it.
    fn build_edge_pair_index(&self) -> EdgePairMap {
        let mut built =
            EdgePairMap::with_capacity_and_hasher(self.edges.len(), rustc_hash::FxBuildHasher);
        let positions = self.nodes.positions();
        for (position, edge_key) in self.edges.keys().enumerate() {
            if let Some(source_idx) = positions.of(edge_key.source.as_str())
                && let Some(target_idx) = positions.of(edge_key.target.as_str())
            {
                built.insert((source_idx, target_idx), position);
            }
        }
        built
    }

    /// br-r37-c1-s8dj1: position of the (source, target) bucket in `edges`,
    /// addressed by node POSITION and never by name.
    /// TRI-STATE: `None` means "no index available, use the name path" -- which
    /// is what the warm-up guard returns until a caller has proved it will probe
    /// enough to amortise the build. `Some(inner)` is an authoritative answer,
    /// `inner` being the bucket position or `None` for absent.
    fn edge_pair_position(&self, source_idx: usize, target_idx: usize) -> Option<Option<usize>> {
        use std::sync::atomic::Ordering;
        if let Ok(guard) = self.edge_pair_index.map.read()
            && let Some((stamp, map)) = guard.as_ref()
            && *stamp == self.revision
        {
            return Some(map.get(&(source_idx, target_idx)).copied());
        }
        let revision = self.revision;
        if self.edge_pair_index.probed_revision.load(Ordering::Relaxed) != revision {
            self.edge_pair_index
                .probed_revision
                .store(revision, Ordering::Relaxed);
            self.edge_pair_index
                .probes_since_build
                .store(1, Ordering::Relaxed);
            return None;
        }
        let seen = self
            .edge_pair_index
            .probes_since_build
            .fetch_add(1, Ordering::Relaxed)
            + 1;
        if seen < pair_index_warmup(self.edges.len()) {
            return None;
        }
        let built = self.build_edge_pair_index();
        let found = built.get(&(source_idx, target_idx)).copied();
        if let Ok(mut guard) = self.edge_pair_index.map.write() {
            *guard = Some((revision, built));
        }
        Some(found)
    }

    /// br-r37-c1-s8dj1: patch the lookaside through an edge ADD rather than
    /// letting the stamp drop it, which is what made the rebuild version a loss
    /// on add-then-check loops.
    ///
    /// Sound because an add MOVES NOTHING: `IndexMap::insert` appends, so a new
    /// pair takes the last position and every existing position stands, and a
    /// node added by this same call appends to `nodes` so every existing node
    /// position stands too.
    ///
    /// Costs nothing on a graph nobody has probed: the liveness check comes
    /// first, and a cold cache returns before any node-key hash is paid.
    fn pair_index_note_edge_added(
        &mut self,
        previous_revision: u64,
        source: &str,
        target: &str,
        new_pair_position: Option<usize>,
    ) {
        match self.edge_pair_index.map.get_mut() {
            Ok(Some((stamp, _))) if *stamp == previous_revision => {}
            _ => return,
        }
        let resolved = match new_pair_position {
            Some(position) => match (
                self.nodes.get_index_of(source),
                self.nodes.get_index_of(target),
            ) {
                (Some(source_idx), Some(target_idx)) => Some(((source_idx, target_idx), position)),
                // Cannot describe the new edge: drop rather than half-patch.
                _ => {
                    if let Ok(slot) = self.edge_pair_index.map.get_mut() {
                        *slot = None;
                    }
                    return;
                }
            },
            None => None,
        };
        let current = self.revision;
        if let Ok(Some((stamp, map))) = self.edge_pair_index.map.get_mut() {
            if let Some((pair, position)) = resolved {
                map.insert(pair, position);
            }
            *stamp = current;
        }
    }

    /// br-r37-c1-s8dj1: is the pair index worth maintaining right now? A graph
    /// nobody has probed carries no map, and must pay nothing for one.
    fn edge_pair_index_is_live(&self, at_revision: u64) -> bool {
        matches!(
            self.edge_pair_index.map.read().as_deref(),
            Ok(Some((stamp, _))) if *stamp == at_revision
        )
    }

    /// br-r37-c1-s8dj1: patch the lookaside through an edge REMOVAL.
    ///
    /// `remove_edge` SWAP-removes the bucket when its last key goes, which moves
    /// exactly ONE entry: whatever sat last now sits in the freed slot. That is a
    /// bounded repair, unlike `remove_node`, which shift-removes from `nodes` and
    /// renumbers every later node — that one still drops the map and rebuilds,
    /// which is proportionate because the removal itself is already O(V+E).
    ///
    /// `dropped_position` is `None` when the bucket survived (a parallel key went
    /// but the pair remains), in which case nothing moved and only the stamp
    /// needs carrying forward.
    fn pair_index_note_edge_removed(
        &mut self,
        previous_revision: u64,
        source: &str,
        target: &str,
        dropped_position: Option<usize>,
    ) {
        match self.edge_pair_index.map.get_mut() {
            Ok(Some((stamp, _))) if *stamp == previous_revision => {}
            _ => return,
        }
        let removed_pair = match dropped_position {
            Some(_) => match (
                self.nodes.get_index_of(source),
                self.nodes.get_index_of(target),
            ) {
                (Some(source_idx), Some(target_idx)) => Some((source_idx, target_idx)),
                _ => {
                    if let Ok(slot) = self.edge_pair_index.map.get_mut() {
                        *slot = None;
                    }
                    return;
                }
            },
            None => None,
        };
        // The entry that swap_remove relocated into the freed slot, if any.
        let relocated = match dropped_position {
            Some(position) if position < self.edges.len() => match self.edges.get_index(position) {
                Some((moved_key, _)) => {
                    let moved_source = moved_key.source.clone();
                    let moved_target = moved_key.target.clone();
                    match (
                        self.nodes.get_index_of(moved_source.as_str()),
                        self.nodes.get_index_of(moved_target.as_str()),
                    ) {
                        (Some(source_idx), Some(target_idx)) => {
                            Some(((source_idx, target_idx), position))
                        }
                        _ => {
                            if let Ok(slot) = self.edge_pair_index.map.get_mut() {
                                *slot = None;
                            }
                            return;
                        }
                    }
                }
                None => None,
            },
            _ => None,
        };
        let current = self.revision;
        if let Ok(Some((stamp, map))) = self.edge_pair_index.map.get_mut() {
            if let Some(pair) = removed_pair {
                map.remove(&pair);
            }
            if let Some((pair, position)) = relocated {
                map.insert(pair, position);
            }
            *stamp = current;
        }
    }

    /// br-r37-c1-s8dj1: now a REAL index path.
    ///
    /// It used to resolve both indices back to names and call `has_edge(&str,
    /// &str)`, which hashes the full-length keys again — so a caller's O(1)
    /// index resolution bought nothing, and `PyMultiDiGraph`'s exact-string
    /// branch (br-r37-c1-ptiz2) inherited the slope it was written to remove.
    ///
    /// An `edges` bucket is dropped as soon as its last key goes (see
    /// `remove_edge`), so membership of the pair map is exactly `has_edge`.
    #[must_use]
    pub fn has_edge_by_indices(&self, source_idx: usize, target_idx: usize) -> bool {
        if let Some(position) = self.edge_pair_position(source_idx, target_idx) {
            return position.is_some();
        }
        match (
            self.get_node_name(source_idx),
            self.get_node_name(target_idx),
        ) {
            (Some(source), Some(target)) => self.has_edge(source, target),
            _ => false,
        }
    }

    /// br-r37-c1-s8dj1: how many PARALLEL edges a pair carries, addressed by
    /// node POSITION. `number_of_edges(u, v)` built two owned canonicals and
    /// hashed the pair to answer what is a bucket length, and measured 0.272x
    /// of networkx at 2000-character keys — the worst surviving row on this
    /// class after the keyed `has_edge` path landed.
    ///
    /// Falls back to the name path while the lookaside is still warming up, so
    /// it inherits the same amortisation guard rather than forcing a build.
    #[must_use]
    pub fn edge_key_count_by_indices(&self, source_idx: usize, target_idx: usize) -> usize {
        if let Some(position) = self.edge_pair_position(source_idx, target_idx) {
            return position
                .and_then(|found| self.edges.get_index(found))
                .map_or(0, |(_, keyed)| keyed.len());
        }
        match (
            self.get_node_name(source_idx),
            self.get_node_name(target_idx),
        ) {
            (Some(source), Some(target)) => self
                .edges
                .get(&DirectedEdgeKeyRef::new(source, target))
                .map_or(0, IndexMap::len),
            _ => 0,
        }
    }

    /// br-r37-c1-s8dj1: attributes of a KEYED directed edge, addressed by node
    /// POSITION — the directed twin of `MultiGraph::edge_attrs_by_indices`, and
    /// the primitive whose absence is why `PyMultiDiGraph::has_edge` left its
    /// keyed form on the string path.
    #[must_use]
    pub fn edge_attrs_by_indices(
        &self,
        source_idx: usize,
        target_idx: usize,
        key: usize,
    ) -> Option<&AttrMap> {
        if let Some(position) = self.edge_pair_position(source_idx, target_idx) {
            let (_, keyed) = self.edges.get_index(position?)?;
            return keyed.get(&key);
        }
        let source = self.get_node_name(source_idx)?;
        let target = self.get_node_name(target_idx)?;
        self.edges
            .get(&DirectedEdgeKeyRef::new(source, target))
            .and_then(|bucket| bucket.get(&key))
    }

    #[must_use]
    pub fn nodes_ordered(&self) -> Vec<&str> {
        self.nodes.keys().map(String::as_str).collect()
    }

    /// Resolve the current node name for an insertion-order index.
    #[must_use]
    #[inline]
    pub fn get_node_name(&self, index: usize) -> Option<&str> {
        self.nodes.get_index(index).map(|(name, _)| name.as_str())
    }

    /// br-cc-nbunchbulk: int membership fast path for `_nbunch_present` — see the
    /// DiGraph accessor.
    #[must_use]
    pub fn node_index_matches_int(&self, idx: usize) -> bool {
        self.nodes
            .get_index(idx)
            .is_some_and(|(k, _)| k.parse::<usize>() == Ok(idx))
    }

    #[must_use]
    pub fn successors(&self, node: &str) -> Option<Vec<&str>> {
        self.successors
            .get(node)
            .map(|neighbors| neighbors.keys().map(String::as_str).collect::<Vec<&str>>())
    }

    #[must_use]
    pub fn successors_iter(&self, node: &str) -> Option<impl Iterator<Item = &str> + '_> {
        self.successors
            .get(node)
            .map(|neighbors| neighbors.keys().map(String::as_str))
    }

    #[must_use]
    pub fn predecessors(&self, node: &str) -> Option<Vec<&str>> {
        self.predecessors
            .get(node)
            .map(|neighbors| neighbors.keys().map(String::as_str).collect::<Vec<&str>>())
    }

    #[must_use]
    pub fn predecessors_iter(&self, node: &str) -> Option<impl Iterator<Item = &str> + '_> {
        self.predecessors
            .get(node)
            .map(|neighbors| neighbors.keys().map(String::as_str))
    }

    #[must_use]
    pub fn neighbors(&self, node: &str) -> Option<Vec<&str>> {
        self.successors(node)
    }

    #[must_use]
    pub fn neighbors_iter(&self, node: &str) -> Option<impl Iterator<Item = &str> + '_> {
        self.successors_iter(node)
    }

    #[must_use]
    pub fn node_attrs(&self, node: &str) -> Option<&AttrMap> {
        self.nodes.get(node)
    }

    #[must_use]
    pub fn edge_attrs(&self, source: &str, target: &str, key: usize) -> Option<&AttrMap> {
        self.edges
            .get(&DirectedEdgeKeyRef::new(source, target))
            .and_then(|edge_bucket| edge_bucket.get(&key))
    }

    #[must_use]
    pub fn edge_attr_values(
        &self,
        source: &str,
        target: &str,
    ) -> Option<impl Iterator<Item = &AttrMap> + '_> {
        self.edges
            .get(&DirectedEdgeKeyRef::new(source, target))
            .map(|edge_bucket| edge_bucket.values())
    }

    #[must_use]
    pub fn evidence_ledger(&self) -> &EvidenceLedger {
        self.runtime_policy.decision_log()
    }

    pub fn drain_decision_records(&mut self) -> Vec<DecisionRecord> {
        self.runtime_policy.drain_decision_records()
    }

    /// Record a bounded recovery performed outside the graph (e.g. a reader
    /// skipping a malformed line in hardened mode) in this graph's ledger.
    pub fn record_external_recovery(
        &mut self,
        operation: impl Into<std::borrow::Cow<'static, str>>,
        rationale: impl Into<std::borrow::Cow<'static, str>>,
    ) {
        self.runtime_policy.record(
            operation,
            DecisionAction::FullValidate,
            0.7,
            rationale,
            Vec::new(),
        );
    }

    #[must_use]
    pub fn runtime_policy(&self) -> &RuntimePolicy {
        &self.runtime_policy
    }

    pub fn set_runtime_policy(&mut self, runtime_policy: RuntimePolicy) {
        self.mode = runtime_policy.mode();
        self.runtime_policy = runtime_policy;
    }

    #[must_use]
    pub fn is_directed(&self) -> bool {
        true
    }

    #[must_use]
    pub fn is_multigraph(&self) -> bool {
        true
    }

    /// Return the out-degree of a node (number of outgoing parallel edges).
    #[must_use]
    pub fn out_degree(&self, node: &str) -> usize {
        self.successors
            .get(node)
            .map_or(0, |succs| succs.values().map(IndexSet::len).sum())
    }

    /// Return the in-degree of a node (number of incoming parallel edges).
    #[must_use]
    pub fn in_degree(&self, node: &str) -> usize {
        self.predecessors
            .get(node)
            .map_or(0, |preds| preds.values().map(IndexSet::len).sum())
    }

    /// Return the degree of a node (in-degree + out-degree).
    #[must_use]
    pub fn degree(&self, node: &str) -> usize {
        self.in_degree(node) + self.out_degree(node)
    }

    /// br-r37-c1-mgisol (cc): native isolate detection for MultiDiGraph. A node
    /// is isolated iff it has no incident edges in either direction (empty/absent
    /// successor AND predecessor rows). A self-loop records the node in both its
    /// own successor and predecessor rows, keeping it non-isolated (matches nx's
    /// degree-2 self-loop convention). Yields nodes in insertion order, identical
    /// to the old `multidigraph_to_simple_digraph` projection path but without the
    /// per-call O(V+E) simple-graph rebuild.
    #[must_use]
    pub fn isolates(&self) -> Vec<String> {
        self.nodes
            .keys()
            .filter(|node| self.is_isolate(node.as_str()))
            .cloned()
            .collect()
    }

    /// br-r37-c1-mgisol (cc): isolate count without the simple-digraph projection.
    #[must_use]
    pub fn number_of_isolates(&self) -> usize {
        self.nodes
            .keys()
            .filter(|node| self.is_isolate(node.as_str()))
            .count()
    }

    /// br-r37-c1-mgisol (cc): O(1) isolate predicate. Absent node -> false
    /// (mirrors `is_isolate_directed(&DiGraph)`); the binding validates presence.
    #[must_use]
    pub fn is_isolate(&self, node: &str) -> bool {
        self.nodes.contains_key(node)
            && self.successors.get(node).is_none_or(IndexMap::is_empty)
            && self.predecessors.get(node).is_none_or(IndexMap::is_empty)
    }

    pub fn add_node(&mut self, node: impl AsRef<str> + Into<String>) -> bool {
        self.add_node_with_attrs(node, AttrMap::new())
    }

    /// br-r37-c1-wa1b9: the storage half of [`Self::add_node_with_attrs`],
    /// without its ledger record. See the undirected `Graph` sibling.
    ///
    /// br-r37-c1-jc9e4: this key used to be materialised THREE times per call —
    /// `node.into()` plus a `clone()` for the `nodes` entry and another for the
    /// `successors` entry — because `entry()` allocates its key eagerly whether
    /// or not the slot is vacant. On a node that already exists all three were
    /// built only to be dropped. The allocation census measured this pair at
    /// 4.000 allocations per call against 1.000 for every other graph type,
    /// which is how the outlier was found. Each map is now probed borrowed and
    /// the key owned only where a slot is genuinely created.
    ///
    /// The `successors` / `predecessors` rows are still created unconditionally
    /// when missing rather than being skipped whenever the node exists: nothing
    /// here may assume that every path which inserts into `nodes` also created
    /// those rows, and `contains_key` + `insert`-when-absent is exactly what
    /// `entry(..).or_default()` did, including appending at the end when vacant.
    fn add_node_with_attrs_unrecorded(
        &mut self,
        node: impl AsRef<str> + Into<String>,
        attrs: AttrMap,
    ) -> (bool, bool, usize) {
        let existed = self.nodes.contains_key(node.as_ref());
        let mut changed = !existed;
        let attrs_count = {
            let bucket = if existed {
                self.nodes
                    .get_mut(node.as_ref())
                    .expect("contains_key reported this node present")
            } else {
                self.nodes.push(node.as_ref().to_owned(), AttrMap::new())
            };
            if !attrs.is_empty()
                && attrs
                    .iter()
                    .any(|(key, value)| bucket.get(key) != Some(value))
            {
                changed = true;
            }
            bucket.extend(attrs);
            bucket.len()
        };
        if !self.successors.contains_key(node.as_ref()) {
            self.successors
                .insert(node.as_ref().to_owned(), IndexMap::default());
        }
        if !self.predecessors.contains_key(node.as_ref()) {
            self.predecessors.insert(node.into(), IndexMap::default());
        }
        if changed {
            self.revision = self.revision.saturating_add(1);
        }
        (changed, existed, attrs_count)
    }

    pub fn add_node_with_attrs(
        &mut self,
        node: impl AsRef<str> + Into<String>,
        attrs: AttrMap,
    ) -> bool {
        let (changed, existed, attrs_count) = self.add_node_with_attrs_unrecorded(node, attrs);
        self.record_decision(
            "add_node",
            0.0,
            false,
            vec![
                EvidenceTerm {
                    signal: "node_preexisting".into(),
                    observed_value: bool_evidence(existed),
                    log_likelihood_ratio: -3.0,
                },
                EvidenceTerm {
                    signal: "attrs_count".into(),
                    observed_value: count_evidence(attrs_count),
                    log_likelihood_ratio: -1.0,
                },
            ],
        );
        changed
    }

    pub fn add_edge(
        &mut self,
        source: impl Into<String>,
        target: impl Into<String>,
    ) -> Result<usize, GraphError> {
        self.add_edge_impl(source, target, None, AttrMap::new())
    }

    /// br-r37-c1-l5ve7: bulk node insert WITH attrs, one ledger record —
    /// MultiDiGraph mirror of DiGraph::extend_nodes_with_attrs_unrecorded.
    /// Existing nodes merge attrs (extend).
    pub fn extend_nodes_with_attrs_unrecorded<I>(&mut self, nodes: I) -> usize
    where
        I: IntoIterator<Item = (String, AttrMap)>,
    {
        let mut inserted = 0usize;
        for (node, attrs) in nodes {
            if let Some(existing) = self.nodes.get_mut(&node) {
                existing.extend(attrs);
                continue;
            }
            self.nodes.insert(node.clone(), attrs);
            self.successors.entry(node.clone()).or_default();
            self.predecessors.entry(node).or_default();
            inserted += 1;
        }
        if inserted > 0 {
            self.revision = self
                .revision
                .saturating_add(u64::try_from(inserted).unwrap_or(u64::MAX));
            self.record_decision(
                "extend_nodes_with_attrs_unrecorded",
                0.0,
                false,
                vec![EvidenceTerm {
                    signal: "batch_node_count".into(),
                    observed_value: count_evidence(inserted),
                    log_likelihood_ratio: -1.0,
                }],
            );
        }
        inserted
    }

    /// Bulk node insert without attrs, one summary ledger record.
    pub fn extend_nodes_unrecorded<I, N>(&mut self, nodes: I) -> usize
    where
        I: IntoIterator<Item = N>,
        N: Into<String>,
    {
        let iterator = nodes.into_iter();
        let (lower_bound, _) = iterator.size_hint();
        self.nodes.reserve(lower_bound);
        self.successors.reserve(lower_bound);
        self.predecessors.reserve(lower_bound);

        let mut inserted = 0usize;
        for node in iterator {
            let node = node.into();
            if self.nodes.contains_key(&node) {
                continue;
            }
            self.nodes.insert(node.clone(), AttrMap::new());
            self.successors.entry(node.clone()).or_default();
            self.predecessors.entry(node).or_default();
            inserted += 1;
        }
        if inserted > 0 {
            self.revision = self
                .revision
                .saturating_add(u64::try_from(inserted).unwrap_or(u64::MAX));
            self.record_decision(
                "extend_nodes_unrecorded",
                0.0,
                false,
                vec![EvidenceTerm {
                    signal: "batch_node_count".into(),
                    observed_value: count_evidence(inserted),
                    log_likelihood_ratio: -1.0,
                }],
            );
        }
        inserted
    }

    /// br-r37-c1-l5ve7: bulk KEYED edge insert WITH attrs, one ledger
    /// record — for native copy/convert paths that replicate a source
    /// multigraph's exact internal keys (add_edge_impl pays TWO
    /// record_decision calls per edge). Endpoint nodes are created on
    /// demand; an existing (u, v, key) cell merges attrs (extend),
    /// matching add_edge_with_key_and_attrs.
    pub fn extend_keyed_edges_with_attrs_unrecorded<I>(&mut self, edges: I) -> usize
    where
        I: IntoIterator<Item = (String, String, usize, AttrMap)>,
    {
        let mut inserted = 0usize;
        for (source, target, key, attrs) in edges {
            if !self.nodes.contains_key(&source) {
                self.nodes.insert(source.clone(), AttrMap::new());
                self.successors.entry(source.clone()).or_default();
                self.predecessors.entry(source.clone()).or_default();
            }
            if source != target && !self.nodes.contains_key(&target) {
                self.nodes.insert(target.clone(), AttrMap::new());
                self.successors.entry(target.clone()).or_default();
                self.predecessors.entry(target.clone()).or_default();
            }
            let edge_key = DirectedEdgeKey::new(&source, &target);
            let bucket = self.edges.entry(edge_key).or_default();
            if !bucket.contains_key(&key) {
                self.edge_count += 1;
                inserted += 1;
            }
            bucket.entry(key).or_default().extend(attrs);
            self.successors
                .entry(source.clone())
                .or_default()
                .entry(target.clone())
                .or_default()
                .insert(key);
            self.predecessors
                .entry(target)
                .or_default()
                .entry(source)
                .or_default()
                .insert(key);
        }
        if inserted > 0 {
            self.revision = self
                .revision
                .saturating_add(u64::try_from(inserted).unwrap_or(u64::MAX));
            self.record_decision(
                "extend_keyed_edges_with_attrs_unrecorded",
                0.0,
                false,
                vec![EvidenceTerm {
                    signal: "batch_edge_count".into(),
                    observed_value: count_evidence(inserted),
                    log_likelihood_ratio: -1.0,
                }],
            );
        }
        inserted
    }

    /// Bulk-add keyed attributed directed multiedges into a fresh graph when the
    /// caller has already assigned NetworkX-compatible node indices and per-pair
    /// keys. This avoids hashing/copying canonical endpoint strings for every
    /// edge in fresh `MultiDiGraph.add_edges_from` batches.
    #[must_use]
    pub fn extend_fresh_index_keyed_edges_with_attrs_unrecorded<I, N>(
        &mut self,
        nodes: N,
        edges: I,
    ) -> usize
    where
        I: IntoIterator<Item = (usize, usize, usize, AttrMap)>,
        N: IntoIterator<Item = String>,
    {
        if !self.nodes.is_empty()
            || !self.successors.is_empty()
            || !self.predecessors.is_empty()
            || !self.edges.is_empty()
        {
            return 0;
        }

        let node_labels: Vec<String> = nodes.into_iter().collect();
        for node in &node_labels {
            self.nodes.insert(node.clone(), AttrMap::new());
            self.successors
                .insert(node.clone(), crate::FxIndexMap::default());
            self.predecessors
                .insert(node.clone(), crate::FxIndexMap::default());
        }

        let node_count = node_labels.len();
        let mut inserted = 0usize;
        let mut merged_changed = false;
        for (source_idx, target_idx, key, attrs) in edges {
            let Some(source) = node_labels.get(source_idx) else {
                continue;
            };
            let Some(target) = node_labels.get(target_idx) else {
                continue;
            };
            let edge_key = DirectedEdgeKey::new(source, target);
            let bucket = self.edges.entry(edge_key).or_default();
            if !bucket.contains_key(&key) {
                self.edge_count += 1;
                inserted += 1;
            }
            let edge_attrs = bucket.entry(key).or_default();
            if !attrs.is_empty()
                && attrs
                    .iter()
                    .any(|(attr_key, value)| edge_attrs.get(attr_key) != Some(value))
            {
                merged_changed = true;
            }
            edge_attrs.extend(attrs);
            self.successors
                .get_mut(source.as_str())
                .expect("fresh node row exists")
                .entry(target.clone())
                .or_default()
                .insert(key);
            self.predecessors
                .get_mut(target.as_str())
                .expect("fresh node row exists")
                .entry(source.clone())
                .or_default()
                .insert(key);
        }

        if node_count > 0 || inserted > 0 || merged_changed {
            self.revision = self.revision.saturating_add(
                u64::try_from(node_count.saturating_add(inserted)).unwrap_or(u64::MAX),
            );
            self.record_decision(
                "extend_fresh_index_keyed_edges_with_attrs_unrecorded",
                0.0,
                false,
                vec![
                    EvidenceTerm {
                        signal: "batch_node_count".into(),
                        observed_value: count_evidence(node_count),
                        log_likelihood_ratio: -1.0,
                    },
                    EvidenceTerm {
                        signal: "batch_edge_count".into(),
                        observed_value: count_evidence(inserted),
                        log_likelihood_ratio: -1.0,
                    },
                ],
            );
        }

        inserted
    }

    pub fn add_edge_with_attrs(
        &mut self,
        source: impl Into<String>,
        target: impl Into<String>,
        attrs: AttrMap,
    ) -> Result<usize, GraphError> {
        self.add_edge_impl(source, target, None, attrs)
    }

    pub fn add_edge_with_key_and_attrs(
        &mut self,
        source: impl Into<String>,
        target: impl Into<String>,
        key: usize,
        attrs: AttrMap,
    ) -> Result<usize, GraphError> {
        self.add_edge_impl(source, target, Some(key), attrs)
    }

    fn add_edge_impl(
        &mut self,
        source: impl Into<String>,
        target: impl Into<String>,
        explicit_key: Option<usize>,
        attrs: AttrMap,
    ) -> Result<usize, GraphError> {
        let source = source.into();
        let target = target.into();
        // br-r37-c1-s8dj1: the revision AT ENTRY, before the node inserts below.
        // Those call `add_node_with_attrs_unrecorded`, which bumps the revision
        // too -- capturing after them left the pair index stale-looking and the
        // patch never fired, so every add still forced an O(E) rebuild.
        // Everything this function does to either index space is an APPEND, so a
        // map valid at entry is still valid to patch here.
        let previous_revision = self.revision;

        let unknown_feature = attrs
            .keys()
            .any(|key| key.starts_with("__fnx_incompatible"));
        let self_loop = source == target;
        let incompatibility_probability = if unknown_feature {
            1.0
        } else if self_loop {
            0.22
        } else {
            0.08
        };

        // br-r37-c1-eo88t: the gate decision earns a ledger record only when it
        // REFUSES. A refusal returns from here, so the outcome record at the end
        // of this function never runs and the gate record is the whole audit
        // trail for the rejected mutation. On the allow path the outcome record
        // carries the same operation/mode/action/incompatibility_probability
        // plus strictly richer evidence, so recording the gate as well wrote two
        // near-identical records for every edge on the hottest mutation path.
        let action = self
            .runtime_policy
            .action_for(incompatibility_probability, unknown_feature);

        if action == DecisionAction::FailClosed || action == DecisionAction::FullValidate {
            self.record_decision(
                "add_edge",
                incompatibility_probability,
                unknown_feature,
                vec![EvidenceTerm {
                    signal: "unknown_incompatible_feature".into(),
                    observed_value: bool_evidence(unknown_feature),
                    log_likelihood_ratio: 12.0,
                }],
            );
            return Err(GraphError::FailClosed {
                operation: "add_edge",
                reason: "incompatible edge metadata".to_owned(),
            });
        }

        // br-r37-c1-wa1b9: internal autocreation files no record of its own —
        // see the undirected `Graph` sibling for the ruling.
        let mut source_autocreated = false;
        if !self.nodes.contains_key(&source) {
            let _ = self.add_node_with_attrs_unrecorded(source.clone(), AttrMap::new());
            source_autocreated = true;
        }
        let mut target_autocreated = false;
        if self_loop {
            target_autocreated = source_autocreated;
        } else if !self.nodes.contains_key(&target) {
            let _ = self.add_node_with_attrs_unrecorded(target.clone(), AttrMap::new());
            target_autocreated = true;
        }

        let edge_key = DirectedEdgeKey::new(&source, &target);
        let key = explicit_key.unwrap_or_else(|| {
            let edge_bucket = self.edges.get(&edge_key);
            let mut k = edge_bucket.map_or(0, |b| b.len());
            if let Some(b) = edge_bucket {
                while b.contains_key(&k) {
                    k += 1;
                }
            }
            k
        });
        // br-r37-c1-s8dj1: captured BEFORE the insert, so the pair-index patch
        // below can tell a new bucket from an added parallel key without paying
        // a second hash of the pair.
        let edges_len_before = self.edges.len();
        let mut changed;
        let edge_attr_count = {
            let edge_bucket = self.edges.entry(edge_key.clone()).or_default();
            let is_new = !edge_bucket.contains_key(&key);
            if is_new {
                self.edge_count += 1;
            }
            changed = is_new;
            let edge_attrs = edge_bucket.entry(key).or_default();
            if edge_attrs.is_empty() {
                // br-r37-c1-jc9e4: a fresh keyed bucket TAKES the caller's map
                // whole. Same lever and same byte-identity argument as the
                // `Graph` sibling; `attrs` here is the function's own owned
                // parameter, so the move is unambiguous.
                if !attrs.is_empty() {
                    changed = true;
                }
                *edge_attrs = attrs;
            } else {
                if !attrs.is_empty()
                    && attrs
                        .iter()
                        .any(|(attr_key, value)| edge_attrs.get(attr_key) != Some(value))
                {
                    changed = true;
                }
                edge_attrs.extend(attrs);
            }
            edge_bucket.len()
        };

        self.successors
            .entry(source.clone())
            .or_default()
            .entry(target.clone())
            .or_default()
            .insert(key);
        self.predecessors
            .entry(target.clone())
            .or_default()
            .entry(source.clone())
            .or_default()
            .insert(key);
        if changed {
            self.revision = self.revision.saturating_add(1);
        }
        // br-r37-c1-s8dj1: `insert` APPENDS, so a bucket created by this call
        // sits at the LAST position and no existing position moved.
        let new_pair_position = if self.edges.len() > edges_len_before {
            Some(self.edges.len() - 1)
        } else {
            None
        };
        self.pair_index_note_edge_added(previous_revision, &source, &target, new_pair_position);

        self.record_decision(
            "add_edge",
            incompatibility_probability,
            unknown_feature,
            vec![
                EvidenceTerm {
                    signal: "edge_key".into(),
                    observed_value: key.to_string().into(),
                    log_likelihood_ratio: -2.0,
                },
                EvidenceTerm {
                    signal: "edge_attr_count".into(),
                    observed_value: count_evidence(edge_attr_count),
                    log_likelihood_ratio: -2.0,
                },
                EvidenceTerm {
                    signal: "source_autocreated".into(),
                    observed_value: bool_evidence(source_autocreated),
                    log_likelihood_ratio: -1.25,
                },
                EvidenceTerm {
                    signal: "target_autocreated".into(),
                    observed_value: bool_evidence(target_autocreated),
                    log_likelihood_ratio: -1.25,
                },
            ],
        );

        Ok(key)
    }

    /// br-r37-c1-sjf4t: overwrite the attribute map for an existing
    /// (source, target, key) edge. Returns whether the edge existed.
    pub fn replace_edge_attrs(
        &mut self,
        source: &str,
        target: &str,
        key: usize,
        attrs: AttrMap,
    ) -> bool {
        let edge_key = DirectedEdgeKey::new(source, target);
        if let Some(bucket) = self.edges.get_mut(&edge_key)
            && let Some(slot) = bucket.get_mut(&key)
        {
            if *slot != attrs {
                *slot = attrs;
                self.revision = self.revision.saturating_add(1);
            }
            return true;
        }
        false
    }

    /// Overwrite one attribute value on an existing keyed edge. This is the
    /// hot path for algorithms that clone graph topology and patch one derived
    /// edge field without rebuilding every edge's full attribute map.
    pub fn set_edge_attr_value(
        &mut self,
        source: &str,
        target: &str,
        key: usize,
        attr_key: &str,
        value: CgseValue,
    ) -> bool {
        let edge_key = DirectedEdgeKey::new(source, target);
        if let Some(bucket) = self.edges.get_mut(&edge_key)
            && let Some(attrs) = bucket.get_mut(&key)
        {
            if attrs.get(attr_key) != Some(&value) {
                attrs.insert(attr_key.to_owned(), value);
                self.revision = self.revision.saturating_add(1);
            }
            return true;
        }
        false
    }

    /// Overwrite one attribute on all keyed edges in `edges_ordered_borrowed`
    /// order. Returns false if the caller supplied the wrong value count.
    pub fn set_ordered_edge_attr_values<I>(&mut self, attr_key: &str, values: I) -> bool
    where
        I: IntoIterator<Item = CgseValue>,
    {
        let mut values = values.into_iter();
        let mut changed = false;
        let mut visited = 0usize;

        for node in self.nodes.keys() {
            if let Some(neighbors) = self.successors.get(node) {
                for target in neighbors.keys() {
                    let pair = DirectedEdgeKeyRef::new(node, target);
                    if let Some(edge_bucket) = self.edges.get_mut(&pair) {
                        for attrs in edge_bucket.values_mut() {
                            let Some(value) = values.next() else {
                                return false;
                            };
                            if attrs.get(attr_key) != Some(&value) {
                                attrs.insert(attr_key.to_owned(), value);
                                changed = true;
                            }
                            visited = visited.saturating_add(1);
                        }
                    }
                }
            }
        }

        if values.next().is_some() || visited != self.edge_count {
            return false;
        }
        if changed {
            self.revision = self.revision.saturating_add(1);
        }
        true
    }

    /// br-r37-c1-sjf4t: overwrite the attribute map for an existing node.
    pub fn replace_node_attrs(&mut self, node: &str, attrs: AttrMap) -> bool {
        if let Some(slot) = self.nodes.get_mut(node) {
            if *slot != attrs {
                *slot = attrs;
                self.revision = self.revision.saturating_add(1);
            }
            true
        } else {
            false
        }
    }

    pub fn remove_edge(&mut self, source: &str, target: &str, key: Option<usize>) -> bool {
        // br-r37-c1-s8dj1: revision at entry, for the pair-index patch below.
        let previous_revision = self.revision;
        let edge_key = DirectedEdgeKeyRef::new(source, target);
        let removal_key = key.or_else(|| {
            self.edges
                .get(&edge_key)
                .and_then(|edge_bucket| edge_bucket.keys().next_back().copied())
        });

        let Some(removal_key) = removal_key else {
            return false;
        };

        let removed = self
            .edges
            .get_mut(&edge_key)
            .is_some_and(|edge_bucket| edge_bucket.shift_remove(&removal_key).is_some());
        if !removed {
            return false;
        }

        self.edge_count -= 1;

        let should_drop_bucket = self.edges.get(&edge_key).is_some_and(IndexMap::is_empty);
        // br-r37-c1-s8dj1: the position the bucket occupies BEFORE swap_remove
        // frees it. Resolved only when a live pair index will actually use it.
        let dropped_position =
            if should_drop_bucket && self.edge_pair_index_is_live(previous_revision) {
                self.edges.get_index_of(&edge_key)
            } else {
                None
            };
        if should_drop_bucket {
            // br-r37-c1-vbwpl (cc): swap_remove (O(1)) on the OUTER node-pair map
            // — its storage order is never observed (edges_ordered walks
            // successors and looks pairs up by key; remove_node already
            // swap_removes this same map). The inner bucket above keeps
            // shift_remove because per-pair key order IS observed. Turns
            // remove_edges_from from O(k*pairs) into O(k).
            self.edges.swap_remove(&edge_key);
        }

        self.remove_successor_key(source, target, removal_key);
        self.remove_predecessor_key(target, source, removal_key);
        self.revision = self.revision.saturating_add(1);
        self.pair_index_note_edge_removed(previous_revision, source, target, dropped_position);
        true
    }

    pub fn clear_edges(&mut self) {
        if self.edge_count == 0 {
            return;
        }

        self.edges.clear();
        for row in self.successors.values_mut() {
            row.clear();
        }
        for row in self.predecessors.values_mut() {
            row.clear();
        }
        self.edge_count = 0;
        self.revision = self.revision.saturating_add(1);
    }

    pub fn remove_node(&mut self, node: &str) -> bool {
        if !self.nodes.contains_key(node) {
            return false;
        }

        // br-r37-c1-p6bxu: drop each incident edge bucket with O(1) `swap_remove`
        // (the `edges` IndexMap order is never observed externally — every public
        // consumer reads via `edges_ordered`, which walks node->successor order;
        // no internal consumer iterates the map order). Out-edges are known from
        // `successors`, in-edges from `predecessors`; a self-loop's (node,node)
        // bucket appears in both but `swap_remove` returns it only once, so it is
        // counted exactly once. Removal is O(degree) instead of the
        // O(|distinct pairs|) `retain` scan — matching nx.
        //
        // yr2oc.1: the node's own rows are taken out first (O(1) each, the row
        // maps are unordered), so the walk borrows the neighbour names from
        // them instead of cloning every one, and the node table tombstones
        // instead of shifting.
        let mut removed_count = 0usize;
        let succs = self.successors.remove(node).unwrap_or_default();
        let preds = self.predecessors.remove(node).unwrap_or_default();
        for target in succs.keys() {
            if target != node
                && let Some(row) = self.predecessors.get_mut(target.as_str())
            {
                row.shift_remove(node);
            }
            if let Some(bucket) = self
                .edges
                .swap_remove(&DirectedEdgeKeyRef::new(node, target))
            {
                removed_count += bucket.len();
            }
        }
        for source in preds.keys() {
            if source != node
                && let Some(row) = self.successors.get_mut(source.as_str())
            {
                row.shift_remove(node);
            }
            if let Some(bucket) = self
                .edges
                .swap_remove(&DirectedEdgeKeyRef::new(source, node))
            {
                removed_count += bucket.len();
            }
        }
        self.edge_count -= removed_count;
        self.nodes.remove(node);
        self.revision = self.revision.saturating_add(1);
        true
    }

    /// br-r37-c1-mgrnf: batch node removal. A small batch walks only the removed
    /// nodes' rows, O(sum of their degrees); a large one prunes every row in one
    /// O(|V|+|E|) pass. Returns `(removed_node_count,
    /// removed_edge_instance_count)`. The `revision` bump invalidates
    /// `csr_cache` exactly as repeated `remove_node` would.
    pub fn remove_nodes_from<'a, I>(&mut self, nodes: I) -> (usize, usize)
    where
        I: IntoIterator<Item = &'a str>,
    {
        let remove_set: rustc_hash::FxHashSet<&str> = nodes
            .into_iter()
            .filter(|node| self.nodes.contains_key(node))
            .collect();
        if remove_set.is_empty() {
            return (0, 0);
        }

        let old_node_count = self.nodes.len();
        let old_edge_count = self.edge_count;

        // br-r37-c1-mgrnf-incident: SMALL-fraction fast path — walk only the removed
        // nodes' out/in adjacency (like `remove_node`), pruning the removed node from
        // each surviving neighbour's opposite row and dropping incident edge buckets
        // O(1), then drop each removed node's rows and table entry in O(1)
        // (yr2oc.1). O(sum_removed_degrees) vs the whole-graph O(|V|+|E|) retain
        // (removing 10 nodes from a 2000/10000 graph was ~100x nx). Large removals
        // fall through (repeated hub shift_remove).
        let mut removed_instances = 0usize;
        if remove_set.len().saturating_mul(4) <= old_node_count {
            for &rn in &remove_set {
                // Out-edges: rn -> target. Drop rn from a surviving target's preds.
                let targets: Vec<String> = match self.successors.get(rn) {
                    Some(row) => row.keys().cloned().collect(),
                    None => Vec::new(),
                };
                for t in &targets {
                    if t.as_str() != rn
                        && !remove_set.contains(t.as_str())
                        && let Some(preds) = self.predecessors.get_mut(t)
                    {
                        preds.shift_remove(rn);
                    }
                    if let Some(bucket) = self.edges.swap_remove(&DirectedEdgeKeyRef::new(rn, t)) {
                        removed_instances += bucket.len();
                    }
                }
                // In-edges: source -> rn. Drop rn from a surviving source's succs.
                let sources: Vec<String> = match self.predecessors.get(rn) {
                    Some(row) => row.keys().cloned().collect(),
                    None => Vec::new(),
                };
                for s in &sources {
                    if s.as_str() != rn
                        && !remove_set.contains(s.as_str())
                        && let Some(succs) = self.successors.get_mut(s)
                    {
                        succs.shift_remove(rn);
                    }
                    // Self-loop (rn,rn) already dropped in the out-edges pass, so
                    // swap_remove returns None here — no double count.
                    if let Some(bucket) = self.edges.swap_remove(&DirectedEdgeKeyRef::new(s, rn)) {
                        removed_instances += bucket.len();
                    }
                }
            }
            self.edge_count -= removed_instances;
            for &rn in &remove_set {
                self.successors.remove(rn);
                self.predecessors.remove(rn);
                self.nodes.remove(rn);
            }
            let removed_nodes = old_node_count - self.nodes.len();
            let removed_edges = old_edge_count - self.edge_count;
            self.revision = self
                .revision
                .saturating_add(u64::try_from(removed_nodes).unwrap_or(u64::MAX));
            return (removed_nodes, removed_edges);
        }

        // Drop every edge bucket incident to a removed node (source OR target) in
        // one pass, tallying parallel-edge instances so `edge_count` stays exact.
        // `edges` map order is unobserved externally (consumers walk `edges_ordered`
        // = successor order), so `retain` is order-safe — same rationale as
        // `remove_node`.
        self.edges.retain(|key, bucket| {
            let keep = !remove_set.contains(key.source.as_str())
                && !remove_set.contains(key.target.as_str());
            if !keep {
                removed_instances += bucket.len();
            }
            keep
        });
        self.edge_count -= removed_instances;

        // successors + predecessors: drop the removed nodes' own rows, then prune
        // references to removed nodes from every surviving row. Order-preserving
        // `retain` within each row, so byte-identical to repeated `remove_node`.
        self.successors
            .retain(|node, _| !remove_set.contains(node.as_str()));
        for row in self.successors.values_mut() {
            row.retain(|target, _| !remove_set.contains(target.as_str()));
        }
        self.predecessors
            .retain(|node, _| !remove_set.contains(node.as_str()));
        for row in self.predecessors.values_mut() {
            row.retain(|source, _| !remove_set.contains(source.as_str()));
        }
        for &rn in &remove_set {
            self.nodes.remove(rn);
        }

        let removed_nodes = old_node_count - self.nodes.len();
        let removed_edges = old_edge_count - self.edge_count;
        self.revision = self
            .revision
            .saturating_add(u64::try_from(removed_nodes).unwrap_or(u64::MAX));
        (removed_nodes, removed_edges)
    }

    #[must_use]
    pub fn edges_ordered(&self) -> Vec<MultiDiEdgeSnapshot> {
        let mut ordered = Vec::with_capacity(self.edge_count());

        for node in self.nodes.keys() {
            if let Some(neighbors) = self.successors.get(node) {
                for target in neighbors.keys() {
                    let pair = DirectedEdgeKey::new(node, target);
                    if let Some(edge_bucket) = self.edges.get(&pair) {
                        for (key, attrs) in edge_bucket {
                            ordered.push(MultiDiEdgeSnapshot {
                                source: node.clone(),
                                target: target.clone(),
                                key: *key,
                                attrs: attrs.clone(),
                            });
                        }
                    }
                }
            }
        }

        ordered
    }

    #[must_use]
    pub fn edges_ordered_borrowed(&self) -> Vec<(&str, &str, usize, &AttrMap)> {
        let mut ordered = Vec::with_capacity(self.edge_count());

        for node in self.nodes.keys() {
            if let Some(neighbors) = self.successors.get(node) {
                for target in neighbors.keys() {
                    let pair = DirectedEdgeKeyRef::new(node, target);
                    if let Some(edge_bucket) = self.edges.get(&pair) {
                        for (key, attrs) in edge_bucket {
                            ordered.push((node.as_str(), target.as_str(), *key, attrs));
                        }
                    }
                }
            }
        }

        ordered
    }

    /// br-r37-c1-wsize (cc): integer `size(weight)` from the store — directed
    /// multigraph twin of `Graph::weighted_size_int`. Each parallel directed edge
    /// is one bucket entry contributing its weight once (size halves degree's
    /// in+out double count). Returns `None` on any non-integer weight (caller uses
    /// the exact degree path); missing weight defaults to nx's int `1`.
    #[must_use]
    pub fn weighted_size_int(&self, weight: &str) -> Option<i128> {
        let mut total: i128 = 0;
        for bucket in self.edges.values() {
            for attrs in bucket.values() {
                let value = match attrs.get(weight) {
                    Some(CgseValue::Int(v)) => i128::from(*v),
                    Some(_) => return None,
                    None => 1,
                };
                total = total.checked_add(value)?;
            }
        }
        Some(total)
    }

    pub fn try_for_each_edge_ordered_borrowed<E>(
        &self,
        mut visit: impl FnMut(&str, &str, usize, &AttrMap) -> Result<(), E>,
    ) -> Result<(), E> {
        for node in self.nodes.keys() {
            if let Some(neighbors) = self.successors.get(node) {
                for target in neighbors.keys() {
                    let pair = DirectedEdgeKeyRef::new(node, target);
                    if let Some(edge_bucket) = self.edges.get(&pair) {
                        for (key, attrs) in edge_bucket {
                            visit(node.as_str(), target.as_str(), *key, attrs)?;
                        }
                    }
                }
            }
        }

        Ok(())
    }

    pub fn try_for_each_indexed_edge_ordered_borrowed<E>(
        &self,
        mut visit: impl FnMut(usize, usize, &str, &str, usize, &AttrMap) -> Result<(), E>,
    ) -> Result<(), E> {
        let positions = self.nodes.positions();
        for (source_index, node) in self.nodes.keys().enumerate() {
            if let Some(neighbors) = self.successors.get(node) {
                for target in neighbors.keys() {
                    let Some(target_index) = positions.of(target.as_str()) else {
                        continue;
                    };
                    let pair = DirectedEdgeKeyRef::new(node, target);
                    if let Some(edge_bucket) = self.edges.get(&pair) {
                        for (key, attrs) in edge_bucket {
                            visit(
                                source_index,
                                target_index,
                                node.as_str(),
                                target.as_str(),
                                *key,
                                attrs,
                            )?;
                        }
                    }
                }
            }
        }

        Ok(())
    }

    /// Return a reversed copy with NetworkX-compatible node/edge row order.
    ///
    /// This is the directed-multigraph sibling of `DiGraph::reversed`, but it
    /// cannot simply swap `successors` and `predecessors`: NetworkX materializes
    /// `MultiDiGraph.reverse(copy=True)` by walking `G.edges(keys=True, data=True)`
    /// in source-node order and adding `(v, u, key)` to the result. Predecessor
    /// rows may have been formed by historical edge insertion order, which is not
    /// always the same as that source-major edge stream. Rebuild the two row maps
    /// from the observed edge stream while avoiding the expensive per-edge
    /// `add_edge` policy path.
    #[must_use]
    pub fn reversed(&self) -> Self {
        let mut successors =
            MultiDiRows::with_capacity_and_hasher(self.nodes.len(), rustc_hash::FxBuildHasher);
        let mut predecessors =
            MultiDiRows::with_capacity_and_hasher(self.nodes.len(), rustc_hash::FxBuildHasher);
        for node in self.nodes.keys() {
            successors.insert(node.clone(), crate::FxIndexMap::default());
            predecessors.insert(node.clone(), crate::FxIndexMap::default());
        }

        let mut edges: crate::FxIndexMap<DirectedEdgeKey, crate::FxIndexMap<usize, AttrMap>> =
            crate::FxIndexMap::with_capacity_and_hasher(
                self.edges.len(),
                rustc_hash::FxBuildHasher,
            );
        let mut edge_count = 0usize;
        for source in self.nodes.keys() {
            let Some(neighbors) = self.successors.get(source) else {
                continue;
            };
            for target in neighbors.keys() {
                let pair = DirectedEdgeKeyRef::new(source, target);
                let Some(edge_bucket) = self.edges.get(&pair) else {
                    continue;
                };
                let reversed_pair = DirectedEdgeKey::new(target, source);
                let reversed_bucket = edges.entry(reversed_pair).or_default();
                let reversed_succ_keys = successors
                    .get_mut(target.as_str())
                    .expect("target row exists")
                    .entry(source.clone())
                    .or_default();
                let reversed_pred_keys = predecessors
                    .get_mut(source.as_str())
                    .expect("source row exists")
                    .entry(target.clone())
                    .or_default();
                for (key, attrs) in edge_bucket {
                    reversed_bucket.insert(*key, attrs.clone());
                    reversed_succ_keys.insert(*key);
                    reversed_pred_keys.insert(*key);
                    edge_count += 1;
                }
            }
        }

        Self {
            mode: self.mode,
            revision: 0,
            nodes: self.nodes.clone(),
            successors,
            predecessors,
            edges,
            runtime_policy: self.runtime_policy.clone(),
            edge_count,
            csr_cache: std::sync::Arc::default(),
            edge_pair_index: EdgePairIndex::default(),
        }
    }

    /// Out-edges of a single source node in nx adjacency order
    /// (successors[node].keys() then key-bucket order), borrowing attrs.
    ///
    /// Equivalent to ``edges_ordered_borrowed()`` filtered to one source,
    /// but O(out-deg(node)) instead of O(E) — and it clones nothing. Used
    /// by ``MultiDiGraph.edges(nbunch)`` to avoid scanning/cloning every
    /// edge when only a handful of sources are requested.
    #[must_use]
    pub fn out_edges_ordered_borrowed<'a>(
        &'a self,
        node: &str,
    ) -> Vec<(&'a str, &'a str, usize, &'a AttrMap)> {
        let mut ordered = Vec::new();
        if let Some((src, neighbors)) = self.successors.get_key_value(node) {
            for target in neighbors.keys() {
                let pair = DirectedEdgeKeyRef::new(src, target);
                if let Some(edge_bucket) = self.edges.get(&pair) {
                    for (key, attrs) in edge_bucket {
                        ordered.push((src.as_str(), target.as_str(), *key, attrs));
                    }
                }
            }
        }
        ordered
    }

    #[must_use]
    pub fn snapshot(&self) -> MultiDiGraphSnapshot {
        let node_attrs: std::collections::BTreeMap<String, AttrMap> = self
            .nodes
            .iter()
            .filter(|(_, attrs)| !attrs.is_empty())
            .map(|(name, attrs)| (name.clone(), attrs.clone()))
            .collect();
        MultiDiGraphSnapshot {
            mode: self.mode,
            nodes: self.nodes.keys().cloned().collect(),
            node_attrs,
            edges: self.edges_ordered(),
        }
    }

    fn remove_successor_key(&mut self, source: &str, target: &str, key: usize) {
        let mut drop_neighbor = false;
        if let Some(neighbors) = self.successors.get_mut(source)
            && let Some(keys) = neighbors.get_mut(target)
        {
            keys.shift_remove(&key);
            drop_neighbor = keys.is_empty();
        }
        if drop_neighbor && let Some(neighbors) = self.successors.get_mut(source) {
            neighbors.shift_remove(target);
        }
    }

    fn remove_predecessor_key(&mut self, target: &str, source: &str, key: usize) {
        let mut drop_neighbor = false;
        if let Some(neighbors) = self.predecessors.get_mut(target)
            && let Some(keys) = neighbors.get_mut(source)
        {
            keys.shift_remove(&key);
            drop_neighbor = keys.is_empty();
        }
        if drop_neighbor && let Some(neighbors) = self.predecessors.get_mut(target) {
            neighbors.shift_remove(source);
        }
    }

    fn record_decision(
        &mut self,
        operation: &'static str,
        incompatibility_probability: f64,
        unknown_incompatible_feature: bool,
        evidence: Vec<EvidenceTerm>,
    ) -> DecisionAction {
        let action = self
            .runtime_policy
            .action_for(incompatibility_probability, unknown_incompatible_feature);
        self.runtime_policy.record(
            operation,
            action,
            incompatibility_probability,
            "argmin expected loss over {allow,full_validate,fail_closed}",
            evidence,
        );
        action
    }
}

// ===========================================================================
// Tests
// ===========================================================================

#[cfg(test)]
mod tests {
    use super::*;
    use fnx_runtime::{
        CgseValue, CompatibilityMode, DecisionAction, DecisionRecord, RuntimePolicy,
    };
    use proptest::prelude::*;

    fn node_name(id: u8) -> String {
        format!("n{}", id % 8)
    }

    fn single_attr(key: &str, value: &str) -> AttrMap {
        let mut attrs = AttrMap::new();
        attrs.insert(key.to_owned(), CgseValue::from(value));
        attrs
    }

    fn assert_runtime_policy_preserved(source: &RuntimePolicy, result: &RuntimePolicy) {
        assert_eq!(result.mode(), source.mode());
        assert_eq!(result.allowlist(), source.allowlist());
        assert_eq!(result.loss_matrix(), source.loss_matrix());
        assert!(result.posterior().observation_count >= source.posterior().observation_count);
        assert!(
            result
                .decision_log()
                .records()
                .starts_with(source.decision_log().records())
        );
    }

    // -- Invariant checker --------------------------------------------------

    fn assert_digraph_core_invariants(g: &DiGraph) {
        // Every edge in the edge map must be reflected in successors/predecessors.
        for (&(source_idx, target_idx), _attrs) in &g.edges {
            let source = g
                .node_order
                .name(source_idx)
                .expect("edge source index should be a node");
            let target = g
                .node_order
                .name(target_idx)
                .expect("edge target index should be a node");
            assert!(
                g.has_node(source),
                "edge source {} should be a node",
                source
            );
            assert!(
                g.has_node(target),
                "edge target {} should be a node",
                target
            );
            let succs = g
                .successors(source)
                .expect("source should have successors bucket");
            assert!(
                succs.contains(&target),
                "{} should be in successors of {}",
                target,
                source
            );
            let preds = g
                .predecessors(target)
                .expect("target should have predecessors bucket");
            assert!(
                preds.contains(&source),
                "{} should be in predecessors of {}",
                source,
                target
            );
        }

        // Every successor entry should have a corresponding edge.
        let mut edge_count_from_adj = 0usize;
        for node in g.nodes_ordered() {
            let succs = g
                .successors(node)
                .expect("node should have successors bucket");
            for s in &succs {
                assert!(
                    g.has_edge(node, s),
                    "successor {} of {} should have directed edge",
                    s,
                    node
                );
                edge_count_from_adj += 1;
            }
            // Every predecessor entry should have a corresponding edge.
            let preds = g
                .predecessors(node)
                .expect("node should have predecessors bucket");
            for p in &preds {
                assert!(
                    g.has_edge(p, node),
                    "predecessor {} of {} should have directed edge",
                    p,
                    node
                );
            }
        }
        assert_eq!(g.edge_count(), edge_count_from_adj);
    }

    #[test]
    fn reorder_rows_like_restores_pred_rows_and_swaps_for_a_reverse() {
        // source: succ x [b, a]; pred x [a, c, b]
        let mut source = DiGraph::strict();
        for (u, v) in [("a", "x"), ("c", "x"), ("b", "x"), ("x", "b"), ("x", "a")] {
            let _ = source.add_edge(u, v);
        }
        // Copy-shaped rebuild over [x, a, b, c]: succ rows right, pred x
        // filled in walk order [a, b, c].
        let mut copy = DiGraph::strict();
        for node in ["x", "a", "b", "c"] {
            let _ = copy.add_node(node);
        }
        for (u, v) in [("x", "b"), ("x", "a"), ("a", "x"), ("b", "x"), ("c", "x")] {
            let _ = copy.add_edge(u, v);
        }
        assert_eq!(copy.predecessors("x"), Some(vec!["a", "b", "c"]));
        let _ = copy.csr();
        let before = copy.revision();
        copy.reorder_rows_like(&source, false);
        assert_eq!(copy.predecessors("x"), Some(vec!["a", "c", "b"]));
        assert_eq!(copy.successors("x"), Some(vec!["b", "a"]));
        assert!(copy.revision() > before);
        // The CSR is rebuilt from the reordered rows, not served stale.
        let csr = copy.csr();
        let x = copy.get_node_index("x").unwrap();
        let names: Vec<&str> = csr
            .predecessors(x)
            .iter()
            .map(|&i| copy.get_node_name(i as usize).unwrap())
            .collect();
        assert_eq!(names, vec!["a", "c", "b"]);

        // The reverse view's copy: succ x = source pred x [a, c, b], pred x
        // filled in walk order [a, b]; its pred row is source's SUCC row.
        let mut reverse = DiGraph::strict();
        for node in ["x", "a", "b", "c"] {
            let _ = reverse.add_node(node);
        }
        for (u, v) in [("x", "a"), ("x", "c"), ("x", "b"), ("a", "x"), ("b", "x")] {
            let _ = reverse.add_edge(u, v);
        }
        assert_eq!(reverse.predecessors("x"), Some(vec!["a", "b"]));
        reverse.reorder_rows_like(&source, true);
        assert_eq!(reverse.predecessors("x"), Some(vec!["b", "a"]));
        assert_eq!(reverse.successors("x"), Some(vec!["a", "c", "b"]));
    }

    #[test]
    fn set_row_orders_sets_both_families_or_neither() {
        let mut g = DiGraph::strict();
        for (u, v) in [("a", "b"), ("a", "c"), ("b", "a"), ("c", "a")] {
            let _ = g.add_edge(u, v);
        }
        // positions: a 0, b 1, c 2; succ a [b, c], pred a [b, c]
        let _ = g.csr();
        assert!(g.set_row_orders(
            &[vec![2, 1], vec![0], vec![0]],
            &[vec![2, 1], vec![0], vec![0]]
        ));
        assert_eq!(g.successors("a"), Some(vec!["c", "b"]));
        assert_eq!(g.predecessors("a"), Some(vec!["c", "b"]));
        let a = g.get_node_index("a").unwrap();
        assert_eq!(g.csr().successors(a), &[2, 1]);
        // A valid succ family with a bad pred family changes neither.
        assert!(!g.set_row_orders(
            &[vec![1, 2], vec![0], vec![0]],
            &[vec![1, 1], vec![0], vec![0]]
        ));
        assert_eq!(g.successors("a"), Some(vec!["c", "b"]));
        assert_eq!(g.predecessors("a"), Some(vec!["c", "b"]));
    }

    #[test]
    fn multidigraph_reorder_rows_like_restores_pred_rows_and_swaps_for_a_reverse() {
        // br-r37-c1-36v6r: source succ x [b, a]; pred x [a, c, b]; the arc
        // c -> x has two keys.
        let mut source = MultiDiGraph::strict();
        for (u, v) in [
            ("a", "x"),
            ("c", "x"),
            ("b", "x"),
            ("x", "b"),
            ("x", "a"),
            ("c", "x"),
        ] {
            let _ = source.add_edge(u, v);
        }
        // Copy-shaped rebuild over [x, a, b, c]: succ rows right, pred x
        // filled in walk order [a, b, c].
        let mut copy = MultiDiGraph::strict();
        for node in ["x", "a", "b", "c"] {
            let _ = copy.add_node(node);
        }
        for (u, v) in [
            ("x", "b"),
            ("x", "a"),
            ("a", "x"),
            ("b", "x"),
            ("c", "x"),
            ("c", "x"),
        ] {
            let _ = copy.add_edge(u, v);
        }
        assert_eq!(copy.predecessors("x"), Some(vec!["a", "b", "c"]));
        let _ = copy.csr();
        let before = copy.revision();
        copy.reorder_rows_like(&source, false);
        assert_eq!(copy.predecessors("x"), Some(vec!["a", "c", "b"]));
        assert_eq!(copy.successors("x"), Some(vec!["b", "a"]));
        assert_eq!(copy.edge_keys("c", "x"), Some(vec![0, 1]));
        assert!(copy.revision() > before);
        // The CSR is rebuilt from the reordered rows, not served stale.
        let x = copy.get_node_index("x").unwrap();
        let csr = copy.csr();
        let names: Vec<&str> = csr
            .predecessors(x)
            .iter()
            .map(|&i| copy.get_node_name(i as usize).unwrap())
            .collect();
        assert_eq!(names, vec!["a", "c", "b"]);
        let after = copy.revision();
        copy.reorder_rows_like(&source, false);
        assert_eq!(copy.revision(), after, "an ordered graph is left alone");

        // The reverse view's copy: its succ x is source's pred x [a, c, b]
        // and its pred x source's succ x [b, a].
        let mut reverse = MultiDiGraph::strict();
        for node in ["x", "a", "b", "c"] {
            let _ = reverse.add_node(node);
        }
        for (u, v) in [("x", "a"), ("x", "b"), ("x", "c"), ("a", "x"), ("b", "x")] {
            let _ = reverse.add_edge(u, v);
        }
        reverse.reorder_rows_like(&source, true);
        assert_eq!(reverse.successors("x"), Some(vec!["a", "c", "b"]));
        assert_eq!(reverse.predecessors("x"), Some(vec!["b", "a"]));
    }

    #[test]
    fn multidigraph_set_row_orders_orders_named_rows_and_moves_the_revision() {
        let mut g = MultiDiGraph::strict();
        for (u, v) in [("a", "b"), ("a", "c"), ("a", "b"), ("b", "a"), ("c", "a")] {
            let _ = g.add_edge(u, v);
        }
        let _ = g.csr();
        let before = g.revision();
        let row = |node: &str, order: &[&str]| {
            (
                node.to_owned(),
                order.iter().map(|&n| n.to_owned()).collect::<Vec<_>>(),
            )
        };
        g.set_row_orders(&[row("a", &["c", "b"])], &[row("a", &["c", "b"])]);
        assert_eq!(g.successors("a"), Some(vec!["c", "b"]));
        assert_eq!(g.predecessors("a"), Some(vec!["c", "b"]));
        assert_eq!(g.edge_keys("a", "b"), Some(vec![0, 1]));
        assert!(g.revision() > before);
        let a = g.get_node_index("a").unwrap();
        let csr = g.csr();
        let names: Vec<&str> = csr
            .successors(a)
            .iter()
            .map(|&i| g.get_node_name(i as usize).unwrap())
            .collect();
        assert_eq!(names, vec!["c", "b"]);
    }

    #[test]
    fn add_node_with_attrs_reports_change_on_existing_node() {
        let mut graph = DiGraph::strict();
        assert!(graph.add_node_with_attrs("a", single_attr("color", "red")));
        let r1 = graph.revision();
        assert!(graph.add_node_with_attrs("a", single_attr("color", "blue")));
        let r2 = graph.revision();
        assert!(r2 > r1);
        let expected = CgseValue::from("blue");
        assert_eq!(graph.node_attrs("a").unwrap().get("color"), Some(&expected));
    }

    #[test]
    fn multidigraph_add_node_with_attrs_reports_change_on_existing_node() {
        let mut graph = MultiDiGraph::strict();
        assert!(graph.add_node_with_attrs("a", single_attr("color", "red")));
        let r1 = graph.revision();
        assert!(graph.add_node_with_attrs("a", single_attr("color", "blue")));
        let r2 = graph.revision();
        assert!(r2 > r1);
        let expected = CgseValue::from("blue");
        assert_eq!(graph.node_attrs("a").unwrap().get("color"), Some(&expected));
    }

    fn assert_multidigraph_core_invariants(g: &MultiDiGraph) {
        let mut edge_instances = std::collections::BTreeSet::new();
        for node in g.nodes_ordered() {
            let successors = g
                .successors(node)
                .expect("multidigraph nodes should always have successor buckets");
            for target in successors {
                assert!(g.has_node(target));
                assert!(g.has_edge(node, target));
                let preds = g
                    .predecessors(target)
                    .expect("target should always have predecessor buckets");
                assert!(preds.contains(&node));
                let keys = g
                    .edge_keys(node, target)
                    .expect("parallel directed edge bucket should exist");
                for key in keys {
                    edge_instances.insert((node.to_owned(), target.to_owned(), key));
                }
            }
        }
        assert_eq!(g.edge_count(), edge_instances.len());
    }

    fn assert_decision_record_schema(record: &DecisionRecord, expected_mode: CompatibilityMode) {
        assert!(record.ts_unix_ms > 0);
        assert!(!record.operation.trim().is_empty());
        assert_eq!(record.mode, expected_mode);
        assert!((0.0..=1.0).contains(&record.incompatibility_probability));
        assert!(!record.rationale.trim().is_empty());
        assert!(!record.evidence.is_empty());
        for term in &record.evidence {
            assert!(!term.signal.trim().is_empty());
            assert!(!term.observed_value.trim().is_empty());
        }
    }

    // -- Basic operations ---------------------------------------------------

    #[test]
    fn add_directed_edge_autocreates_nodes() {
        let mut g = DiGraph::strict();
        g.add_edge("a", "b").unwrap();

        assert_eq!(g.node_count(), 2);
        assert_eq!(g.edge_count(), 1);
        assert!(g.has_edge("a", "b"));
        assert!(!g.has_edge("b", "a")); // directed: reverse does NOT exist
        assert_digraph_core_invariants(&g);
    }

    #[test]
    fn directed_edge_asymmetry() {
        let mut g = DiGraph::strict();
        g.add_edge("a", "b").unwrap();
        g.add_edge("b", "a").unwrap();

        assert_eq!(g.edge_count(), 2);
        assert!(g.has_edge("a", "b"));
        assert!(g.has_edge("b", "a"));
        assert_digraph_core_invariants(&g);
    }

    #[test]
    fn successors_and_predecessors() {
        let mut g = DiGraph::strict();
        g.add_edge("a", "b").unwrap();
        g.add_edge("a", "c").unwrap();
        g.add_edge("d", "a").unwrap();

        assert_eq!(g.successors("a"), Some(vec!["b", "c"]));
        assert_eq!(g.predecessors("a"), Some(vec!["d"]));
        assert_eq!(g.out_degree("a"), 2);
        assert_eq!(g.in_degree("a"), 1);
        assert_eq!(g.degree("a"), 3);
        assert_digraph_core_invariants(&g);
    }

    #[test]
    fn degree_single_lookup_preserves_missing_and_self_loop_semantics() {
        let mut g = DiGraph::strict();
        g.add_edge("a", "a").expect("self-loop add should succeed");
        g.add_edge("a", "b").expect("edge add should succeed");
        g.add_edge("c", "a").expect("edge add should succeed");

        for node in ["a", "b", "c", "missing"] {
            assert_eq!(g.degree(node), g.in_degree(node) + g.out_degree(node));
        }
        assert_eq!(g.degree("a"), 4);
        assert_eq!(g.degree("missing"), 0);
    }

    #[test]
    fn neighbors_returns_successors() {
        let mut g = DiGraph::strict();
        g.add_edge("a", "b").unwrap();
        g.add_edge("c", "a").unwrap();

        // neighbors() = successors() per NetworkX convention
        assert_eq!(g.neighbors("a"), Some(vec!["b"]));
        assert_digraph_core_invariants(&g);
    }

    #[test]
    fn in_edges_and_out_edges() {
        let mut g = DiGraph::strict();
        g.add_edge("x", "y").unwrap();
        g.add_edge("z", "y").unwrap();
        g.add_edge("y", "w").unwrap();

        assert_eq!(g.out_edges("y"), vec![("y", "w")]);
        assert_eq!(g.in_edges("y"), vec![("x", "y"), ("z", "y")]);
        assert_digraph_core_invariants(&g);
    }

    #[test]
    fn remove_directed_edge() {
        let mut g = DiGraph::strict();
        g.add_edge("a", "b").unwrap();
        g.add_edge("b", "a").unwrap();

        assert!(g.remove_edge("a", "b"));
        assert!(!g.has_edge("a", "b"));
        assert!(g.has_edge("b", "a")); // reverse still exists
        assert_eq!(g.edge_count(), 1);
        assert_digraph_core_invariants(&g);
    }

    #[test]
    fn remove_node_removes_all_incident_directed_edges() {
        let mut g = DiGraph::strict();
        g.add_edge("a", "b").unwrap();
        g.add_edge("b", "c").unwrap();
        g.add_edge("c", "a").unwrap();
        g.add_edge("d", "b").unwrap();

        assert!(g.remove_node("b"));
        assert_eq!(g.node_count(), 3);
        assert!(!g.has_edge("a", "b"));
        assert!(!g.has_edge("b", "c"));
        assert!(!g.has_edge("d", "b"));
        assert!(g.has_edge("c", "a")); // not incident to b
        assert_digraph_core_invariants(&g);
    }

    /// yr2oc.1: every reader that speaks positions agrees with the name-based
    /// readers (the slot store): successor and predecessor rows, `edges_indexed`,
    /// and the CSR.
    fn assert_digraph_positions_agree_with_names(g: &DiGraph) {
        let names: Vec<String> = g.nodes_ordered().into_iter().map(str::to_owned).collect();
        let name_of = |p: usize| g.get_node_name(p).expect("position resolves");
        let csr = g.csr();
        for (position, name) in names.iter().enumerate() {
            assert_eq!(g.get_node_index(name), Some(position));
            let succ: Vec<&str> = g
                .successors_indices(position)
                .expect("live position")
                .iter()
                .map(|&p| name_of(p))
                .collect();
            assert_eq!(
                succ,
                g.successors(name).expect("live node"),
                "succ of {name}"
            );
            let pred: Vec<&str> = g
                .predecessors_indices(position)
                .expect("live position")
                .iter()
                .map(|&p| name_of(p))
                .collect();
            assert_eq!(
                pred,
                g.predecessors(name).expect("live node"),
                "pred of {name}"
            );
            let csr_succ: Vec<&str> = csr
                .successors(position)
                .iter()
                .map(|&p| name_of(p as usize))
                .collect();
            assert_eq!(csr_succ, succ);
            // yr2oc.4: the slot-read edge attributes follow the same rows.
            let out_attrs: Vec<&AttrMap> = g.out_edge_attrs(name).expect("live node").collect();
            let expected_out: Vec<&AttrMap> = succ
                .iter()
                .map(|&t| g.edge_attrs(name, t).expect("edge"))
                .collect();
            assert_eq!(out_attrs, expected_out, "out-edges of {name}");
            let in_attrs: Vec<&AttrMap> = g.in_edge_attrs(name).expect("live node").collect();
            let expected_in: Vec<&AttrMap> = pred
                .iter()
                .map(|&s| g.edge_attrs(s, name).expect("edge"))
                .collect();
            assert_eq!(in_attrs, expected_in, "in-edges of {name}");
        }
        assert!(g.successors_indices(names.len()).is_none());
        for ((source, target), attrs) in g.edges_indexed() {
            assert_eq!(Some(attrs), g.edge_attrs(name_of(source), name_of(target)));
        }
        assert_eq!(g.edges_indexed().count(), g.edge_count());
    }

    /// yr2oc.1: the directed twin of the Graph test — a run of removals keeps
    /// slots apart from positions, the positional readers stay in step, and
    /// the first insertion compacts.
    #[test]
    fn removal_runs_keep_directed_positional_readers_in_step_with_names() {
        let mut state = 0x2545_f491_4f6c_dd1d_u64;
        let mut next = |bound: usize| {
            state ^= state << 13;
            state ^= state >> 7;
            state ^= state << 17;
            usize::try_from(state % bound as u64).unwrap()
        };
        for round in 0..40 {
            let n = 12 + round % 20;
            let mut g = DiGraph::strict();
            for i in 0..n {
                let _ = g.add_node(format!("n{i}"));
            }
            for _ in 0..3 * n {
                let (u, v) = (next(n), next(n));
                let mut attrs = AttrMap::new();
                attrs.insert(
                    "w".into(),
                    CgseValue::Int(i64::try_from(u * 100 + v).unwrap()),
                );
                g.add_edge_with_attrs(format!("n{u}"), format!("n{v}"), attrs)
                    .expect("edge");
            }
            assert!(g.is_compact());
            for _ in 0..n / 3 {
                let names: Vec<String> = g.nodes_ordered().into_iter().map(str::to_owned).collect();
                let victim = names[next(names.len() - 1)].clone();
                assert!(g.remove_node(&victim));
                assert!(!g.is_compact(), "round {round}: slots should stay apart");
                assert_digraph_positions_agree_with_names(&g);
                assert_digraph_core_invariants(&g);
            }
            g.add_edge("fresh", "n0").expect("edge");
            assert!(g.is_compact());
            assert_digraph_positions_agree_with_names(&g);
        }
    }

    /// yr2oc.1: every positional reader of a MultiDiGraph (node index and
    /// name, the CSR, the index-addressed edge reads and the indexed edge walk)
    /// agrees with the node order and the name-keyed rows.
    fn assert_multidigraph_positions_agree_with_names(g: &MultiDiGraph) {
        let names = g.nodes_ordered();
        assert_eq!(g.node_count(), names.len());
        for (position, &name) in names.iter().enumerate() {
            assert_eq!(g.get_node_index(name), Some(position), "{name}");
            assert_eq!(g.get_node_name(position), Some(name));
        }
        assert_eq!(g.get_node_name(names.len()), None);
        let csr = g.csr();
        let named = |row: &[u32]| row.iter().map(|&i| names[i as usize]).collect::<Vec<_>>();
        for (position, &name) in names.iter().enumerate() {
            assert_eq!(Some(named(csr.successors(position))), g.successors(name));
            assert_eq!(
                Some(named(csr.predecessors(position))),
                g.predecessors(name)
            );
        }
        // Enough probes to warm the pair index, so both of its paths are read.
        for (s, &source) in names.iter().enumerate() {
            for (t, &target) in names.iter().enumerate() {
                assert_eq!(g.has_edge_by_indices(s, t), g.has_edge(source, target));
                assert_eq!(
                    g.edge_key_count_by_indices(s, t),
                    g.edge_keys(source, target).map_or(0, |keys| keys.len())
                );
            }
        }
        let mut indexed = Vec::new();
        g.try_for_each_indexed_edge_ordered_borrowed(|s, t, source, target, key, _| {
            assert_eq!((names[s], names[t]), (source, target));
            indexed.push((source.to_owned(), target.to_owned(), key));
            Ok::<(), ()>(())
        })
        .expect("the walk never fails");
        let borrowed: Vec<_> = g
            .edges_ordered_borrowed()
            .into_iter()
            .map(|(source, target, key, _)| (source.to_owned(), target.to_owned(), key))
            .collect();
        assert_eq!(indexed, borrowed);
    }

    /// yr2oc.1: runs of O(1) MultiDiGraph removals (single and batched, small
    /// and large batches), interleaved with insertions, keep the node order,
    /// the node attributes, the edges and every positional reader in step with
    /// a name-level model.
    #[test]
    fn multidigraph_removal_runs_keep_positional_readers_in_step_with_names() {
        use std::collections::BTreeMap;
        let mut state = 0x9e37_79b9_7f4a_7c15_u64;
        let mut next = |bound: usize| {
            state ^= state << 13;
            state ^= state >> 7;
            state ^= state << 17;
            usize::try_from(state % bound as u64).unwrap()
        };
        let tagged = |tag: usize| {
            let mut attrs = AttrMap::new();
            attrs.insert("tag".into(), CgseValue::Int(i64::try_from(tag).unwrap()));
            attrs
        };
        for round in 0..30 {
            let n = 10 + round % 17;
            let mut g = MultiDiGraph::strict();
            let mut order: Vec<String> = Vec::new();
            let mut tags: BTreeMap<String, usize> = BTreeMap::new();
            let mut edges: BTreeMap<(String, String), usize> = BTreeMap::new();
            for i in 0..n {
                let name = format!("n{i}");
                g.add_node_with_attrs(name.clone(), tagged(i));
                tags.insert(name.clone(), i);
                order.push(name);
            }
            for _ in 0..3 * n {
                let (u, v) = (format!("n{}", next(n)), format!("n{}", next(n)));
                *edges.entry((u.clone(), v.clone())).or_default() += 1;
                g.add_edge(u, v).expect("edge");
            }
            let mut fresh = 0;
            let forget = |victim: &str,
                          order: &mut Vec<String>,
                          tags: &mut BTreeMap<String, usize>,
                          edges: &mut BTreeMap<(String, String), usize>| {
                order.retain(|name| name != victim);
                tags.remove(victim);
                edges.retain(|(u, v), _| u != victim && v != victim);
            };
            for _ in 0..4 * n {
                match next(6) {
                    0..=2 if !order.is_empty() => {
                        let victim = order[next(order.len())].clone();
                        assert!(g.remove_node(&victim));
                        assert!(!g.remove_node(&victim));
                        forget(&victim, &mut order, &mut tags, &mut edges);
                    }
                    3 | 4 => {
                        let u = format!("f{fresh}");
                        g.add_node_with_attrs(u.clone(), tagged(1000 + fresh));
                        tags.insert(u.clone(), 1000 + fresh);
                        order.push(u.clone());
                        fresh += 1;
                        let v = if next(4) == 0 {
                            u.clone()
                        } else {
                            order[next(order.len())].clone()
                        };
                        let (source, target) = if next(2) == 0 { (u, v) } else { (v, u) };
                        *edges.entry((source.clone(), target.clone())).or_default() += 1;
                        g.add_edge(source, target).expect("edge");
                    }
                    5 if order.len() > 4 => {
                        // Under a quarter of the nodes: the per-node batch path.
                        let victim = order[next(order.len())].clone();
                        let removed = g.remove_nodes_from([victim.as_str(), "absent"]);
                        assert_eq!(removed.0, 1);
                        forget(&victim, &mut order, &mut tags, &mut edges);
                    }
                    _ => {}
                }
                assert_eq!(g.nodes_ordered(), order, "round {round}");
                for (name, &tag) in &tags {
                    assert_eq!(g.node_attrs(name), Some(&tagged(tag)), "{name}");
                }
                assert_eq!(g.edge_count(), edges.values().sum::<usize>());
                for ((u, v), &count) in &edges {
                    assert_eq!(g.edge_keys(u, v).map(|keys| keys.len()), Some(count));
                }
                assert_multidigraph_core_invariants(&g);
                assert_multidigraph_positions_agree_with_names(&g);
            }
            // Half the nodes at once: the whole-graph batch path.
            let doomed: Vec<String> = order.iter().step_by(2).cloned().collect();
            let removed = g.remove_nodes_from(doomed.iter().map(String::as_str));
            assert_eq!(removed.0, doomed.len());
            for victim in &doomed {
                forget(victim, &mut order, &mut tags, &mut edges);
            }
            assert_eq!(g.nodes_ordered(), order);
            assert_eq!(g.edge_count(), edges.values().sum::<usize>());
            assert_multidigraph_core_invariants(&g);
            assert_multidigraph_positions_agree_with_names(&g);
        }
    }

    #[test]
    fn multidigraph_counts_parallel_selfloops_only() {
        let mut g = MultiDiGraph::strict();
        let _ = g.add_edge("a", "a").expect("self-loop add should succeed");
        let _ = g.add_edge("a", "a").expect("self-loop add should succeed");
        let _ = g.add_edge("b", "b").expect("self-loop add should succeed");
        let _ = g.add_edge("a", "b").expect("edge add should succeed");
        let _ = g.add_edge("b", "a").expect("edge add should succeed");

        assert_eq!(g.number_of_selfloops(), 3);
        assert_multidigraph_core_invariants(&g);
    }

    #[test]
    fn multidigraph_parallel_edge_counts_lists_only_repeated_pairs() {
        let mut g = MultiDiGraph::strict();
        for (u, v) in [
            ("a", "b"),
            ("b", "c"),
            ("a", "b"),
            ("c", "c"),
            ("c", "c"),
            ("a", "b"),
        ] {
            let _ = g.add_edge(u, v).expect("edge add should succeed");
        }
        let _ = g.add_edge("b", "a").expect("edge add should succeed");
        let counts: Vec<_> = g.parallel_edge_counts().collect();
        assert_eq!(counts, vec![("a", "b", 3), ("c", "c", 2)]);

        let mut simple = MultiDiGraph::strict();
        let _ = simple.add_edge("a", "b").expect("edge add should succeed");
        let _ = simple.add_edge("b", "a").expect("edge add should succeed");
        assert_eq!(simple.parallel_edge_counts().count(), 0);
    }

    // br-r37-c1-p6bxu: A/B substrate bench for MultiDiGraph::remove_node
    // (O(degree) swap_remove vs the old O(|E|) retain). Ignored by default; run
    // with `cargo test -p fnx-classes --release ab_bench_multidigraph_remove_node
    // -- --ignored --nocapture`. Also asserts byte-exact parity (node_count,
    // edge_count, edges_ordered, incl. self-loops) between the two paths.
    #[test]
    #[ignore]
    fn ab_bench_multidigraph_remove_node() {
        use std::time::Instant;
        const N: usize = 1000;
        const M: usize = 8000;
        const ITERS: usize = 500;
        let build = || {
            let mut g = MultiDiGraph::new(CompatibilityMode::Strict);
            for i in 0..N {
                let _ = g.add_node(i.to_string());
            }
            let mut s: u64 = 0x9E3779B97F4A7C15;
            let mut next = || {
                s = s
                    .wrapping_mul(6364136223846793005)
                    .wrapping_add(1442695040888963407);
                (s >> 33) as usize % N
            };
            for _ in 0..M {
                let a = next();
                let b = next();
                let _ = g.add_edge(a.to_string(), b.to_string());
            }
            g
        };

        let mut gnew = build();
        let victims: Vec<String> = gnew.nodes.keys().take(ITERS).cloned().collect();
        let t = Instant::now();
        for node in &victims {
            gnew.remove_node(node);
        }
        let new_t = t.elapsed();

        // OLD path: full O(|E|) retain per removal.
        let mut gold = build();
        let t = Instant::now();
        for node in &victims {
            if !gold.nodes.contains_key(node) {
                continue;
            }
            if let Some(succs) = gold.successors.get(node) {
                let targets: Vec<String> = succs.keys().cloned().collect();
                for target in targets {
                    if target != *node
                        && let Some(preds) = gold.predecessors.get_mut(&target)
                    {
                        preds.shift_remove(node.as_str());
                    }
                }
            }
            if let Some(preds) = gold.predecessors.get(node) {
                let sources: Vec<String> = preds.keys().cloned().collect();
                for source in sources {
                    if source != *node
                        && let Some(succs) = gold.successors.get_mut(&source)
                    {
                        succs.shift_remove(node.as_str());
                    }
                }
            }
            let mut rc = 0usize;
            let nd = node.clone();
            gold.edges.retain(|k, bucket| {
                let keep = k.source != nd && k.target != nd;
                if !keep {
                    rc += bucket.len();
                }
                keep
            });
            gold.edge_count -= rc;
            gold.successors.remove(node);
            gold.predecessors.remove(node);
            gold.nodes.remove(node);
        }
        let old_t = t.elapsed();

        assert_eq!(gnew.node_count(), gold.node_count());
        assert_eq!(gnew.edge_count(), gold.edge_count());
        assert_eq!(gnew.edges_ordered(), gold.edges_ordered());
        eprintln!(
            "MultiDiGraph remove_node x{ITERS}: retain {old_t:?} -> swap_remove {new_t:?} = {:.2}x",
            old_t.as_secs_f64() / new_t.as_secs_f64()
        );
    }

    #[test]
    fn self_loop_directed() {
        let mut g = DiGraph::strict();
        g.add_edge("a", "a").unwrap();

        assert_eq!(g.edge_count(), 1);
        assert!(g.has_edge("a", "a"));
        assert_eq!(g.out_degree("a"), 1);
        assert_eq!(g.in_degree("a"), 1);
        assert_digraph_core_invariants(&g);
    }

    #[test]
    fn edge_attrs_directed() {
        let mut g = DiGraph::strict();
        let mut attrs = AttrMap::new();
        attrs.insert("weight".to_owned(), "5".into());
        g.add_edge_with_attrs("a", "b", attrs).unwrap();

        assert_eq!(
            g.edge_attrs("a", "b").unwrap().get("weight"),
            Some(&CgseValue::String("5".to_owned()))
        );
        assert!(g.edge_attrs("b", "a").is_none()); // reverse has no attrs
    }

    #[test]
    fn extend_edges_unrecorded_preserves_direction_and_records_once() {
        let mut g = DiGraph::strict();
        g.add_node("a");
        g.add_node("b");
        g.add_node("c");
        let before = g.evidence_ledger().records().len();

        let inserted = g.extend_edges_unrecorded([("a", "b"), ("b", "a"), ("a", "b")]);

        assert_eq!(inserted, 2);
        assert_eq!(g.edge_count(), 2);
        assert!(g.has_edge("a", "b"));
        assert!(g.has_edge("b", "a"));
        assert!(!g.has_edge("a", "c"));
        let records = g.evidence_ledger().records();
        assert_eq!(records.len(), before + 1);
        assert_eq!(
            records.last().map(|record| record.operation.as_ref()),
            Some("extend_edges_unrecorded")
        );
    }

    #[test]
    fn row_staged_attr_edges_preserve_orders_and_duplicate_merges() {
        let mut g = DiGraph::strict();
        let before = g.evidence_ledger().records().len();

        let inserted = g.extend_prepared_edges_with_attrs_row_staged_unrecorded(
            ["b", "x", "a", "t"]
                .into_iter()
                .map(std::borrow::ToOwned::to_owned),
            vec![
                ("b".to_owned(), "x".to_owned(), single_attr("first", "bx")),
                ("a".to_owned(), "t".to_owned(), single_attr("first", "at")),
                ("b".to_owned(), "t".to_owned(), single_attr("first", "bt")),
                (
                    "a".to_owned(),
                    "t".to_owned(),
                    single_attr("second", "merge"),
                ),
                ("t".to_owned(), "t".to_owned(), single_attr("loop", "yes")),
            ],
        );

        assert_eq!(inserted, 4);
        assert_eq!(g.nodes_ordered(), vec!["b", "x", "a", "t"]);
        assert_eq!(
            g.edges_ordered_borrowed()
                .into_iter()
                .map(|(u, v, _)| (u, v))
                .collect::<Vec<_>>(),
            vec![("b", "x"), ("b", "t"), ("a", "t"), ("t", "t")]
        );
        assert_eq!(
            g.edges_ordered_indices(),
            vec![(0, 1), (0, 3), (2, 3), (3, 3)]
        );
        assert_eq!(g.successors("b"), Some(vec!["x", "t"]));
        assert_eq!(g.predecessors("t"), Some(vec!["a", "b", "t"]));
        let merged = g.edge_attrs("a", "t").expect("merged edge should exist");
        assert_eq!(merged.get("first"), Some(&CgseValue::from("at")));
        assert_eq!(merged.get("second"), Some(&CgseValue::from("merge")));
        assert_digraph_core_invariants(&g);
        let records = g.evidence_ledger().records();
        assert_eq!(records.len(), before + 1);
        assert_eq!(
            records.last().map(|record| record.operation.as_ref()),
            Some("extend_edges_unrecorded")
        );
    }

    #[test]
    fn repeated_edge_merges_attrs() {
        let mut g = DiGraph::strict();
        let mut first = AttrMap::new();
        first.insert("weight".to_owned(), "1".into());
        g.add_edge_with_attrs("a", "b", first).unwrap();

        let mut second = AttrMap::new();
        second.insert("color".to_owned(), "red".into());
        g.add_edge_with_attrs("a", "b", second).unwrap();

        assert_eq!(g.edge_count(), 1);
        let attrs = g.edge_attrs("a", "b").unwrap();
        assert_eq!(
            attrs.get("weight"),
            Some(&CgseValue::String("1".to_owned()))
        );
        assert_eq!(
            attrs.get("color"),
            Some(&CgseValue::String("red".to_owned()))
        );
    }

    #[test]
    fn edges_ordered_preserves_direction() {
        let mut g = DiGraph::strict();
        g.add_edge("b", "a").unwrap();
        g.add_edge("a", "c").unwrap();

        let pairs: Vec<(String, String)> = g
            .edges_ordered()
            .into_iter()
            .map(|e| (e.left, e.right))
            .collect();
        // b was added first as source, so b→a first, then a→c
        assert_eq!(
            pairs,
            vec![
                ("b".to_owned(), "a".to_owned()),
                ("a".to_owned(), "c".to_owned()),
            ]
        );
    }

    #[test]
    fn type_identity() {
        let g = DiGraph::strict();
        assert!(g.is_directed());
        assert!(!g.is_multigraph());
    }

    #[test]
    fn to_undirected_merges_edges() {
        let mut g = DiGraph::strict();
        g.add_edge("a", "b").unwrap();
        g.add_edge("b", "a").unwrap();
        g.add_edge("b", "c").unwrap();

        let ug = g.to_undirected();
        assert_eq!(ug.node_count(), 3);
        assert_eq!(ug.edge_count(), 2); // (a,b) merged, plus (b,c)
        assert!(ug.has_edge("a", "b"));
        assert!(ug.has_edge("b", "a")); // undirected: same edge
        assert!(ug.has_edge("b", "c"));
    }

    #[test]
    fn to_undirected_preserves_runtime_policy() {
        let mut g = DiGraph::hardened();
        g.add_edge("a", "b").unwrap();
        g.add_edge("b", "a").unwrap();
        let expected_policy = g.runtime_policy().clone();

        let ug = g.to_undirected();

        assert_runtime_policy_preserved(&expected_policy, ug.runtime_policy());
    }

    /// br-r37-c1-toundirbatch: paired-interleaved median A/B for DiGraph::to_undirected —
    /// inline per-edge baseline vs the shipped batch, one binary / one worker with a NULL
    /// control. A complete digraph K100 (every ordered pair a->b) makes 9900 directed edges
    /// collapse to 4950 undirected, so HALF the add_edge_with_attrs calls are merge-dups and
    /// each paid a per-edge policy record. Byte-identity asserted first.
    /// `#[ignore]`; run with
    /// `cargo test --release -p fnx-classes --lib to_undirected_batch_ab -- --ignored --nocapture`.
    #[test]
    #[ignore = "measurement; run with --release --ignored --nocapture"]
    fn to_undirected_batch_ab() {
        use std::hint::black_box;
        use std::time::Instant;

        let mut input = DiGraph::strict();
        for i in 0..200 {
            input.add_node(format!("{i}"));
        }
        for a in 0..200 {
            for b in 0..200 {
                if a != b {
                    input.add_edge(format!("{a}"), format!("{b}")).unwrap();
                }
            }
        }

        let build = |g: &DiGraph, batch: bool| -> crate::Graph {
            let mut out = crate::Graph::with_runtime_policy(g.runtime_policy.clone());
            for (node, idx) in g.node_order.iter() {
                let attrs = &g.node_attrs[idx];
                out.add_node_with_attrs(node.clone(), attrs.clone());
            }
            if batch {
                let edges: Vec<(String, String, AttrMap)> = g
                    .edges_ordered()
                    .into_iter()
                    .map(|snap| (snap.left, snap.right, snap.attrs))
                    .collect();
                let _ = out.extend_edges_with_attrs_unrecorded(edges);
            } else {
                for snap in g.edges_ordered() {
                    let _ = out.add_edge_with_attrs(snap.left, snap.right, snap.attrs);
                }
            }
            out
        };

        let old = build(&input, false);
        let new = build(&input, true);
        assert_eq!(
            old.edges_ordered_borrowed(),
            new.edges_ordered_borrowed(),
            "batch to_undirected must equal the per-edge baseline"
        );
        assert_eq!(old.nodes_ordered(), new.nodes_ordered());

        let time = |batch: bool| -> f64 {
            let t0 = Instant::now();
            black_box(build(&input, batch));
            t0.elapsed().as_secs_f64()
        };
        for _ in 0..3 {
            black_box(time(true));
            black_box(time(false));
        }
        let median = |v: &[f64]| {
            let mut s = v.to_vec();
            s.sort_by(|a, b| a.partial_cmp(b).unwrap());
            s[s.len() / 2]
        };
        let rounds = 61usize;
        let paired = |cand: bool, base_arm: bool| -> Vec<f64> {
            let mut v = Vec::with_capacity(rounds);
            for round in 0..rounds {
                let (tb, tc) = if round.is_multiple_of(2) {
                    let bt = time(base_arm);
                    let ct = time(cand);
                    (bt, ct)
                } else {
                    let ct = time(cand);
                    let bt = time(base_arm);
                    (bt, ct)
                };
                v.push(tb / tc);
            }
            v
        };
        let report = |name: &str, ratios: &[f64]| {
            let wins = ratios.iter().filter(|&&r| r > 1.0).count();
            let mut sorted = ratios.to_vec();
            sorted.sort_by(|a, b| a.partial_cmp(b).unwrap());
            println!(
                "TOUNDIR_BATCH_AB {name}: median={:.4}x win_rate={wins}/{rounds} \
                 p5_p95=[{:.4},{:.4}]",
                median(ratios),
                sorted[rounds * 5 / 100],
                sorted[rounds * 95 / 100],
            );
        };
        println!("TOUNDIR_BATCH_AB K200_complete rounds={rounds} (>1 = batch faster)");
        report("BATCH_vs_string", &paired(true, false));
        report("NULL_batch_vs_batch", &paired(true, true));
    }

    #[test]
    fn snapshot_roundtrip() {
        let mut g = DiGraph::strict();
        let mut attrs = AttrMap::new();
        attrs.insert("weight".to_owned(), "3".into());
        g.add_edge_with_attrs("a", "b", attrs).unwrap();
        g.add_edge("b", "c").unwrap();
        g.add_edge("c", "a").unwrap();

        let snap = g.snapshot();
        let mut replayed = DiGraph::new(snap.mode);
        for node in &snap.nodes {
            let _ = replayed.add_node(node.clone());
        }
        for edge in &snap.edges {
            replayed
                .add_edge_with_attrs(edge.left.clone(), edge.right.clone(), edge.attrs.clone())
                .unwrap();
        }

        assert_eq!(replayed.snapshot(), snap);
        assert_digraph_core_invariants(&replayed);
    }

    #[test]
    fn multidigraph_tracks_parallel_edges_with_distinct_keys() {
        let mut graph = MultiDiGraph::strict();
        let first = graph.add_edge("a", "b").expect("edge add should succeed");
        let second = graph.add_edge("a", "b").expect("edge add should succeed");

        assert_ne!(first, second);
        assert_eq!(graph.edge_count(), 2);
        assert_eq!(graph.edge_keys("a", "b"), Some(vec![0, 1]));
        assert_multidigraph_core_invariants(&graph);
    }

    #[test]
    fn multidigraph_remove_node_clears_incoming_and_outgoing_parallel_edges() {
        let mut graph = MultiDiGraph::strict();
        let _ = graph.add_edge("a", "b").expect("edge add should succeed");
        let _ = graph.add_edge("a", "b").expect("edge add should succeed");
        let _ = graph.add_edge("b", "a").expect("edge add should succeed");
        let _ = graph.add_edge("b", "c").expect("edge add should succeed");

        assert!(graph.remove_node("b"));
        assert_eq!(graph.edge_count(), 0);
        assert!(!graph.has_node("b"));
        assert_eq!(graph.successors("a"), Some(vec![]));
        assert_eq!(graph.predecessors("a"), Some(vec![]));
        assert_multidigraph_core_invariants(&graph);
    }

    #[test]
    fn multidigraph_clear_edges_preserves_nodes_attrs_and_rows() {
        let mut graph = MultiDiGraph::strict();
        assert!(graph.add_node_with_attrs("a", single_attr("color", "red")));
        assert!(graph.add_node_with_attrs("b", single_attr("color", "blue")));
        let _ = graph
            .add_edge_with_key_and_attrs("a", "b", 7, single_attr("weight", "3"))
            .expect("edge add should succeed");
        let _ = graph
            .add_edge_with_key_and_attrs("b", "a", 2, single_attr("weight", "5"))
            .expect("edge add should succeed");
        let before_revision = graph.revision();

        graph.clear_edges();

        assert_eq!(graph.node_count(), 2);
        assert_eq!(graph.nodes_ordered(), vec!["a", "b"]);
        assert_eq!(graph.edge_count(), 0);
        assert_eq!(graph.edges_ordered(), Vec::<MultiDiEdgeSnapshot>::new());
        assert_eq!(graph.successors("a"), Some(vec![]));
        assert_eq!(graph.predecessors("a"), Some(vec![]));
        assert_eq!(
            graph.node_attrs("a").unwrap().get("color"),
            Some(&CgseValue::from("red"))
        );
        assert!(graph.revision() > before_revision);
        assert_multidigraph_core_invariants(&graph);
    }

    #[test]
    fn multidigraph_roundtrips_sparse_snapshot_keys() {
        let mut graph = MultiDiGraph::strict();
        assert_eq!(
            graph.add_edge("a", "b").expect("edge add should succeed"),
            0
        );
        assert_eq!(
            graph.add_edge("a", "b").expect("edge add should succeed"),
            1
        );
        assert_eq!(
            graph.add_edge("a", "b").expect("edge add should succeed"),
            2
        );
        assert!(graph.remove_edge("a", "b", Some(1)));
        assert_eq!(
            graph.add_edge("a", "b").expect("edge add should succeed"),
            3
        );

        let snapshot = graph.snapshot();
        assert_eq!(
            snapshot
                .edges
                .iter()
                .map(|edge| edge.key)
                .collect::<Vec<_>>(),
            vec![0, 2, 3]
        );

        let mut replayed = MultiDiGraph::new(snapshot.mode);
        for node in &snapshot.nodes {
            let _ = replayed.add_node(node.clone());
        }
        for edge in &snapshot.edges {
            replayed
                .add_edge_with_key_and_attrs(
                    edge.source.clone(),
                    edge.target.clone(),
                    edge.key,
                    edge.attrs.clone(),
                )
                .expect("snapshot replay should preserve explicit keys");
        }

        assert_eq!(replayed.snapshot(), snapshot);
        assert_multidigraph_core_invariants(&replayed);
    }

    #[test]
    fn runtime_policy_tracks_hardened_multidigraph_recovery() {
        let mut g = MultiDiGraph::hardened();
        g.add_edge("a", "b")
            .expect("hardened multidigraph edge add should succeed");

        assert_eq!(g.runtime_policy().mode(), CompatibilityMode::Hardened);
        assert!(!g.runtime_policy().decision_log().records().is_empty());
        assert!(g.runtime_policy().posterior().observation_count >= 1);
    }

    /// br-r37-c1-eo88t: directed mirror of the simple/multi pairing in
    /// `fnx-classes/src/lib.rs`. An ALLOWED edge add writes exactly one
    /// `add_edge` ledger record and it is the outcome one; a REFUSED add still
    /// writes its gate record, which is the rejected mutation's only audit
    /// trail because the outcome record never runs.
    fn add_edge_evidence_signals(records: &[DecisionRecord]) -> Vec<Vec<&str>> {
        records
            .iter()
            .filter(|record| record.operation == "add_edge")
            .map(|record| {
                record
                    .evidence
                    .iter()
                    .map(|term| term.signal.as_ref())
                    .collect()
            })
            .collect()
    }

    /// br-r37-c1-4c29a: `MultiDiGraph`'s adjacency rows moved from the default
    /// SipHash to `FxBuildHasher`. That is only admissible because `IndexMap`
    /// and `IndexSet` are INSERTION-ordered, so the hasher cannot reach
    /// iteration order — and iteration order is CGSE-observable here, since
    /// `edges_ordered` walks these rows. Asserted rather than asserted-by-faith:
    /// the endpoints below are inserted in an order that is neither sorted nor
    /// hash order, and must come back exactly as inserted.
    #[test]
    fn multidigraph_adjacency_iterates_in_insertion_order_under_fxhash() {
        let mut graph = MultiDiGraph::strict();
        for (source, target) in [
            ("zeta", "mid"),
            ("zeta", "alpha"),
            ("zeta", "omega"),
            ("zeta", "beta"),
            ("alpha", "zeta"),
        ] {
            graph
                .add_edge_with_attrs(source, target, AttrMap::new())
                .expect("edge is allowed");
        }
        // Parallel edges on an already-seen pair must append, not reorder.
        graph
            .add_edge_with_attrs("zeta", "alpha", AttrMap::new())
            .expect("parallel edge is allowed");

        let observed: Vec<(String, String, usize)> = graph
            .edges_ordered_borrowed()
            .into_iter()
            .map(|(source, target, key, _)| (source.to_owned(), target.to_owned(), key))
            .collect();
        assert_eq!(
            observed,
            vec![
                ("zeta".to_owned(), "mid".to_owned(), 0),
                ("zeta".to_owned(), "alpha".to_owned(), 0),
                ("zeta".to_owned(), "alpha".to_owned(), 1),
                ("zeta".to_owned(), "omega".to_owned(), 0),
                ("zeta".to_owned(), "beta".to_owned(), 0),
                ("alpha".to_owned(), "zeta".to_owned(), 0),
            ],
            "adjacency rows must stay insertion-ordered regardless of hasher"
        );
    }

    /// br-r37-c1-jc9e4: directed-multi mirror of the `Graph` fresh-bucket lock.
    /// `add_edge_impl` now TAKES the caller's attribute map when the keyed
    /// bucket is empty. The multigraph-only risk is that the take could land on
    /// the wrong parallel edge, so each key is asserted separately.
    #[test]
    fn multidigraph_fresh_bucket_take_matches_merge_semantics() {
        fn weight(value: i64) -> AttrMap {
            let mut attrs = AttrMap::new();
            attrs.insert("w".to_owned(), CgseValue::Int(value));
            attrs
        }

        let mut graph = MultiDiGraph::strict();
        let k0 = graph
            .add_edge_with_attrs("a", "b", weight(1))
            .expect("edge is allowed");
        // A PARALLEL edge gets its own fresh bucket and must not disturb k0.
        let k1 = graph
            .add_edge_with_attrs("a", "b", weight(2))
            .expect("parallel edge is allowed");
        assert_ne!(k0, k1);
        assert_eq!(
            graph.edge_attrs("a", "b", k0).and_then(|a| a.get("w")),
            Some(&CgseValue::Int(1))
        );
        assert_eq!(
            graph.edge_attrs("a", "b", k1).and_then(|a| a.get("w")),
            Some(&CgseValue::Int(2))
        );

        // A keyed edge created with EMPTY attrs still takes the empty-bucket
        // branch when attrs arrive later.
        let k2 = graph
            .add_edge_with_attrs("c", "d", AttrMap::new())
            .expect("edge is allowed");
        assert!(
            graph
                .edge_attrs("c", "d", k2)
                .expect("edge exists")
                .is_empty()
        );
        graph
            .add_edge_with_key_and_attrs("c", "d", k2, weight(7))
            .expect("edge is allowed");
        assert_eq!(
            graph.edge_attrs("c", "d", k2).and_then(|a| a.get("w")),
            Some(&CgseValue::Int(7))
        );

        // Non-empty bucket still merges rather than being replaced.
        let mut extra = AttrMap::new();
        extra.insert("k".to_owned(), CgseValue::Int(3));
        graph
            .add_edge_with_key_and_attrs("c", "d", k2, extra)
            .expect("edge is allowed");
        let stored = graph.edge_attrs("c", "d", k2).expect("edge exists");
        assert_eq!(stored.get("w"), Some(&CgseValue::Int(7)));
        assert_eq!(stored.get("k"), Some(&CgseValue::Int(3)));
    }

    /// br-r37-c1-jc9e4: `MultiDiGraph::add_node_with_attrs_unrecorded` stopped
    /// using `entry(node.clone()).or_default()` for the `successors` /
    /// `predecessors` rows and now probes borrowed, inserting only when the row
    /// is genuinely absent. The risk that swap introduces is a row that fails to
    /// get created, which would surface later as a lost edge rather than at the
    /// call itself, so this exercises row creation and row preservation
    /// directly through the degrees.
    #[test]
    fn multidigraph_add_node_creates_and_preserves_adjacency_rows() {
        let mut graph = MultiDiGraph::strict();

        // A node created by add_node — not by an edge — must still get usable
        // successor AND predecessor rows.
        graph.add_node("x");
        graph
            .add_edge_with_attrs("x", "y", AttrMap::new())
            .expect("edge is allowed");
        graph
            .add_edge_with_attrs("w", "x", AttrMap::new())
            .expect("edge is allowed");
        assert_eq!(graph.out_degree("x"), 1);
        assert_eq!(graph.in_degree("x"), 1);

        // Re-adding an EXISTING node now takes the borrowed path that skips the
        // insert entirely. It must not disturb the rows it skips.
        for _ in 0..5 {
            graph.add_node("x");
        }
        assert_eq!(graph.out_degree("x"), 1);
        assert_eq!(graph.in_degree("x"), 1);
        assert_eq!(graph.node_count(), 3);

        // Parallel edges still accumulate on the preserved row.
        graph
            .add_edge_with_attrs("x", "y", AttrMap::new())
            .expect("parallel edge is allowed");
        assert_eq!(graph.out_degree("x"), 2);
    }

    /// br-r37-c1-jc9e4: directed mirror of the `Graph` lock. Pins the endpoint
    /// resolution cases that collapsing the per-endpoint `contains_key` +
    /// `get_index_of` pair restructured — in particular the self-loop branch,
    /// which now returns the source index instead of probing for the target and
    /// still has to report `target_autocreated` as a COPY of
    /// `source_autocreated`. Direction is asserted alongside, because the
    /// successor/predecessor rows are now filled from the reused indices.
    #[test]
    fn digraph_add_edge_endpoint_resolution_preserves_autocreation_and_direction() {
        fn autocreated(graph: &DiGraph) -> (String, String) {
            let record = graph
                .evidence_ledger()
                .records()
                .last()
                .expect("add_edge files an outcome record");
            let read = |signal: &str| {
                record
                    .evidence
                    .iter()
                    .find(|term| term.signal == signal)
                    .map(|term| term.observed_value.to_string())
                    .unwrap_or_else(|| panic!("{signal} must be present"))
            };
            (read("source_autocreated"), read("target_autocreated"))
        }

        // Self-loop that CREATES its node: both flags true, one node added.
        let mut graph = DiGraph::strict();
        graph
            .add_edge_with_attrs("s", "s", AttrMap::new())
            .expect("self-loop is allowed");
        assert_eq!(autocreated(&graph), ("true".to_owned(), "true".to_owned()));
        assert_eq!(graph.node_count(), 1);

        // Self-loop on an EXISTING node: both flags false, nothing added.
        graph
            .add_edge_with_attrs("s", "s", AttrMap::new())
            .expect("re-adding a self-loop is allowed");
        assert_eq!(
            autocreated(&graph),
            ("false".to_owned(), "false".to_owned())
        );
        assert_eq!(graph.node_count(), 1);

        // Source exists, target is new — the asymmetric case.
        graph
            .add_edge_with_attrs("s", "t", AttrMap::new())
            .expect("edge is allowed");
        assert_eq!(autocreated(&graph), ("false".to_owned(), "true".to_owned()));

        // Both new: source resolved and inserted BEFORE target, so insertion
        // order is source-then-target and earlier indices are unmoved.
        graph
            .add_edge_with_attrs("z", "a", AttrMap::new())
            .expect("edge is allowed");
        assert_eq!(autocreated(&graph), ("true".to_owned(), "true".to_owned()));
        assert_eq!(graph.get_node_index("s"), Some(0));
        assert_eq!(graph.get_node_index("t"), Some(1));
        assert_eq!(graph.get_node_index("z"), Some(2));
        assert_eq!(graph.get_node_index("a"), Some(3));

        // Direction survives filling succ/pred from the reused indices.
        assert!(graph.has_edge("z", "a"));
        assert!(!graph.has_edge("a", "z"));
        assert!(graph.has_edge("s", "t"));
        assert!(!graph.has_edge("t", "s"));
    }

    #[test]
    fn allowed_digraph_add_edge_records_only_the_outcome_decision() {
        let mut g = DiGraph::strict();
        g.add_edge_with_attrs("a", "b", AttrMap::new())
            .expect("edge add should succeed");

        assert_eq!(
            add_edge_evidence_signals(g.evidence_ledger().records()),
            vec![vec![
                "self_loop",
                "edge_attr_count",
                "source_autocreated",
                "target_autocreated",
            ]]
        );
        assert_eq!(
            g.evidence_ledger()
                .records()
                .iter()
                .filter(|record| record.operation == "add_edge")
                .map(|record| record.action)
                .collect::<Vec<_>>(),
            vec![DecisionAction::Allow]
        );
    }

    #[test]
    fn allowed_multidigraph_add_edge_records_only_the_outcome_decision() {
        let mut g = MultiDiGraph::strict();
        let _ = g
            .add_edge_with_attrs("a", "b", AttrMap::new())
            .expect("edge add should succeed");

        assert_eq!(
            add_edge_evidence_signals(g.evidence_ledger().records()),
            vec![vec![
                "edge_key",
                "edge_attr_count",
                "source_autocreated",
                "target_autocreated",
            ]]
        );
    }

    #[test]
    fn refused_multidigraph_add_edge_still_records_its_gate_decision() {
        let mut g = MultiDiGraph::strict();
        let mut attrs = AttrMap::new();
        attrs.insert("__fnx_incompatible_decoder".to_owned(), "v2".into());
        let err = g
            .add_edge_with_attrs("a", "b", attrs)
            .expect_err("strict mode should fail closed");
        assert_eq!(
            err,
            GraphError::FailClosed {
                operation: "add_edge",
                reason: "incompatible edge metadata".to_owned(),
            }
        );

        assert_eq!(
            add_edge_evidence_signals(g.evidence_ledger().records()),
            vec![vec!["unknown_incompatible_feature"]]
        );
        assert_eq!(
            g.evidence_ledger().records().last().map(|r| r.action),
            Some(DecisionAction::FailClosed)
        );
    }

    #[test]
    fn strict_mode_fails_closed_for_incompatible_attrs() {
        let mut g = DiGraph::strict();
        let mut attrs = AttrMap::new();
        attrs.insert("__fnx_incompatible_decoder".to_owned(), "v2".into());
        let err = g
            .add_edge_with_attrs("a", "b", attrs)
            .expect_err("strict mode should fail closed");

        assert_eq!(
            err,
            GraphError::FailClosed {
                operation: "add_edge",
                reason: "incompatible edge metadata".to_owned(),
            }
        );
    }

    #[test]
    fn revision_increments_on_mutations() {
        let mut g = DiGraph::strict();
        let r0 = g.revision();
        let _ = g.add_node("a");
        let r1 = g.revision();
        assert!(r1 > r0);

        g.add_edge("a", "b").unwrap();
        let r2 = g.revision();
        assert!(r2 > r1);

        let _ = g.remove_edge("a", "b");
        let r3 = g.revision();
        assert!(r3 > r2);
    }

    #[test]
    fn hardened_self_loop_records_allow() {
        let mut g = DiGraph::hardened();
        g.add_edge("loop", "loop").unwrap();

        let record = g
            .evidence_ledger()
            .records()
            .iter()
            .rev()
            .find(|r| r.operation == "add_edge")
            .expect("add_edge should emit ledger row");
        assert_decision_record_schema(record, CompatibilityMode::Hardened);
        assert_eq!(record.action, DecisionAction::Allow);
    }

    #[test]
    fn runtime_policy_tracks_hardened_digraph_recovery() {
        let mut g = DiGraph::hardened();
        g.add_edge("a", "b")
            .expect("hardened digraph edge add should succeed");

        assert_eq!(g.runtime_policy().mode(), CompatibilityMode::Hardened);
        assert!(!g.runtime_policy().decision_log().records().is_empty());
        assert!(g.runtime_policy().posterior().observation_count >= 1);
    }

    /// br-r37-c1-p1tvg: the index probe answers the ORIENTED pair, and only
    /// for live nodes. Probing the undirected pair or a freed index would
    /// each pass a test that only checked present edges.
    #[test]
    fn has_edge_by_indices_is_oriented_and_range_checked() {
        let mut g = DiGraph::strict();
        g.add_edge("tail", "head").expect("edge add should succeed");
        g.add_edge("head", "third")
            .expect("edge add should succeed");

        let (tail, head, third) = (
            g.get_node_index("tail").expect("tail index"),
            g.get_node_index("head").expect("head index"),
            g.get_node_index("third").expect("third index"),
        );
        assert!(g.has_edge_by_indices(tail, head));
        assert!(!g.has_edge_by_indices(head, tail));
        assert!(!g.has_edge_by_indices(tail, third));
        assert!(!g.has_edge_by_indices(third, g.node_count() + 7));
        assert!(!g.has_edge_by_indices(g.node_count() + 7, third));

        // Removal renumbers: "third" takes the freed slot, so a stale key
        // left behind by the remap would answer for the wrong node pair.
        assert!(g.remove_node("tail"), "node removal should succeed");
        let head_after = g.get_node_index("head").expect("head index after removal");
        let third_after = g
            .get_node_index("third")
            .expect("third index after removal");
        assert!(g.has_edge_by_indices(head_after, third_after));
        assert!(!g.has_edge_by_indices(third_after, head_after));
        assert!(!g.has_edge_by_indices(head_after, g.node_count()));
    }

    // -- Proptest -----------------------------------------------------------

    proptest! {
        /// br-r37-c1-p1tvg: the index probe must agree with the String path
        /// on EVERY pair — present, absent, reversed, and out of range —
        /// through arbitrary add/remove sequences that renumber indices.
        #[test]
        fn prop_has_edge_by_indices_matches_the_name_path(
            ops in prop::collection::vec((0_u8..6, 0_u8..6, 0_u8..10), 1..60),
        ) {
            let mut g = DiGraph::strict();
            for (src_id, tgt_id, action) in ops {
                let (source, target) = (node_name(src_id), node_name(tgt_id));
                match action {
                    0..=6 => {
                        prop_assert!(g.add_edge(source, target).is_ok());
                    }
                    7 | 8 => {
                        let _ = g.remove_edge(&source, &target);
                    }
                    _ => {
                        let _ = g.remove_node(&source);
                    }
                }
            }

            let names: Vec<String> = g.nodes_ordered().iter().map(|n| (*n).to_owned()).collect();
            for (i, source) in names.iter().enumerate() {
                for (j, target) in names.iter().enumerate() {
                    prop_assert_eq!(
                        g.has_edge_by_indices(i, j),
                        g.has_edge(source, target),
                        "index pair ({}, {}) disagrees with ({}, {})",
                        i, j, source, target
                    );
                }
                prop_assert!(!g.has_edge_by_indices(i, names.len()));
                prop_assert!(!g.has_edge_by_indices(names.len(), i));
            }
        }

        #[test]
        fn prop_digraph_invariants_under_mixed_mutations(
            ops in prop::collection::vec((0_u8..8, 0_u8..8, any::<bool>()), 1..80),
        ) {
            let mut g = DiGraph::strict();
            let mut last_rev = g.revision();

            for (src_id, tgt_id, is_add) in ops {
                let src = node_name(src_id);
                let tgt = node_name(tgt_id);
                if is_add {
                    prop_assert!(g.add_edge(src, tgt).is_ok());
                } else {
                    let _ = g.remove_edge(&src, &tgt);
                }
                let rev = g.revision();
                prop_assert!(rev >= last_rev);
                last_rev = rev;
                assert_digraph_core_invariants(&g);
            }
        }

        #[test]
        fn prop_digraph_snapshot_deterministic(
            ops in prop::collection::vec((0_u8..8, 0_u8..8, 0_u8..3), 0..64),
        ) {
            let mut g1 = DiGraph::hardened();
            let mut g2 = DiGraph::hardened();

            for (src_id, tgt_id, attrs_variant) in ops {
                let src = node_name(src_id);
                let tgt = node_name(tgt_id);
                let mut attrs = AttrMap::new();
                if attrs_variant == 1 {
                    attrs.insert("weight".to_owned(), (src_id % 5).to_string().into());
                } else if attrs_variant == 2 {
                    attrs.insert("tag".to_owned(), format!("k{}", tgt_id % 4).into());
                }
                prop_assert!(g1.add_edge_with_attrs(src.clone(), tgt.clone(), attrs.clone()).is_ok());
                prop_assert!(g2.add_edge_with_attrs(src, tgt, attrs).is_ok());
            }

            prop_assert_eq!(g1.snapshot(), g2.snapshot());
        }

        #[test]
        fn prop_remove_node_clears_all_directed_edges(
            ops in prop::collection::vec((0_u8..8, 0_u8..8), 1..64),
            target_id in 0_u8..8,
        ) {
            let mut g = DiGraph::strict();
            for (src_id, tgt_id) in ops {
                prop_assert!(g.add_edge(node_name(src_id), node_name(tgt_id)).is_ok());
            }

            let target = node_name(target_id);
            let removed = g.remove_node(&target);
            if removed {
                prop_assert!(!g.has_node(&target));
                for node in g.nodes_ordered() {
                    prop_assert!(!g.has_edge(node, &target));
                    prop_assert!(!g.has_edge(&target, node));
                }
            }
            assert_digraph_core_invariants(&g);
        }

        #[test]
        fn prop_directed_edge_count_equals_successor_sum(
            ops in prop::collection::vec((0_u8..8, 0_u8..8), 1..64),
        ) {
            let mut g = DiGraph::strict();
            for (src_id, tgt_id) in ops {
                prop_assert!(g.add_edge(node_name(src_id), node_name(tgt_id)).is_ok());
            }

            let total_out: usize = g.nodes_ordered().iter().map(|n| g.out_degree(n)).sum();
            let total_in: usize = g.nodes_ordered().iter().map(|n| g.in_degree(n)).sum();
            prop_assert_eq!(g.edge_count(), total_out);
            prop_assert_eq!(g.edge_count(), total_in);
        }
    }
}