// === crates/m1nd-core/src/graph.rs ===

use smallvec::SmallVec;
use std::collections::HashMap;
use std::sync::atomic::{AtomicU32, Ordering};

use crate::error::{M1ndError, M1ndResult};
use crate::types::*;

// ---------------------------------------------------------------------------
// StringInterner — zero-allocation string comparison (04-SPEC Section 1.5)
// Replaces: Python string equality everywhere
// ---------------------------------------------------------------------------

/// Intern table mapping strings to unique u32 handles.
/// Thread-safe for reads; mutation requires &mut.
pub struct StringInterner {
    strings: Vec<String>,
    index: HashMap<String, InternedStr>,
}

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

impl StringInterner {
    pub fn new() -> Self {
        Self {
            strings: Vec::new(),
            index: HashMap::new(),
        }
    }

    pub fn with_capacity(cap: usize) -> Self {
        Self {
            strings: Vec::with_capacity(cap),
            index: HashMap::with_capacity(cap),
        }
    }

    /// Intern `s`, returning its handle. Idempotent.
    pub fn get_or_intern(&mut self, s: &str) -> InternedStr {
        if let Some(&idx) = self.index.get(s) {
            return idx;
        }
        let idx = InternedStr(self.strings.len() as u32);
        self.strings.push(s.to_owned());
        self.index.insert(s.to_owned(), idx);
        idx
    }

    /// Resolve handle back to string. Panics if idx out of range.
    pub fn resolve(&self, idx: InternedStr) -> &str {
        &self.strings[idx.0 as usize]
    }

    /// Try to resolve without panicking.
    pub fn try_resolve(&self, idx: InternedStr) -> Option<&str> {
        self.strings.get(idx.0 as usize).map(|s| s.as_str())
    }

    /// Lookup without interning. Returns `None` if not present.
    pub fn lookup(&self, s: &str) -> Option<InternedStr> {
        self.index.get(s).copied()
    }

    pub fn len(&self) -> usize {
        self.strings.len()
    }

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

// ---------------------------------------------------------------------------
// CsrGraph — Compressed Sparse Row (04-SPEC Section 1.1)
// Replaces: engine_fast.py CSRAdjacency
// ---------------------------------------------------------------------------

/// Raw edge data stored before CSR construction.
#[derive(Clone)]
pub struct PendingEdge {
    pub source: NodeId,
    pub target: NodeId,
    pub weight: FiniteF32,
    pub inhibitory: bool,
    pub relation: InternedStr,
    pub direction: EdgeDirection,
    pub causal_strength: FiniteF32,
}

/// Compressed Sparse Row graph with forward and reverse adjacency.
/// For node `i`, outgoing edges span `offsets[i]..offsets[i+1]`
/// into targets, weights, inhibitory, relations, directions, causal_strengths.
pub struct CsrGraph {
    // --- Forward CSR ---
    /// Length: num_nodes + 1. offsets[num_nodes] == total_edges.
    pub offsets: Vec<u64>,
    /// Length: total_edges. Target node for each edge.
    pub targets: Vec<NodeId>,
    /// Length: total_edges. Edge weight — atomic for lock-free plasticity updates (FM-ACT-021).
    pub weights: Vec<AtomicU32>, // bit-reinterpreted f32; use FiniteF32 for reads
    /// Length: total_edges. true = inhibitory edge.
    pub inhibitory: Vec<bool>,
    /// Length: total_edges. Relation type (interned).
    pub relations: Vec<InternedStr>,
    /// Length: total_edges. Forward or Bidirectional.
    pub directions: Vec<EdgeDirection>,
    /// Length: total_edges. Causal strength in [0.0, 1.0].
    pub causal_strengths: Vec<FiniteF32>,

    // --- Reverse CSR (built at finalize) ---
    /// Length: num_nodes + 1.
    pub rev_offsets: Vec<u64>,
    /// Length: total_edges. Source node for each reverse edge.
    pub rev_sources: Vec<NodeId>,
    /// Length: total_edges. Index into forward arrays for this reverse edge.
    pub rev_edge_idx: Vec<EdgeIdx>,

    /// Pre-finalize edge staging area.
    pub pending_edges: Vec<PendingEdge>,
}

impl CsrGraph {
    /// Create an empty CSR with no nodes/edges.
    pub fn empty() -> Self {
        Self {
            offsets: Vec::new(),
            targets: Vec::new(),
            weights: Vec::new(),
            inhibitory: Vec::new(),
            relations: Vec::new(),
            directions: Vec::new(),
            causal_strengths: Vec::new(),
            rev_offsets: Vec::new(),
            rev_sources: Vec::new(),
            rev_edge_idx: Vec::new(),
            pending_edges: Vec::new(),
        }
    }

    /// Number of edges in the forward CSR.
    pub fn num_edges(&self) -> usize {
        if self.offsets.is_empty() {
            0
        } else {
            *self.offsets.last().unwrap() as usize
        }
    }

