Skip to main content

ciri_toolkit/
scan2.rs

1//! Scan2: second-pass rescue, candidate validation, and FSJ counting.
2//!
3//! This stage consumes the Java-compatible BSJ output from Scan1, rebuilds
4//! the de-duplicated candidate indexes expected by Java CIRI3, and then revisits
5//! the input alignments to rescue additional support while counting FSJ evidence.
6
7use crate::circ_catalog::CircCatalogRecord;
8use crate::is_bsj_hg2::{report_scan2_hg_profile, IsBSJHg2};
9use crate::misd::misd;
10use crate::runtime::{
11    emit_perf_line, emit_trace_line, scan2_profile_enabled, should_trace_read, with_trace_hg2_scope,
12};
13use crate::utils::{
14    alignment_short_cs, bam_shard_count, bsj_is_summary_priority, bsj_payload_start,
15    clip_sequence_payload, local_clip_evidence_lines, part_path, reverse_complement,
16    AlignmentRecord,
17};
18use anyhow::{bail, Result};
19use indicatif::{ProgressBar, ProgressStyle};
20use memmap2::Mmap;
21use noodles::sam::{
22    self,
23    alignment::{
24        record::{
25            data::field::{Tag, Value},
26            Data as _, Sequence as _,
27        },
28        Record as _,
29    },
30};
31use rayon::prelude::*;
32use std::borrow::Cow;
33use std::collections::{HashMap, HashSet};
34use std::fmt::Write as FmtWrite;
35use std::fs::File;
36use std::io::{BufRead, BufReader, BufWriter, Seek, Write};
37use std::sync::atomic::{AtomicU64, Ordering};
38use std::sync::mpsc;
39use std::thread;
40use std::time::Duration;
41use std::time::Instant;
42
43const NON_BSJ_SEGMENT_MAPQ_THRES: i32 = 5;
44const NON_BSJ_SEGMENT_MAX_SPAN: i32 = 200000;
45
46/// Core logic for the second scan pass.
47pub struct Scan2 {
48    /// Minimum mapping quality for a read.
49    pub min_mapq_uni: i32,
50    /// Minimum linear range size for competition check.
51    pub linear_range_size_min: i32,
52    /// Index for BSJ candidates (indexed by Site1).
53    ///
54    /// Java parity note:
55    /// - Backed by de-duplicated `siteInfor` records (site1..sum_q), not raw BSJ1 lines.
56    /// - Candidate payload layout is `[site1, site2, strand, signal1, signal2, sum_q]`.
57    pub index1: HashMap<String, Vec<CandidateBreakpoint>>,
58    /// Index for BSJ candidates (indexed by Site2).
59    pub index2: HashMap<String, Vec<CandidateBreakpoint>>,
60    /// Java parity gate: existing buckets for Site1 index.
61    pub site_array1: HashMap<String, HashSet<i32>>,
62    /// Java parity gate: existing buckets for Site2 index.
63    pub site_array2: HashMap<String, HashSet<i32>>,
64    /// Map to store Forward Spliced Junction (FSJ) counts.
65    pub fsj_map: HashMap<String, i32>,
66    /// Memory limit per thread for Page Cache eviction.
67    pub mem_limit: u64,
68    /// Read IDs already assigned in Scan1 (Java parity: skip in Scan2).
69    pub scan1_ids: HashSet<String>,
70    /// Java Scan2 bucket size (`seqLen`) used for directional candidate traversal.
71    pub seq_len: i32,
72    /// Unique read IDs rescued by Scan2 and appended to the final BSJ file.
73    pub rescued_reads: usize,
74    /// Total unique BSJ-supporting read IDs in the final merged BSJ file.
75    pub final_bsj_reads: usize,
76}
77
78/// Temporary sidecar paths produced while Scan2 feeds the segments stage.
79///
80/// These files are not part of the CIRI3-compatible BSJ output contract. They
81/// let the CLI hand large read-level extension evidence directly to the
82/// post-Summary segments stage without first merging it into a second huge text
83/// file that would immediately be mmap-parsed again.
84#[derive(Default)]
85pub struct Scan2SegmentArtifacts {
86    /// Shard-local non-BSJ topology sidecars retained for direct segments input.
87    pub non_bsj_segment_evidence_paths: Vec<String>,
88}
89
90/// Returns a stable text payload for one optional `XA:Z` field.
91///
92/// Scan2 writes `*` for missing alternatives, but otherwise preserves raw BWA
93/// XA syntax so the post-Summary segments phase can parse alternatives without
94/// a second pass over BAM/SAM.
95fn xa_payload(raw: &str) -> &str {
96    if raw.is_empty() {
97        "*"
98    } else {
99        raw
100    }
101}
102
103/// Extracts a raw BWA `XA:Z` payload from one SAM record.
104///
105/// XA never feeds Scan2 rescue or FSJ counting. It is carried only as sidecar
106/// evidence for later circ-context segment chain selection.
107fn raw_xa_from_sam_record(record: &sam::Record) -> Result<String> {
108    let tag = Tag::new(b'X', b'A');
109    let data = record.data();
110    let Some(value) = data.get(&tag).transpose()? else {
111        return Ok(String::new());
112    };
113    let Value::String(raw) = value else {
114        return Ok(String::new());
115    };
116    let Ok(raw) = std::str::from_utf8(raw.as_ref()) else {
117        return Ok(String::new());
118    };
119    Ok(raw.to_string())
120}
121
122/// Extracts a raw BWA `XA:Z` payload from one BAM record.
123///
124/// Malformed optional tags are treated as absent because XA is a post-Summary
125/// representation hint, not required evidence for CIRI3 parity decisions.
126fn raw_xa_from_bam_record(record: &noodles::bam::Record) -> Result<String> {
127    let tag = Tag::new(b'X', b'A');
128    let data = record.data();
129    let Some(value) = data.get(&tag).transpose()? else {
130        return Ok(String::new());
131    };
132    let Value::String(raw) = value else {
133        return Ok(String::new());
134    };
135    let Ok(raw) = std::str::from_utf8(raw.as_ref()) else {
136        return Ok(String::new());
137    };
138    Ok(raw.to_string())
139}
140
141/// Returns a fixed-width reference window, padding out-of-range bases with `N`.
142///
143/// External BED catalogs can legally point close to chromosome ends. Scan2 needs
144/// two signal bases in the same payload slots as Scan1-derived candidates, so
145/// padding keeps the payload well-formed while the downstream sequence validator
146/// still decides whether any read truly supports the boundary.
147fn reference_window_or_n(seq: &str, start: i32, end: i32) -> String {
148    let bytes = seq.as_bytes();
149    let mut out = String::with_capacity((end - start).max(0) as usize);
150    for pos in start..end {
151        if pos >= 0 {
152            let idx = pos as usize;
153            if idx < bytes.len() {
154                out.push(bytes[idx] as char);
155                continue;
156            }
157        }
158        out.push('N');
159    }
160    out
161}
162
163#[derive(Clone)]
164pub struct CandidateBreakpoint {
165    /// Genomic coordinate used for bucket/range lookup.
166    pub site: i32,
167    /// Insertion order after Java-style de-duplication.
168    /// Used as a stable tie-breaker when two candidates share the same `site`.
169    pub order: usize,
170    /// Candidate payload in Java `siteInfor` layout:
171    /// `[site1, site2, strand, signal1, signal2, sum_q]`.
172    pub data: Vec<String>,
173}
174
175type OwnedAlignmentRecord = AlignmentRecord<'static>;
176type OwnedStandMap = HashMap<i32, (char, Cow<'static, str>)>;
177
178struct SamOwnedScan2Group {
179    read_id: String,
180    alignments: Vec<OwnedAlignmentRecord>,
181    all_alignments: Vec<OwnedAlignmentRecord>,
182    stand_map: OwnedStandMap,
183}
184
185#[derive(Clone, Copy)]
186struct NonBsjMsid {
187    kind: i32,
188    clip1: i32,
189    clip2: i32,
190    ref_len: i32,
191}
192
193impl NonBsjMsid {
194    fn invalid(ref_len: i32) -> Self {
195        Self {
196            kind: 0,
197            clip1: 0,
198            clip2: 0,
199            ref_len,
200        }
201    }
202}
203
204/// Advises the kernel that a processed `mmap` slice can be evicted from cache.
205///
206/// The offset is aligned manually because `madvise` requires page alignment on
207/// Linux. Without that alignment the call can fail silently from the pipeline's
208/// point of view and the full BAM/SAM page cache footprint keeps growing.
209fn advise_dontneed(mmap: &Mmap, offset: usize, len: usize) {
210    if len == 0 {
211        return;
212    }
213    let page_size = 4096usize;
214    let aligned_offset = (offset / page_size) * page_size;
215    let aligned_len = ((offset + len + page_size - 1) / page_size) * page_size - aligned_offset;
216    unsafe {
217        let ptr = mmap.as_ptr().add(aligned_offset);
218        libc::madvise(ptr as *mut libc::c_void, aligned_len, libc::MADV_DONTNEED);
219    }
220}
221
222/// Optional Scan2 profiler used when CLI `--perf` or legacy
223/// `CIRI_PROFILE_SCAN2=1` is enabled.
224///
225/// The counters stay intentionally coarse so the profiler can be left inside the
226/// candidate hot path without materially perturbing release timings.
227#[derive(Default)]
228pub(crate) struct Scan2Profile {
229    wall_total_ns: AtomicU64,
230    shard_total_ns: AtomicU64,
231    group_process_ns: AtomicU64,
232    validator_call_ns: AtomicU64,
233    display_ns: AtomicU64,
234    segment_sidecar_ns: AtomicU64,
235    non_bsj_sidecar_ns: AtomicU64,
236    write_ns: AtomicU64,
237    merge_ns: AtomicU64,
238    records: AtomicU64,
239    groups: AtomicU64,
240    candidate_checks: AtomicU64,
241    candidate_hits: AtomicU64,
242    segment_sidecar_rows: AtomicU64,
243    segment_sidecar_bytes: AtomicU64,
244    non_bsj_sidecar_rows: AtomicU64,
245    non_bsj_sidecar_bytes: AtomicU64,
246}
247
248/// Mate-level Scan1 claims used to gate Scan2 display-only rescue rows.
249///
250/// The display path is intentionally outside Summary counting, but it runs for
251/// every Scan2 read group. Storing claims as a compact per-read bitmask avoids
252/// allocating `read_id\tmate` strings in that hot path and lets fully claimed
253/// groups skip display reconstruction before any per-mate alignment indexing.
254struct Scan2DisplayClaims {
255    by_read: HashMap<String, u8>,
256}
257
258impl Scan2DisplayClaims {
259    /// Creates an empty display claim index.
260    fn new() -> Self {
261        Self {
262            by_read: HashMap::new(),
263        }
264    }
265
266    /// Records that Scan1 already owns one displayed mate for `read_id`.
267    fn insert(&mut self, read_id: &str, mate_label: &str) {
268        let bit = match mate_label {
269            "R1" => 0b01,
270            "R2" => 0b10,
271            _ => 0,
272        };
273        if bit != 0 {
274            *self.by_read.entry(read_id.to_string()).or_insert(0) |= bit;
275        }
276    }
277
278    /// Tests whether the display path should skip this mate.
279    fn contains_mate(&self, read_id: &str, mate_label: &str) -> bool {
280        let bit = match mate_label {
281            "R1" => 0b01,
282            "R2" => 0b10,
283            _ => 0,
284        };
285        bit != 0
286            && self
287                .by_read
288                .get(read_id)
289                .is_some_and(|mask| mask & bit != 0)
290    }
291
292    /// Tests whether both mates are already represented by Scan1 display rows.
293    fn contains_both_mates(&self, read_id: &str) -> bool {
294        self.by_read
295            .get(read_id)
296            .is_some_and(|mask| mask & 0b11 == 0b11)
297    }
298}
299
300impl Scan2Profile {
301    /// Checks whether release profiling is enabled for the current Scan2 run.
302    fn enabled_from_env() -> bool {
303        scan2_profile_enabled()
304    }
305
306    /// Emits the aggregated Scan2 timing summary.
307    fn report(&self) {
308        let wall_total_ns = self.wall_total_ns.load(Ordering::Relaxed);
309        let shard_total_ns = self.shard_total_ns.load(Ordering::Relaxed);
310        let group_process_ns = self.group_process_ns.load(Ordering::Relaxed);
311        let validator_call_ns = self.validator_call_ns.load(Ordering::Relaxed);
312        let display_ns = self.display_ns.load(Ordering::Relaxed);
313        let segment_sidecar_ns = self.segment_sidecar_ns.load(Ordering::Relaxed);
314        let non_bsj_sidecar_ns = self.non_bsj_sidecar_ns.load(Ordering::Relaxed);
315        let write_ns = self.write_ns.load(Ordering::Relaxed);
316        let merge_ns = self.merge_ns.load(Ordering::Relaxed);
317        let records = self.records.load(Ordering::Relaxed);
318        let groups = self.groups.load(Ordering::Relaxed);
319        let candidate_checks = self.candidate_checks.load(Ordering::Relaxed);
320        let candidate_hits = self.candidate_hits.load(Ordering::Relaxed);
321        let segment_sidecar_rows = self.segment_sidecar_rows.load(Ordering::Relaxed);
322        let segment_sidecar_bytes = self.segment_sidecar_bytes.load(Ordering::Relaxed);
323        let non_bsj_sidecar_rows = self.non_bsj_sidecar_rows.load(Ordering::Relaxed);
324        let non_bsj_sidecar_bytes = self.non_bsj_sidecar_bytes.load(Ordering::Relaxed);
325        let accounted_ns = group_process_ns
326            .saturating_add(display_ns)
327            .saturating_add(segment_sidecar_ns)
328            .saturating_add(non_bsj_sidecar_ns)
329            .saturating_add(write_ns);
330        let other_shard_ns = shard_total_ns.saturating_sub(accounted_ns);
331        let pct = |part: u64, whole: u64| -> f64 {
332            if whole == 0 {
333                0.0
334            } else {
335                part as f64 * 100.0 / whole as f64
336            }
337        };
338        emit_perf_line(&format!(
339            "[PROFILE_SCAN2] wall_ms={:.3} shard_work_ms={:.3} merge_ms={:.3} records={} groups={} candidate_checks={} candidate_hits={}",
340            wall_total_ns as f64 / 1_000_000.0,
341            shard_total_ns as f64 / 1_000_000.0,
342            merge_ns as f64 / 1_000_000.0,
343            records,
344            groups,
345            candidate_checks,
346            candidate_hits,
347        ));
348        emit_perf_line(&format!(
349            "[PROFILE_SCAN2] shard_breakdown_ms group_process={:.3} ({:.1}%) validator={:.3} ({:.1}% of group) display={:.3} ({:.1}%) segment_sidecar={:.3} ({:.1}%) non_bsj_sidecar={:.3} ({:.1}%) write={:.3} ({:.1}%) other={:.3} ({:.1}%)",
350            group_process_ns as f64 / 1_000_000.0,
351            pct(group_process_ns, shard_total_ns),
352            validator_call_ns as f64 / 1_000_000.0,
353            pct(validator_call_ns, group_process_ns),
354            display_ns as f64 / 1_000_000.0,
355            pct(display_ns, shard_total_ns),
356            segment_sidecar_ns as f64 / 1_000_000.0,
357            pct(segment_sidecar_ns, shard_total_ns),
358            non_bsj_sidecar_ns as f64 / 1_000_000.0,
359            pct(non_bsj_sidecar_ns, shard_total_ns),
360            write_ns as f64 / 1_000_000.0,
361            pct(write_ns, shard_total_ns),
362            other_shard_ns as f64 / 1_000_000.0,
363            pct(other_shard_ns, shard_total_ns),
364        ));
365        emit_perf_line(&format!(
366            "[PROFILE_SCAN2_SIDECAR] bsj_rows={} bsj_bytes={} non_bsj_rows={} non_bsj_bytes={}",
367            segment_sidecar_rows,
368            segment_sidecar_bytes,
369            non_bsj_sidecar_rows,
370            non_bsj_sidecar_bytes,
371        ));
372    }
373}
374
375impl Scan2 {
376    /// Mirrors Java `String.hashCode` so Scan2 can rebuild Java-like `HashSet`
377    /// iteration order when constructing the candidate index.
378    #[inline]
379    fn java_string_hash(s: &str) -> i32 {
380        // Java String.hashCode(): h = 31*h + ch (32-bit overflow).
381        let mut h: i32 = 0;
382        for ch in s.chars() {
383            h = h.wrapping_mul(31).wrapping_add(ch as i32);
384        }
385        h
386    }
387
388    /// Applies the same spread step Java `HashMap` uses before bucket selection.
389    #[inline]
390    fn java_hash_spread(h: i32) -> u32 {
391        let x = h as u32;
392        x ^ (x >> 16)
393    }
394
395    /// Computes the minimum Java-style `HashSet` backing capacity for `n` items.
396    #[inline]
397    fn java_hashset_capacity(n: usize) -> usize {
398        // Java HashMap defaults: initial capacity 16, load factor 0.75.
399        // Capacity is power-of-two and grows so that size <= cap * 0.75.
400        let mut cap = 16usize;
401        while n > ((cap as f64) * 0.75f64) as usize {
402            cap = cap.saturating_mul(2);
403        }
404        cap.max(1)
405    }
406
407    /// Creates a Scan2 runner with CIRI3-compatible thresholds and bucket size.
408    pub fn new(min_mapq_uni: i32, linear_range_size_min: i32, seq_len: i32) -> Self {
409        Self {
410            min_mapq_uni,
411            linear_range_size_min,
412            index1: HashMap::new(),
413            index2: HashMap::new(),
414            site_array1: HashMap::new(),
415            site_array2: HashMap::new(),
416            fsj_map: HashMap::new(),
417            mem_limit: 2 * 1024 * 1024 * 1024,
418            scan1_ids: HashSet::new(),
419            seq_len,
420            rescued_reads: 0,
421            final_bsj_reads: 0,
422        }
423    }
424
425    /// Binary-search lower bound used by the sorted candidate buckets.
426    fn lower_bound(list: &[CandidateBreakpoint], key: i32) -> usize {
427        let mut lo = 0usize;
428        let mut hi = list.len();
429        while lo < hi {
430            let mid = (lo + hi) / 2;
431            if list[mid].site < key {
432                lo = mid + 1;
433            } else {
434                hi = mid;
435            }
436        }
437        lo
438    }
439
440    /// Returns the candidate slice covering one Java bucket.
441    ///
442    /// The index vectors are sorted once during `build_index`, so hot-path bucket
443    /// scans can stay branch-light and avoid re-filtering the whole chromosome list.
444    fn bucket_range(list: &[CandidateBreakpoint], bucket: i32, bucket_size: i32) -> (usize, usize) {
445        if bucket < 0 || bucket_size <= 0 {
446            return (0, 0);
447        }
448        let start = bucket.saturating_mul(bucket_size);
449        let end_exclusive = (bucket + 1).saturating_mul(bucket_size);
450        let l = Self::lower_bound(list, start);
451        let r = Self::lower_bound(list, end_exclusive);
452        (l, r)
453    }
454
455    /// Overrides the per-thread page-cache eviction threshold.
456    pub fn set_mem_limit(&mut self, limit: u64) {
457        self.mem_limit = limit;
458    }
459
460    /// Drops Scan2-only indexing state once rescue is complete.
461    ///
462    /// Summary only needs the merged `fsj_map`, so the large candidate indexes and
463    /// Scan1 read-id filter can be released explicitly before the final stage to
464    /// keep peak RSS lower on whole-genome inputs.
465    pub fn release_working_set(&mut self) {
466        self.index1.clear();
467        self.index1.shrink_to_fit();
468        self.index2.clear();
469        self.index2.shrink_to_fit();
470        self.site_array1.clear();
471        self.site_array1.shrink_to_fit();
472        self.site_array2.clear();
473        self.site_array2.shrink_to_fit();
474        self.scan1_ids.clear();
475        self.scan1_ids.shrink_to_fit();
476    }
477
478    /// Returns the temporary path used for one shard-local FSJ spill file.
479    #[inline]
480    fn shard_fsj_path(output_fsj: &str, shard_idx: usize) -> String {
481        part_path(output_fsj, shard_idx)
482    }
483
484    /// Writes one shard-local FSJ map to disk so the main thread can merge it
485    /// later without retaining every shard map in memory at once.
486    fn write_fsj_shard(path: &str, local_fsj: &HashMap<String, i32>) -> Result<()> {
487        let mut writer = BufWriter::with_capacity(256 * 1024, File::create(path)?);
488        for (key, count) in local_fsj {
489            writeln!(writer, "{}\t{}", key, count)?;
490        }
491        writer.flush()?;
492        Ok(())
493    }
494
495    /// Loads Scan1 BSJ1 output and builds the de-duplicated Scan2 candidate index.
496    ///
497    /// This function intentionally mirrors Java's unique-site indexing semantics:
498    /// the traversal order of candidates must match Java, because Scan2 returns on
499    /// the first valid non-`2` tag it encounters.
500    ///
501    /// Two details here are easy to "simplify" incorrectly:
502    /// - Scan2 de-duplicates by circ site payload before building the index, so
503    ///   repeated BSJ1 rows for the same circ do not create extra candidates.
504    /// - The insertion order is preserved through `order`, then reused after the
505    ///   final per-bucket sort, because Java's `HashSet -> ArrayList` path still
506    ///   leaves a stable first-hit order once the same inputs are replayed.
507    pub fn build_index(&mut self, bsj1_file: &str) -> Result<()> {
508        // Java parity: Scan2 index is built from unique circ sites, not all BSJ1 rows.
509        // Equivalent Java flow:
510        //   BSJ1 -> scan1IdMap + chrCircSiteMap(HashSet) -> siteArray/siteMap.
511        let file = File::open(bsj1_file)?;
512        let reader = BufReader::new(file);
513        let mut chr_circ_site_seen: HashMap<String, HashSet<String>> = HashMap::new();
514        let mut chr_circ_site_insertion: HashMap<String, Vec<String>> = HashMap::new();
515        for line_res in reader.lines() {
516            let line = line_res?;
517            let p: Vec<&str> = line.split('\t').collect();
518            if p.len() < 10 || !bsj_is_summary_priority(&p) {
519                continue;
520            }
521            let payload_start = bsj_payload_start(&p);
522            if p.len() <= payload_start + 8 {
523                continue;
524            }
525            self.scan1_ids.insert(p[0].to_string());
526            let chr = p[payload_start + 2].to_string();
527            let site_infor = p[payload_start + 3..].join("\t");
528            let seen = chr_circ_site_seen
529                .entry(chr.clone())
530                .or_insert_with(HashSet::new);
531            if seen.insert(site_infor.clone()) {
532                chr_circ_site_insertion
533                    .entry(chr)
534                    .or_insert_with(Vec::new)
535                    .push(site_infor);
536            }
537        }
538
539        let mut order_counter: usize = 0;
540        for (chr, insertion_list) in chr_circ_site_insertion {
541            // Java parity: emulate HashSet iteration order:
542            // bucket index ascending, and insertion order within each bucket.
543            let cap = Self::java_hashset_capacity(insertion_list.len());
544            let mut buckets: Vec<Vec<usize>> = vec![Vec::new(); cap];
545            for (idx, s) in insertion_list.iter().enumerate() {
546                let spread = Self::java_hash_spread(Self::java_string_hash(s));
547                let bucket_idx = (spread as usize) & (cap - 1);
548                buckets[bucket_idx].push(idx);
549            }
550
551            for bucket in buckets {
552                for idx in bucket {
553                    let arr: Vec<String> = insertion_list[idx]
554                        .split('\t')
555                        .map(|s| s.to_string())
556                        .collect();
557                    if arr.len() < 6 {
558                        continue;
559                    }
560                    let site1 = arr[0].parse().unwrap_or(0);
561                    let site2 = arr[1].parse().unwrap_or(0);
562                    self.fsj_map
563                        .entry(format!("{}\t{}\t{}", chr, arr[0], arr[1]))
564                        .or_insert(0);
565                    self.index1
566                        .entry(chr.clone())
567                        .or_insert_with(Vec::new)
568                        .push(CandidateBreakpoint {
569                            site: site1,
570                            order: order_counter,
571                            data: arr.clone(),
572                        });
573                    self.index2
574                        .entry(chr.clone())
575                        .or_insert_with(Vec::new)
576                        .push(CandidateBreakpoint {
577                            site: site2,
578                            order: order_counter,
579                            data: arr,
580                        });
581                    let bucket_size = self.seq_len.max(1);
582                    self.site_array1
583                        .entry(chr.clone())
584                        .or_insert_with(HashSet::new)
585                        .insert(site1 / bucket_size);
586                    self.site_array2
587                        .entry(chr.clone())
588                        .or_insert_with(HashSet::new)
589                        .insert(site2 / bucket_size);
590                    order_counter += 1;
591                }
592            }
593        }
594        for list in self.index1.values_mut() {
595            list.sort_by_key(|x| (x.site, x.order));
596        }
597        for list in self.index2.values_mut() {
598            list.sort_by_key(|x| (x.site, x.order));
599        }
600        Ok(())
601    }
602
603    /// Adds cohort-level external BSJ sites to the Scan2 candidate indexes.
604    ///
605    /// `--circ` candidates come from BED6 catalog coordinates rather than from
606    /// Scan1 read evidence. They are therefore appended after the Java-parity
607    /// Scan1 candidates: default runs stay unchanged, while a second-pass run can
608    /// test additional cohort-supported BSJ boundaries against this sample's own
609    /// reads. The left/right signal fields are reconstructed from the reference
610    /// so the existing `is_bsj_hg2` validator can remain the single source of
611    /// sequence-match decisions.
612    pub fn add_external_circ_candidates(
613        &mut self,
614        candidates: &[CircCatalogRecord],
615        reference: &HashMap<String, String>,
616    ) -> Result<usize> {
617        let mut seen = HashSet::new();
618        let mut next_order = 0usize;
619        for (chrom, list) in &self.index1 {
620            for candidate in list {
621                next_order = next_order.max(candidate.order + 1);
622                if candidate.data.len() >= 3 {
623                    seen.insert(format!(
624                        "{}\t{}\t{}\t{}",
625                        chrom, candidate.data[0], candidate.data[1], candidate.data[2]
626                    ));
627                }
628            }
629        }
630
631        let mut added = 0usize;
632        let bucket_size = self.seq_len.max(1);
633        for candidate in candidates {
634            if candidate.start < 1 || candidate.end < candidate.start {
635                bail!(
636                    "invalid external circ candidate {}:{}-{}",
637                    candidate.chrom,
638                    candidate.start,
639                    candidate.end
640                );
641            }
642            let Some(chr_seq) = reference.get(&candidate.chrom) else {
643                bail!(
644                    "external circ candidate {}:{}-{} uses chromosome absent from reference",
645                    candidate.chrom,
646                    candidate.start,
647                    candidate.end
648                );
649            };
650            if candidate.end as usize > chr_seq.len() {
651                bail!(
652                    "external circ candidate {}:{}-{} exceeds reference length {}",
653                    candidate.chrom,
654                    candidate.start,
655                    candidate.end,
656                    chr_seq.len()
657                );
658            }
659            let dedup_key = format!(
660                "{}\t{}\t{}\t{}",
661                candidate.chrom, candidate.start, candidate.end, candidate.strand
662            );
663            if !seen.insert(dedup_key) {
664                continue;
665            }
666            let signal_left =
667                reference_window_or_n(chr_seq, candidate.start - 3, candidate.start - 1);
668            let signal_right = reference_window_or_n(chr_seq, candidate.end, candidate.end + 2);
669            let payload = vec![
670                candidate.start.to_string(),
671                candidate.end.to_string(),
672                candidate.strand.clone(),
673                signal_left,
674                signal_right,
675                "1".to_string(),
676            ];
677            self.fsj_map
678                .entry(format!(
679                    "{}\t{}\t{}",
680                    candidate.chrom, candidate.start, candidate.end
681                ))
682                .or_insert(0);
683            self.index1
684                .entry(candidate.chrom.clone())
685                .or_insert_with(Vec::new)
686                .push(CandidateBreakpoint {
687                    site: candidate.start,
688                    order: next_order,
689                    data: payload.clone(),
690                });
691            self.index2
692                .entry(candidate.chrom.clone())
693                .or_insert_with(Vec::new)
694                .push(CandidateBreakpoint {
695                    site: candidate.end,
696                    order: next_order,
697                    data: payload,
698                });
699            self.site_array1
700                .entry(candidate.chrom.clone())
701                .or_insert_with(HashSet::new)
702                .insert(candidate.start / bucket_size);
703            self.site_array2
704                .entry(candidate.chrom.clone())
705                .or_insert_with(HashSet::new)
706                .insert(candidate.end / bucket_size);
707            next_order += 1;
708            added += 1;
709        }
710
711        for list in self.index1.values_mut() {
712            list.sort_by_key(|x| (x.site, x.order));
713        }
714        for list in self.index2.values_mut() {
715            list.sort_by_key(|x| (x.site, x.order));
716        }
717        Ok(added)
718    }
719
720    /// Builds a mate-level display index from the post-Summary Scan1 display rows.
721    ///
722    /// This path is intentionally isolated from the parity index above. It keeps
723    /// the user-facing `.bsj` display self-consistent without feeding any of the
724    /// extra mate-level rows back into `.out`.
725    fn build_display_index(&mut self, bsj1_display_file: &str) -> Result<Scan2DisplayClaims> {
726        self.index1.clear();
727        self.index2.clear();
728        self.site_array1.clear();
729        self.site_array2.clear();
730        let file = File::open(bsj1_display_file)?;
731        let reader = BufReader::new(file);
732        let mut claims = Scan2DisplayClaims::new();
733        let mut chr_circ_site_seen: HashMap<String, HashSet<String>> = HashMap::new();
734        let mut chr_circ_site_insertion: HashMap<String, Vec<String>> = HashMap::new();
735        for line_res in reader.lines() {
736            let line = line_res?;
737            let p: Vec<&str> = line.split('\t').collect();
738            if p.len() < 10 {
739                continue;
740            }
741            let payload_start = bsj_payload_start(&p);
742            let mate_label = if payload_start == 3 { p[1] } else { "NA" };
743            if p.len() <= payload_start + 8 {
744                continue;
745            }
746            claims.insert(p[0], mate_label);
747            let chr = p[payload_start + 2].to_string();
748            let site_infor = p[payload_start + 3..].join("\t");
749            let seen = chr_circ_site_seen
750                .entry(chr.clone())
751                .or_insert_with(HashSet::new);
752            if seen.insert(site_infor.clone()) {
753                chr_circ_site_insertion
754                    .entry(chr)
755                    .or_insert_with(Vec::new)
756                    .push(site_infor);
757            }
758        }
759
760        let mut order_counter: usize = 0;
761        for (chr, sites) in chr_circ_site_insertion {
762            let mut list1 = Vec::new();
763            let mut list2 = Vec::new();
764            let mut set1 = HashSet::new();
765            let mut set2 = HashSet::new();
766            for site in sites {
767                let data: Vec<String> = site.split('\t').map(|s| s.to_string()).collect();
768                if data.len() != 6 {
769                    continue;
770                }
771                let site1 = data[0].parse::<i32>().unwrap_or(0);
772                let site2 = data[1].parse::<i32>().unwrap_or(0);
773                let num1 = site1 / self.seq_len.max(1);
774                let num2 = site2 / self.seq_len.max(1);
775                set1.insert(num1);
776                set2.insert(num2);
777                list1.push(CandidateBreakpoint {
778                    site: site1,
779                    order: order_counter,
780                    data: data.clone(),
781                });
782                list2.push(CandidateBreakpoint {
783                    site: site2,
784                    order: order_counter,
785                    data,
786                });
787                order_counter += 1;
788            }
789            list1.sort_by_key(|c| (c.site, c.order));
790            list2.sort_by_key(|c| (c.site, c.order));
791            self.index1.insert(chr.clone(), list1);
792            self.index2.insert(chr.clone(), list2);
793            self.site_array1.insert(chr.clone(), set1);
794            self.site_array2.insert(chr, set2);
795        }
796        Ok(claims)
797    }
798
799    /// Collects FSJ keys overlapped by a linear alignment span.
800    ///
801    /// This mirrors Java `GetFSJClass.getFSJ(...)`: only the two buckets touched
802    /// by `[start_tem, end_tem]` are scanned, with reverse traversal on `num1`
803    /// and forward traversal on `num2`. The broader lower-bound scan is cheaper
804    /// to write but not Java-compatible, and it over-counts FSJs on chr1 while
805    /// leaving BSJ rescue unchanged.
806    ///
807    /// `style` is kept in Java's original encoding (`0`, `1`, `-1`, `10`) on
808    /// purpose. Those magic-looking values control which bucket edge is treated
809    /// as inclusive for full-match, MS, SM, and SMS alignments, and normalizing
810    /// them into a more abstract enum made earlier parity checks harder to audit.
811    fn collect_fsj_keys_in_range(
812        &self,
813        chr: &str,
814        start_tem: i32,
815        end_tem: i32,
816        style: i32,
817        out: &mut HashSet<String>,
818    ) {
819        let bucket_size = self.seq_len.max(1);
820        let num1 = start_tem / bucket_size;
821        let num2 = end_tem / bucket_size;
822        if let Some(list1) = self.index1.get(chr) {
823            if self.site_array1.get(chr).is_some_and(|s| s.contains(&num1)) {
824                let (l1, r1) = Self::bucket_range(list1, num1, bucket_size);
825                for idx in (l1..r1).rev() {
826                    let cand = &list1[idx];
827                    if cand.site >= start_tem {
828                        if cand.site <= end_tem || style == 10 || style == 1 {
829                            out.insert(format!("{}\t{}\t{}", chr, cand.data[0], cand.data[1]));
830                        }
831                    } else {
832                        break;
833                    }
834                }
835            }
836            if num2 != num1 && self.site_array1.get(chr).is_some_and(|s| s.contains(&num2)) {
837                let (l2, r2) = Self::bucket_range(list1, num2, bucket_size);
838                for idx in l2..r2 {
839                    let cand = &list1[idx];
840                    if cand.site <= end_tem {
841                        if cand.site >= start_tem || style == 10 || style == -1 {
842                            out.insert(format!("{}\t{}\t{}", chr, cand.data[0], cand.data[1]));
843                        }
844                    } else {
845                        break;
846                    }
847                }
848            }
849        }
850        if let Some(list2) = self.index2.get(chr) {
851            if self.site_array2.get(chr).is_some_and(|s| s.contains(&num1)) {
852                let (l1, r1) = Self::bucket_range(list2, num1, bucket_size);
853                for idx in (l1..r1).rev() {
854                    let cand = &list2[idx];
855                    if cand.site >= start_tem {
856                        if cand.site <= end_tem || style == 10 || style == 1 {
857                            out.insert(format!("{}\t{}\t{}", chr, cand.data[0], cand.data[1]));
858                        }
859                    } else {
860                        break;
861                    }
862                }
863            }
864            if num2 != num1 && self.site_array2.get(chr).is_some_and(|s| s.contains(&num2)) {
865                let (l2, r2) = Self::bucket_range(list2, num2, bucket_size);
866                for idx in l2..r2 {
867                    let cand = &list2[idx];
868                    if cand.site <= end_tem {
869                        if cand.site >= start_tem || style == 10 || style == -1 {
870                            out.insert(format!("{}\t{}\t{}", chr, cand.data[0], cand.data[1]));
871                        }
872                    } else {
873                        break;
874                    }
875                }
876            }
877        }
878    }
879
880    /// Dispatches to the SAM or BAM Scan2 implementation.
881    pub fn run(
882        &mut self,
883        sam_file: &str,
884        output_bsj2: &str,
885        output_fsj: &str,
886        chr_tcga_map: &HashMap<String, String>,
887    ) -> Result<()> {
888        self.run_with_display(sam_file, output_bsj2, output_fsj, None, None, chr_tcga_map)
889    }
890
891    /// Runs Scan2 and optionally emits mate-level `priority=0` rescue rows into
892    /// the same `.bsj2` stream during the scan.
893    pub fn run_with_display(
894        &mut self,
895        sam_file: &str,
896        output_bsj2: &str,
897        output_fsj: &str,
898        scan1_display_path: Option<&str>,
899        display_output_path: Option<&str>,
900        chr_tcga_map: &HashMap<String, String>,
901    ) -> Result<()> {
902        self.run_with_display_and_segments(
903            sam_file,
904            output_bsj2,
905            output_fsj,
906            scan1_display_path,
907            display_output_path,
908            None,
909            None,
910            chr_tcga_map,
911        )
912        .map(|_| ())
913    }
914
915    /// Runs Scan2 while also writing a segments-evidence sidecar for rescued BSJ reads.
916    ///
917    /// The extra sidecar is independent from `.bsj2` and FSJ counting. It records
918    /// mapper alignment blocks for read groups with Scan2 BSJ evidence so the
919    /// post-Summary segments stage can reconstruct confirmed BSJ rows without
920    /// scanning the input alignment file again.
921    pub fn run_with_display_and_segments(
922        &mut self,
923        sam_file: &str,
924        output_bsj2: &str,
925        output_fsj: &str,
926        scan1_display_path: Option<&str>,
927        display_output_path: Option<&str>,
928        segments_output_path: Option<&str>,
929        non_bsj_segments_output_path: Option<&str>,
930        chr_tcga_map: &HashMap<String, String>,
931    ) -> Result<Scan2SegmentArtifacts> {
932        use crate::sam_bam::{detect_format, InputFormat};
933        let display_scan2 = if let Some(scan1_path) = scan1_display_path {
934            let mut helper =
935                Scan2::new(self.min_mapq_uni, self.linear_range_size_min, self.seq_len);
936            let claims = helper.build_display_index(scan1_path)?;
937            Some((helper, claims))
938        } else {
939            None
940        };
941        let format = detect_format(sam_file)?;
942        match format {
943            InputFormat::Sam => self.run_sam_with_display(
944                sam_file,
945                output_bsj2,
946                output_fsj,
947                display_output_path,
948                segments_output_path,
949                non_bsj_segments_output_path,
950                display_scan2.as_ref(),
951                chr_tcga_map,
952            ),
953            InputFormat::Bam => self.run_bam_with_display(
954                sam_file,
955                output_bsj2,
956                output_fsj,
957                display_output_path,
958                segments_output_path,
959                non_bsj_segments_output_path,
960                display_scan2.as_ref(),
961                chr_tcga_map,
962            ),
963        }
964    }
965
966    /// Runs Scan2 on SAM input.
967    ///
968    /// The SAM path preserves queryname groups before batch-parallel processing,
969    /// matching the BAM path's read-group semantics while avoiding shard-boundary
970    /// FSJ duplication.
971    pub fn run_sam(
972        &mut self,
973        sam_file: &str,
974        output_bsj2: &str,
975        output_fsj: &str,
976        chr_tcga_map: &HashMap<String, String>,
977    ) -> Result<()> {
978        self.run_sam_with_display(
979            sam_file,
980            output_bsj2,
981            output_fsj,
982            None,
983            None,
984            None,
985            None,
986            chr_tcga_map,
987        )
988        .map(|_| ())
989    }
990
991    fn run_sam_with_display(
992        &mut self,
993        sam_file: &str,
994        output_bsj2: &str,
995        output_fsj: &str,
996        display_output_path: Option<&str>,
997        segments_output_path: Option<&str>,
998        non_bsj_segments_output_path: Option<&str>,
999        display_scan2: Option<&(Scan2, Scan2DisplayClaims)>,
1000        chr_tcga_map: &HashMap<String, String>,
1001    ) -> Result<Scan2SegmentArtifacts> {
1002        let run_started = Instant::now();
1003        let profile = if Scan2Profile::enabled_from_env() {
1004            Some(Scan2Profile::default())
1005        } else {
1006            None
1007        };
1008        let profile_ref = profile.as_ref();
1009        let file_size = std::fs::metadata(sam_file)?.len();
1010
1011        let pb = ProgressBar::new(file_size as u64);
1012        pb.set_style(ProgressStyle::default_bar().template("{spinner:.green} [{elapsed_precise}] [{bar:40.cyan/blue}] {bytes}/{total_bytes} {percent:>3}% ({eta}) {msg}")?.progress_chars("#>-"));
1013        pb.set_message("");
1014        let shard_out = part_path(output_bsj2, 0);
1015        let display_shard_out = display_output_path.map(|path| part_path(path, 0));
1016        let segments_shard_out = segments_output_path.map(|path| part_path(path, 0));
1017        let non_bsj_segments_shard_out =
1018            non_bsj_segments_output_path.map(|path| part_path(path, 0));
1019        let fsj_out = Self::shard_fsj_path(output_fsj, 0);
1020        self.process_sam_file_to_file(
1021            sam_file,
1022            chr_tcga_map,
1023            &pb,
1024            &shard_out,
1025            &fsj_out,
1026            display_scan2,
1027            display_shard_out.as_deref(),
1028            segments_shard_out.as_deref(),
1029            non_bsj_segments_shard_out.as_deref(),
1030            profile_ref,
1031        )?;
1032
1033        pb.set_style(ProgressStyle::default_bar().template("{spinner:.green} [{elapsed_precise}] [{bar:40.cyan/blue}] {bytes}/{total_bytes} {msg}")?.progress_chars("#>-"));
1034        pb.finish_with_message("Completed");
1035        let merge_started = Instant::now();
1036        self.merge_shards_and_fsj(output_bsj2, output_fsj, 1)?;
1037        if let Some(profile) = profile_ref {
1038            profile
1039                .merge_ns
1040                .fetch_add(merge_started.elapsed().as_nanos() as u64, Ordering::Relaxed);
1041            profile
1042                .wall_total_ns
1043                .fetch_add(run_started.elapsed().as_nanos() as u64, Ordering::Relaxed);
1044            profile.report();
1045            report_scan2_hg_profile();
1046        }
1047        if let Some(path) = display_output_path {
1048            self.merge_display_shards(path, 1)?;
1049        }
1050        if let Some(path) = segments_output_path {
1051            self.merge_segment_shards(path, 1)?;
1052        }
1053        let non_bsj_segment_evidence_paths = non_bsj_segments_output_path
1054            .map(|path| vec![part_path(path, 0)])
1055            .unwrap_or_default();
1056        Ok(Scan2SegmentArtifacts {
1057            non_bsj_segment_evidence_paths,
1058        })
1059    }
1060
1061    /// Merges per-shard Scan2 BSJ2 outputs and FSJ spill files.
1062    ///
1063    /// `rescued_reads` and `final_bsj_reads` are stage-level accounting only.
1064    /// The user-facing "final BSJ reads" summary is recomputed later from the
1065    /// clustered `.out`, because Summary can still merge circ families without
1066    /// changing the raw `.bsj1/.bsj2` membership.
1067    fn merge_shards_and_fsj(
1068        &mut self,
1069        output_bsj2: &str,
1070        output_fsj: &str,
1071        num_threads: usize,
1072    ) -> Result<()> {
1073        let mut writer = BufWriter::with_capacity(1024 * 1024, File::create(output_bsj2)?);
1074        let mut rescued_ids = HashSet::new();
1075        for i in 0..num_threads {
1076            let shard_path = part_path(output_bsj2, i);
1077            if let Ok(shard_file) = File::open(&shard_path) {
1078                let mut shard_reader = BufReader::new(shard_file);
1079                let mut line = String::new();
1080                while shard_reader.read_line(&mut line)? != 0 {
1081                    let trimmed = line.trim_end();
1082                    if !trimmed.is_empty() {
1083                        let parts: Vec<&str> = trimmed.split('\t').collect();
1084                        if !parts.is_empty() && bsj_is_summary_priority(&parts) {
1085                            rescued_ids.insert(parts[0].to_string());
1086                        }
1087                        writeln!(writer, "{}", trimmed)?;
1088                    }
1089                    line.clear();
1090                }
1091            }
1092            let _ = std::fs::remove_file(shard_path);
1093        }
1094        for i in 0..num_threads {
1095            let fsj_path = Self::shard_fsj_path(output_fsj, i);
1096            if let Ok(fsj_file) = File::open(&fsj_path) {
1097                let mut fsj_reader = BufReader::new(fsj_file);
1098                let mut line = String::new();
1099                while fsj_reader.read_line(&mut line)? != 0 {
1100                    let trimmed = line.trim_end();
1101                    if !trimmed.is_empty() {
1102                        let mut parts = trimmed.rsplitn(2, '\t');
1103                        let count = parts
1104                            .next()
1105                            .and_then(|s| s.parse::<i32>().ok())
1106                            .unwrap_or(0);
1107                        if let Some(key) = parts.next() {
1108                            *self.fsj_map.entry(key.to_string()).or_insert(0) += count;
1109                        }
1110                    }
1111                    line.clear();
1112                }
1113            }
1114            let _ = std::fs::remove_file(fsj_path);
1115        }
1116        writer.flush()?;
1117        self.rescued_reads = rescued_ids.len();
1118        self.final_bsj_reads = self.scan1_ids.len() + self.rescued_reads;
1119        Ok(())
1120    }
1121
1122    /// Runs Scan2 on BAM input using BGZF-aware sharding.
1123    pub fn run_bam(
1124        &mut self,
1125        bam_file: &str,
1126        output_bsj2: &str,
1127        output_fsj: &str,
1128        chr_tcga_map: &HashMap<String, String>,
1129    ) -> Result<()> {
1130        self.run_bam_with_display(
1131            bam_file,
1132            output_bsj2,
1133            output_fsj,
1134            None,
1135            None,
1136            None,
1137            None,
1138            chr_tcga_map,
1139        )
1140        .map(|_| ())
1141    }
1142
1143    fn run_bam_with_display(
1144        &mut self,
1145        bam_file: &str,
1146        output_bsj2: &str,
1147        output_fsj: &str,
1148        display_output_path: Option<&str>,
1149        segments_output_path: Option<&str>,
1150        non_bsj_segments_output_path: Option<&str>,
1151        display_scan2: Option<&(Scan2, Scan2DisplayClaims)>,
1152        chr_tcga_map: &HashMap<String, String>,
1153    ) -> Result<Scan2SegmentArtifacts> {
1154        use noodles::bam;
1155        let run_started = Instant::now();
1156        let profile = if Scan2Profile::enabled_from_env() {
1157            Some(Scan2Profile::default())
1158        } else {
1159            None
1160        };
1161        let profile_ref = profile.as_ref();
1162        let file = File::open(bam_file)?;
1163        let file_size = std::fs::metadata(bam_file)?.len();
1164        let mmap = unsafe { Mmap::map(&file)? };
1165        let num_threads = bam_shard_count(mmap.len(), rayon::current_num_threads());
1166        let shard_size = mmap.len() / num_threads;
1167        let mut reader = bam::io::Reader::new(file);
1168        let header = reader.read_header()?;
1169
1170        let pb = ProgressBar::new(file_size);
1171        pb.set_style(
1172            ProgressStyle::default_bar()
1173                .template("{spinner:.green} [{elapsed_precise}] [{bar:40.cyan/blue}] {bytes}/{total_bytes} {percent:>3}% ({eta}) {msg}")?
1174                .progress_chars("#>-"),
1175        );
1176        pb.set_message("");
1177        pb.enable_steady_tick(Duration::from_millis(120));
1178
1179        unsafe {
1180            libc::madvise(
1181                mmap.as_ptr() as *mut libc::c_void,
1182                mmap.len(),
1183                libc::MADV_SEQUENTIAL,
1184            );
1185        }
1186
1187        let header_ref = &header;
1188        (0..num_threads).into_par_iter().try_for_each(|i| {
1189            let start = i * shard_size;
1190            let end = if i == num_threads - 1 {
1191                mmap.len()
1192            } else {
1193                (i + 1) * shard_size
1194            };
1195            let shard_out = part_path(output_bsj2, i);
1196            let display_shard_out = display_output_path.map(|path| part_path(path, i));
1197            let segments_shard_out = segments_output_path.map(|path| part_path(path, i));
1198            let non_bsj_segments_shard_out =
1199                non_bsj_segments_output_path.map(|path| part_path(path, i));
1200            let fsj_out = Self::shard_fsj_path(output_fsj, i);
1201            self.process_bam_shard_to_file(
1202                &mmap,
1203                start,
1204                end,
1205                header_ref,
1206                chr_tcga_map,
1207                &pb,
1208                &shard_out,
1209                &fsj_out,
1210                display_scan2,
1211                display_shard_out.as_deref(),
1212                segments_shard_out.as_deref(),
1213                non_bsj_segments_shard_out.as_deref(),
1214                profile_ref,
1215            )
1216        })?;
1217
1218        pb.set_position(file_size);
1219        pb.set_style(
1220            ProgressStyle::default_bar()
1221                .template("{spinner:.green} [{elapsed_precise}] [{bar:40.cyan/blue}] {bytes}/{total_bytes} {msg}")?
1222                .progress_chars("#>-"),
1223        );
1224        pb.finish_with_message("Completed");
1225        let merge_started = Instant::now();
1226        self.merge_shards_and_fsj(output_bsj2, output_fsj, num_threads)?;
1227        if let Some(profile) = profile_ref {
1228            profile
1229                .merge_ns
1230                .fetch_add(merge_started.elapsed().as_nanos() as u64, Ordering::Relaxed);
1231            profile
1232                .wall_total_ns
1233                .fetch_add(run_started.elapsed().as_nanos() as u64, Ordering::Relaxed);
1234            profile.report();
1235            report_scan2_hg_profile();
1236        }
1237        if let Some(path) = display_output_path {
1238            self.merge_display_shards(path, num_threads)?;
1239        }
1240        if let Some(path) = segments_output_path {
1241            self.merge_segment_shards(path, num_threads)?;
1242        }
1243        let non_bsj_segment_evidence_paths = non_bsj_segments_output_path
1244            .map(|path| {
1245                (0..num_threads)
1246                    .map(|idx| part_path(path, idx))
1247                    .collect::<Vec<_>>()
1248            })
1249            .unwrap_or_default();
1250        Ok(Scan2SegmentArtifacts {
1251            non_bsj_segment_evidence_paths,
1252        })
1253    }
1254
1255    /// Merges display shard files in shard order into one final temporary file.
1256    fn merge_display_shards(&self, output_path: &str, num_threads: usize) -> Result<()> {
1257        let mut writer = BufWriter::with_capacity(1024 * 1024, File::create(output_path)?);
1258        for i in 0..num_threads {
1259            let shard_path = part_path(output_path, i);
1260            if let Ok(shard_file) = File::open(&shard_path) {
1261                let mut reader = BufReader::new(shard_file);
1262                let mut line = String::new();
1263                loop {
1264                    line.clear();
1265                    if reader.read_line(&mut line)? == 0 {
1266                        break;
1267                    }
1268                    writer.write_all(line.as_bytes())?;
1269                }
1270            }
1271            let _ = std::fs::remove_file(&shard_path);
1272        }
1273        writer.flush()?;
1274        Ok(())
1275    }
1276
1277    /// Merges shard-local Scan2 segment-evidence sidecars in stable shard order.
1278    ///
1279    /// The merged file is not consumed by Summary. It is only a post-Summary
1280    /// reconstruction input, so keeping it separate from `.bsj2` preserves the
1281    /// CIRI3-compatible rescue/counting contract.
1282    fn merge_segment_shards(&self, output_path: &str, num_threads: usize) -> Result<()> {
1283        let mut writer = BufWriter::with_capacity(1024 * 1024, File::create(output_path)?);
1284        for i in 0..num_threads {
1285            let shard_path = part_path(output_path, i);
1286            if let Ok(shard_file) = File::open(&shard_path) {
1287                let mut reader = BufReader::new(shard_file);
1288                let mut line = String::new();
1289                while reader.read_line(&mut line)? != 0 {
1290                    writer.write_all(line.as_bytes())?;
1291                    line.clear();
1292                }
1293            }
1294            let _ = std::fs::remove_file(&shard_path);
1295        }
1296        writer.flush()?;
1297        Ok(())
1298    }
1299
1300    /// Formats read-group alignments for the Scan2 `<prefix>.segments2` sidecar.
1301    ///
1302    /// These rows intentionally mirror Scan1's sidecar protocol: raw mapper
1303    /// blocks are captured only for read groups with Scan2 BSJ evidence, and the
1304    /// post-Summary segments stage later applies circ confirmation, mate labels,
1305    /// and chain-level CIGAR reconstruction.
1306    fn segment_evidence_lines<'a>(
1307        read_id: &str,
1308        stage: &str,
1309        alignments: &[AlignmentRecord<'a>],
1310        reference: &HashMap<String, String>,
1311    ) -> Vec<String> {
1312        let mut rows = Vec::new();
1313        for aln in alignments {
1314            if aln.chrom.as_ref() == "*" || aln.cigar.as_ref() == "*" {
1315                continue;
1316            }
1317            let mate = if aln.flag & 0x40 != 0 { "R1" } else { "R2" };
1318            let clips = clip_sequence_payload(aln.cigar.as_ref(), aln.seq.as_ref());
1319            let cs = alignment_short_cs(
1320                aln.cigar.as_ref(),
1321                aln.seq.as_ref(),
1322                aln.chrom.as_ref(),
1323                aln.pos,
1324                reference,
1325            );
1326            rows.push(format!(
1327                "{}\t{}\t{}\t{}\t{}\t{}\t{}\t{}\t{}\t{}\t{}\t{}",
1328                read_id,
1329                stage,
1330                mate,
1331                aln.flag,
1332                aln.chrom,
1333                aln.pos,
1334                aln.mapq,
1335                aln.cigar,
1336                aln.seq.len(),
1337                clips,
1338                cs,
1339                xa_payload(aln.xa.as_ref())
1340            ));
1341        }
1342        rows
1343    }
1344
1345    /// Formats Scan2 read-level non-BSJ topology candidates.
1346    ///
1347    /// This sidecar is deliberately independent of the BSJ candidate index and
1348    /// Summary circRNA spans. It only uses read-level split/outward geometry to
1349    /// decide whether a group is worth re-evaluating after Summary. Soft clips
1350    /// are stored as compact `L:/R:` payloads so local clip placement remains
1351    /// available without repeating full read sequences across every alignment.
1352    /// The short `N` stage and compact alignment payload also carries
1353    /// short-form cs so sequence-aware `.segments.bam` can be reconstructed
1354    /// without rescanning FASTQ/BAM.
1355    fn non_bsj_segment_evidence_lines<'a>(
1356        read_id: &str,
1357        alignments: &[AlignmentRecord<'a>],
1358        reference: &HashMap<String, String>,
1359    ) -> Vec<String> {
1360        if !Self::may_support_non_bsj_segments(alignments) {
1361            return Vec::new();
1362        }
1363        let mut records = Vec::new();
1364        for aln in alignments {
1365            if aln.chrom.as_ref() == "*" || aln.cigar.as_ref() == "*" {
1366                continue;
1367            }
1368            let clips = clip_sequence_payload(aln.cigar.as_ref(), aln.seq.as_ref());
1369            let cs = alignment_short_cs(
1370                aln.cigar.as_ref(),
1371                aln.seq.as_ref(),
1372                aln.chrom.as_ref(),
1373                aln.pos,
1374                reference,
1375            );
1376            records.push(format!(
1377                "{}|{}|{}|{}|{}|{}|{}",
1378                aln.flag, aln.chrom, aln.pos, aln.mapq, aln.cigar, clips, cs
1379            ));
1380        }
1381        if records.is_empty() {
1382            Vec::new()
1383        } else {
1384            vec![format!("{}\t{}\t{}", read_id, "N", records.join(";"))]
1385        }
1386    }
1387
1388    /// Cheap read-level filter for supplemental backward/outward candidates.
1389    fn may_support_non_bsj_segments<'a>(alignments: &[AlignmentRecord<'a>]) -> bool {
1390        Self::may_support_outward_segments(alignments)
1391            || Self::may_support_backward_segments(alignments)
1392    }
1393
1394    /// Tests the 5' overlap plus 3' outward pair geometry without circ gating.
1395    fn may_support_outward_segments<'a>(alignments: &[AlignmentRecord<'a>]) -> bool {
1396        let mut r1: Option<&AlignmentRecord<'_>> = None;
1397        let mut r2: Option<&AlignmentRecord<'_>> = None;
1398        for aln in alignments {
1399            if aln.flag & 0x4 != 0 || aln.flag & 0x100 != 0 || aln.flag & 0x800 != 0 {
1400                continue;
1401            }
1402            match aln.flag & 0x40 != 0 {
1403                true if r1.is_none() => r1 = Some(aln),
1404                true => return false,
1405                false if r2.is_none() => r2 = Some(aln),
1406                false => return false,
1407            }
1408        }
1409        let (Some(r1), Some(r2)) = (r1, r2) else {
1410            return false;
1411        };
1412        if r1.chrom != r2.chrom
1413            || r1.chrom.as_ref() == "*"
1414            || r1.mapq < NON_BSJ_SEGMENT_MAPQ_THRES
1415            || r2.mapq < NON_BSJ_SEGMENT_MAPQ_THRES
1416        {
1417            return false;
1418        }
1419        let Some(r1_span) = Self::alignment_ref_span(r1) else {
1420            return false;
1421        };
1422        let Some(r2_span) = Self::alignment_ref_span(r2) else {
1423            return false;
1424        };
1425        let r1_reverse = r1.flag & 0x10 != 0;
1426        let r2_reverse = r2.flag & 0x10 != 0;
1427        if r1_reverse == r2_reverse {
1428            return false;
1429        }
1430        let (reverse_span, forward_span) = if r1_reverse {
1431            (r1_span, r2_span)
1432        } else {
1433            (r2_span, r1_span)
1434        };
1435        reverse_span.0 < forward_span.0
1436            && forward_span.0 <= reverse_span.1
1437            && reverse_span.1 < forward_span.1
1438            && forward_span.1 - reverse_span.0 + 1 <= NON_BSJ_SEGMENT_MAX_SPAN
1439    }
1440
1441    /// Tests whether split alignments can form a CIRI-AS-style backward candidate.
1442    ///
1443    /// The filter intentionally mirrors the cheap, circ-independent part of
1444    /// `mapping_check2`: same chromosome and strand are necessary but not enough;
1445    /// the paired CIGAR classifiers must also satisfy a backward junction
1446    /// geometry. This keeps Scan2 from writing ordinary supplementary-heavy
1447    /// groups into the non-BSJ sidecar while still allowing reads with no known
1448    /// BSJ or circRNA assignment.
1449    fn may_support_backward_segments<'a>(alignments: &[AlignmentRecord<'a>]) -> bool {
1450        for i in 0..alignments.len().saturating_sub(1) {
1451            let left = &alignments[i];
1452            if left.chrom.as_ref() == "*" || left.cigar.as_ref() == "*" {
1453                continue;
1454            }
1455            for right in &alignments[i + 1..] {
1456                if left.chrom != right.chrom
1457                    || right.cigar.as_ref() == "*"
1458                    || (left.flag & 0x10 != 0) != (right.flag & 0x10 != 0)
1459                {
1460                    continue;
1461                }
1462                let read_len = left.seq.len().max(right.seq.len()) as i32;
1463                if read_len <= 0 {
1464                    continue;
1465                }
1466                let left_msid = Self::non_bsj_msid(left.cigar.as_ref(), read_len);
1467                let right_msid = Self::non_bsj_msid(right.cigar.as_ref(), read_len);
1468                if Self::non_bsj_backward_pair_possible(
1469                    left, right, left_msid, right_msid, read_len,
1470                ) || Self::non_bsj_backward_pair_possible(
1471                    right, left, right_msid, left_msid, read_len,
1472                ) {
1473                    return true;
1474                }
1475            }
1476        }
1477        false
1478    }
1479
1480    /// Tests the CIRI-AS backward candidate geometry without circ-boundary checks.
1481    fn non_bsj_backward_pair_possible<'a>(
1482        left: &AlignmentRecord<'a>,
1483        right: &AlignmentRecord<'a>,
1484        left_msid: NonBsjMsid,
1485        right_msid: NonBsjMsid,
1486        read_len: i32,
1487    ) -> bool {
1488        let product = left_msid.kind * right_msid.kind;
1489        if product == -1 {
1490            let cir_scale = left_msid.kind * (left.pos + left_msid.clip2)
1491                + right_msid.kind * (right.pos + right_msid.clip2);
1492            return (left_msid.clip1 - right_msid.clip1).abs() <= 6 && cir_scale < 0;
1493        }
1494        if product.abs() != 10 {
1495            return false;
1496        }
1497        let (mx, my, rx, ry) = if left_msid.kind <= right_msid.kind {
1498            (left_msid, right_msid, left, right)
1499        } else {
1500            (right_msid, left_msid, right, left)
1501        };
1502        if mx.kind == -1 {
1503            let cir_scale = ry.pos + my.ref_len - 1 - rx.pos;
1504            (read_len - my.clip2 - mx.clip1).abs() <= 6 && cir_scale < 0
1505        } else {
1506            let cir_scale = rx.pos + mx.ref_len - 1 - ry.pos;
1507            (mx.clip1 - my.clip1).abs() <= 6 && cir_scale < 0
1508        }
1509    }
1510
1511    /// Classifies CIGARs using the subset of CIRI-AS `MSID` needed for prefiltering.
1512    fn non_bsj_msid(cigar: &str, read_len: i32) -> NonBsjMsid {
1513        if cigar == "*" || cigar.is_empty() {
1514            return NonBsjMsid::invalid(-1);
1515        }
1516        let mut counts = Vec::new();
1517        let mut styles = Vec::new();
1518        let mut count = 0i32;
1519        let mut has_count = false;
1520        for ch in cigar.chars() {
1521            if ch.is_ascii_digit() {
1522                let Some(next) = count
1523                    .checked_mul(10)
1524                    .and_then(|value| value.checked_add(ch.to_digit(10).unwrap_or(0) as i32))
1525                else {
1526                    return NonBsjMsid::invalid(-2);
1527                };
1528                count = next;
1529                has_count = true;
1530            } else {
1531                if !has_count {
1532                    return NonBsjMsid::invalid(-2);
1533                }
1534                counts.push(count);
1535                styles.push(if ch == 'H' { 'S' } else { ch });
1536                count = 0;
1537                has_count = false;
1538            }
1539        }
1540        if has_count || styles.is_empty() {
1541            return NonBsjMsid::invalid(-2);
1542        }
1543
1544        if counts.len() == 1 {
1545            return if styles[0] == 'M' && counts[0] == read_len {
1546                NonBsjMsid {
1547                    kind: 0,
1548                    clip1: 0,
1549                    clip2: 0,
1550                    ref_len: read_len,
1551                }
1552            } else {
1553                NonBsjMsid::invalid(-1)
1554            };
1555        }
1556
1557        match counts.len() {
1558            2 => match (styles[0], styles[1]) {
1559                ('M', 'S') => NonBsjMsid {
1560                    kind: 1,
1561                    clip1: counts[0],
1562                    clip2: counts[0] - 1,
1563                    ref_len: counts[0],
1564                },
1565                ('S', 'M') => NonBsjMsid {
1566                    kind: -1,
1567                    clip1: counts[0],
1568                    clip2: 0,
1569                    ref_len: counts[1],
1570                },
1571                _ => NonBsjMsid::invalid(-2),
1572            },
1573            3 => match (styles[0], styles[1], styles[2]) {
1574                ('S', _, 'S') => NonBsjMsid {
1575                    kind: 10,
1576                    clip1: counts[0],
1577                    clip2: counts[2],
1578                    ref_len: counts[1],
1579                },
1580                ('M', 'D', 'M') => NonBsjMsid {
1581                    kind: 0,
1582                    clip1: 0,
1583                    clip2: 0,
1584                    ref_len: read_len + counts[1],
1585                },
1586                ('M', 'I', 'M') => NonBsjMsid {
1587                    kind: 0,
1588                    clip1: 0,
1589                    clip2: 0,
1590                    ref_len: read_len - counts[1],
1591                },
1592                _ => NonBsjMsid::invalid(-2),
1593            },
1594            _ if styles[0] == 'M' && *styles.last().unwrap() == 'S' => {
1595                let (m_sum, d_sum) =
1596                    Self::non_bsj_sum_md(&styles[..styles.len() - 1], &counts[..counts.len() - 1]);
1597                NonBsjMsid {
1598                    kind: 1,
1599                    clip1: read_len - counts[counts.len() - 1],
1600                    clip2: m_sum + d_sum - 1,
1601                    ref_len: m_sum + d_sum,
1602                }
1603            }
1604            _ if styles[0] == 'S' && *styles.last().unwrap() == 'M' => {
1605                let (m_sum, d_sum) = Self::non_bsj_sum_md(&styles[1..], &counts[1..]);
1606                NonBsjMsid {
1607                    kind: -1,
1608                    clip1: counts[0],
1609                    clip2: 0,
1610                    ref_len: m_sum + d_sum,
1611                }
1612            }
1613            _ if styles[0] == 'M' && *styles.last().unwrap() == 'M' => {
1614                let (m_sum, d_sum) = Self::non_bsj_sum_md(&styles, &counts);
1615                NonBsjMsid {
1616                    kind: 0,
1617                    clip1: 0,
1618                    clip2: 0,
1619                    ref_len: m_sum + d_sum,
1620                }
1621            }
1622            _ if styles[0] == 'S' && *styles.last().unwrap() == 'S' => {
1623                let (m_sum, d_sum) = Self::non_bsj_sum_md(
1624                    &styles[1..styles.len() - 1],
1625                    &counts[1..counts.len() - 1],
1626                );
1627                NonBsjMsid {
1628                    kind: 10,
1629                    clip1: counts[0],
1630                    clip2: counts[counts.len() - 1],
1631                    ref_len: m_sum + d_sum,
1632                }
1633            }
1634            _ => NonBsjMsid::invalid(-2),
1635        }
1636    }
1637
1638    /// Sums reference match and deletion lengths for the local MSID classifier.
1639    fn non_bsj_sum_md(styles: &[char], counts: &[i32]) -> (i32, i32) {
1640        let mut match_sum = 0;
1641        let mut deletion_sum = 0;
1642        for (style, count) in styles.iter().zip(counts.iter()) {
1643            match style {
1644                'M' | '=' | 'X' => match_sum += count,
1645                'D' => deletion_sum += count,
1646                _ => {}
1647            }
1648        }
1649        (match_sum, deletion_sum)
1650    }
1651
1652    /// Returns the genomic span covered by reference-consuming CIGAR operators.
1653    fn alignment_ref_span(aln: &AlignmentRecord<'_>) -> Option<(i32, i32)> {
1654        let mut ref_len = 0i32;
1655        let mut count = 0i32;
1656        let mut has_count = false;
1657        for op in aln.cigar.as_ref().chars() {
1658            if op.is_ascii_digit() {
1659                count = count
1660                    .checked_mul(10)?
1661                    .checked_add(op.to_digit(10)? as i32)?;
1662                has_count = true;
1663                continue;
1664            }
1665            if !has_count {
1666                return None;
1667            }
1668            if matches!(op, 'M' | 'D' | 'N' | '=' | 'X') {
1669                ref_len = ref_len.checked_add(count)?;
1670            }
1671            count = 0;
1672            has_count = false;
1673        }
1674        if has_count || ref_len <= 0 {
1675            return None;
1676        }
1677        Some((aln.pos, aln.pos + ref_len - 1))
1678    }
1679
1680    /// Formats mapper rows plus validator-accepted local clip pseudo rows.
1681    fn segment_evidence_lines_with_local<'a>(
1682        read_id: &str,
1683        stage: &str,
1684        local_stage: &str,
1685        alignments: &[AlignmentRecord<'a>],
1686        bsj_lines: &[String],
1687        reference: &HashMap<String, String>,
1688    ) -> Vec<String> {
1689        let mut rows = Self::segment_evidence_lines(read_id, stage, alignments, reference);
1690        let refs: Vec<&AlignmentRecord<'_>> = alignments.iter().collect();
1691        rows.extend(local_clip_evidence_lines(
1692            read_id,
1693            local_stage,
1694            &refs,
1695            bsj_lines,
1696            reference,
1697            10,
1698        ));
1699        rows
1700    }
1701
1702    /// Builds a `(read_id, mate_label, legacy_payload)` key from a priority row.
1703    ///
1704    /// The key lets the expanded Scan2 writer avoid emitting the same mate-level
1705    /// rescue twice when the display path rediscovers the Java first-hit row.
1706    fn scan2_priority_key(line: &str) -> Option<String> {
1707        let mut parts = line.splitn(4, '\t');
1708        let read_id = parts.next()?;
1709        let mate_label = parts.next()?;
1710        let _priority = parts.next()?;
1711        let payload = parts.next()?;
1712        Some(format!("{read_id}\t{mate_label}\t{payload}"))
1713    }
1714
1715    /// Adds a `priority` field to a display row of the form
1716    /// `read_id, mate_label, legacy_payload...`.
1717    fn scan2_display_key_and_priority_line(line: &str, priority: &str) -> Option<(String, String)> {
1718        let mut parts = line.splitn(3, '\t');
1719        let read_id = parts.next()?;
1720        let mate_label = parts.next()?;
1721        let payload = parts.next()?;
1722        Some((
1723            format!("{read_id}\t{mate_label}\t{payload}"),
1724            format!("{read_id}\t{mate_label}\t{priority}\t{payload}"),
1725        ))
1726    }
1727
1728    /// Streams a SAM file into one shard output while batching grouped reads for
1729    /// parallel Scan2 rescue evaluation.
1730    fn process_sam_file_to_file(
1731        &self,
1732        sam_file: &str,
1733        chr_tcga_map: &HashMap<String, String>,
1734        pb: &ProgressBar,
1735        out_path: &str,
1736        fsj_path: &str,
1737        display_scan2: Option<&(Scan2, Scan2DisplayClaims)>,
1738        display_out_path: Option<&str>,
1739        segments_out_path: Option<&str>,
1740        non_bsj_segments_out_path: Option<&str>,
1741        profile: Option<&Scan2Profile>,
1742    ) -> Result<()> {
1743        let mut writer = BufWriter::with_capacity(256 * 1024, File::create(out_path)?);
1744        let mut display_writer = if let Some(path) = display_out_path {
1745            Some(BufWriter::with_capacity(256 * 1024, File::create(path)?))
1746        } else {
1747            None
1748        };
1749        let mut segments_writer = if let Some(path) = segments_out_path {
1750            Some(BufWriter::with_capacity(256 * 1024, File::create(path)?))
1751        } else {
1752            None
1753        };
1754        let mut non_bsj_segments_writer = if let Some(path) = non_bsj_segments_out_path {
1755            Some(BufWriter::with_capacity(256 * 1024, File::create(path)?))
1756        } else {
1757            None
1758        };
1759        let batch_size = (rayon::current_num_threads().max(1) * 256).max(1024);
1760        let (tx, rx) =
1761            mpsc::sync_channel::<Vec<SamOwnedScan2Group>>(rayon::current_num_threads().max(2));
1762        let pb_clone = pb.clone();
1763
1764        let local_fsj = thread::scope(|scope| -> Result<HashMap<String, i32>> {
1765            let producer = scope.spawn(|| {
1766                self.stream_sam_group_batches_for_scan2(
1767                    sam_file, &pb_clone, tx, batch_size, profile,
1768                )
1769            });
1770            let mut merged_fsj = HashMap::new();
1771            for mut batch in rx {
1772                self.process_sam_group_batch_for_scan2(
1773                    &mut writer,
1774                    display_writer.as_mut(),
1775                    segments_writer.as_mut(),
1776                    non_bsj_segments_writer.as_mut(),
1777                    &mut batch,
1778                    &mut merged_fsj,
1779                    chr_tcga_map,
1780                    display_scan2,
1781                    profile,
1782                )?;
1783            }
1784            producer
1785                .join()
1786                .map_err(|_| anyhow::anyhow!("SAM Scan2 group producer thread panicked"))??;
1787            Ok(merged_fsj)
1788        })?;
1789
1790        writer.flush()?;
1791        if let Some(writer) = display_writer.as_mut() {
1792            writer.flush()?;
1793        }
1794        if let Some(writer) = segments_writer.as_mut() {
1795            writer.flush()?;
1796        }
1797        if let Some(writer) = non_bsj_segments_writer.as_mut() {
1798            writer.flush()?;
1799        }
1800        Self::write_fsj_shard(fsj_path, &local_fsj)
1801    }
1802
1803    /// Sequentially parses SAM records into grouped Scan2 inputs.
1804    fn stream_sam_group_batches_for_scan2(
1805        &self,
1806        sam_file: &str,
1807        pb: &ProgressBar,
1808        tx: mpsc::SyncSender<Vec<SamOwnedScan2Group>>,
1809        batch_size: usize,
1810        profile: Option<&Scan2Profile>,
1811    ) -> Result<()> {
1812        let sam_reader = BufReader::with_capacity(1024 * 1024, File::open(sam_file)?);
1813        let mut reader = sam::io::Reader::new(sam_reader);
1814        let header = reader.read_header()?;
1815        let mut record = sam::Record::default();
1816        let mut current_id: Vec<u8> = Vec::new();
1817        let mut alignments: Vec<OwnedAlignmentRecord> = Vec::with_capacity(16);
1818        let mut all_alignments: Vec<OwnedAlignmentRecord> = Vec::with_capacity(16);
1819        let mut stand_map: OwnedStandMap = HashMap::with_capacity(4);
1820        let mut one_read_key: i32 = -1;
1821        let mut batch = Vec::with_capacity(batch_size);
1822        let mut records_since_progress = 0usize;
1823        let mut last_progress_pos = reader.get_mut().stream_position()?;
1824        let mut cigar_buf = String::with_capacity(64);
1825        let mut seq_buf = String::with_capacity(256);
1826
1827        while reader.read_record(&mut record)? != 0 {
1828            let read_id = record
1829                .name()
1830                .ok_or_else(|| anyhow::anyhow!("Missing read name"))?;
1831            if let Some(profile) = profile {
1832                profile.records.fetch_add(1, Ordering::Relaxed);
1833            }
1834            if read_id != current_id.as_slice() {
1835                if !current_id.is_empty() {
1836                    self.push_sam_owned_group_for_scan2(
1837                        &mut batch,
1838                        &current_id,
1839                        &mut alignments,
1840                        &mut all_alignments,
1841                        &mut stand_map,
1842                    );
1843                    if batch.len() >= batch_size {
1844                        tx.send(std::mem::take(&mut batch))
1845                            .map_err(|_| anyhow::anyhow!("SAM Scan2 consumer dropped"))?;
1846                        batch = Vec::with_capacity(batch_size);
1847                    }
1848                }
1849                current_id.clear();
1850                current_id.extend_from_slice(read_id);
1851                one_read_key = -1;
1852            }
1853
1854            let flag = i32::from(u16::from(record.flags()?));
1855            let chrom = match record.reference_sequence(&header) {
1856                Some(Ok((name, _))) => String::from_utf8_lossy(name).to_string(),
1857                _ => "*".to_string(),
1858            };
1859            let start_pos = record
1860                .alignment_start()
1861                .transpose()?
1862                .map(|p| p.get() as i32)
1863                .unwrap_or(0);
1864            let mapq = record
1865                .mapping_quality()
1866                .transpose()?
1867                .map(u8::from)
1868                .unwrap_or(0) as i32;
1869            cigar_buf.clear();
1870            for result in record.cigar().iter() {
1871                let op = result?;
1872                use noodles::sam::alignment::record::cigar::op::Kind;
1873                let op_char = match op.kind() {
1874                    Kind::Match => 'M',
1875                    Kind::Insertion => 'I',
1876                    Kind::Deletion => 'D',
1877                    Kind::Skip => 'N',
1878                    Kind::SoftClip => 'S',
1879                    Kind::HardClip => 'H',
1880                    Kind::Pad => 'P',
1881                    Kind::SequenceMatch => '=',
1882                    Kind::SequenceMismatch => 'X',
1883                };
1884                let _ = write!(&mut cigar_buf, "{}{}", op.len(), op_char);
1885            }
1886            seq_buf.clear();
1887            for b in record.sequence().iter() {
1888                seq_buf.push(char::from(b));
1889            }
1890            let seq = seq_buf.clone();
1891            let s_idx = if flag & 0x40 != 0 { 1 } else { 0 };
1892            let st_c = if flag & 0x10 != 0 { '1' } else { '0' };
1893            if s_idx != one_read_key {
1894                one_read_key = s_idx;
1895                alignments.retain(|a| {
1896                    let idx = if a.flag & 0x40 != 0 { 1 } else { 0 };
1897                    idx != s_idx
1898                });
1899                if !seq.is_empty() && seq != "*" {
1900                    stand_map.insert(s_idx, (st_c, Cow::Owned(seq.clone())));
1901                }
1902            }
1903            let alignment = AlignmentRecord {
1904                flag,
1905                chrom: Cow::Owned(chrom),
1906                pos: start_pos,
1907                mapq,
1908                cigar: Cow::Owned(cigar_buf.clone()),
1909                seq: Cow::Owned(seq),
1910                xa: Cow::Owned(raw_xa_from_sam_record(&record)?),
1911            };
1912            all_alignments.push(alignment.clone());
1913            alignments.push(alignment);
1914
1915            records_since_progress += 1;
1916            if records_since_progress >= 4096 {
1917                let pos = reader.get_mut().stream_position()?;
1918                if pos > last_progress_pos {
1919                    pb.inc(pos - last_progress_pos);
1920                    last_progress_pos = pos;
1921                }
1922                records_since_progress = 0;
1923            }
1924        }
1925
1926        if !current_id.is_empty() {
1927            self.push_sam_owned_group_for_scan2(
1928                &mut batch,
1929                &current_id,
1930                &mut alignments,
1931                &mut all_alignments,
1932                &mut stand_map,
1933            );
1934        }
1935        if !batch.is_empty() {
1936            tx.send(batch)
1937                .map_err(|_| anyhow::anyhow!("SAM Scan2 consumer dropped"))?;
1938        }
1939
1940        let final_pos = reader.get_mut().stream_position()?;
1941        if final_pos > last_progress_pos {
1942            pb.inc(final_pos - last_progress_pos);
1943        }
1944        Ok(())
1945    }
1946
1947    /// Moves the currently accumulated SAM Scan2 group into the worker batch.
1948    fn push_sam_owned_group_for_scan2(
1949        &self,
1950        batch: &mut Vec<SamOwnedScan2Group>,
1951        current_id: &[u8],
1952        alignments: &mut Vec<OwnedAlignmentRecord>,
1953        all_alignments: &mut Vec<OwnedAlignmentRecord>,
1954        stand_map: &mut OwnedStandMap,
1955    ) {
1956        batch.push(SamOwnedScan2Group {
1957            read_id: String::from_utf8_lossy(current_id).into_owned(),
1958            alignments: std::mem::take(alignments),
1959            all_alignments: std::mem::take(all_alignments),
1960            stand_map: std::mem::take(stand_map),
1961        });
1962        *alignments = Vec::with_capacity(16);
1963        *all_alignments = Vec::with_capacity(16);
1964        *stand_map = HashMap::with_capacity(4);
1965    }
1966
1967    /// Evaluates one batch of grouped SAM reads in parallel and writes Scan2 output.
1968    fn process_sam_group_batch_for_scan2(
1969        &self,
1970        writer: &mut BufWriter<File>,
1971        mut display_writer: Option<&mut BufWriter<File>>,
1972        mut segments_writer: Option<&mut BufWriter<File>>,
1973        mut non_bsj_segments_writer: Option<&mut BufWriter<File>>,
1974        batch: &mut Vec<SamOwnedScan2Group>,
1975        merged_fsj: &mut HashMap<String, i32>,
1976        chr_tcga_map: &HashMap<String, String>,
1977        display_scan2: Option<&(Scan2, Scan2DisplayClaims)>,
1978        profile: Option<&Scan2Profile>,
1979    ) -> Result<()> {
1980        let groups = std::mem::take(batch);
1981        let results: Vec<(
1982            Vec<String>,
1983            HashMap<String, i32>,
1984            Vec<String>,
1985            Vec<String>,
1986            Vec<String>,
1987        )> = groups
1988            .into_par_iter()
1989            .map(|owned| {
1990                let mut local_lines = Vec::new();
1991                let mut local_fsj = HashMap::new();
1992                let mut validator = IsBSJHg2::new(self.linear_range_size_min, self.min_mapq_uni);
1993                let _ = self.process_group_view(
1994                    &owned.read_id,
1995                    &owned.alignments,
1996                    &owned.stand_map,
1997                    &mut local_lines,
1998                    &mut local_fsj,
1999                    chr_tcga_map,
2000                    &mut validator,
2001                    profile,
2002                );
2003                let display_lines = if let Some((display_helper, scan1_claims)) = display_scan2 {
2004                    let mut display_validator =
2005                        IsBSJHg2::new(self.linear_range_size_min, self.min_mapq_uni);
2006                    display_helper
2007                        .process_group_view_display(
2008                            &owned.read_id,
2009                            &owned.alignments,
2010                            &owned.stand_map,
2011                            scan1_claims,
2012                            chr_tcga_map,
2013                            &mut display_validator,
2014                        )
2015                        .unwrap_or_default()
2016                } else {
2017                    Vec::new()
2018                };
2019                let has_bsj = !local_lines.is_empty() || !display_lines.is_empty();
2020                let evidence = if has_bsj {
2021                    let mut bsj_lines = local_lines.clone();
2022                    bsj_lines.extend(display_lines.iter().cloned());
2023                    Self::segment_evidence_lines_with_local(
2024                        &owned.read_id,
2025                        "scan2",
2026                        "scan2_local",
2027                        &owned.alignments,
2028                        &bsj_lines,
2029                        chr_tcga_map,
2030                    )
2031                } else {
2032                    Vec::new()
2033                };
2034                let scan1_claimed = display_scan2.as_ref().is_some_and(|(_, scan1_claims)| {
2035                    scan1_claims.by_read.contains_key(owned.read_id.as_str())
2036                });
2037                let non_bsj_evidence =
2038                    if local_lines.is_empty() && display_lines.is_empty() && !scan1_claimed {
2039                        Self::non_bsj_segment_evidence_lines(
2040                            &owned.read_id,
2041                            &owned.all_alignments,
2042                            chr_tcga_map,
2043                        )
2044                    } else {
2045                        Vec::new()
2046                    };
2047                (
2048                    local_lines,
2049                    local_fsj,
2050                    display_lines,
2051                    evidence,
2052                    non_bsj_evidence,
2053                )
2054            })
2055            .collect();
2056
2057        for (lines, fsj_map, display_lines, evidence, non_bsj_evidence) in results {
2058            let main_keys: HashSet<String> = lines
2059                .iter()
2060                .filter_map(|line| Self::scan2_priority_key(line))
2061                .collect();
2062            for line in lines {
2063                writeln!(writer, "{}", line)?;
2064            }
2065            if let Some(display_writer) = display_writer.as_deref_mut() {
2066                for line in display_lines {
2067                    writeln!(display_writer, "{}", line)?;
2068                }
2069            } else {
2070                for line in display_lines {
2071                    if let Some((key, priority_line)) =
2072                        Self::scan2_display_key_and_priority_line(&line, "0")
2073                    {
2074                        if !main_keys.contains(&key) {
2075                            writeln!(writer, "{}", priority_line)?;
2076                        }
2077                    }
2078                }
2079            }
2080            for (key, count) in fsj_map {
2081                *merged_fsj.entry(key).or_insert(0) += count;
2082            }
2083            if let Some(segments_writer) = segments_writer.as_deref_mut() {
2084                let mut rows = 0_u64;
2085                let mut bytes = 0_u64;
2086                for line in evidence {
2087                    rows += 1;
2088                    bytes += line.len() as u64 + 1;
2089                    writeln!(segments_writer, "{}", line)?;
2090                }
2091                if let Some(profile) = profile {
2092                    profile
2093                        .segment_sidecar_rows
2094                        .fetch_add(rows, Ordering::Relaxed);
2095                    profile
2096                        .segment_sidecar_bytes
2097                        .fetch_add(bytes, Ordering::Relaxed);
2098                }
2099            }
2100            if let Some(non_bsj_segments_writer) = non_bsj_segments_writer.as_deref_mut() {
2101                let mut rows = 0_u64;
2102                let mut bytes = 0_u64;
2103                for line in non_bsj_evidence {
2104                    rows += 1;
2105                    bytes += line.len() as u64 + 1;
2106                    writeln!(non_bsj_segments_writer, "{}", line)?;
2107                }
2108                if let Some(profile) = profile {
2109                    profile
2110                        .non_bsj_sidecar_rows
2111                        .fetch_add(rows, Ordering::Relaxed);
2112                    profile
2113                        .non_bsj_sidecar_bytes
2114                        .fetch_add(bytes, Ordering::Relaxed);
2115                }
2116            }
2117        }
2118        Ok(())
2119    }
2120
2121    /// Processes one BAM shard and emits rescued BSJ2 lines plus one local FSJ spill file.
2122    ///
2123    /// The small buffer reuses inside this function are intentional low-risk
2124    /// optimizations: they reduce per-record allocation churn without changing the
2125    /// order or content of any candidate checks.
2126    fn process_bam_shard_to_file(
2127        &self,
2128        mmap: &Mmap,
2129        start: usize,
2130        end: usize,
2131        header: &sam::Header,
2132        chr_tcga_map: &HashMap<String, String>,
2133        pb: &ProgressBar,
2134        out_path: &str,
2135        fsj_path: &str,
2136        display_scan2: Option<&(Scan2, Scan2DisplayClaims)>,
2137        display_out_path: Option<&str>,
2138        segments_out_path: Option<&str>,
2139        non_bsj_segments_out_path: Option<&str>,
2140        profile: Option<&Scan2Profile>,
2141    ) -> Result<()> {
2142        use noodles::bam;
2143        let shard_started = profile.map(|_| Instant::now());
2144        let mut writer = BufWriter::with_capacity(256 * 1024, File::create(out_path)?);
2145        let mut display_writer = if let Some(path) = display_out_path {
2146            Some(BufWriter::with_capacity(256 * 1024, File::create(path)?))
2147        } else {
2148            None
2149        };
2150        let mut segments_writer = if let Some(path) = segments_out_path {
2151            Some(BufWriter::with_capacity(256 * 1024, File::create(path)?))
2152        } else {
2153            None
2154        };
2155        let mut non_bsj_segments_writer = if let Some(path) = non_bsj_segments_out_path {
2156            Some(BufWriter::with_capacity(256 * 1024, File::create(path)?))
2157        } else {
2158            None
2159        };
2160        let mut local_fsj = HashMap::new();
2161        let mut validator = IsBSJHg2::new(self.linear_range_size_min, self.min_mapq_uni);
2162        let mut display_validator = display_scan2
2163            .as_ref()
2164            .map(|_| IsBSJHg2::new(self.linear_range_size_min, self.min_mapq_uni));
2165        let block_start = if start == 0 {
2166            0
2167        } else {
2168            let mut found = None;
2169            for i in start..mmap.len().saturating_sub(3) {
2170                if &mmap[i..i + 4] == b"\x1f\x8b\x08\x04" {
2171                    found = Some(i);
2172                    break;
2173                }
2174            }
2175            found.unwrap_or(mmap.len())
2176        };
2177        let pos = if start == 0 {
2178            0
2179        } else {
2180            // Decode from the previous BGZF block so we can distinguish a true
2181            // new read group from one that started in the earlier shard.
2182            let mut found = None;
2183            let mut i = block_start
2184                .saturating_sub(1)
2185                .min(mmap.len().saturating_sub(4));
2186            loop {
2187                if i + 3 < mmap.len() && &mmap[i..i + 4] == b"\x1f\x8b\x08\x04" {
2188                    found = Some(i);
2189                    break;
2190                }
2191                if i == 0 {
2192                    break;
2193                }
2194                i -= 1;
2195            }
2196            found.unwrap_or(0)
2197        };
2198        if pos >= mmap.len() {
2199            writer.flush()?;
2200            if let Some(display_writer) = display_writer.as_mut() {
2201                display_writer.flush()?;
2202            }
2203            if let Some(segments_writer) = segments_writer.as_mut() {
2204                segments_writer.flush()?;
2205            }
2206            if let Some(non_bsj_segments_writer) = non_bsj_segments_writer.as_mut() {
2207                non_bsj_segments_writer.flush()?;
2208            }
2209            return Self::write_fsj_shard(fsj_path, &local_fsj);
2210        }
2211
2212        let mut reader = bam::io::Reader::new(&mmap[pos..]);
2213        if start == 0 {
2214            let _ = reader.read_header()?;
2215        }
2216        let mut record = bam::Record::default();
2217        let mut current_id: Vec<u8> = Vec::new();
2218        let mut alignments: Vec<AlignmentRecord> = Vec::with_capacity(16);
2219        let mut all_alignments: Vec<AlignmentRecord> = Vec::with_capacity(16);
2220        let mut stand_map: HashMap<i32, (char, Cow<str>)> = HashMap::with_capacity(4);
2221        let mut one_read_key: i32 = -1;
2222        let mut crossed_start = start == 0;
2223        let mut leading_partial_id: Option<Vec<u8>> = None;
2224        let mut last_compressed_pos = block_start.saturating_sub(pos);
2225        let mut last_evicted_pos = pos;
2226        // Keep page-cache eviction on a small sliding window. Large thresholds let
2227        // the whole mapped BAM drift into RSS before any MADV_DONTNEED call fires.
2228        let eviction_threshold =
2229            (self.mem_limit as usize / 8).clamp(8 * 1024 * 1024, 64 * 1024 * 1024);
2230        let mut cigar_buf = String::with_capacity(64);
2231        let mut seq_buf = String::with_capacity(256);
2232        let mut res_batch = Vec::new();
2233
2234        while reader.read_record(&mut record)? != 0 {
2235            if let Some(profile) = profile {
2236                profile.records.fetch_add(1, Ordering::Relaxed);
2237            }
2238            let curr_c_pos = reader.get_ref().virtual_position().compressed() as usize;
2239            pb.inc(curr_c_pos.saturating_sub(last_compressed_pos) as u64);
2240            last_compressed_pos = curr_c_pos;
2241
2242            let evicted_rel = last_evicted_pos.saturating_sub(pos);
2243            if curr_c_pos.saturating_sub(evicted_rel) > eviction_threshold {
2244                advise_dontneed(
2245                    mmap,
2246                    last_evicted_pos,
2247                    curr_c_pos.saturating_sub(evicted_rel),
2248                );
2249                last_evicted_pos = pos + curr_c_pos;
2250            }
2251
2252            let read_id = record
2253                .name()
2254                .ok_or_else(|| anyhow::anyhow!("Missing read name"))?;
2255            let abs_c_pos = pos + curr_c_pos;
2256            if !crossed_start {
2257                if abs_c_pos < block_start {
2258                    leading_partial_id = Some(read_id.to_vec());
2259                    continue;
2260                }
2261                crossed_start = true;
2262            }
2263            if let Some(partial_id) = &leading_partial_id {
2264                if read_id == partial_id.as_slice() {
2265                    // Let the earlier shard own the cross-boundary read group
2266                    // completely; otherwise Scan2 can silently drop the first
2267                    // full read group visible in this shard.
2268                    continue;
2269                }
2270                leading_partial_id = None;
2271            }
2272
2273            if read_id != current_id.as_slice() {
2274                if !current_id.is_empty() {
2275                    let id_str = String::from_utf8_lossy(&current_id);
2276                    res_batch.clear();
2277                    self.process_group_view(
2278                        &id_str,
2279                        &alignments,
2280                        &stand_map,
2281                        &mut res_batch,
2282                        &mut local_fsj,
2283                        chr_tcga_map,
2284                        &mut validator,
2285                        profile,
2286                    )?;
2287                    let write_started = profile.map(|_| Instant::now());
2288                    let main_keys: HashSet<String> = res_batch
2289                        .iter()
2290                        .filter_map(|line| Self::scan2_priority_key(line))
2291                        .collect();
2292                    for line in &res_batch {
2293                        writeln!(writer, "{}", line)?;
2294                    }
2295                    if let (Some(profile), Some(write_started)) = (profile, write_started) {
2296                        profile.write_ns.fetch_add(
2297                            write_started.elapsed().as_nanos() as u64,
2298                            Ordering::Relaxed,
2299                        );
2300                    }
2301                    let display_started = profile.map(|_| Instant::now());
2302                    let display_lines =
2303                        if let (Some((display_helper, scan1_claims)), Some(display_validator)) =
2304                            (display_scan2, display_validator.as_mut())
2305                        {
2306                            display_helper.process_group_view_display(
2307                                &id_str,
2308                                &alignments,
2309                                &stand_map,
2310                                scan1_claims,
2311                                chr_tcga_map,
2312                                display_validator,
2313                            )?
2314                        } else {
2315                            Vec::new()
2316                        };
2317                    if let (Some(profile), Some(display_started)) = (profile, display_started) {
2318                        profile.display_ns.fetch_add(
2319                            display_started.elapsed().as_nanos() as u64,
2320                            Ordering::Relaxed,
2321                        );
2322                    }
2323                    if !display_lines.is_empty() {
2324                        let write_started = profile.map(|_| Instant::now());
2325                        if let Some(display_writer) = display_writer.as_mut() {
2326                            for line in &display_lines {
2327                                writeln!(display_writer, "{}", line)?;
2328                            }
2329                        } else {
2330                            for line in &display_lines {
2331                                if let Some((key, priority_line)) =
2332                                    Self::scan2_display_key_and_priority_line(line, "0")
2333                                {
2334                                    if !main_keys.contains(&key) {
2335                                        writeln!(writer, "{}", priority_line)?;
2336                                    }
2337                                }
2338                            }
2339                        }
2340                        if let (Some(profile), Some(write_started)) = (profile, write_started) {
2341                            profile.write_ns.fetch_add(
2342                                write_started.elapsed().as_nanos() as u64,
2343                                Ordering::Relaxed,
2344                            );
2345                        }
2346                    }
2347                    let scan1_claimed = display_scan2.as_ref().is_some_and(|(_, scan1_claims)| {
2348                        scan1_claims.by_read.contains_key(id_str.as_ref())
2349                    });
2350                    let has_bsj =
2351                        !res_batch.is_empty() || !display_lines.is_empty() || scan1_claimed;
2352                    if let Some(segments_writer) = segments_writer.as_mut() {
2353                        let segment_started = profile.map(|_| Instant::now());
2354                        if has_bsj {
2355                            let mut bsj_lines = res_batch.clone();
2356                            bsj_lines.extend(display_lines.iter().cloned());
2357                            let mut rows = 0_u64;
2358                            let mut bytes = 0_u64;
2359                            for line in Self::segment_evidence_lines_with_local(
2360                                &id_str,
2361                                "scan2",
2362                                "scan2_local",
2363                                &alignments,
2364                                &bsj_lines,
2365                                chr_tcga_map,
2366                            ) {
2367                                rows += 1;
2368                                bytes += line.len() as u64 + 1;
2369                                writeln!(segments_writer, "{}", line)?;
2370                            }
2371                            if let Some(profile) = profile {
2372                                profile
2373                                    .segment_sidecar_rows
2374                                    .fetch_add(rows, Ordering::Relaxed);
2375                                profile
2376                                    .segment_sidecar_bytes
2377                                    .fetch_add(bytes, Ordering::Relaxed);
2378                            }
2379                        }
2380                        if let (Some(profile), Some(segment_started)) = (profile, segment_started) {
2381                            profile.segment_sidecar_ns.fetch_add(
2382                                segment_started.elapsed().as_nanos() as u64,
2383                                Ordering::Relaxed,
2384                            );
2385                        }
2386                    }
2387                    if !has_bsj {
2388                        if let Some(non_bsj_segments_writer) = non_bsj_segments_writer.as_mut() {
2389                            let non_bsj_started = profile.map(|_| Instant::now());
2390                            let mut rows = 0_u64;
2391                            let mut bytes = 0_u64;
2392                            for line in Self::non_bsj_segment_evidence_lines(
2393                                &id_str,
2394                                &all_alignments,
2395                                chr_tcga_map,
2396                            ) {
2397                                rows += 1;
2398                                bytes += line.len() as u64 + 1;
2399                                writeln!(non_bsj_segments_writer, "{}", line)?;
2400                            }
2401                            if let (Some(profile), Some(non_bsj_started)) =
2402                                (profile, non_bsj_started)
2403                            {
2404                                profile.non_bsj_sidecar_ns.fetch_add(
2405                                    non_bsj_started.elapsed().as_nanos() as u64,
2406                                    Ordering::Relaxed,
2407                                );
2408                                profile
2409                                    .non_bsj_sidecar_rows
2410                                    .fetch_add(rows, Ordering::Relaxed);
2411                                profile
2412                                    .non_bsj_sidecar_bytes
2413                                    .fetch_add(bytes, Ordering::Relaxed);
2414                            }
2415                        }
2416                    }
2417                    if abs_c_pos > end {
2418                        current_id.clear();
2419                        break;
2420                    }
2421                }
2422                current_id.clear();
2423                current_id.extend_from_slice(read_id);
2424                alignments.clear();
2425                all_alignments.clear();
2426                stand_map.clear();
2427                one_read_key = -1;
2428            }
2429
2430            let flag = i32::from(u16::from(record.flags()));
2431            let chrom = match record.reference_sequence(header) {
2432                Some(Ok((name, _))) => String::from_utf8_lossy(name).to_string(),
2433                _ => "*".to_string(),
2434            };
2435            let start_pos = record
2436                .alignment_start()
2437                .transpose()?
2438                .map(|p| p.get() as i32)
2439                .unwrap_or(0);
2440            let mapq = record.mapping_quality().map(u8::from).unwrap_or(0) as i32;
2441            cigar_buf.clear();
2442            for result in record.cigar().iter() {
2443                let op = result?;
2444                use noodles::sam::alignment::record::cigar::op::Kind;
2445                let op_char = match op.kind() {
2446                    Kind::Match => 'M',
2447                    Kind::Insertion => 'I',
2448                    Kind::Deletion => 'D',
2449                    Kind::Skip => 'N',
2450                    Kind::SoftClip => 'S',
2451                    Kind::HardClip => 'H',
2452                    Kind::Pad => 'P',
2453                    Kind::SequenceMatch => '=',
2454                    Kind::SequenceMismatch => 'X',
2455                };
2456                let _ = write!(&mut cigar_buf, "{}{}", op.len(), op_char);
2457            }
2458            seq_buf.clear();
2459            for b in record.sequence().iter() {
2460                seq_buf.push(char::from(b));
2461            }
2462            let seq = seq_buf.clone();
2463            let s_idx = if flag & 0x40 != 0 { 1 } else { 0 };
2464            let st_c = if flag & 0x10 != 0 { '1' } else { '0' };
2465            if s_idx != one_read_key {
2466                one_read_key = s_idx;
2467                // Java parity: BAM Scan2 rebuilds the per-mate alignment list each
2468                // time the iterator switches between R1 and R2 within the same
2469                // read group. Keeping every earlier alignment for that mate looks
2470                // harmless, but it changes which candidate payloads contribute to
2471                // `temFSJId` on complex supplementary-heavy reads and was one of
2472                // the reasons full hg38 FSJ counting drifted while chr1 stayed
2473                // clean. This overwrite is therefore intentional and verified.
2474                alignments.retain(|a| {
2475                    let idx = if a.flag & 0x40 != 0 { 1 } else { 0 };
2476                    idx != s_idx
2477                });
2478                // Java parity: `standMap` is overwritten on mate switches with the
2479                // current record's sequence; it is not a "longest-sequence wins"
2480                // cache in Scan2. This only affects Scan2 candidate validation,
2481                // not Scan1 representative-sequence handling. The distinction is
2482                // easy to miss because Scan1 does keep the longest representative
2483                // sequence, but carrying that policy into Scan2 perturbs FSJ-only
2484                // behavior on supplementary-heavy BAM families.
2485                if !seq.is_empty() && seq != "*" {
2486                    stand_map.insert(s_idx, (st_c, Cow::Owned(seq.clone())));
2487                }
2488            }
2489            let alignment = AlignmentRecord {
2490                flag,
2491                chrom: Cow::Owned(chrom),
2492                pos: start_pos,
2493                mapq,
2494                cigar: Cow::Owned(cigar_buf.clone()),
2495                seq: Cow::Owned(seq),
2496                xa: Cow::Owned(raw_xa_from_bam_record(&record)?),
2497            };
2498            all_alignments.push(alignment.clone());
2499            alignments.push(alignment);
2500        }
2501        if !current_id.is_empty() {
2502            let id_str = String::from_utf8_lossy(&current_id);
2503            res_batch.clear();
2504            self.process_group_view(
2505                &id_str,
2506                &alignments,
2507                &stand_map,
2508                &mut res_batch,
2509                &mut local_fsj,
2510                chr_tcga_map,
2511                &mut validator,
2512                profile,
2513            )?;
2514            let write_started = profile.map(|_| Instant::now());
2515            let main_keys: HashSet<String> = res_batch
2516                .iter()
2517                .filter_map(|line| Self::scan2_priority_key(line))
2518                .collect();
2519            for line in &res_batch {
2520                writeln!(writer, "{}", line)?;
2521            }
2522            if let (Some(profile), Some(write_started)) = (profile, write_started) {
2523                profile
2524                    .write_ns
2525                    .fetch_add(write_started.elapsed().as_nanos() as u64, Ordering::Relaxed);
2526            }
2527            let display_started = profile.map(|_| Instant::now());
2528            let display_lines =
2529                if let (Some((display_helper, scan1_claims)), Some(display_validator)) =
2530                    (display_scan2, display_validator.as_mut())
2531                {
2532                    display_helper.process_group_view_display(
2533                        &id_str,
2534                        &alignments,
2535                        &stand_map,
2536                        scan1_claims,
2537                        chr_tcga_map,
2538                        display_validator,
2539                    )?
2540                } else {
2541                    Vec::new()
2542                };
2543            if let (Some(profile), Some(display_started)) = (profile, display_started) {
2544                profile.display_ns.fetch_add(
2545                    display_started.elapsed().as_nanos() as u64,
2546                    Ordering::Relaxed,
2547                );
2548            }
2549            if !display_lines.is_empty() {
2550                let write_started = profile.map(|_| Instant::now());
2551                if let Some(display_writer) = display_writer.as_mut() {
2552                    for line in &display_lines {
2553                        writeln!(display_writer, "{}", line)?;
2554                    }
2555                } else {
2556                    for line in &display_lines {
2557                        if let Some((key, priority_line)) =
2558                            Self::scan2_display_key_and_priority_line(line, "0")
2559                        {
2560                            if !main_keys.contains(&key) {
2561                                writeln!(writer, "{}", priority_line)?;
2562                            }
2563                        }
2564                    }
2565                }
2566                if let (Some(profile), Some(write_started)) = (profile, write_started) {
2567                    profile
2568                        .write_ns
2569                        .fetch_add(write_started.elapsed().as_nanos() as u64, Ordering::Relaxed);
2570                }
2571            }
2572            let scan1_claimed = display_scan2.as_ref().is_some_and(|(_, scan1_claims)| {
2573                scan1_claims.by_read.contains_key(id_str.as_ref())
2574            });
2575            let has_bsj = !res_batch.is_empty() || !display_lines.is_empty() || scan1_claimed;
2576            if let Some(segments_writer) = segments_writer.as_mut() {
2577                let segment_started = profile.map(|_| Instant::now());
2578                if has_bsj {
2579                    let mut bsj_lines = res_batch.clone();
2580                    bsj_lines.extend(display_lines.iter().cloned());
2581                    let mut rows = 0_u64;
2582                    let mut bytes = 0_u64;
2583                    for line in Self::segment_evidence_lines_with_local(
2584                        &id_str,
2585                        "scan2",
2586                        "scan2_local",
2587                        &alignments,
2588                        &bsj_lines,
2589                        chr_tcga_map,
2590                    ) {
2591                        rows += 1;
2592                        bytes += line.len() as u64 + 1;
2593                        writeln!(segments_writer, "{}", line)?;
2594                    }
2595                    if let Some(profile) = profile {
2596                        profile
2597                            .segment_sidecar_rows
2598                            .fetch_add(rows, Ordering::Relaxed);
2599                        profile
2600                            .segment_sidecar_bytes
2601                            .fetch_add(bytes, Ordering::Relaxed);
2602                    }
2603                }
2604                if let (Some(profile), Some(segment_started)) = (profile, segment_started) {
2605                    profile.segment_sidecar_ns.fetch_add(
2606                        segment_started.elapsed().as_nanos() as u64,
2607                        Ordering::Relaxed,
2608                    );
2609                }
2610            }
2611            if !has_bsj {
2612                if let Some(non_bsj_segments_writer) = non_bsj_segments_writer.as_mut() {
2613                    let non_bsj_started = profile.map(|_| Instant::now());
2614                    let mut rows = 0_u64;
2615                    let mut bytes = 0_u64;
2616                    for line in
2617                        Self::non_bsj_segment_evidence_lines(&id_str, &all_alignments, chr_tcga_map)
2618                    {
2619                        rows += 1;
2620                        bytes += line.len() as u64 + 1;
2621                        writeln!(non_bsj_segments_writer, "{}", line)?;
2622                    }
2623                    if let (Some(profile), Some(non_bsj_started)) = (profile, non_bsj_started) {
2624                        profile.non_bsj_sidecar_ns.fetch_add(
2625                            non_bsj_started.elapsed().as_nanos() as u64,
2626                            Ordering::Relaxed,
2627                        );
2628                        profile
2629                            .non_bsj_sidecar_rows
2630                            .fetch_add(rows, Ordering::Relaxed);
2631                        profile
2632                            .non_bsj_sidecar_bytes
2633                            .fetch_add(bytes, Ordering::Relaxed);
2634                    }
2635                }
2636            }
2637        }
2638        let evicted_rel = last_evicted_pos.saturating_sub(pos);
2639        if last_compressed_pos > evicted_rel {
2640            advise_dontneed(mmap, last_evicted_pos, last_compressed_pos - evicted_rel);
2641        }
2642        writer.flush()?;
2643        if let Some(display_writer) = display_writer.as_mut() {
2644            display_writer.flush()?;
2645        }
2646        if let Some(segments_writer) = segments_writer.as_mut() {
2647            segments_writer.flush()?;
2648        }
2649        if let Some(non_bsj_segments_writer) = non_bsj_segments_writer.as_mut() {
2650            non_bsj_segments_writer.flush()?;
2651        }
2652        if let (Some(profile), Some(shard_started)) = (profile, shard_started) {
2653            profile
2654                .shard_total_ns
2655                .fetch_add(shard_started.elapsed().as_nanos() as u64, Ordering::Relaxed);
2656        }
2657        Self::write_fsj_shard(fsj_path, &local_fsj)
2658    }
2659
2660    /// Rescues BSJ candidates and counts overlapping FSJs for one read group.
2661    ///
2662    /// This is the Scan2 hot path. The preserved optimizations are intentionally
2663    /// modest:
2664    /// - mate-segment lookup is hoisted once per segment side
2665    /// - oriented read/mate strings are computed once per alignment, not once per
2666    ///   candidate
2667    /// - bucket traversal order remains Java-compatible
2668    pub(crate) fn process_group_view<'a>(
2669        &self,
2670        id: &str,
2671        alignments: &[AlignmentRecord<'a>],
2672        stand_map: &HashMap<i32, (char, Cow<'a, str>)>,
2673        results: &mut Vec<String>,
2674        local_fsj: &mut HashMap<String, i32>,
2675        chr_tcga_map: &HashMap<String, String>,
2676        is_bsj_hg2: &mut IsBSJHg2,
2677        profile: Option<&Scan2Profile>,
2678    ) -> Result<()> {
2679        let started = profile.map(|_| Instant::now());
2680        let trace_read = should_trace_read(id);
2681        let trace_all_candidates =
2682            trace_read && std::env::var("CIRI_TRACE_ALL_CANDS").ok().as_deref() == Some("1");
2683        if self.scan1_ids.contains(id) {
2684            return Ok(());
2685        }
2686        let mut segments: HashMap<i32, Vec<&AlignmentRecord<'a>>> = HashMap::new();
2687        for aln in alignments {
2688            segments
2689                .entry(if aln.flag & 0x40 != 0 { 1 } else { 0 })
2690                .or_default()
2691                .push(aln);
2692        }
2693        // Java counts FSJ support per read group, not per alignment record. The
2694        // temporary set intentionally de-duplicates all linear evidence gathered
2695        // from one read before the local shard counter is incremented.
2696        let mut tem_fsj_keys = HashSet::new();
2697        for seg_idx in [0_i32, 1_i32] {
2698            let Some(seg_alns) = segments.get(&seg_idx) else {
2699                continue;
2700            };
2701            let (read_strand, seq) = match stand_map.get(&seg_idx) {
2702                Some(&(st, ref s)) => (st, s.as_ref()),
2703                None => continue,
2704            };
2705            let slen = seq.len() as i32;
2706            // Mate alignment lookup is invariant for all candidates generated from
2707            // this read side, so keep the reference once and reuse it below.
2708            let mate_seg_alns = segments.get(&(1 - seg_idx));
2709            for aln in seg_alns {
2710                let chr = aln.chrom.as_ref();
2711                if !self.index1.contains_key(chr) {
2712                    continue;
2713                }
2714                let cigar_ref: Cow<'_, str> = if aln.cigar.contains('H') {
2715                    Cow::Owned(aln.cigar.replace('H', "S"))
2716                } else {
2717                    Cow::Borrowed(aln.cigar.as_ref())
2718                };
2719                if cigar_ref == "*" {
2720                    continue;
2721                }
2722                let c = misd(cigar_ref.as_ref(), slen);
2723                if cigar_ref == format!("{}M", slen) {
2724                    let start_tem = aln.pos + 6;
2725                    let end_tem = aln.pos + slen - 7;
2726                    self.collect_fsj_keys_in_range(chr, start_tem, end_tem, 0, &mut tem_fsj_keys);
2727                    continue;
2728                }
2729                if c[0] == -1 || c[0] == 10 {
2730                    if let Some(list) = self.index1.get(chr) {
2731                        let curr_strand = if aln.flag & 0x10 != 0 { '1' } else { '0' };
2732                        // Orientation depends on the alignment strand of the
2733                        // current record, so the normalized strings are hoisted to
2734                        // alignment scope but not further.
2735                        let str_e_owned = if aln.flag & 0x10 != 0 {
2736                            if read_strand == '0' {
2737                                reverse_complement(seq)
2738                            } else {
2739                                seq.to_string()
2740                            }
2741                        } else if read_strand == '0' {
2742                            seq.to_string()
2743                        } else {
2744                            reverse_complement(seq)
2745                        };
2746                        let p_str_owned =
2747                            if let Some(&(p_strand, ref p_seq)) = stand_map.get(&(1 - seg_idx)) {
2748                                if p_strand != curr_strand {
2749                                    p_seq.to_string()
2750                                } else {
2751                                    reverse_complement(p_seq)
2752                                }
2753                            } else {
2754                                String::new()
2755                            };
2756                        let new_num_site = aln.pos;
2757                        // Java parity: fixed seqLen from constructor.
2758                        let bucket_size = self.seq_len.max(1);
2759                        let num1 = (new_num_site - 6) / bucket_size;
2760                        let num2 = (new_num_site + 6) / bucket_size;
2761
2762                        let mut eval_candidate = |cand: &CandidateBreakpoint| -> Result<bool> {
2763                            let e_idx = c[1] + (cand.site - new_num_site);
2764                            if e_idx <= 0 || e_idx > slen {
2765                                return Ok(false);
2766                            }
2767                            let str_f = &str_e_owned[0..e_idx as usize];
2768                            let str3 = if c[0] == 10 {
2769                                let si = slen - c[2];
2770                                if si >= 0 && si <= slen {
2771                                    &str_e_owned[si as usize..]
2772                                } else {
2773                                    ""
2774                                }
2775                            } else {
2776                                "*"
2777                            };
2778                            let site1 = cand.data[0].parse::<i32>().unwrap_or(0);
2779                            let site2 = cand.data[1].parse::<i32>().unwrap_or(0);
2780                            let mut s2_ok = if mate_seg_alns.is_some() { 0 } else { 1 };
2781                            if let Some(mate_seg_alns) = mate_seg_alns {
2782                                for mate_aln in mate_seg_alns {
2783                                    if mate_aln.chrom.as_ref() == chr
2784                                        && mate_aln.mapq >= self.min_mapq_uni
2785                                    {
2786                                        let c_ano = misd(&mate_aln.cigar, slen);
2787                                        let mate_strand =
2788                                            if mate_aln.flag & 0x10 != 0 { '1' } else { '0' };
2789                                        if mate_strand != curr_strand
2790                                            && mate_aln.pos >= site1 - 6
2791                                            && mate_aln.pos + c_ano[3] - 1 <= site2 + 6
2792                                        {
2793                                            s2_ok = 1;
2794                                        } else {
2795                                            s2_ok = -1;
2796                                        }
2797                                        break;
2798                                    }
2799                                }
2800                            }
2801                            let circ_c = vec![
2802                                (if aln.flag & 0x10 != 0 { "1" } else { "0" }).to_string(),
2803                                chr.to_string(),
2804                                "sm".to_string(),
2805                                cand.data[0].clone(),
2806                                cand.data[1].clone(),
2807                                str_f.to_string(),
2808                                p_str_owned.clone(),
2809                                str3.to_string(),
2810                                s2_ok.to_string(),
2811                                cand.data[2].clone(),
2812                                cand.data[3].clone(),
2813                                cand.data[4].clone(),
2814                                aln.mapq.to_string(),
2815                            ];
2816                            let validator_started = profile.map(|_| Instant::now());
2817                            let tag = with_trace_hg2_scope(trace_read, || {
2818                                is_bsj_hg2.is_bsj_hg2(&circ_c, chr_tcga_map.get(chr).unwrap())
2819                            });
2820                            if let Some(profile) = profile {
2821                                profile.candidate_checks.fetch_add(1, Ordering::Relaxed);
2822                                if let Some(validator_started) = validator_started {
2823                                    profile.validator_call_ns.fetch_add(
2824                                        validator_started.elapsed().as_nanos() as u64,
2825                                        Ordering::Relaxed,
2826                                    );
2827                                }
2828                            }
2829                            if trace_read {
2830                                emit_trace_line(&format!(
2831                                    "[TRACE_SCAN2_CAND] id={} type=sm seg={} aln_pos={} chr={} site1={} site2={} cand_site={} cigar={} mapq={} s2_ok={} str_len={} pair_len={} tag={}",
2832                                    id,
2833                                    seg_idx,
2834                                    aln.pos,
2835                                    chr,
2836                                    cand.data[0],
2837                                    cand.data[1],
2838                                    cand.site,
2839                                    cigar_ref.as_ref(),
2840                                    aln.mapq,
2841                                    circ_c[8],
2842                                    circ_c[5].len(),
2843                                    circ_c[6].len(),
2844                                    tag
2845                                ));
2846                            }
2847                            if tag == "0" {
2848                                // Java treats validator tag `0` as "this linear
2849                                // candidate is not a rescuable BSJ, but it still
2850                                // supports the circ as an FSJ competitor". These
2851                                // insertions are one of the main places where
2852                                // BSJ parity can already be perfect while FSJ
2853                                // counts still drift.
2854                                tem_fsj_keys
2855                                    .insert(format!("{}\t{}\t{}", chr, cand.data[0], cand.data[1]));
2856                                Ok(false)
2857                            } else if tag != "2" {
2858                                if let Some(profile) = profile {
2859                                    profile.candidate_hits.fetch_add(1, Ordering::Relaxed);
2860                                }
2861                                let tag_body = &tag[0..tag.len() - 1];
2862                                let mate_label = if seg_idx == 1 { "R1" } else { "R2" };
2863                                results.push(format!(
2864                                    "{}\t{}\t1\t{}\t{}\t{}\t{}\t{}\t{}\t{}\t{}\t{}",
2865                                    id,
2866                                    mate_label,
2867                                    cigar_ref.as_ref(),
2868                                    tag_body,
2869                                    chr,
2870                                    cand.data[0],
2871                                    cand.data[1],
2872                                    cand.data[2],
2873                                    cand.data[3],
2874                                    cand.data[4],
2875                                    tag.chars().next_back().unwrap()
2876                                ));
2877                                Ok(true)
2878                            } else {
2879                                Ok(false)
2880                            }
2881                        };
2882
2883                        // Java parity: num1 bucket reverse order.
2884                        if num1 > 0 && self.site_array1.get(chr).is_some_and(|s| s.contains(&num1))
2885                        {
2886                            let (l1, r1) = Self::bucket_range(list, num1, bucket_size);
2887                            for idx in (l1..r1).rev() {
2888                                let cand = &list[idx];
2889                                let bias = cand.site - new_num_site;
2890                                if bias >= -6 {
2891                                    if bias <= 6 && eval_candidate(cand)? && !trace_all_candidates {
2892                                        return Ok(());
2893                                    }
2894                                } else {
2895                                    break;
2896                                }
2897                            }
2898                        }
2899                        // Java parity: num2 bucket forward order.
2900                        if num2 != num1
2901                            && self.site_array1.get(chr).is_some_and(|s| s.contains(&num2))
2902                        {
2903                            let (l2, r2) = Self::bucket_range(list, num2, bucket_size);
2904                            for idx in l2..r2 {
2905                                let cand = &list[idx];
2906                                let bias = cand.site - new_num_site;
2907                                if bias <= 6 {
2908                                    if eval_candidate(cand)? && !trace_all_candidates {
2909                                        return Ok(());
2910                                    }
2911                                } else {
2912                                    break;
2913                                }
2914                            }
2915                        }
2916                    }
2917                }
2918                if c[0] == 1 || c[0] == 10 {
2919                    let new_site = aln.pos + c[3] - 1;
2920                    if let Some(list) = self.index2.get(chr) {
2921                        let curr_strand = if aln.flag & 0x10 != 0 { '1' } else { '0' };
2922                        let str_e_owned = if aln.flag & 0x10 != 0 {
2923                            if read_strand == '0' {
2924                                reverse_complement(seq)
2925                            } else {
2926                                seq.to_string()
2927                            }
2928                        } else if read_strand == '0' {
2929                            seq.to_string()
2930                        } else {
2931                            reverse_complement(seq)
2932                        };
2933                        let p_str_owned =
2934                            if let Some(&(p_strand, ref p_seq)) = stand_map.get(&(1 - seg_idx)) {
2935                                if p_strand != curr_strand {
2936                                    p_seq.to_string()
2937                                } else {
2938                                    reverse_complement(p_seq)
2939                                }
2940                            } else {
2941                                String::new()
2942                            };
2943                        // Java parity: fixed seqLen from constructor.
2944                        let bucket_size = self.seq_len.max(1);
2945                        let num1 = (new_site - 6) / bucket_size;
2946                        let num2 = (new_site + 6) / bucket_size;
2947
2948                        let mut eval_candidate = |cand: &CandidateBreakpoint| -> Result<bool> {
2949                            let s_idx = if c[0] == 10 {
2950                                // Java parity: SMS right-anchor slicing uses
2951                                // `seqLen - tail_soft_clip + bias`, not
2952                                // `start_soft_clip + mapped_len + bias`.
2953                                // The latter drifts whenever the middle CIGAR
2954                                // contains insertions or deletions.
2955                                slen - c[2] + (cand.site - new_site)
2956                            } else {
2957                                c[1] + (cand.site - new_site)
2958                            };
2959                            if s_idx < 0 || s_idx >= slen {
2960                                return Ok(false);
2961                            }
2962                            let str_f = &str_e_owned[s_idx as usize..];
2963                            let str3 = if c[0] == 10 {
2964                                &str_e_owned[0..c[1] as usize]
2965                            } else {
2966                                "*"
2967                            };
2968                            let site1 = cand.data[0].parse::<i32>().unwrap_or(0);
2969                            let site2 = cand.data[1].parse::<i32>().unwrap_or(0);
2970                            let mut s2_ok = if mate_seg_alns.is_some() { 0 } else { 1 };
2971                            if let Some(mate_seg_alns) = mate_seg_alns {
2972                                for mate_aln in mate_seg_alns {
2973                                    if mate_aln.chrom.as_ref() == chr
2974                                        && mate_aln.mapq >= self.min_mapq_uni
2975                                    {
2976                                        let c_ano = misd(&mate_aln.cigar, slen);
2977                                        let mate_strand =
2978                                            if mate_aln.flag & 0x10 != 0 { '1' } else { '0' };
2979                                        if mate_strand != curr_strand
2980                                            && mate_aln.pos >= site1 - 6
2981                                            && mate_aln.pos + c_ano[3] - 1 <= site2 + 6
2982                                        {
2983                                            s2_ok = 1;
2984                                        } else {
2985                                            s2_ok = -1;
2986                                        }
2987                                        break;
2988                                    }
2989                                }
2990                            }
2991                            let circ_c = vec![
2992                                (if aln.flag & 0x10 != 0 { "1" } else { "0" }).to_string(),
2993                                chr.to_string(),
2994                                "ms".to_string(),
2995                                cand.data[0].clone(),
2996                                cand.data[1].clone(),
2997                                str_f.to_string(),
2998                                p_str_owned.clone(),
2999                                str3.to_string(),
3000                                s2_ok.to_string(),
3001                                cand.data[2].clone(),
3002                                cand.data[3].clone(),
3003                                cand.data[4].clone(),
3004                                aln.mapq.to_string(),
3005                            ];
3006                            let validator_started = profile.map(|_| Instant::now());
3007                            let tag = with_trace_hg2_scope(trace_read, || {
3008                                is_bsj_hg2.is_bsj_hg2(&circ_c, chr_tcga_map.get(chr).unwrap())
3009                            });
3010                            if let Some(profile) = profile {
3011                                profile.candidate_checks.fetch_add(1, Ordering::Relaxed);
3012                                if let Some(validator_started) = validator_started {
3013                                    profile.validator_call_ns.fetch_add(
3014                                        validator_started.elapsed().as_nanos() as u64,
3015                                        Ordering::Relaxed,
3016                                    );
3017                                }
3018                            }
3019                            if trace_read {
3020                                emit_trace_line(&format!(
3021                                    "[TRACE_SCAN2_CAND] id={} type=ms seg={} aln_pos={} chr={} site1={} site2={} cand_site={} cigar={} mapq={} s2_ok={} str_len={} pair_len={} tag={}",
3022                                    id,
3023                                    seg_idx,
3024                                    aln.pos,
3025                                    chr,
3026                                    cand.data[0],
3027                                    cand.data[1],
3028                                    cand.site,
3029                                    cigar_ref.as_ref(),
3030                                    aln.mapq,
3031                                    circ_c[8],
3032                                    circ_c[5].len(),
3033                                    circ_c[6].len(),
3034                                    tag
3035                                ));
3036                            }
3037                            if tag == "0" {
3038                                tem_fsj_keys
3039                                    .insert(format!("{}\t{}\t{}", chr, cand.data[0], cand.data[1]));
3040                                Ok(false)
3041                            } else if tag != "2" {
3042                                if let Some(profile) = profile {
3043                                    profile.candidate_hits.fetch_add(1, Ordering::Relaxed);
3044                                }
3045                                let tag_body = &tag[0..tag.len() - 1];
3046                                let mate_label = if seg_idx == 1 { "R1" } else { "R2" };
3047                                results.push(format!(
3048                                    "{}\t{}\t1\t{}\t{}\t{}\t{}\t{}\t{}\t{}\t{}\t{}",
3049                                    id,
3050                                    mate_label,
3051                                    cigar_ref.as_ref(),
3052                                    tag_body,
3053                                    chr,
3054                                    cand.data[0],
3055                                    cand.data[1],
3056                                    cand.data[2],
3057                                    cand.data[3],
3058                                    cand.data[4],
3059                                    tag.chars().next_back().unwrap()
3060                                ));
3061                                Ok(true)
3062                            } else {
3063                                Ok(false)
3064                            }
3065                        };
3066
3067                        // Java parity: num1 bucket reverse order.
3068                        if self.site_array2.get(chr).is_some_and(|s| s.contains(&num1)) {
3069                            let (l1, r1) = Self::bucket_range(list, num1, bucket_size);
3070                            for idx in (l1..r1).rev() {
3071                                let cand = &list[idx];
3072                                let bias = cand.site - new_site;
3073                                if bias >= -6 {
3074                                    if bias <= 6 && eval_candidate(cand)? && !trace_all_candidates {
3075                                        return Ok(());
3076                                    }
3077                                } else {
3078                                    break;
3079                                }
3080                            }
3081                        }
3082                        // Java parity: num2 bucket forward order.
3083                        if num2 != num1
3084                            && self.site_array2.get(chr).is_some_and(|s| s.contains(&num2))
3085                        {
3086                            let (l2, r2) = Self::bucket_range(list, num2, bucket_size);
3087                            for idx in l2..r2 {
3088                                let cand = &list[idx];
3089                                let bias = cand.site - new_site;
3090                                if bias <= 6 {
3091                                    if eval_candidate(cand)? && !trace_all_candidates {
3092                                        return Ok(());
3093                                    }
3094                                } else {
3095                                    break;
3096                                }
3097                            }
3098                        }
3099                    }
3100                }
3101                // Besides explicit tag==0 additions above, every linear segment
3102                // also contributes bucket-gated FSJ overlaps from its covered
3103                // genomic span, exactly like Java `GetFSJClass.getFSJ(...)`.
3104                let start_tem = aln.pos + 6;
3105                let end_tem = aln.pos + c[3] - 7;
3106                self.collect_fsj_keys_in_range(chr, start_tem, end_tem, c[0], &mut tem_fsj_keys);
3107            }
3108        }
3109        for k in tem_fsj_keys {
3110            *local_fsj.entry(k).or_insert(0) += 1;
3111        }
3112        if let (Some(profile), Some(started)) = (profile, started) {
3113            profile
3114                .group_process_ns
3115                .fetch_add(started.elapsed().as_nanos() as u64, Ordering::Relaxed);
3116            profile.groups.fetch_add(1, Ordering::Relaxed);
3117        }
3118        Ok(())
3119    }
3120
3121    /// Enumerates mate-level Scan2 rescue hits for the post-Summary `.bsj`.
3122    ///
3123    /// This helper keeps the expanded mate view out of the parity path. It uses
3124    /// the display-only claim set from Scan1 and returns at most one first-hit
3125    /// rescue per mate.
3126    fn process_group_view_display<'a>(
3127        &self,
3128        id: &str,
3129        alignments: &[AlignmentRecord<'a>],
3130        stand_map: &HashMap<i32, (char, Cow<'a, str>)>,
3131        scan1_claims: &Scan2DisplayClaims,
3132        chr_tcga_map: &HashMap<String, String>,
3133        is_bsj_hg2: &mut IsBSJHg2,
3134    ) -> Result<Vec<String>> {
3135        let trace_read = should_trace_read(id);
3136        let trace_all_candidates =
3137            trace_read && std::env::var("CIRI_TRACE_ALL_CANDS").ok().as_deref() == Some("1");
3138        if scan1_claims.contains_both_mates(id) {
3139            return Ok(Vec::new());
3140        }
3141        let mut results = Vec::new();
3142        let mut segments: HashMap<i32, Vec<&AlignmentRecord<'a>>> = HashMap::new();
3143        for aln in alignments {
3144            segments
3145                .entry(if aln.flag & 0x40 != 0 { 1 } else { 0 })
3146                .or_default()
3147                .push(aln);
3148        }
3149
3150        'mate: for seg_idx in [1_i32, 0_i32] {
3151            let mate_label = if seg_idx == 1 { "R1" } else { "R2" };
3152            if scan1_claims.contains_mate(id, mate_label) {
3153                continue;
3154            }
3155            let Some(seg_alns) = segments.get(&seg_idx) else {
3156                continue;
3157            };
3158            let (read_strand, seq) = match stand_map.get(&seg_idx) {
3159                Some(&(st, ref s)) => (st, s.as_ref()),
3160                None => continue,
3161            };
3162            let slen = seq.len() as i32;
3163            let mate_seg_alns = segments.get(&(1 - seg_idx));
3164            for aln in seg_alns {
3165                let chr = aln.chrom.as_ref();
3166                if !self.index1.contains_key(chr) {
3167                    continue;
3168                }
3169                let cigar_ref: Cow<'_, str> = if aln.cigar.contains('H') {
3170                    Cow::Owned(aln.cigar.replace('H', "S"))
3171                } else {
3172                    Cow::Borrowed(aln.cigar.as_ref())
3173                };
3174                if cigar_ref == "*" {
3175                    continue;
3176                }
3177                let c = misd(cigar_ref.as_ref(), slen);
3178                if c[0] == -1 || c[0] == 10 {
3179                    if let Some(list) = self.index1.get(chr) {
3180                        let curr_strand = if aln.flag & 0x10 != 0 { '1' } else { '0' };
3181                        let str_e_owned = if aln.flag & 0x10 != 0 {
3182                            if read_strand == '0' {
3183                                reverse_complement(seq)
3184                            } else {
3185                                seq.to_string()
3186                            }
3187                        } else if read_strand == '0' {
3188                            seq.to_string()
3189                        } else {
3190                            reverse_complement(seq)
3191                        };
3192                        let p_str_owned =
3193                            if let Some(&(p_strand, ref p_seq)) = stand_map.get(&(1 - seg_idx)) {
3194                                if p_strand != curr_strand {
3195                                    p_seq.to_string()
3196                                } else {
3197                                    reverse_complement(p_seq)
3198                                }
3199                            } else {
3200                                String::new()
3201                            };
3202                        let new_num_site = aln.pos;
3203                        let bucket_size = self.seq_len.max(1);
3204                        let num1 = (new_num_site - 6) / bucket_size;
3205                        let num2 = (new_num_site + 6) / bucket_size;
3206
3207                        let mut eval_candidate = |cand: &CandidateBreakpoint| -> Result<bool> {
3208                            let e_idx = c[1] + (cand.site - new_num_site);
3209                            if e_idx <= 0 || e_idx > slen {
3210                                return Ok(false);
3211                            }
3212                            let str_f = &str_e_owned[0..e_idx as usize];
3213                            let str3 = if c[0] == 10 {
3214                                let si = slen - c[2];
3215                                if si >= 0 && si <= slen {
3216                                    &str_e_owned[si as usize..]
3217                                } else {
3218                                    ""
3219                                }
3220                            } else {
3221                                "*"
3222                            };
3223                            let site1 = cand.data[0].parse::<i32>().unwrap_or(0);
3224                            let site2 = cand.data[1].parse::<i32>().unwrap_or(0);
3225                            let mut s2_ok = if mate_seg_alns.is_some() { 0 } else { 1 };
3226                            if let Some(mate_seg_alns) = mate_seg_alns {
3227                                for mate_aln in mate_seg_alns {
3228                                    if mate_aln.chrom.as_ref() == chr
3229                                        && mate_aln.mapq >= self.min_mapq_uni
3230                                    {
3231                                        let c_ano = misd(&mate_aln.cigar, slen);
3232                                        let mate_strand =
3233                                            if mate_aln.flag & 0x10 != 0 { '1' } else { '0' };
3234                                        if mate_strand != curr_strand
3235                                            && mate_aln.pos >= site1 - 6
3236                                            && mate_aln.pos + c_ano[3] - 1 <= site2 + 6
3237                                        {
3238                                            s2_ok = 1;
3239                                        } else {
3240                                            s2_ok = -1;
3241                                        }
3242                                        break;
3243                                    }
3244                                }
3245                            }
3246                            let circ_c = vec![
3247                                (if aln.flag & 0x10 != 0 { "1" } else { "0" }).to_string(),
3248                                chr.to_string(),
3249                                "sm".to_string(),
3250                                cand.data[0].clone(),
3251                                cand.data[1].clone(),
3252                                str_f.to_string(),
3253                                p_str_owned.clone(),
3254                                str3.to_string(),
3255                                s2_ok.to_string(),
3256                                cand.data[2].clone(),
3257                                cand.data[3].clone(),
3258                                cand.data[4].clone(),
3259                                aln.mapq.to_string(),
3260                            ];
3261                            let tag = with_trace_hg2_scope(trace_read, || {
3262                                is_bsj_hg2.is_bsj_hg2(&circ_c, chr_tcga_map.get(chr).unwrap())
3263                            });
3264                            if trace_read {
3265                                emit_trace_line(&format!(
3266                                    "[TRACE_SCAN2_DISPLAY] id={} type=sm seg={} aln_pos={} chr={} site1={} site2={} cand_site={} cigar={} mapq={} tag={}",
3267                                    id,
3268                                    seg_idx,
3269                                    aln.pos,
3270                                    chr,
3271                                    cand.data[0],
3272                                    cand.data[1],
3273                                    cand.site,
3274                                    cigar_ref.as_ref(),
3275                                    aln.mapq,
3276                                    tag
3277                                ));
3278                            }
3279                            if tag != "0" && tag != "2" {
3280                                let tag_body = &tag[0..tag.len() - 1];
3281                                results.push(format!(
3282                                    "{}\t{}\t{}\t{}\t{}\t{}\t{}\t{}\t{}\t{}\t{}",
3283                                    id,
3284                                    mate_label,
3285                                    cigar_ref.as_ref(),
3286                                    tag_body,
3287                                    chr,
3288                                    cand.data[0],
3289                                    cand.data[1],
3290                                    cand.data[2],
3291                                    cand.data[3],
3292                                    cand.data[4],
3293                                    tag.chars().next_back().unwrap()
3294                                ));
3295                                Ok(true)
3296                            } else {
3297                                Ok(false)
3298                            }
3299                        };
3300
3301                        if num1 > 0 && self.site_array1.get(chr).is_some_and(|s| s.contains(&num1))
3302                        {
3303                            let (l1, r1) = Self::bucket_range(list, num1, bucket_size);
3304                            for idx in (l1..r1).rev() {
3305                                let cand = &list[idx];
3306                                let bias = cand.site - new_num_site;
3307                                if bias >= -6 {
3308                                    if bias <= 6 && eval_candidate(cand)? && !trace_all_candidates {
3309                                        continue 'mate;
3310                                    }
3311                                } else {
3312                                    break;
3313                                }
3314                            }
3315                        }
3316                        if num2 != num1
3317                            && self.site_array1.get(chr).is_some_and(|s| s.contains(&num2))
3318                        {
3319                            let (l2, r2) = Self::bucket_range(list, num2, bucket_size);
3320                            for idx in l2..r2 {
3321                                let cand = &list[idx];
3322                                let bias = cand.site - new_num_site;
3323                                if bias <= 6 {
3324                                    if eval_candidate(cand)? && !trace_all_candidates {
3325                                        continue 'mate;
3326                                    }
3327                                } else {
3328                                    break;
3329                                }
3330                            }
3331                        }
3332                    }
3333                }
3334                if c[0] == 1 || c[0] == 10 {
3335                    let new_site = aln.pos + c[3] - 1;
3336                    if let Some(list) = self.index2.get(chr) {
3337                        let curr_strand = if aln.flag & 0x10 != 0 { '1' } else { '0' };
3338                        let str_e_owned = if aln.flag & 0x10 != 0 {
3339                            if read_strand == '0' {
3340                                reverse_complement(seq)
3341                            } else {
3342                                seq.to_string()
3343                            }
3344                        } else if read_strand == '0' {
3345                            seq.to_string()
3346                        } else {
3347                            reverse_complement(seq)
3348                        };
3349                        let p_str_owned =
3350                            if let Some(&(p_strand, ref p_seq)) = stand_map.get(&(1 - seg_idx)) {
3351                                if p_strand != curr_strand {
3352                                    p_seq.to_string()
3353                                } else {
3354                                    reverse_complement(p_seq)
3355                                }
3356                            } else {
3357                                String::new()
3358                            };
3359                        let bucket_size = self.seq_len.max(1);
3360                        let num1 = (new_site - 6) / bucket_size;
3361                        let num2 = (new_site + 6) / bucket_size;
3362
3363                        let mut eval_candidate = |cand: &CandidateBreakpoint| -> Result<bool> {
3364                            let s_idx = if c[0] == 10 {
3365                                slen - c[2] + (cand.site - new_site)
3366                            } else {
3367                                c[1] + (cand.site - new_site)
3368                            };
3369                            if s_idx < 0 || s_idx >= slen {
3370                                return Ok(false);
3371                            }
3372                            let str_f = &str_e_owned[s_idx as usize..];
3373                            let str3 = if c[0] == 10 {
3374                                &str_e_owned[0..c[1] as usize]
3375                            } else {
3376                                "*"
3377                            };
3378                            let site1 = cand.data[0].parse::<i32>().unwrap_or(0);
3379                            let site2 = cand.data[1].parse::<i32>().unwrap_or(0);
3380                            let mut s2_ok = if mate_seg_alns.is_some() { 0 } else { 1 };
3381                            if let Some(mate_seg_alns) = mate_seg_alns {
3382                                for mate_aln in mate_seg_alns {
3383                                    if mate_aln.chrom.as_ref() == chr
3384                                        && mate_aln.mapq >= self.min_mapq_uni
3385                                    {
3386                                        let c_ano = misd(&mate_aln.cigar, slen);
3387                                        let mate_strand =
3388                                            if mate_aln.flag & 0x10 != 0 { '1' } else { '0' };
3389                                        if mate_strand != curr_strand
3390                                            && mate_aln.pos >= site1 - 6
3391                                            && mate_aln.pos + c_ano[3] - 1 <= site2 + 6
3392                                        {
3393                                            s2_ok = 1;
3394                                        } else {
3395                                            s2_ok = -1;
3396                                        }
3397                                        break;
3398                                    }
3399                                }
3400                            }
3401                            let circ_c = vec![
3402                                (if aln.flag & 0x10 != 0 { "1" } else { "0" }).to_string(),
3403                                chr.to_string(),
3404                                "ms".to_string(),
3405                                cand.data[0].clone(),
3406                                cand.data[1].clone(),
3407                                str_f.to_string(),
3408                                p_str_owned.clone(),
3409                                str3.to_string(),
3410                                s2_ok.to_string(),
3411                                cand.data[2].clone(),
3412                                cand.data[3].clone(),
3413                                cand.data[4].clone(),
3414                                aln.mapq.to_string(),
3415                            ];
3416                            let tag = with_trace_hg2_scope(trace_read, || {
3417                                is_bsj_hg2.is_bsj_hg2(&circ_c, chr_tcga_map.get(chr).unwrap())
3418                            });
3419                            if trace_read {
3420                                emit_trace_line(&format!(
3421                                    "[TRACE_SCAN2_DISPLAY] id={} type=ms seg={} aln_pos={} chr={} site1={} site2={} cand_site={} cigar={} mapq={} tag={}",
3422                                    id,
3423                                    seg_idx,
3424                                    aln.pos,
3425                                    chr,
3426                                    cand.data[0],
3427                                    cand.data[1],
3428                                    cand.site,
3429                                    cigar_ref.as_ref(),
3430                                    aln.mapq,
3431                                    tag
3432                                ));
3433                            }
3434                            if tag != "0" && tag != "2" {
3435                                let tag_body = &tag[0..tag.len() - 1];
3436                                results.push(format!(
3437                                    "{}\t{}\t{}\t{}\t{}\t{}\t{}\t{}\t{}\t{}\t{}",
3438                                    id,
3439                                    mate_label,
3440                                    cigar_ref.as_ref(),
3441                                    tag_body,
3442                                    chr,
3443                                    cand.data[0],
3444                                    cand.data[1],
3445                                    cand.data[2],
3446                                    cand.data[3],
3447                                    cand.data[4],
3448                                    tag.chars().next_back().unwrap()
3449                                ));
3450                                Ok(true)
3451                            } else {
3452                                Ok(false)
3453                            }
3454                        };
3455
3456                        if self.site_array2.get(chr).is_some_and(|s| s.contains(&num1)) {
3457                            let (l1, r1) = Self::bucket_range(list, num1, bucket_size);
3458                            for idx in (l1..r1).rev() {
3459                                let cand = &list[idx];
3460                                let bias = cand.site - new_site;
3461                                if bias >= -6 {
3462                                    if bias <= 6 && eval_candidate(cand)? && !trace_all_candidates {
3463                                        continue 'mate;
3464                                    }
3465                                } else {
3466                                    break;
3467                                }
3468                            }
3469                        }
3470                        if num2 != num1
3471                            && self.site_array2.get(chr).is_some_and(|s| s.contains(&num2))
3472                        {
3473                            let (l2, r2) = Self::bucket_range(list, num2, bucket_size);
3474                            for idx in l2..r2 {
3475                                let cand = &list[idx];
3476                                let bias = cand.site - new_site;
3477                                if bias <= 6 {
3478                                    if eval_candidate(cand)? && !trace_all_candidates {
3479                                        continue 'mate;
3480                                    }
3481                                } else {
3482                                    break;
3483                                }
3484                            }
3485                        }
3486                    }
3487                }
3488            }
3489        }
3490
3491        Ok(results)
3492    }
3493}