Skip to main content

_diffctx/
graph.rs

1use std::cmp::{Ordering, Reverse};
2use std::collections::BinaryHeap;
3
4use rayon::prelude::*;
5use rustc_hash::{FxHashMap, FxHashSet};
6
7use crate::config::graph_filtering::GRAPH_FILTERING;
8use crate::config::scoring::EGO;
9use crate::types::{Fragment, FragmentId};
10
11#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
12pub enum EdgeCategory {
13    Semantic,
14    Structural,
15    Sibling,
16    Config,
17    ConfigGeneric,
18    Document,
19    Similarity,
20    History,
21    TestEdge,
22    Generic,
23}
24
25impl EdgeCategory {
26    pub fn from_str(s: &str) -> Self {
27        match s {
28            "semantic" => Self::Semantic,
29            "structural" => Self::Structural,
30            "sibling" => Self::Sibling,
31            "config" => Self::Config,
32            "config_generic" => Self::ConfigGeneric,
33            "document" => Self::Document,
34            "similarity" => Self::Similarity,
35            "history" => Self::History,
36            "test_edge" => Self::TestEdge,
37            _ => Self::Generic,
38        }
39    }
40
41    pub fn as_str(self) -> &'static str {
42        match self {
43            Self::Semantic => "semantic",
44            Self::Structural => "structural",
45            Self::Sibling => "sibling",
46            Self::Config => "config",
47            Self::ConfigGeneric => "config_generic",
48            Self::Document => "document",
49            Self::Similarity => "similarity",
50            Self::History => "history",
51            Self::TestEdge => "test_edge",
52            Self::Generic => "generic",
53        }
54    }
55
56    fn is_suppression_exempt(self) -> bool {
57        matches!(self, Self::Semantic | Self::Structural | Self::TestEdge)
58    }
59}
60
61pub struct CsrGraph {
62    pub n: usize,
63    pub indptr: Vec<u32>,
64    pub indices: Vec<u32>,
65    pub weights: Vec<f64>,
66    pub out_weight_sum: Vec<f64>,
67    pub node_to_idx: FxHashMap<FragmentId, u32>,
68    pub idx_to_node: Vec<FragmentId>,
69}
70
71/// Statistics from the per-source out-edge cap that runs in
72/// `build_graph` after `apply_hub_suppression`. Surfaced to Python
73/// via `LatencyBreakdown` so calibration runs can quantify how often
74/// the cap fires and how many edges it discards.
75#[derive(Default, Clone)]
76pub struct EdgeCapStats {
77    /// Edge count after merge + hub suppression, before cap.
78    pub edges_before_cap: usize,
79    /// Edge count after cap.
80    pub edges_after_cap: usize,
81    /// Edges discarded by the cap (lowest-weight neighbors of overfull nodes).
82    pub edges_dropped_by_cap: usize,
83    /// Number of source nodes that had > `max_per_node` outgoing edges
84    /// and therefore had their neighbor list truncated.
85    pub nodes_capped: usize,
86    /// The `max_per_node` value actually applied (after env-var override).
87    pub max_out_edges_per_node: usize,
88    /// Per-category (raw emissions, first-seen deduped) counts from
89    /// pass 1 of the two-pass edge build, sorted by category name.
90    /// Names the builder category responsible for near-dense emission
91    /// blowups (#116). Empty on the materialized-edge path.
92    pub emissions_by_category: Vec<(EdgeCategory, u64, u64)>,
93}
94
95/// A single node-interned edge. 24 bytes instead of a string-keyed
96/// hashmap entry — the difference between ~400 MB and several GB on
97/// multi-million-edge repositories (vendored Go monorepos, vscode),
98/// which is what used to OOM-kill memory-limited benchmark runners.
99#[derive(Clone, Copy)]
100pub struct CompactEdge {
101    pub src: u32,
102    pub dst: u32,
103    pub weight: f64,
104    pub category: EdgeCategory,
105    /// The RAW builder weight placed this edge in the naming class (a
106    /// semantic channel that spells out a symbol or file: import, call,
107    /// type/member ref), as opposed to a diffuse endorsement (0.05 markers,
108    /// tags fallback, similarity). Classified before hub suppression so a
109    /// damped import into a registry hub keeps its naming status.
110    pub naming: bool,
111}
112
113/// Node-interned edge list: the memory-bounded intermediate between
114/// edge collection and graph construction. `idx_to_node` is the sorted,
115/// deduplicated fragment-id universe, so CSR node indexing built from it
116/// is identical to the ordering `Graph::freeze` produces.
117pub struct CompactEdges {
118    pub node_to_idx: FxHashMap<FragmentId, u32>,
119    pub idx_to_node: Vec<FragmentId>,
120    pub edges: Vec<CompactEdge>,
121}
122
123pub fn intern_fragment_nodes(
124    fragments: &[Fragment],
125) -> (FxHashMap<FragmentId, u32>, Vec<FragmentId>) {
126    let mut idx_to_node: Vec<FragmentId> = fragments.iter().map(|f| f.id.clone()).collect();
127    idx_to_node.sort();
128    idx_to_node.dedup();
129    let node_to_idx = idx_to_node
130        .iter()
131        .enumerate()
132        .map(|(i, n)| (n.clone(), i as u32))
133        .collect();
134    (node_to_idx, idx_to_node)
135}
136
137/// Merge duplicate (src, dst) entries: weight = max across duplicates,
138/// category = first occurrence in input order (builder order), matching
139/// the historical `EdgeDict` max-merge + `or_insert` category semantics.
140/// Requires a stable sort so first-in-input stays first-in-group —
141/// `par_sort_by` is a stable parallel merge sort, so its output is
142/// bit-identical to `sort_by` while cutting the dominant serial phase of a
143/// tens-of-millions-edge build (#196).
144pub fn dedup_compact_edges(edges: &mut Vec<CompactEdge>) {
145    use rayon::prelude::*;
146    edges.par_sort_by(|a, b| (a.src, a.dst).cmp(&(b.src, b.dst)));
147    let mut out = 0usize;
148    let mut i = 0usize;
149    while i < edges.len() {
150        let mut merged = edges[i];
151        let mut j = i + 1;
152        while j < edges.len() && edges[j].src == merged.src && edges[j].dst == merged.dst {
153            if edges[j].weight > merged.weight {
154                merged.weight = edges[j].weight;
155            }
156            merged.naming |= edges[j].naming;
157            j += 1;
158        }
159        edges[out] = merged;
160        out += 1;
161        i = j;
162    }
163    edges.truncate(out);
164}
165
166/// Compact (src, dst) -> category lookup over the pre-cap edge set.
167/// Replaces the FragmentId-pair-keyed hashmap that used to retain the
168/// full uncapped edge universe for the lifetime of the pipeline.
169#[derive(Default)]
170pub struct EdgeCategoryTable {
171    node_to_idx: FxHashMap<FragmentId, u32>,
172    idx_to_node: Vec<FragmentId>,
173    entries: Vec<(u32, u32, EdgeCategory)>,
174    sorted: bool,
175}
176
177impl EdgeCategoryTable {
178    fn from_sorted_parts(
179        node_to_idx: FxHashMap<FragmentId, u32>,
180        idx_to_node: Vec<FragmentId>,
181        entries: Vec<(u32, u32, EdgeCategory)>,
182    ) -> Self {
183        debug_assert!(
184            entries
185                .windows(2)
186                .all(|w| (w[0].0, w[0].1) < (w[1].0, w[1].1))
187        );
188        Self {
189            node_to_idx,
190            idx_to_node,
191            entries,
192            sorted: true,
193        }
194    }
195
196    fn intern(&mut self, id: FragmentId) -> u32 {
197        if let Some(&i) = self.node_to_idx.get(&id) {
198            return i;
199        }
200        let i = self.idx_to_node.len() as u32;
201        self.idx_to_node.push(id.clone());
202        self.node_to_idx.insert(id, i);
203        i
204    }
205
206    pub fn insert(&mut self, src: FragmentId, dst: FragmentId, category: EdgeCategory) {
207        let s = self.intern(src);
208        let d = self.intern(dst);
209        self.entries.push((s, d, category));
210        self.sorted = false;
211    }
212
213    /// Stable-sort + last-wins dedup, matching hashmap overwrite
214    /// semantics for repeated `insert` of the same key.
215    pub fn ensure_sorted(&mut self) {
216        if self.sorted {
217            return;
218        }
219        self.entries.sort_by_key(|e| (e.0, e.1));
220        let mut out = 0usize;
221        let mut i = 0usize;
222        while i < self.entries.len() {
223            let mut j = i;
224            while j + 1 < self.entries.len()
225                && self.entries[j + 1].0 == self.entries[i].0
226                && self.entries[j + 1].1 == self.entries[i].1
227            {
228                j += 1;
229            }
230            self.entries[out] = self.entries[j];
231            out += 1;
232            i = j + 1;
233        }
234        self.entries.truncate(out);
235        self.sorted = true;
236    }
237
238    pub fn get(&self, src: &FragmentId, dst: &FragmentId) -> Option<EdgeCategory> {
239        debug_assert!(self.sorted, "EdgeCategoryTable queried before freeze");
240        let s = *self.node_to_idx.get(src)?;
241        let d = *self.node_to_idx.get(dst)?;
242        self.entries
243            .binary_search_by_key(&(s, d), |e| (e.0, e.1))
244            .ok()
245            .map(|k| self.entries[k].2)
246    }
247
248    pub fn for_each<F: FnMut(&FragmentId, &FragmentId, EdgeCategory)>(&self, mut f: F) {
249        for &(s, d, c) in &self.entries {
250            f(
251                &self.idx_to_node[s as usize],
252                &self.idx_to_node[d as usize],
253                c,
254            );
255        }
256    }
257}
258
259pub struct Graph {
260    nodes: FxHashSet<FragmentId>,
261    fwd: FxHashMap<FragmentId, FxHashMap<FragmentId, f64>>,
262    rev: FxHashMap<FragmentId, FxHashMap<FragmentId, f64>>,
263    edge_categories: EdgeCategoryTable,
264    csr_cache: Option<(CsrGraph, CsrGraph)>,
265    pub cap_stats: EdgeCapStats,
266}
267
268impl Graph {
269    pub fn new() -> Self {
270        Self {
271            nodes: FxHashSet::default(),
272            fwd: FxHashMap::default(),
273            rev: FxHashMap::default(),
274            edge_categories: EdgeCategoryTable::default(),
275            csr_cache: None,
276            cap_stats: EdgeCapStats::default(),
277        }
278    }
279
280    pub fn edge_category(&self, src: &FragmentId, dst: &FragmentId) -> Option<EdgeCategory> {
281        self.edge_categories.get(src, dst)
282    }
283
284    pub fn for_each_categorized_edge<F: FnMut(&FragmentId, &FragmentId, EdgeCategory)>(
285        &self,
286        f: F,
287    ) {
288        self.edge_categories.for_each(f)
289    }
290
291    pub fn insert_edge_category(&mut self, src: FragmentId, dst: FragmentId, cat: EdgeCategory) {
292        self.edge_categories.insert(src, dst, cat);
293    }
294
295    pub fn categorized_edge_count(&self) -> usize {
296        self.edge_categories.entries.len()
297    }
298
299    pub fn add_node(&mut self, node: FragmentId) {
300        self.nodes.insert(node);
301    }
302
303    pub fn add_edge(&mut self, src: FragmentId, dst: FragmentId, weight: f64) {
304        if weight.is_nan() || weight.is_infinite() || weight <= 0.0 {
305            return;
306        }
307        if src == dst {
308            return;
309        }
310        debug_assert!(
311            self.csr_cache.is_none(),
312            "add_edge called after Graph was frozen"
313        );
314
315        let fwd_nbrs = self.fwd.entry(src.clone()).or_default();
316        let existing = fwd_nbrs.get(&dst).copied().unwrap_or(0.0);
317        let new_weight = existing.max(weight);
318        fwd_nbrs.insert(dst.clone(), new_weight);
319
320        let rev_nbrs = self.rev.entry(dst).or_default();
321        rev_nbrs.insert(src, new_weight);
322    }
323
324    pub fn node_count(&self) -> usize {
325        self.nodes.len()
326    }
327
328    pub fn nodes(&self) -> impl Iterator<Item = &FragmentId> {
329        self.nodes.iter()
330    }
331
332    pub fn edge_count(&self) -> usize {
333        if let Some((fwd, _)) = &self.csr_cache {
334            return fwd.indices.len();
335        }
336        self.fwd.values().map(|nbrs| nbrs.len()).sum()
337    }
338
339    /// Convert the build-time hashmap representation into CSR and drop the hashmaps.
340    /// After freeze, fwd/rev are empty and all reads go through CSR.
341    pub fn freeze(&mut self) {
342        self.edge_categories.ensure_sorted();
343        if self.csr_cache.is_some() {
344            return;
345        }
346
347        let mut nodes: Vec<FragmentId> = self.nodes.iter().cloned().collect();
348        nodes.sort();
349
350        let node_to_idx: FxHashMap<FragmentId, u32> = nodes
351            .iter()
352            .enumerate()
353            .map(|(i, n)| (n.clone(), i as u32))
354            .collect();
355
356        let fwd = std::mem::take(&mut self.fwd);
357        let rev = std::mem::take(&mut self.rev);
358
359        let fwd_csr = build_csr_owned(fwd, &nodes, &node_to_idx);
360        let rev_csr = build_csr_owned(rev, &nodes, &node_to_idx);
361
362        self.csr_cache = Some((fwd_csr, rev_csr));
363    }
364
365    pub fn to_csr(&mut self) -> &(CsrGraph, CsrGraph) {
366        self.freeze();
367        self.csr_cache.as_ref().unwrap()
368    }
369
370    pub fn fwd_csr(&self) -> Option<&CsrGraph> {
371        self.csr_cache.as_ref().map(|(f, _)| f)
372    }
373
374    pub fn rev_csr(&self) -> Option<&CsrGraph> {
375        self.csr_cache.as_ref().map(|(_, r)| r)
376    }
377
378    /// Look up the weight of edge `src -> dst` in the forward CSR.
379    pub fn forward_edge_weight(&self, src: &FragmentId, dst: &FragmentId) -> Option<f64> {
380        let fwd = self.fwd_csr()?;
381        let src_idx = *fwd.node_to_idx.get(src)? as usize;
382        let dst_idx = *fwd.node_to_idx.get(dst)?;
383        let s = fwd.indptr[src_idx] as usize;
384        let e = fwd.indptr[src_idx + 1] as usize;
385        for k in s..e {
386            if fwd.indices[k] == dst_idx {
387                return Some(fwd.weights[k]);
388            }
389        }
390        None
391    }
392
393    /// Invoke `f(neighbor_id, weight)` for each forward neighbor of `node`.
394    pub fn for_each_forward_neighbor<F: FnMut(&FragmentId, f64)>(
395        &self,
396        node: &FragmentId,
397        mut f: F,
398    ) {
399        let fwd = match self.fwd_csr() {
400            Some(c) => c,
401            None => return,
402        };
403        let idx = match fwd.node_to_idx.get(node) {
404            Some(&i) => i as usize,
405            None => return,
406        };
407        let s = fwd.indptr[idx] as usize;
408        let e = fwd.indptr[idx + 1] as usize;
409        for k in s..e {
410            let dst_idx = fwd.indices[k] as usize;
411            f(&fwd.idx_to_node[dst_idx], fwd.weights[k]);
412        }
413    }
414
415    pub fn ego_graph(
416        &self,
417        seeds: &FxHashSet<FragmentId>,
418        radius: usize,
419    ) -> FxHashMap<FragmentId, f64> {
420        let (fwd, rev) = match &self.csr_cache {
421            Some(c) => c,
422            None => return FxHashMap::default(),
423        };
424        if fwd.n == 0 {
425            return FxHashMap::default();
426        }
427
428        let mut valid_seed_idxs: Vec<u32> = seeds
429            .iter()
430            .filter_map(|s| fwd.node_to_idx.get(s).copied())
431            .collect();
432        valid_seed_idxs.sort_unstable();
433
434        let per_seed: Vec<Vec<(u32, u32, f64)>> = valid_seed_idxs
435            .par_iter()
436            .map(|&seed_idx| bfs_from_seed_with_path_weight(fwd, rev, seed_idx, radius))
437            .collect();
438
439        let gamma = EGO.per_hop_decay;
440        let mut scores: FxHashMap<u32, f64> = FxHashMap::default();
441        for visits in per_seed {
442            for (idx, dist, w_path) in visits {
443                let contribution = gamma.powi(dist as i32) * w_path;
444                *scores.entry(idx).or_insert(0.0) += contribution;
445            }
446        }
447
448        scores
449            .into_iter()
450            .map(|(idx, score)| (fwd.idx_to_node[idx as usize].clone(), score))
451            .collect()
452    }
453}
454
455/// BFS over `fwd ∪ rev` from a single seed, tracking both the shortest
456/// hop distance and the max-product edge-weight path of that length.
457///
458/// Implements the paper's `R_ego` kernel (§4.4.2):
459/// `R_ego(v) = Σ_{u∈E_0} 1[d_hop(u,v) ≤ L] · γ^{d_hop} · W_path(u,v)`
460/// where `W_path(u,v) = max_π ∏_{(a,b)∈π} w_{ab}` over paths of length
461/// equal to `d_hop`. The `Σ` over seeds is performed in `ego_graph`;
462/// per-seed shortest-distance + max-product is computed here.
463fn bfs_from_seed_with_path_weight(
464    fwd: &CsrGraph,
465    rev: &CsrGraph,
466    seed_idx: u32,
467    radius: usize,
468) -> Vec<(u32, u32, f64)> {
469    let n = fwd.n;
470    let mut dist = vec![u32::MAX; n];
471    let mut max_w = vec![0.0_f64; n];
472    dist[seed_idx as usize] = 0;
473    max_w[seed_idx as usize] = 1.0;
474    let mut frontier: Vec<u32> = vec![seed_idx];
475
476    for step in 0..radius {
477        let new_dist = (step + 1) as u32;
478        let mut next: Vec<u32> = Vec::new();
479        for &u in &frontier {
480            let ui = u as usize;
481            let w_u = max_w[ui];
482            for csr in [fwd, rev] {
483                let s = csr.indptr[ui] as usize;
484                let e = csr.indptr[ui + 1] as usize;
485                for k in s..e {
486                    let v = csr.indices[k];
487                    let w_uv = csr.weights[k];
488                    let candidate = w_u * w_uv;
489                    let vi = v as usize;
490                    if dist[vi] == u32::MAX {
491                        dist[vi] = new_dist;
492                        max_w[vi] = candidate;
493                        next.push(v);
494                    } else if dist[vi] == new_dist && candidate > max_w[vi] {
495                        max_w[vi] = candidate;
496                    }
497                }
498            }
499        }
500        frontier = next;
501    }
502
503    let mut result = Vec::new();
504    for i in 0..n {
505        if dist[i] != u32::MAX {
506            result.push((i as u32, dist[i], max_w[i]));
507        }
508    }
509    result
510}
511
512fn build_csr_owned(
513    adj: FxHashMap<FragmentId, FxHashMap<FragmentId, f64>>,
514    nodes: &[FragmentId],
515    node_to_idx: &FxHashMap<FragmentId, u32>,
516) -> CsrGraph {
517    let n = nodes.len();
518    let total_edges: usize = adj.values().map(|v| v.len()).sum();
519
520    let mut indptr = vec![0u32; n + 1];
521    let mut indices = Vec::with_capacity(total_edges);
522    let mut weights = Vec::with_capacity(total_edges);
523
524    for (i, node) in nodes.iter().enumerate() {
525        if let Some(nbrs) = adj.get(node) {
526            let mut edges: Vec<(u32, f64)> = nbrs
527                .iter()
528                .filter_map(|(dst, &w)| node_to_idx.get(dst).map(|&idx| (idx, w)))
529                .collect();
530            edges.sort_by_key(|&(idx, _)| idx);
531            for (idx, w) in edges {
532                indices.push(idx);
533                weights.push(w);
534            }
535        }
536        indptr[i + 1] = indices.len() as u32;
537    }
538
539    let mut out_weight_sum = vec![0.0f64; n];
540    for i in 0..n {
541        let s = indptr[i] as usize;
542        let e = indptr[i + 1] as usize;
543        if e > s {
544            out_weight_sum[i] = weights[s..e].iter().sum();
545        }
546    }
547
548    CsrGraph {
549        n,
550        indptr,
551        indices,
552        weights,
553        out_weight_sum,
554        node_to_idx: node_to_idx.clone(),
555        idx_to_node: nodes.to_vec(),
556    }
557}
558
559/// Hub-suppression damping factors, precomputed from dedup-aware degree
560/// counters so the damped weight of any edge can be evaluated without
561/// materializing the edge universe. The per-edge arithmetic is identical
562/// to the historical in-place suppression loops: at most one division,
563/// by `ln(1+in_degree)` for non-exempt edges into hub destinations or by
564/// `sqrt(fan)` for Semantic edges out of wide-fan sources.
565pub struct SuppressionFactors {
566    in_degree: Vec<u32>,
567    d_p95: f64,
568    sem_file_deg: Vec<u32>,
569}
570
571impl SuppressionFactors {
572    pub fn from_counters(in_degree: Vec<u32>, sem_file_deg: Vec<u32>) -> Self {
573        let mut degrees_sorted: Vec<u32> = in_degree.iter().copied().filter(|&d| d > 0).collect();
574        degrees_sorted.sort_unstable();
575        let d_p95 = if degrees_sorted.is_empty() {
576            0.0
577        } else {
578            let n = degrees_sorted.len();
579            let idx = ((n as f64 * 0.95).ceil() as usize)
580                .saturating_sub(1)
581                .min(n - 1);
582            degrees_sorted[idx] as f64
583        };
584        Self {
585            in_degree,
586            d_p95,
587            sem_file_deg,
588        }
589    }
590
591    fn from_edges(edges: &[CompactEdge], idx_to_node: &[FragmentId]) -> Self {
592        let n_nodes = idx_to_node.len();
593        let mut in_degree = vec![0u32; n_nodes];
594        for e in edges {
595            in_degree[e.dst as usize] += 1;
596        }
597
598        let mut sem_out_files: FxHashMap<u32, FxHashSet<&str>> = FxHashMap::default();
599        for e in edges {
600            if e.category == EdgeCategory::Semantic {
601                sem_out_files
602                    .entry(e.src)
603                    .or_default()
604                    .insert(idx_to_node[e.dst as usize].path.as_ref());
605            }
606        }
607        let mut sem_file_deg = vec![0u32; n_nodes];
608        for (&src, files) in &sem_out_files {
609            sem_file_deg[src as usize] = files.len() as u32;
610        }
611
612        Self::from_counters(in_degree, sem_file_deg)
613    }
614
615    pub fn damp(&self, weight: f64, category: EdgeCategory, src: u32, dst: u32) -> f64 {
616        let mut w = weight;
617        let dst_deg = self.in_degree[dst as usize] as f64;
618        if dst_deg > self.d_p95 && !category.is_suppression_exempt() {
619            w /= dst_deg.ln_1p().max(1.0);
620        }
621        if category == EdgeCategory::Semantic {
622            let src_deg = self.sem_file_deg[src as usize];
623            if src_deg >= GRAPH_FILTERING.hub_out_degree_threshold as u32 {
624                w /= (src_deg as f64).sqrt();
625            }
626        }
627        w
628    }
629}
630
631fn apply_hub_suppression(edges: &mut [CompactEdge], idx_to_node: &[FragmentId]) {
632    if edges.is_empty() {
633        return;
634    }
635    let factors = SuppressionFactors::from_edges(edges, idx_to_node);
636    for e in edges.iter_mut() {
637        e.weight = factors.damp(e.weight, e.category, e.src, e.dst);
638    }
639}
640
641/// Default top-K out-edges per source node. Calibrated against the
642/// observed edge density distribution: typical Python file emits
643/// 5-20 semantic + 2-5 structural + ≤20 sibling edges (~30-50 normal),
644/// so K=64 preserves all legitimate edges while clamping pathological
645/// dense nodes (e.g. utility hubs in django/material-ui that radiate
646/// into thousands of dependents).
647pub(crate) const DEFAULT_MAX_OUT_EDGES_PER_NODE: usize = 64;
648
649/// Truncate each node's outgoing edge list to the top-K by weight.
650/// Run AFTER `apply_hub_suppression` so the suppression pass sees
651/// the true in-degree distribution; otherwise its IDF damping is
652/// computed against a graph that has already been thinned.
653///
654/// Returns the cap stats for diagnostic surfacing into `LatencyBreakdown`.
655/// Keeps each source's top-K neighbors by weight (ties broken by dst
656/// index for determinism), truncating the rest in place.
657pub(crate) fn cap_out_edges_per_source(
658    edges: &mut Vec<CompactEdge>,
659    max_per_node: usize,
660) -> EdgeCapStats {
661    let edges_before = edges.len();
662
663    edges.sort_unstable_by(|a, b| {
664        a.src
665            .cmp(&b.src)
666            .then_with(|| {
667                b.weight
668                    .partial_cmp(&a.weight)
669                    .unwrap_or(std::cmp::Ordering::Equal)
670            })
671            .then_with(|| a.dst.cmp(&b.dst))
672    });
673
674    let mut nodes_capped = 0;
675    let mut out = 0usize;
676    let mut i = 0usize;
677    while i < edges.len() {
678        let src = edges[i].src;
679        let mut j = i;
680        while j < edges.len() && edges[j].src == src {
681            j += 1;
682        }
683        let group = j - i;
684        if group > max_per_node {
685            nodes_capped += 1;
686        }
687        let take = group.min(max_per_node);
688        for k in i..i + take {
689            edges[out] = edges[k];
690            out += 1;
691        }
692        i = j;
693    }
694    edges.truncate(out);
695
696    EdgeCapStats {
697        edges_before_cap: edges_before,
698        edges_after_cap: edges.len(),
699        edges_dropped_by_cap: edges_before - edges.len(),
700        nodes_capped,
701        max_out_edges_per_node: max_per_node,
702        emissions_by_category: Vec::new(),
703    }
704}
705
706pub(crate) fn read_max_out_edges_per_node() -> usize {
707    std::env::var("DIFFCTX_MAX_EDGES_PER_NODE")
708        .ok()
709        .and_then(|v| v.parse::<usize>().ok())
710        .filter(|&v| v > 0)
711        .unwrap_or(DEFAULT_MAX_OUT_EDGES_PER_NODE)
712}
713
714/// Candidate out-edge ranked by post-suppression weight. The ordering
715/// replicates the sort in `cap_out_edges_per_source` (descending weight,
716/// ties by ascending dst index), so a bounded min-heap of these keeps
717/// exactly the edges the full sort-and-truncate would keep.
718#[derive(Clone, Copy)]
719pub struct RankedCandidate {
720    pub weight: f64,
721    pub dst: u32,
722    pub category: EdgeCategory,
723    pub naming: bool,
724}
725
726impl RankedCandidate {
727    fn rank(&self, other: &Self) -> Ordering {
728        self.weight
729            .partial_cmp(&other.weight)
730            .unwrap_or(Ordering::Equal)
731            .then_with(|| other.dst.cmp(&self.dst))
732    }
733}
734
735impl PartialEq for RankedCandidate {
736    fn eq(&self, other: &Self) -> bool {
737        self.rank(other) == Ordering::Equal
738    }
739}
740
741impl Eq for RankedCandidate {}
742
743impl PartialOrd for RankedCandidate {
744    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
745        Some(self.rank(other))
746    }
747}
748
749impl Ord for RankedCandidate {
750    fn cmp(&self, other: &Self) -> Ordering {
751        self.rank(other)
752    }
753}
754
755pub type SourceTopK = BinaryHeap<Reverse<RankedCandidate>>;
756
757pub fn push_bounded_top_k(heap: &mut SourceTopK, candidate: RankedCandidate, k: usize) {
758    if heap.len() < k {
759        heap.push(Reverse(candidate));
760        return;
761    }
762    if let Some(Reverse(worst)) = heap.peek() {
763        if candidate.rank(worst) == Ordering::Greater {
764            heap.pop();
765            heap.push(Reverse(candidate));
766        }
767    }
768}
769
770/// Output of the two-pass edge construction: the final damped, deduped,
771/// per-source-capped edge set plus cap stats derived from pass-1 counters.
772/// The category table is *not* carried here — `assemble_graph` derives it
773/// from this same post-cap `edges` vector, which is the only way to
774/// guarantee the exported category table and the CSR never disagree.
775pub struct CappedEdges {
776    pub node_to_idx: FxHashMap<FragmentId, u32>,
777    pub idx_to_node: Vec<FragmentId>,
778    pub edges: Vec<CompactEdge>,
779    pub cap_stats: EdgeCapStats,
780}
781
782/// Build a CSR from (src, dst, weight) triples over the interned node
783/// universe. Sorting by (src, dst) reproduces the neighbor ordering the
784/// hashmap-based `freeze` path produced (neighbors sorted by dst index),
785/// so weights, out-weight sums, and traversal results are bit-identical.
786fn build_csr_from_pairs(
787    mut pairs: Vec<(u32, u32, f64)>,
788    idx_to_node: &[FragmentId],
789    node_to_idx: &FxHashMap<FragmentId, u32>,
790) -> CsrGraph {
791    pairs.sort_unstable_by_key(|p| (p.0, p.1));
792    let n = idx_to_node.len();
793
794    let mut indptr = vec![0u32; n + 1];
795    let mut indices = Vec::with_capacity(pairs.len());
796    let mut weights = Vec::with_capacity(pairs.len());
797
798    let mut row = 0usize;
799    for &(src, dst, w) in &pairs {
800        while row < src as usize {
801            row += 1;
802            indptr[row] = indices.len() as u32;
803        }
804        indices.push(dst);
805        weights.push(w);
806    }
807    while row < n {
808        row += 1;
809        indptr[row] = indices.len() as u32;
810    }
811
812    let mut out_weight_sum = vec![0.0f64; n];
813    for i in 0..n {
814        let s = indptr[i] as usize;
815        let e = indptr[i + 1] as usize;
816        if e > s {
817            out_weight_sum[i] = weights[s..e].iter().sum();
818        }
819    }
820
821    CsrGraph {
822        n,
823        indptr,
824        indices,
825        weights,
826        out_weight_sum,
827        node_to_idx: node_to_idx.clone(),
828        idx_to_node: idx_to_node.to_vec(),
829    }
830}
831
832/// Graph-construction path for pre-capped edges from the two-pass
833/// pipeline: suppression and the per-source cap already ran, so only
834/// the category table and CSR assembly remain.
835pub fn build_graph_capped(fragments: &[Fragment], capped: CappedEdges) -> Graph {
836    let CappedEdges {
837        node_to_idx,
838        idx_to_node,
839        edges,
840        cap_stats,
841    } = capped;
842    tracing::debug!(
843        "edge cap K={}: {} -> {} (dropped {} from {} nodes)",
844        cap_stats.max_out_edges_per_node,
845        cap_stats.edges_before_cap,
846        cap_stats.edges_after_cap,
847        cap_stats.edges_dropped_by_cap,
848        cap_stats.nodes_capped,
849    );
850    assemble_graph(fragments, node_to_idx, idx_to_node, edges, cap_stats)
851}
852
853/// Materialized-edge construction path kept for the map-based adapter
854/// and small callers: hub suppression, per-source cap, and CSR assembly
855/// all run on the interned edge array.
856pub fn build_graph_compact(fragments: &[Fragment], compact: CompactEdges) -> Graph {
857    let CompactEdges {
858        node_to_idx,
859        idx_to_node,
860        mut edges,
861    } = compact;
862
863    apply_hub_suppression(&mut edges, &idx_to_node);
864
865    let max_per_node = read_max_out_edges_per_node();
866    let cap_stats = cap_out_edges_per_source(&mut edges, max_per_node);
867    tracing::debug!(
868        "edge cap K={}: {} -> {} (dropped {} from {} nodes)",
869        max_per_node,
870        cap_stats.edges_before_cap,
871        cap_stats.edges_after_cap,
872        cap_stats.edges_dropped_by_cap,
873        cap_stats.nodes_capped,
874    );
875
876    assemble_graph(fragments, node_to_idx, idx_to_node, edges, cap_stats)
877}
878
879/// Self-loops and non-finite/non-positive weights must never reach a live
880/// graph: they used to be rejected only by `Graph::add_edge`'s guard, which
881/// has no production caller (both real construction paths route through
882/// here). An infinite weight makes `out_weight_sum` infinite, every
883/// normalized PPR transition probability NaN, and the score map come back
884/// empty with exit 0 -- silently shipping core-only context.
885fn is_valid_edge(e: &CompactEdge) -> bool {
886    e.src != e.dst && e.weight.is_finite() && e.weight > 0.0
887}
888
889/// Builds both the CSR and the category table from the *same* filtered,
890/// already-capped `edges` vector. Deriving `edge_categories` here instead
891/// of accepting it as a separately pre-cap parameter is what guarantees
892/// `Graph::categorized_edge_count() == Graph::edge_count()` always --
893/// previously the category table was built from the pre-cap edge set in
894/// both callers, so it disagreed with the CSR whenever the cap actually
895/// fired (capped-away edges exported with `weight: 0.0`, `edge_count`
896/// mismatching `len(edges)` in the same JSON document).
897fn assemble_graph(
898    fragments: &[Fragment],
899    node_to_idx: FxHashMap<FragmentId, u32>,
900    idx_to_node: Vec<FragmentId>,
901    edges: Vec<CompactEdge>,
902    cap_stats: EdgeCapStats,
903) -> Graph {
904    let valid_edges: Vec<CompactEdge> = edges.into_iter().filter(is_valid_edge).collect();
905
906    let mut category_entries: Vec<(u32, u32, EdgeCategory)> = valid_edges
907        .iter()
908        .map(|e| (e.src, e.dst, e.category))
909        .collect();
910    category_entries.sort_unstable_by_key(|e| (e.0, e.1));
911
912    let fwd_pairs: Vec<(u32, u32, f64)> = valid_edges
913        .iter()
914        .map(|e| (e.src, e.dst, e.weight))
915        .collect();
916    let rev_pairs: Vec<(u32, u32, f64)> = valid_edges
917        .iter()
918        .map(|e| (e.dst, e.src, e.weight))
919        .collect();
920    drop(valid_edges);
921
922    let fwd_csr = build_csr_from_pairs(fwd_pairs, &idx_to_node, &node_to_idx);
923    let rev_csr = build_csr_from_pairs(rev_pairs, &idx_to_node, &node_to_idx);
924
925    let mut graph = Graph::new();
926    for frag in fragments {
927        graph.nodes.insert(frag.id.clone());
928    }
929    graph.edge_categories =
930        EdgeCategoryTable::from_sorted_parts(node_to_idx, idx_to_node, category_entries);
931    graph.cap_stats = cap_stats;
932    graph.csr_cache = Some((fwd_csr, rev_csr));
933    graph
934}
935
936/// Map-based adapter kept for tests and small callers; converts to the
937/// compact representation and delegates. Edges whose endpoints are not
938/// in `fragments` are dropped (the hashmap path silently dropped them
939/// at CSR construction).
940pub fn build_graph(
941    fragments: &[Fragment],
942    edges: FxHashMap<(FragmentId, FragmentId), f64>,
943    categories: FxHashMap<(FragmentId, FragmentId), EdgeCategory>,
944) -> Graph {
945    let (node_to_idx, idx_to_node) = intern_fragment_nodes(fragments);
946    let mut compact_edges = Vec::with_capacity(edges.len());
947    for ((src, dst), w) in &edges {
948        let s = match node_to_idx.get(src) {
949            Some(&i) => i,
950            None => continue,
951        };
952        let d = match node_to_idx.get(dst) {
953            Some(&i) => i,
954            None => continue,
955        };
956        let category = categories
957            .get(&(src.clone(), dst.clone()))
958            .copied()
959            .unwrap_or(EdgeCategory::Generic);
960        compact_edges.push(CompactEdge {
961            src: s,
962            dst: d,
963            weight: *w,
964            category,
965            naming: false,
966        });
967    }
968    dedup_compact_edges(&mut compact_edges);
969    build_graph_compact(
970        fragments,
971        CompactEdges {
972            node_to_idx,
973            idx_to_node,
974            edges: compact_edges,
975        },
976    )
977}
978
979#[cfg(test)]
980mod tests {
981    use super::*;
982    use std::sync::Arc;
983
984    fn fid(path: &str, start: u32, end: u32) -> FragmentId {
985        FragmentId::new(Arc::from(path), start, end)
986    }
987
988    fn collect_forward(g: &Graph, node: &FragmentId) -> Vec<(FragmentId, f64)> {
989        let mut out = Vec::new();
990        g.for_each_forward_neighbor(node, |nbr, w| out.push((nbr.clone(), w)));
991        out
992    }
993
994    #[test]
995    fn add_edge_takes_max_weight() {
996        let mut g = Graph::new();
997        let a = fid("a.rs", 1, 10);
998        let b = fid("b.rs", 1, 10);
999        g.add_node(a.clone());
1000        g.add_node(b.clone());
1001        g.add_edge(a.clone(), b.clone(), 0.5);
1002        g.add_edge(a.clone(), b.clone(), 0.8);
1003        g.add_edge(a.clone(), b.clone(), 0.3);
1004        g.freeze();
1005
1006        let fwd = collect_forward(&g, &a);
1007        assert_eq!(fwd.len(), 1);
1008        assert!((fwd[0].1 - 0.8).abs() < 1e-9);
1009        assert_eq!(fwd[0].0, b);
1010    }
1011
1012    #[test]
1013    fn add_edge_drops_invalid_weights() {
1014        let mut g = Graph::new();
1015        let a = fid("a.rs", 1, 10);
1016        let b = fid("b.rs", 1, 10);
1017        g.add_node(a.clone());
1018        g.add_node(b.clone());
1019        g.add_edge(a.clone(), b.clone(), f64::NAN);
1020        g.add_edge(a.clone(), b.clone(), f64::INFINITY);
1021        g.add_edge(a.clone(), b.clone(), -1.0);
1022        g.add_edge(a.clone(), b.clone(), 0.0);
1023        g.freeze();
1024
1025        assert!(collect_forward(&g, &a).is_empty());
1026        assert_eq!(g.edge_count(), 0);
1027    }
1028
1029    #[test]
1030    fn csr_round_trip() {
1031        let mut g = Graph::new();
1032        let a = fid("a.rs", 1, 10);
1033        let b = fid("b.rs", 1, 10);
1034        let c = fid("c.rs", 1, 10);
1035        g.add_node(a.clone());
1036        g.add_node(b.clone());
1037        g.add_node(c.clone());
1038        g.add_edge(a.clone(), b.clone(), 1.0);
1039        g.add_edge(b.clone(), c.clone(), 2.0);
1040
1041        let (fwd, _rev) = g.to_csr();
1042        assert_eq!(fwd.n, 3);
1043        assert_eq!(fwd.indptr.len(), 4);
1044        assert!(fwd.out_weight_sum[fwd.node_to_idx[&a] as usize] > 0.0);
1045    }
1046
1047    #[test]
1048    fn ego_graph_scores() {
1049        let mut g = Graph::new();
1050        let a = fid("a.rs", 1, 10);
1051        let b = fid("b.rs", 1, 10);
1052        let c = fid("c.rs", 1, 10);
1053        g.add_node(a.clone());
1054        g.add_node(b.clone());
1055        g.add_node(c.clone());
1056        g.add_edge(a.clone(), b.clone(), 1.0);
1057        g.add_edge(b.clone(), c.clone(), 1.0);
1058        g.freeze();
1059
1060        let mut seeds = FxHashSet::default();
1061        seeds.insert(a.clone());
1062        let scores = g.ego_graph(&seeds, 2);
1063
1064        let gamma = crate::config::scoring::EGO.per_hop_decay;
1065        assert!((scores[&a] - 1.0).abs() < 1e-9);
1066        assert!((scores[&b] - gamma).abs() < 1e-9);
1067        assert!((scores[&c] - gamma * gamma).abs() < 1e-9);
1068    }
1069
1070    #[test]
1071    fn ego_graph_sums_over_seeds() {
1072        let mut g = Graph::new();
1073        let a = fid("a.rs", 1, 10);
1074        let b = fid("b.rs", 1, 10);
1075        let v = fid("v.rs", 1, 10);
1076        g.add_node(a.clone());
1077        g.add_node(b.clone());
1078        g.add_node(v.clone());
1079        g.add_edge(a.clone(), v.clone(), 1.0);
1080        g.add_edge(b.clone(), v.clone(), 1.0);
1081        g.freeze();
1082
1083        let mut seeds = FxHashSet::default();
1084        seeds.insert(a.clone());
1085        seeds.insert(b.clone());
1086        let scores = g.ego_graph(&seeds, 1);
1087
1088        let gamma = crate::config::scoring::EGO.per_hop_decay;
1089        assert!(
1090            (scores[&v] - 2.0 * gamma).abs() < 1e-9,
1091            "v reached by 2 seeds at d=1 must score 2·γ; got {}",
1092            scores[&v]
1093        );
1094    }
1095
1096    #[test]
1097    fn ego_graph_uses_path_weight() {
1098        let mut g = Graph::new();
1099        let a = fid("a.rs", 1, 10);
1100        let b = fid("b.rs", 1, 10);
1101        let c = fid("c.rs", 1, 10);
1102        g.add_node(a.clone());
1103        g.add_node(b.clone());
1104        g.add_node(c.clone());
1105        g.add_edge(a.clone(), b.clone(), 0.7);
1106        g.add_edge(b.clone(), c.clone(), 0.4);
1107        g.freeze();
1108
1109        let mut seeds = FxHashSet::default();
1110        seeds.insert(a.clone());
1111        let scores = g.ego_graph(&seeds, 2);
1112
1113        let gamma = crate::config::scoring::EGO.per_hop_decay;
1114        assert!(
1115            (scores[&b] - gamma * 0.7).abs() < 1e-9,
1116            "1-hop weighted score = γ·0.7; got {}",
1117            scores[&b]
1118        );
1119        assert!(
1120            (scores[&c] - gamma * gamma * 0.7 * 0.4).abs() < 1e-9,
1121            "2-hop product-of-weights score = γ²·0.7·0.4; got {}",
1122            scores[&c]
1123        );
1124    }
1125
1126    #[test]
1127    fn ego_graph_empty() {
1128        let mut g = Graph::new();
1129        g.freeze();
1130        let seeds = FxHashSet::default();
1131        let scores = g.ego_graph(&seeds, 2);
1132        assert!(scores.is_empty());
1133    }
1134
1135    #[test]
1136    fn dedup_compact_edges_max_weight_first_category() {
1137        let mut edges = vec![
1138            CompactEdge {
1139                src: 0,
1140                dst: 1,
1141                weight: 0.5,
1142                category: EdgeCategory::Semantic,
1143                naming: false,
1144            },
1145            CompactEdge {
1146                src: 0,
1147                dst: 1,
1148                weight: 0.8,
1149                category: EdgeCategory::Similarity,
1150                naming: false,
1151            },
1152            CompactEdge {
1153                src: 2,
1154                dst: 1,
1155                weight: 0.3,
1156                category: EdgeCategory::Sibling,
1157                naming: false,
1158            },
1159        ];
1160        dedup_compact_edges(&mut edges);
1161        assert_eq!(edges.len(), 2);
1162        assert!((edges[0].weight - 0.8).abs() < 1e-9);
1163        assert_eq!(edges[0].category, EdgeCategory::Semantic);
1164        assert_eq!(edges[1].category, EdgeCategory::Sibling);
1165    }
1166
1167    #[test]
1168    fn cap_out_edges_per_source_keeps_top_k_ties_broken_by_ascending_dst() {
1169        let mut edges = vec![
1170            CompactEdge {
1171                src: 0,
1172                dst: 100,
1173                weight: 10.0,
1174                category: EdgeCategory::Semantic,
1175                naming: false,
1176            },
1177            CompactEdge {
1178                src: 0,
1179                dst: 50,
1180                weight: 8.0,
1181                category: EdgeCategory::Semantic,
1182                naming: false,
1183            },
1184            CompactEdge {
1185                src: 0,
1186                dst: 30,
1187                weight: 8.0,
1188                category: EdgeCategory::Semantic,
1189                naming: false,
1190            },
1191            CompactEdge {
1192                src: 0,
1193                dst: 70,
1194                weight: 8.0,
1195                category: EdgeCategory::Semantic,
1196                naming: false,
1197            },
1198            CompactEdge {
1199                src: 0,
1200                dst: 5,
1201                weight: 1.0,
1202                category: EdgeCategory::Semantic,
1203                naming: false,
1204            },
1205        ];
1206
1207        let stats = cap_out_edges_per_source(&mut edges, 2);
1208
1209        assert_eq!(stats.edges_before_cap, 5);
1210        assert_eq!(stats.edges_after_cap, 2);
1211        assert_eq!(stats.edges_dropped_by_cap, 3);
1212        assert_eq!(stats.nodes_capped, 1);
1213        assert_eq!(stats.max_out_edges_per_node, 2);
1214
1215        let survivors: Vec<(u32, f64)> = edges.iter().map(|e| (e.dst, e.weight)).collect();
1216        assert_eq!(
1217            survivors,
1218            vec![(100, 10.0), (30, 8.0)],
1219            "of the three weight-8.0 ties (dst 30, 50, 70), only the lowest dst (30) may survive \
1220             the second slot"
1221        );
1222    }
1223
1224    #[test]
1225    fn ranked_candidate_tie_break_matches_cap_out_edges_tie_break() {
1226        // Pins that `RankedCandidate::rank`'s dst tie-break agrees with
1227        // `cap_out_edges_per_source`'s sort tie-break: both must favor the
1228        // lower dst index on equal weight. A regression flipping either
1229        // side would silently change which edges the two-pass path keeps
1230        // relative to the materialized path.
1231        let candidates = [
1232            RankedCandidate {
1233                weight: 5.0,
1234                dst: 30,
1235                category: EdgeCategory::Generic,
1236                naming: false,
1237            },
1238            RankedCandidate {
1239                weight: 5.0,
1240                dst: 10,
1241                category: EdgeCategory::Generic,
1242                naming: false,
1243            },
1244            RankedCandidate {
1245                weight: 5.0,
1246                dst: 20,
1247                category: EdgeCategory::Generic,
1248                naming: false,
1249            },
1250        ];
1251
1252        let mut heap: SourceTopK = BinaryHeap::new();
1253        for &c in &candidates {
1254            push_bounded_top_k(&mut heap, c, 1);
1255        }
1256        let kept_by_heap: Vec<u32> = heap.into_iter().map(|Reverse(c)| c.dst).collect();
1257        assert_eq!(
1258            kept_by_heap,
1259            vec![10],
1260            "heap tie-break must keep the lowest dst"
1261        );
1262
1263        let mut edges: Vec<CompactEdge> = candidates
1264            .iter()
1265            .map(|c| CompactEdge {
1266                src: 0,
1267                dst: c.dst,
1268                weight: c.weight,
1269                category: c.category,
1270                naming: false,
1271            })
1272            .collect();
1273        cap_out_edges_per_source(&mut edges, 1);
1274        let kept_by_sort: Vec<u32> = edges.iter().map(|e| e.dst).collect();
1275
1276        assert_eq!(
1277            kept_by_heap, kept_by_sort,
1278            "RankedCandidate::rank tie-break must agree with cap_out_edges_per_source's sort \
1279             tie-break"
1280        );
1281    }
1282
1283    #[test]
1284    fn two_pass_heap_cap_matches_materialized_cap() {
1285        // Faithful to `edges::collect_capped_edges`'s two passes: pass 1
1286        // fixes a single canonical category per (src, dst) pair (first
1287        // builder, in registration order, to emit it) *before* pass 2's
1288        // per-builder bounded-heap capping runs. This test verifies the
1289        // correctness claim documented there -- per-builder top-K heaps,
1290        // merged + deduped + capped again, must select exactly the same
1291        // surviving edges as capping the fully materialized union
1292        // directly, including the case where the pair's canonical
1293        // category came from a builder whose own (locally weaker) copy
1294        // gets evicted from its own per-builder heap before the merge.
1295        let k = 3usize;
1296        type RawEdge = (u32, u32, f64, EdgeCategory);
1297
1298        let builder_logs: Vec<Vec<RawEdge>> = vec![
1299            vec![
1300                (0, 1, 9.0, EdgeCategory::Semantic),
1301                (0, 2, 9.0, EdgeCategory::Semantic),
1302                (0, 3, 7.0, EdgeCategory::Semantic),
1303                (0, 4, 1.0, EdgeCategory::Semantic),
1304                (1, 5, 4.0, EdgeCategory::Semantic),
1305            ],
1306            vec![
1307                (0, 4, 8.0, EdgeCategory::Structural),
1308                (0, 6, 2.0, EdgeCategory::Structural),
1309                (1, 5, 6.0, EdgeCategory::Structural),
1310                (1, 7, 3.0, EdgeCategory::Structural),
1311            ],
1312            vec![
1313                (0, 2, 9.0, EdgeCategory::Sibling),
1314                (0, 7, 9.0, EdgeCategory::Sibling),
1315                (1, 8, 0.5, EdgeCategory::Sibling),
1316            ],
1317        ];
1318
1319        let mut canonical_category: FxHashMap<(u32, u32), EdgeCategory> = FxHashMap::default();
1320        for log in &builder_logs {
1321            for &(src, dst, _w, cat) in log {
1322                canonical_category.entry((src, dst)).or_insert(cat);
1323            }
1324        }
1325        let canonicalize = |src: u32, dst: u32, fallback: EdgeCategory| {
1326            canonical_category
1327                .get(&(src, dst))
1328                .copied()
1329                .unwrap_or(fallback)
1330        };
1331
1332        let mut two_pass_edges: Vec<CompactEdge> = Vec::new();
1333        for log in &builder_logs {
1334            let mut per_source: FxHashMap<u32, SourceTopK> = FxHashMap::default();
1335            for &(src, dst, weight, cat) in log {
1336                let category = canonicalize(src, dst, cat);
1337                push_bounded_top_k(
1338                    per_source.entry(src).or_default(),
1339                    RankedCandidate {
1340                        weight,
1341                        dst,
1342                        category,
1343                        naming: false,
1344                    },
1345                    k,
1346                );
1347            }
1348            for (src, heap) in per_source {
1349                for Reverse(c) in heap {
1350                    two_pass_edges.push(CompactEdge {
1351                        src,
1352                        dst: c.dst,
1353                        weight: c.weight,
1354                        category: c.category,
1355                        naming: false,
1356                    });
1357                }
1358            }
1359        }
1360        dedup_compact_edges(&mut two_pass_edges);
1361        cap_out_edges_per_source(&mut two_pass_edges, k);
1362
1363        let mut materialized_edges: Vec<CompactEdge> = builder_logs
1364            .iter()
1365            .flatten()
1366            .map(|&(src, dst, weight, cat)| CompactEdge {
1367                src,
1368                dst,
1369                weight,
1370                category: canonicalize(src, dst, cat),
1371                naming: false,
1372            })
1373            .collect();
1374        dedup_compact_edges(&mut materialized_edges);
1375        cap_out_edges_per_source(&mut materialized_edges, k);
1376
1377        let normalize = |edges: &[CompactEdge]| -> Vec<(u32, u32, u64, EdgeCategory)> {
1378            let mut v: Vec<(u32, u32, u64, EdgeCategory)> = edges
1379                .iter()
1380                .map(|e| (e.src, e.dst, e.weight.to_bits(), e.category))
1381                .collect();
1382            v.sort_by_key(|t| (t.0, t.1));
1383            v
1384        };
1385
1386        let two_pass_normalized = normalize(&two_pass_edges);
1387        let materialized_normalized = normalize(&materialized_edges);
1388        assert_eq!(
1389            two_pass_normalized, materialized_normalized,
1390            "two-pass per-builder heap capping must be bit-identical to capping the fully \
1391             materialized edge set"
1392        );
1393        assert!(
1394            !materialized_normalized.is_empty() && materialized_normalized.len() <= 2 * k,
1395            "the cap must actually have fired in this fixture for the comparison to be meaningful"
1396        );
1397    }
1398
1399    fn plain_fragment(id: FragmentId) -> Fragment {
1400        Fragment {
1401            id,
1402            kind: crate::types::FragmentKind::Function,
1403            content: Arc::from(""),
1404            identifiers: FxHashSet::default(),
1405            token_count: 10,
1406            symbol_name: None,
1407        }
1408    }
1409
1410    #[test]
1411    fn assemble_graph_excludes_self_loops_and_non_finite_weights() {
1412        let a = fid("a.rs", 1, 10);
1413        let b = fid("b.rs", 1, 10);
1414        let frags = vec![plain_fragment(a.clone()), plain_fragment(b.clone())];
1415
1416        let mut edges: FxHashMap<(FragmentId, FragmentId), f64> = FxHashMap::default();
1417        let mut cats: FxHashMap<(FragmentId, FragmentId), EdgeCategory> = FxHashMap::default();
1418        edges.insert((a.clone(), b.clone()), 1.0);
1419        cats.insert((a.clone(), b.clone()), EdgeCategory::Semantic);
1420        edges.insert((a.clone(), a.clone()), 5.0);
1421        cats.insert((a.clone(), a.clone()), EdgeCategory::Semantic);
1422        edges.insert((b.clone(), a.clone()), f64::INFINITY);
1423        cats.insert((b.clone(), a.clone()), EdgeCategory::Semantic);
1424
1425        let graph = build_graph(&frags, edges, cats);
1426
1427        assert_eq!(
1428            graph.edge_count(),
1429            1,
1430            "only the valid a->b edge should survive"
1431        );
1432        assert_eq!(graph.categorized_edge_count(), 1);
1433        assert_eq!(graph.forward_edge_weight(&a, &b), Some(1.0));
1434        assert!(
1435            graph.forward_edge_weight(&a, &a).is_none(),
1436            "self-loop must be excluded from the live CSR"
1437        );
1438        assert!(
1439            graph.forward_edge_weight(&b, &a).is_none(),
1440            "infinite weight must be excluded from the live CSR"
1441        );
1442    }
1443
1444    #[test]
1445    fn category_table_stays_aligned_with_csr_when_cap_fires() {
1446        let hub = fid("hub.rs", 1, 10);
1447        let mut frags = vec![plain_fragment(hub.clone())];
1448        let mut edges: FxHashMap<(FragmentId, FragmentId), f64> = FxHashMap::default();
1449        let mut cats: FxHashMap<(FragmentId, FragmentId), EdgeCategory> = FxHashMap::default();
1450
1451        let n_leaves = DEFAULT_MAX_OUT_EDGES_PER_NODE + 10;
1452        for i in 0..n_leaves {
1453            let leaf = fid(&format!("leaf_{i}.rs"), 1, 10);
1454            frags.push(plain_fragment(leaf.clone()));
1455            edges.insert((hub.clone(), leaf.clone()), 1.0 + i as f64);
1456            cats.insert((hub.clone(), leaf.clone()), EdgeCategory::Semantic);
1457        }
1458
1459        let graph = build_graph(&frags, edges, cats);
1460
1461        assert!(
1462            graph.cap_stats.edges_dropped_by_cap > 0,
1463            "the cap must actually fire for this test to be meaningful"
1464        );
1465        assert_eq!(
1466            graph.categorized_edge_count(),
1467            graph.edge_count(),
1468            "the category table must be capped alongside the CSR"
1469        );
1470
1471        let mut checked = 0usize;
1472        graph.for_each_categorized_edge(|src, dst, _cat| {
1473            let w = graph.forward_edge_weight(src, dst);
1474            assert!(
1475                w.is_some_and(|w| w > 0.0),
1476                "every categorized edge must exist in the CSR with a positive weight; {src} -> {dst} did not"
1477            );
1478            checked += 1;
1479        });
1480        assert_eq!(checked, graph.edge_count());
1481    }
1482}