    /// Outgoing edge range for `node`.
    #[inline]
    pub fn out_range(&self, node: NodeId) -> std::ops::Range<usize> {
        let lo = self.offsets[node.as_usize()] as usize;
        let hi = self.offsets[node.as_usize() + 1] as usize;
        lo..hi
    }

    /// Incoming edge range for `node` (reverse CSR).
    #[inline]
    pub fn in_range(&self, node: NodeId) -> std::ops::Range<usize> {
        let lo = self.rev_offsets[node.as_usize()] as usize;
        let hi = self.rev_offsets[node.as_usize() + 1] as usize;
        lo..hi
    }

    /// Read weight atomically as FiniteF32 (FM-ACT-021).
    #[inline]
    pub fn read_weight(&self, edge: EdgeIdx) -> FiniteF32 {
        let bits = self.weights[edge.as_usize()].load(Ordering::Relaxed);
        FiniteF32::new(f32::from_bits(bits))
    }

    /// Atomic CAS max on edge weight. Returns Ok on success, Err after retry limit (FM-ACT-019).
    /// Replaces: engine_fast.py direct weight assignment (now lock-free).
    pub fn atomic_max_weight(
        &self,
        edge: EdgeIdx,
        new_val: FiniteF32,
        max_retries: u32,
    ) -> M1ndResult<()> {
        let slot = &self.weights[edge.as_usize()];
        let new_bits = new_val.get().to_bits();
        for _ in 0..max_retries {
            let old_bits = slot.load(Ordering::Relaxed);
            let old_val = f32::from_bits(old_bits);
            if old_val >= new_val.get() {
                return Ok(());
            }
            if slot
                .compare_exchange_weak(old_bits, new_bits, Ordering::Release, Ordering::Relaxed)
                .is_ok()
            {
                return Ok(());
            }
        }
        Err(M1ndError::CasRetryExhausted {
            edge,
            limit: max_retries,
        })
    }

    /// Atomic CAS write on edge weight (for plasticity). Returns Ok on success, Err after retry limit.
    pub fn atomic_write_weight(
        &self,
        edge: EdgeIdx,
        new_val: FiniteF32,
        max_retries: u32,
    ) -> M1ndResult<()> {
        let slot = &self.weights[edge.as_usize()];
        let new_bits = new_val.get().to_bits();
        for _ in 0..max_retries {
            let old_bits = slot.load(Ordering::Relaxed);
            match slot.compare_exchange_weak(
                old_bits,
                new_bits,
                Ordering::Release,
                Ordering::Relaxed,
            ) {
                Ok(_) => return Ok(()),
                Err(_) => continue,
            }
        }
        Err(M1ndError::CasRetryExhausted {
            edge,
            limit: max_retries,
        })
    }
}

// ---------------------------------------------------------------------------
// PlasticityNode — per-node homeostatic metadata (04-SPEC Section 1.2)
// ---------------------------------------------------------------------------

#[derive(Clone, Copy, Debug, Default)]
pub struct PlasticityNode {
    /// Sum of incoming edge weights (for homeostatic normalisation).
    pub incoming_weight_sum: FiniteF32,
    /// Homeostatic ceiling for this node.
    /// Default: HOMEOSTATIC_CEILING = 5.0 from plasticity.py
    pub ceiling: FiniteF32,
}

// ---------------------------------------------------------------------------
// NodeProvenance — cold-path source metadata for nodes
// ---------------------------------------------------------------------------

#[derive(Clone, Copy, Debug, Default)]
pub struct NodeProvenance {
    pub source_path: Option<InternedStr>,
    pub line_start: u32,
    pub line_end: u32,
    pub excerpt: Option<InternedStr>,
    pub namespace: Option<InternedStr>,
    pub canonical: bool,
}

#[derive(Clone, Debug, Default)]
pub struct NodeProvenanceInput<'a> {
    pub source_path: Option<&'a str>,
    pub line_start: Option<u32>,
    pub line_end: Option<u32>,
    pub excerpt: Option<&'a str>,
    pub namespace: Option<&'a str>,
    pub canonical: bool,
}

#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct ResolvedNodeProvenance {
    pub source_path: Option<String>,
    pub line_start: Option<u32>,
    pub line_end: Option<u32>,
    pub excerpt: Option<String>,
    pub namespace: Option<String>,
    pub canonical: bool,
}

impl ResolvedNodeProvenance {
    pub fn is_empty(&self) -> bool {
        self.source_path.is_none()
            && self.line_start.is_none()
            && self.line_end.is_none()
            && self.excerpt.is_none()
            && self.namespace.is_none()
            && !self.canonical
    }
}

// ---------------------------------------------------------------------------
// NodeStorage — SoA layout (04-SPEC Section 1.2)
// Replaces: engine_v2.py Node dataclass, engine_fast.py FastNode
// ---------------------------------------------------------------------------

/// All per-node data in Struct-of-Arrays layout for cache-friendly access.
pub struct NodeStorage {
    pub count: u32,

    // --- Hot path: activation engine reads every query ---
    /// Activation levels [structural, semantic, temporal, causal] per node.
    /// Packed as [f32; 4] because all 4 dims accessed together per node.
    pub activation: Vec<[FiniteF32; 4]>,
    /// PageRank score, computed once at finalize.
    pub pagerank: Vec<FiniteF32>,

    // --- Warm path: plasticity reads per query ---
    pub plasticity: Vec<PlasticityNode>,

    // --- Cold path: seed finding, display, export ---
    /// Interned label index.
    pub label: Vec<InternedStr>,
    /// Node type tag.
    pub node_type: Vec<NodeType>,
    /// Tag set: SmallVec of interned string indices.
    pub tags: Vec<SmallVec<[InternedStr; 6]>>,
    /// Last modification time (Unix seconds).
    pub last_modified: Vec<f64>,
    /// Change frequency normalised [0.0, 1.0].
    pub change_frequency: Vec<FiniteF32>,
    /// Provenance / source metadata for cold-path inspection.
    pub provenance: Vec<NodeProvenance>,
}

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

impl NodeStorage {
    pub fn new() -> Self {
        Self {
            count: 0,
            activation: Vec::new(),
            pagerank: Vec::new(),
            plasticity: Vec::new(),
            label: Vec::new(),
            node_type: Vec::new(),
            tags: Vec::new(),
            last_modified: Vec::new(),
            change_frequency: Vec::new(),
            provenance: Vec::new(),
        }
    }

    pub fn with_capacity(cap: usize) -> Self {
        Self {
            count: 0,
            activation: Vec::with_capacity(cap),
            pagerank: Vec::with_capacity(cap),
            plasticity: Vec::with_capacity(cap),
            label: Vec::with_capacity(cap),
            node_type: Vec::with_capacity(cap),
            tags: Vec::with_capacity(cap),
            last_modified: Vec::with_capacity(cap),
            change_frequency: Vec::with_capacity(cap),
            provenance: Vec::with_capacity(cap),
        }
    }
}

// ---------------------------------------------------------------------------
// EdgePlasticity — per-edge Hebbian state (04-SPEC Section 1.3)
// Replaces: plasticity.py SynapticState per-edge tracking
// ---------------------------------------------------------------------------

/// Per-edge plasticity metadata. Parallel arrays alongside CSR edges.
pub struct EdgePlasticity {
    /// Original weight at graph construction.
    pub original_weight: Vec<FiniteF32>,
    /// Current weight (canonical — CSR AtomicWeights mirror this).
    pub current_weight: Vec<FiniteF32>,
    /// Number of times this edge was strengthened.
    pub strengthen_count: Vec<u16>,
    /// Number of times this edge was weakened.
    pub weaken_count: Vec<u16>,
    /// Whether LTP (long-term potentiation) was applied.
    pub ltp_applied: Vec<bool>,
    /// Whether LTD (long-term depression) was applied.
    pub ltd_applied: Vec<bool>,
    /// Query index at which this edge was last used.
    pub last_used_query: Vec<u32>,
}

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

impl EdgePlasticity {
    pub fn new() -> Self {
        Self {
            original_weight: Vec::new(),
            current_weight: Vec::new(),
            strengthen_count: Vec::new(),
            weaken_count: Vec::new(),
            ltp_applied: Vec::new(),
            ltd_applied: Vec::new(),
            last_used_query: Vec::new(),
        }
    }

    pub fn with_capacity(cap: usize) -> Self {
        Self {
            original_weight: Vec::with_capacity(cap),
            current_weight: Vec::with_capacity(cap),
            strengthen_count: Vec::with_capacity(cap),
            weaken_count: Vec::with_capacity(cap),
            ltp_applied: Vec::with_capacity(cap),
            ltd_applied: Vec::with_capacity(cap),
            last_used_query: Vec::with_capacity(cap),
        }
    }
}

// ---------------------------------------------------------------------------
// Graph — top-level property graph (04-SPEC Section 1.6)
// Replaces: engine_v2.py PropertyGraph + engine_fast.py FastPropertyGraph
// ---------------------------------------------------------------------------

/// The complete property graph. Owns all storage.
/// Mutation methods increment `generation` for desync detection (FM-PL-006).
pub struct Graph {
    pub nodes: NodeStorage,
    pub csr: CsrGraph,
    pub edge_plasticity: EdgePlasticity,
    pub strings: StringInterner,
    /// Maps interned external ID -> internal NodeId.
    pub id_to_node: HashMap<InternedStr, NodeId>,
    /// Monotonic counter incremented on every structural mutation.
    pub generation: Generation,
    pub pagerank_computed: bool,
    pub finalized: bool,
}

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

impl Graph {
    pub fn new() -> Self {
        Self {
            nodes: NodeStorage::new(),
            csr: CsrGraph::empty(),
            edge_plasticity: EdgePlasticity::new(),
            strings: StringInterner::new(),
            id_to_node: HashMap::new(),
            generation: Generation(0),
            pagerank_computed: false,
            finalized: false,
        }
    }

    pub fn with_capacity(node_cap: usize, edge_cap: usize) -> Self {
        Self {
            nodes: NodeStorage::with_capacity(node_cap),
            csr: CsrGraph::empty(),
            edge_plasticity: EdgePlasticity::with_capacity(edge_cap),
            strings: StringInterner::with_capacity(node_cap),
            id_to_node: HashMap::with_capacity(node_cap),
            generation: Generation(0),
            pagerank_computed: false,
            finalized: false,
        }
    }

    /// Add a node. Returns its NodeId. Increments generation.
    /// Replaces: engine_v2.py PropertyGraph.add_node()
    pub fn add_node(
        &mut self,
        external_id: &str,
        label: &str,
        node_type: NodeType,
        tags: &[&str],
        last_modified: f64,
        change_frequency: f32,
    ) -> M1ndResult<NodeId> {
        // FM-ACT-016: duplicate node check
        let ext_interned = self.strings.get_or_intern(external_id);
        if let Some(&existing) = self.id_to_node.get(&ext_interned) {
            return Err(M1ndError::DuplicateNode(existing));
        }

        let id = NodeId::new(self.nodes.count);
        self.nodes.count += 1;

        let label_interned = self.strings.get_or_intern(label);
        let tag_interned: SmallVec<[InternedStr; 6]> =
            tags.iter().map(|t| self.strings.get_or_intern(t)).collect();

        self.nodes.label.push(label_interned);
        self.nodes.node_type.push(node_type);
        self.nodes.tags.push(tag_interned);
        self.nodes.last_modified.push(last_modified);
        self.nodes
            .change_frequency
            .push(FiniteF32::new(change_frequency));
        self.nodes.activation.push([FiniteF32::ZERO; 4]);
        self.nodes.pagerank.push(FiniteF32::ZERO);
        self.nodes.plasticity.push(PlasticityNode::default());
        self.nodes.provenance.push(NodeProvenance::default());

        self.id_to_node.insert(ext_interned, id);
        self.generation = self.generation.next();
        self.finalized = false;
        Ok(id)
    }

    /// Add an edge. Validates source/target existence (FM-ACT-011). Increments generation.
    /// Replaces: engine_v2.py PropertyGraph.add_edge()
    #[allow(clippy::too_many_arguments)]
    pub fn add_edge(
        &mut self,
        source: NodeId,
        target: NodeId,
        relation: &str,
        weight: FiniteF32,
        direction: EdgeDirection,
        inhibitory: bool,
        causal_strength: FiniteF32,
    ) -> M1ndResult<EdgeIdx> {
        // FM-ACT-011: dangling edge check
        if source.as_usize() >= self.nodes.count as usize {
            return Err(M1ndError::DanglingEdge {
                edge: EdgeIdx::new(self.edge_plasticity.original_weight.len() as u32),
                node: source,
            });
        }
        if target.as_usize() >= self.nodes.count as usize {
            return Err(M1ndError::DanglingEdge {
                edge: EdgeIdx::new(self.edge_plasticity.original_weight.len() as u32),
                node: target,
            });
        }

        let edge_idx = EdgeIdx::new(self.edge_plasticity.original_weight.len() as u32);
        let rel_interned = self.strings.get_or_intern(relation);

        // Store in pending edge list (will be turned into CSR on finalize)
        self.edge_plasticity.original_weight.push(weight);
        self.edge_plasticity.current_weight.push(weight);
        self.edge_plasticity.strengthen_count.push(0);
        self.edge_plasticity.weaken_count.push(0);
        self.edge_plasticity.ltp_applied.push(false);
        self.edge_plasticity.ltd_applied.push(false);
        self.edge_plasticity.last_used_query.push(0);

        // Store raw edge data for CSR building later
        self.csr.pending_edges.push(PendingEdge {
            source,
            target,
            weight,
            inhibitory,
            relation: rel_interned,
            direction,
            causal_strength,
        });

        self.generation = self.generation.next();
        self.finalized = false;
        Ok(edge_idx)
    }

    /// Harvest edges already materialized in the forward CSR back into the
    /// `pending_edges` staging area so a re-`finalize()` does not discard them.
    ///
    /// Invariant relied upon: after a previous `finalize()`, `edge_plasticity` is laid
    /// out in CSR-slot order (length == `csr.num_edges()`), and any edges added since
    /// via `add_edge` appended their plasticity *after* that block (with one entry per
    /// `add_edge`, un-expanded). This method rebuilds `pending_edges` and
    /// `edge_plasticity` as `[harvested canonical CSR edges] ++ [newly-added edges]`,
    /// carrying each harvested edge's current (plasticity-tracked) weight. Bidirectional
    /// edges are de-duplicated to their canonical (`source <= target`) slot, exactly as
    /// `save_graph` does, so the subsequent rebuild re-expands them correctly.
    fn rehydrate_csr_into_pending(&mut self) {
        let csr_total = self.csr.num_edges();

        // The CSR `offsets` array was sized for the node count at the *previous*
        // finalize. Nodes added since then are not represented in it, so iterate only
        // over the range the CSR actually covers (`offsets.len() - 1`). Newly-added
        // nodes have no outgoing CSR edges to harvest by definition.
        let csr_nodes = self.csr.offsets.len().saturating_sub(1);

        // Stash the newly-added (post-finalize) pending edges; they keep their tail
        // slots in edge_plasticity.
        let new_pending = std::mem::take(&mut self.csr.pending_edges);

        let mut harvested: Vec<PendingEdge> = Vec::with_capacity(csr_total);
        let mut harvested_plasticity = EdgePlasticity::with_capacity(csr_total);

        for src in 0..csr_nodes {
            let lo = self.csr.offsets[src] as usize;
            let hi = self.csr.offsets[src + 1] as usize;
            for j in lo..hi {
                let tgt = self.csr.targets[j].as_usize();
                let direction = self.csr.directions[j];
                // For bidirectional edges keep only the canonical direction; the rebuild
                // re-creates the reverse copy.
                if direction == EdgeDirection::Bidirectional && src > tgt {
                    continue;
                }
                harvested.push(PendingEdge {
                    source: NodeId::new(src as u32),
                    target: self.csr.targets[j],
                    // Use the live (current) weight so learned plasticity survives.
                    weight: self.csr.read_weight(EdgeIdx::new(j as u32)),
                    inhibitory: self.csr.inhibitory[j],
                    relation: self.csr.relations[j],
                    direction,
                    causal_strength: self.csr.causal_strengths[j],
                });
                harvested_plasticity
                    .original_weight
                    .push(self.edge_plasticity.original_weight[j]);
                harvested_plasticity
                    .current_weight
                    .push(self.edge_plasticity.current_weight[j]);
                harvested_plasticity
                    .strengthen_count
                    .push(self.edge_plasticity.strengthen_count[j]);
                harvested_plasticity
                    .weaken_count
                    .push(self.edge_plasticity.weaken_count[j]);
                harvested_plasticity
                    .ltp_applied
                    .push(self.edge_plasticity.ltp_applied[j]);
                harvested_plasticity
                    .ltd_applied
                    .push(self.edge_plasticity.ltd_applied[j]);
                harvested_plasticity
                    .last_used_query
                    .push(self.edge_plasticity.last_used_query[j]);
            }
        }

        // Rebuild edge_plasticity as [harvested canonical] ++ [newly-added tail].
        // Start from the harvested block (already in [0..harvested.len()) order)...
        let mut combined_plasticity = harvested_plasticity;
        combined_plasticity
            .original_weight
            .reserve(new_pending.len());
        // The newly-added edges occupied the tail slots [csr_total..] of the old
        // edge_plasticity arrays — append them in order.
        combined_plasticity
            .original_weight
            .extend_from_slice(&self.edge_plasticity.original_weight[csr_total..]);
        combined_plasticity
            .current_weight
            .extend_from_slice(&self.edge_plasticity.current_weight[csr_total..]);
        combined_plasticity
            .strengthen_count
            .extend_from_slice(&self.edge_plasticity.strengthen_count[csr_total..]);
        combined_plasticity
            .weaken_count
            .extend_from_slice(&self.edge_plasticity.weaken_count[csr_total..]);
        combined_plasticity
            .ltp_applied
            .extend_from_slice(&self.edge_plasticity.ltp_applied[csr_total..]);
        combined_plasticity
            .ltd_applied
            .extend_from_slice(&self.edge_plasticity.ltd_applied[csr_total..]);
        combined_plasticity
            .last_used_query
            .extend_from_slice(&self.edge_plasticity.last_used_query[csr_total..]);

        // pending_edges = [harvested canonical] ++ [newly-added]
        harvested.extend(new_pending);
        self.csr.pending_edges = harvested;
        self.edge_plasticity = combined_plasticity;
    }

    /// Build CSR forward + reverse adjacency. Compute PageRank.
    /// Must be called before any query. Sets `finalized = true`.
    /// Replaces: engine_fast.py FastPropertyGraph.finalize()
    pub fn finalize(&mut self) -> M1ndResult<()> {
        if self.finalized {
            return Ok(());
        }
        let n = self.nodes.count as usize;

        // Re-finalization safety: `finalize()` rebuilds the CSR *exclusively* from
        // `pending_edges`. After the first finalize, `pending_edges` is empty and all
        // edges live only in the CSR arrays. Any subsequent mutation that flips
        // `finalized` back to false (e.g. `add_node`) followed by another `finalize()`
        // would therefore wipe every edge already materialized in the CSR.
        //
        // To make finalize idempotent and non-destructive, harvest the edges that are
        // already in the CSR back into `pending_edges` (carrying their *current*
        // plasticity-tracked weights), realigning the plasticity staging arrays so the
        // insertion-index ↔ plasticity-slot mapping below stays correct. Harvested
        // edges go first (matching how their plasticity is laid out from the previous
        // finalize), followed by the newly-added pending edges.
        if self.csr.num_edges() > 0 {
            self.rehydrate_csr_into_pending();
        }

        // Build forward CSR from pending edges
        // Sort edges by source for CSR layout, preserving original insertion index
        let edges = std::mem::take(&mut self.csr.pending_edges);
        // Pair each edge with its original insertion index (into edge_plasticity)
        let mut indexed_edges: Vec<(usize, PendingEdge)> = edges.into_iter().enumerate().collect();
        indexed_edges.sort_by_key(|(_, e)| e.source.0);

        let total_edges = indexed_edges.len();

        let mut offsets = vec![0u64; n + 1];
        let mut targets = Vec::with_capacity(total_edges);
        let mut weights = Vec::with_capacity(total_edges);
        let mut inhibitory = Vec::with_capacity(total_edges);
        let mut relations = Vec::with_capacity(total_edges);
        let mut directions = Vec::with_capacity(total_edges);
        let mut causal_strengths = Vec::with_capacity(total_edges);

        // Count edges per source
        for (_, e) in &indexed_edges {
            offsets[e.source.as_usize() + 1] += 1;
            // Bidirectional edges also get a reverse entry in forward CSR
            if e.direction == EdgeDirection::Bidirectional {
                offsets[e.target.as_usize() + 1] += 1;
            }
        }
        // Prefix sum
        for i in 1..=n {
            offsets[i] += offsets[i - 1];
        }

        let total_csr_edges = *offsets.last().unwrap_or(&0) as usize;
        targets.resize(total_csr_edges, NodeId::default());
        weights.extend((0..total_csr_edges).map(|_| AtomicU32::new(0)));
        inhibitory.resize(total_csr_edges, false);
        relations.resize(total_csr_edges, InternedStr::default());
        directions.resize(total_csr_edges, EdgeDirection::Forward);
        causal_strengths.resize(total_csr_edges, FiniteF32::ZERO);

        // Fill using write cursors, tracking original->CSR mapping for plasticity rebuild
        // Each entry: (original_insertion_idx, csr_position)
        let mut plasticity_mapping: Vec<(usize, usize)> = Vec::with_capacity(total_csr_edges);

        let mut cursors = offsets[..n].to_vec();

        for &(orig_idx, ref e) in &indexed_edges {
            let src = e.source.as_usize();
            let pos = cursors[src] as usize;
            targets[pos] = e.target;
            weights[pos] = AtomicU32::new(e.weight.get().to_bits());
            inhibitory[pos] = e.inhibitory;
            relations[pos] = e.relation;
            directions[pos] = e.direction;
            causal_strengths[pos] = e.causal_strength;
            cursors[src] += 1;
            plasticity_mapping.push((orig_idx, pos));

            if e.direction == EdgeDirection::Bidirectional {
                let tgt = e.target.as_usize();
                let pos2 = cursors[tgt] as usize;
                targets[pos2] = e.source;
                weights[pos2] = AtomicU32::new(e.weight.get().to_bits());
                inhibitory[pos2] = e.inhibitory;
                relations[pos2] = e.relation;
                directions[pos2] = e.direction;
                causal_strengths[pos2] = e.causal_strength;
                cursors[tgt] += 1;
                // Bidirectional reverse direction gets same plasticity data (cloned)
                plasticity_mapping.push((orig_idx, pos2));
            }
        }

        // Rebuild edge_plasticity arrays to match CSR order and count
        let old_plasticity = &self.edge_plasticity;
        let mut new_plasticity = EdgePlasticity::with_capacity(total_csr_edges);
        new_plasticity
            .original_weight
            .resize(total_csr_edges, FiniteF32::ZERO);
        new_plasticity
            .current_weight
            .resize(total_csr_edges, FiniteF32::ZERO);
        new_plasticity.strengthen_count.resize(total_csr_edges, 0);
        new_plasticity.weaken_count.resize(total_csr_edges, 0);
        new_plasticity.ltp_applied.resize(total_csr_edges, false);
        new_plasticity.ltd_applied.resize(total_csr_edges, false);
        new_plasticity.last_used_query.resize(total_csr_edges, 0);

        for &(orig_idx, csr_pos) in &plasticity_mapping {
            new_plasticity.original_weight[csr_pos] = old_plasticity.original_weight[orig_idx];
            new_plasticity.current_weight[csr_pos] = old_plasticity.current_weight[orig_idx];
            new_plasticity.strengthen_count[csr_pos] = old_plasticity.strengthen_count[orig_idx];
            new_plasticity.weaken_count[csr_pos] = old_plasticity.weaken_count[orig_idx];
            new_plasticity.ltp_applied[csr_pos] = old_plasticity.ltp_applied[orig_idx];
            new_plasticity.ltd_applied[csr_pos] = old_plasticity.ltd_applied[orig_idx];
            new_plasticity.last_used_query[csr_pos] = old_plasticity.last_used_query[orig_idx];
        }

        self.edge_plasticity = new_plasticity;

        // Build reverse CSR (in-edges)
        // Count in-degree per node
        let mut rev_offsets = vec![0u64; n + 1];
        for i in 0..n {
            let lo = offsets[i] as usize;
            let hi = offsets[i + 1] as usize;
            for tgt_node in &targets[lo..hi] {
                let tgt = tgt_node.as_usize();
                rev_offsets[tgt + 1] += 1;
            }
        }
        for i in 1..=n {
            rev_offsets[i] += rev_offsets[i - 1];
        }

        let total_rev = *rev_offsets.last().unwrap_or(&0) as usize;
        let mut rev_sources = vec![NodeId::default(); total_rev];
        let mut rev_edge_idx = vec![EdgeIdx::default(); total_rev];

        let mut rev_cursors = rev_offsets[..n].to_vec();
        #[allow(clippy::needless_range_loop)]
        for src in 0..n {
            let lo = offsets[src] as usize;
            let hi = offsets[src + 1] as usize;
            for j in lo..hi {
                let tgt = targets[j].as_usize();
                let pos = rev_cursors[tgt] as usize;
                rev_sources[pos] = NodeId::new(src as u32);
                rev_edge_idx[pos] = EdgeIdx::new(j as u32);
                rev_cursors[tgt] += 1;
            }
        }

        self.csr = CsrGraph {
            offsets,
            targets,
            weights,
            inhibitory,
            relations,
            directions,
            causal_strengths,
            rev_offsets,
            rev_sources,
            rev_edge_idx,
            pending_edges: Vec::new(),
        };

        // Compute PageRank
        self.compute_pagerank(0.85, 50, 1e-6);

        self.finalized = true;
        Ok(())
    }

    /// Number of nodes.
    pub fn num_nodes(&self) -> u32 {
        self.nodes.count
    }

    /// Number of edges (forward CSR).
    pub fn num_edges(&self) -> usize {
        self.csr.num_edges()
    }

    /// Resolve external string ID to NodeId.
    pub fn resolve_id(&self, external_id: &str) -> Option<NodeId> {
        let interned = self.strings.lookup(external_id)?;
        self.id_to_node.get(&interned).copied()
    }

    pub fn set_node_provenance(&mut self, node: NodeId, provenance: NodeProvenanceInput<'_>) {
        let idx = node.as_usize();
        if idx >= self.nodes.count as usize {
            return;
        }

        self.nodes.provenance[idx] = NodeProvenance {
            source_path: provenance
                .source_path
                .filter(|value| !value.is_empty())
                .map(|value| self.strings.get_or_intern(value)),
            line_start: provenance.line_start.unwrap_or(0),
            line_end: provenance.line_end.or(provenance.line_start).unwrap_or(0),
            excerpt: provenance
                .excerpt
                .filter(|value| !value.is_empty())
                .map(|value| self.strings.get_or_intern(value)),
            namespace: provenance
                .namespace
                .filter(|value| !value.is_empty())
                .map(|value| self.strings.get_or_intern(value)),
            canonical: provenance.canonical,
        };
    }

    /// Add tags to a node in place (X-RAY write path / `xray.retag`). Idempotent:
    /// tags already present are skipped. Returns how many were actually added.
    ///
    /// Tags are a cold-path node column (mirrors [`Graph::set_node_provenance`]):
    /// they feed neither CSR edges nor PageRank, so no re-`finalize` is required.
    /// New tag strings are interned into the shared pool.
    pub fn add_node_tags(&mut self, node: NodeId, tags: &[&str]) -> usize {
        let idx = node.as_usize();
        if idx >= self.nodes.count as usize {
            return 0;
        }
        let mut added = 0;
        for tag in tags {
            let interned = self.strings.get_or_intern(tag);
            if !self.nodes.tags[idx].contains(&interned) {
                self.nodes.tags[idx].push(interned);
                added += 1;
            }
        }
        added
    }

    /// Remove tags from a node in place. Returns how many were removed. A tag that
    /// is absent from the intern pool can match nothing and is a no-op.
    pub fn remove_node_tags(&mut self, node: NodeId, tags: &[&str]) -> usize {
        let idx = node.as_usize();
        if idx >= self.nodes.count as usize {
            return 0;
        }
        let targets: SmallVec<[InternedStr; 6]> = tags
            .iter()
            .filter_map(|tag| self.strings.lookup(tag))
            .collect();
        if targets.is_empty() {
            return 0;
        }
        let before = self.nodes.tags[idx].len();
        self.nodes.tags[idx].retain(|tag| !targets.contains(tag));
        before - self.nodes.tags[idx].len()
    }

    /// Replace a node's entire tag set in place. Returns the new tag count.
    pub fn set_node_tags(&mut self, node: NodeId, tags: &[&str]) -> usize {
        let idx = node.as_usize();
        if idx >= self.nodes.count as usize {
            return 0;
        }
        self.nodes.tags[idx] = tags
            .iter()
            .map(|tag| self.strings.get_or_intern(tag))
            .collect();
        self.nodes.tags[idx].len()
    }

    /// Resolve a node's tags to strings (cold-path inspection / round-trip checks).
    pub fn node_tags(&self, node: NodeId) -> Vec<&str> {
        let idx = node.as_usize();
        if idx >= self.nodes.count as usize {
            return Vec::new();
        }
        self.nodes.tags[idx]
            .iter()
            .map(|&tag| self.strings.resolve(tag))
            .collect()
    }

    pub fn merge_node_provenance(&mut self, node: NodeId, incoming: NodeProvenanceInput<'_>) {
        let idx = node.as_usize();
        if idx >= self.nodes.count as usize {
            return;
        }

        let current = self.resolve_node_provenance(node);
        let line_start = current.line_start.or(incoming.line_start);
        let line_end = match (current.line_end, incoming.line_end.or(incoming.line_start)) {
            (Some(existing), Some(extra)) => Some(existing.max(extra)),
            (Some(existing), None) => Some(existing),
            (None, Some(extra)) => Some(extra),
            (None, None) => line_start,
        };

        self.set_node_provenance(
            node,
            NodeProvenanceInput {
                source_path: current.source_path.as_deref().or(incoming.source_path),
                line_start,
                line_end,
                excerpt: current.excerpt.as_deref().or(incoming.excerpt),
                namespace: current.namespace.as_deref().or(incoming.namespace),
                canonical: current.canonical || incoming.canonical,
            },
        );
    }

    pub fn resolve_node_provenance(&self, node: NodeId) -> ResolvedNodeProvenance {
        let idx = node.as_usize();
        if idx >= self.nodes.count as usize {
            return ResolvedNodeProvenance::default();
        }

        let provenance = self.nodes.provenance[idx];
        ResolvedNodeProvenance {
            source_path: provenance
                .source_path
                .and_then(|value| self.strings.try_resolve(value).map(str::to_owned)),
            line_start: (provenance.line_start > 0).then_some(provenance.line_start),
            line_end: (provenance.line_end > 0).then_some(provenance.line_end),
            excerpt: provenance
                .excerpt
                .and_then(|value| self.strings.try_resolve(value).map(str::to_owned)),
            namespace: provenance
                .namespace
                .and_then(|value| self.strings.try_resolve(value).map(str::to_owned)),
            canonical: provenance.canonical,
        }
    }

    /// Average out-degree.
    pub fn avg_degree(&self) -> f32 {
        if self.nodes.count == 0 {
            0.0
        } else {
            self.csr.num_edges() as f32 / self.nodes.count as f32
        }
    }

    /// Iterative PageRank on CSR. Power iteration until convergence.
    /// Replaces: engine_v2.py PropertyGraph.compute_pagerank()
    /// DEC-040: damping=0.85, iterations=50, convergence=1e-6
    fn compute_pagerank(&mut self, damping: f32, max_iterations: u32, convergence: f32) {
        let n = self.nodes.count as usize;
        if n == 0 {
            self.pagerank_computed = true;
            return;
        }

        let nf = n as f32;
        let base = (1.0 - damping) / nf;
        let mut pr = vec![1.0f32 / nf; n];
        let mut new_pr = vec![0.0f32; n];

        // Precompute out-degree from forward CSR
        let mut out_degree = vec![0u32; n];
        for (i, deg) in out_degree.iter_mut().enumerate().take(n) {
            let lo = self.csr.offsets[i] as usize;
            let hi = self.csr.offsets[i + 1] as usize;
            *deg = (hi - lo) as u32;
        }

        for _iter in 0..max_iterations {
            new_pr.fill(base);

            // For each node i, accumulate contribution from in-neighbors
            for (i, new_rank) in new_pr.iter_mut().enumerate().take(n) {
                let lo = self.csr.rev_offsets[i] as usize;
                let hi = self.csr.rev_offsets[i + 1] as usize;
                let mut rank_sum = 0.0f32;
                for j in lo..hi {
                    let src = self.csr.rev_sources[j].as_usize();
                    let deg = out_degree[src];
                    if deg > 0 {
                        rank_sum += pr[src] / deg as f32;
                    }
                }
                *new_rank += damping * rank_sum;
            }

            // Check convergence (L1 norm)
            let mut delta = 0.0f32;
            for i in 0..n {
                delta += (new_pr[i] - pr[i]).abs();
            }
            std::mem::swap(&mut pr, &mut new_pr);
            if delta < convergence {
                break;
            }
        }

        // Normalize to [0, 1] by max value
        let max_pr = pr.iter().cloned().fold(0.0f32, f32::max);
        if max_pr > 0.0 {
            for (i, &rank) in pr.iter().enumerate().take(n) {
                self.nodes.pagerank[i] = FiniteF32::new(rank / max_pr);
            }
        }
        self.pagerank_computed = true;
    }
}

// ---------------------------------------------------------------------------
// SharedGraph — concurrent access (04-SPEC Section 5.1)
// Uses parking_lot for fairness (prevents write starvation from queries).
// ---------------------------------------------------------------------------

/// Shared graph handle. Many readers (queries), one writer (ingestion/plasticity).
pub type SharedGraph = std::sync::Arc<parking_lot::RwLock<Graph>>;
