fix(codec): avoid MatchError on dovetail FR pairs with asymmetric isFrPair

CodecConsensusCaller filtered primaries with a per-record `isFrPair` check
then pattern-matched each read-name bucket as `Seq(rec, mate)`. For dovetail
FR pairs whose aligned ends coincide (e.g. soft-clipping producing TLEN=+/-1),
htsjdk's `SamPairUtil.getPairOrientation` returns FR for R1 but RF for R2,
which kept one mate and dropped the other and produced a singleton bucket that
blew up with `scala.MatchError`.

Group primaries by read name first and keep only buckets that form a single
primary FR pair, where the FR check is computed at the pair level from the
unclipped 5' positions of both records and is therefore symmetric in its
arguments. Templates that do not form a clean primary FR pair (singletons,
tandems, RF, cross-contig) are now rejected cleanly with a new
`NotPrimaryFrPair` reason instead of crashing.

Preserve BAM-iteration order of templates when grouping by read name. The
previous `groupBy(_.name).toSeq` produced a template order that depended on
the JVM's `HashMap` layout, which propagated through `filterToMostCommonAlignment`
and caused `maxBy(_.cigar.lengthOnTarget)` to break ties in hash order when
selecting `longestR1Alignment` / `longestR2Alignment`. That could pair
records from different templates, producing spurious `IndelErrorBetweenStrands`
rejections when the resulting overlap geometry did not agree across the
strands. A new `orderedPrimaryPairs` helper groups by read name but orders
templates by first-occurrence in the input.

Includes unit tests for the new helpers and a regression test that reproduces
the exact geometry (68S53M8S / 28S48M53S, aligned ends coinciding) observed in
real CODEC data.

Author: nh13

src/main/scala/com/fulcrumgenomics/umi/CodecConsensusCaller.scala

@@ -170,10 +170,24 @@ class CodecConsensusCaller(readNamePrefix: String,
       Nil
     }
     else {
-      // Extract just the primary alignments and then clip where they extend past the ends of their mates
-      // TODO: Remove the isFrPair here and handle chimeric reads or reads with one unmapped etc.
-      val primaries = pairs.filterNot(r => r.secondary || r.supplementary).filter(_.isFrPair)
-      primaries.groupBy(_.name).foreach { case (_, Seq(rec, mate)) => clipper.clipExtendingPastMateEnds(rec, mate) }
+      // Group the primary alignments by read name and retain only templates that form a single
+      // primary FR pair. The FR check is performed at the pair level (with both records in hand)
+      // because htsjdk's per-record `SamPairUtil.getPairOrientation` can disagree between mates
+      // on dovetail pairs whose aligned ends coincide (TLEN=+/-1), which previously caused a
+      // `scala.MatchError` on a singleton read-name bucket here.
+      // Template order is preserved from BAM-iteration order (not hash order) so that downstream
+      // `maxBy` tie-breaks on `cigar.lengthOnTarget` are stable across JVMs.
+      // TODO: handle chimeric reads or reads with one unmapped etc.
+      val primaryPairs = CodecConsensusCaller.orderedPrimaryPairs(
+        pairs.filterNot(r => r.secondary || r.supplementary)
+      )
+      val (frPairs, nonFrPairs) = primaryPairs.partition {
+        case (_, Seq(a, b)) => CodecConsensusCaller.isPrimaryFrPair(a, b)
+        case _              => false
+      }
+      rejectRecords(nonFrPairs.flatMap(_._2), RejectionReason.NotPrimaryFrPair)
+      frPairs.foreach { case (_, Seq(rec, mate)) => clipper.clipExtendingPastMateEnds(rec, mate) }
+      val primaries = frPairs.flatMap(_._2)
 
       val r1s = filterToMostCommonAlignment(primaries.filter(_.firstOfPair).map(toSourceReadForCodec))
       val r2s = filterToMostCommonAlignment(primaries.filter(_.secondOfPair).map(toSourceReadForCodec))
@@ -375,3 +389,36 @@ class CodecConsensusCaller(readNamePrefix: String,
     )
   }
 }
+
+object CodecConsensusCaller {
+  /** Groups primary records by read name while preserving BAM-iteration order of templates.
+    *
+    * Scala's `Seq.groupBy` returns a hash-keyed `Map`, so calling `.toSeq` on it produces a
+    * template order that depends on the JVM's `String#hashCode` and the `HashMap` layout.
+    * Downstream code selects the longest R1 and R2 via `maxBy(_.cigar.lengthOnTarget)`, which
+    * returns the first element in iteration order on ties; relying on hash order there yields
+    * non-obvious, JVM-dependent winners. This helper instead orders templates by the first
+    * occurrence of each read name in the input (i.e. BAM-iteration order), so tie-breaks are
+    * stable and match the order the records were observed.
+    */
+  private[umi] def orderedPrimaryPairs(records: Seq[SamRecord]): Seq[(String, Seq[SamRecord])] = {
+    val byName = records.groupBy(_.name)
+    records.iterator.map(_.name).distinct.map(name => (name, byName(name))).toSeq
+  }
+
+  /** Returns true if the two primary records form an FR pair. The check is symmetric in its
+    * arguments so that both records of a pair always produce the same classification; htsjdk's
+    * per-record `SamPairUtil.getPairOrientation` can disagree between mates on dovetail pairs
+    * whose aligned ends coincide, which is why this is computed from unclipped 5' positions
+    * rather than delegated to `SamRecord.isFrPair`.
+    */
+  private[umi] def isPrimaryFrPair(a: SamRecord, b: SamRecord): Boolean = {
+    if (!a.mapped || !b.mapped) false
+    else if (a.refIndex != b.refIndex) false
+    else if (a.positiveStrand == b.positiveStrand) false
+    else {
+      val (fwd, rev) = if (a.positiveStrand) (a, b) else (b, a)
+      fwd.unclippedStart <= rev.unclippedEnd
+    }
+  }
+}

src/main/scala/com/fulcrumgenomics/umi/UmiConsensusCaller.scala

@@ -85,6 +85,7 @@ object UmiConsensusCaller {
     case object IndelErrorBetweenStrands extends _RejectionReason("indel_error_between_strands", "Indel error between top/bottom strands", false, false, true)
     case object HighDuplexDisagreement   extends _RejectionReason("high_duplex_disagreement",    "Too many errors between top/bottoms strands", false, false, true)
     case object ClipOverlapFailed        extends _RejectionReason("clip_overlap_failed",         "See https://github.com/fulcrumgenomics/fgbio/issues/1090", false, false, true)
+    case object NotPrimaryFrPair         extends _RejectionReason("not_primary_fr_pair",         "Template did not have a single primary FR pair of reads", false, false, true)
   }
 
   /** Metric class for outputting consensus calling statistics. */

src/test/scala/com/fulcrumgenomics/umi/CodecConsensusCallerTest.scala

@@ -207,6 +207,38 @@ class CodecConsensusCallerTest extends UnitSpec with OptionValues {
     caller.consensusReadsFromSamRecords(rfPair) shouldBe Seq()
   }
 
+  it should "not throw on an FR dovetail pair whose aligned ends coincide (htsjdk asymmetric isFrPair)" in {
+    // Reproduces a real template from a HEK293T CODEC dataset in which the aligned end of the
+    // reverse-strand read coincides with the aligned start of the forward-strand read and
+    // soft-clipping produces TLEN=+/-1. For this geometry htsjdk's
+    // `SamPairUtil.getPairOrientation` returns FR for R1 and RF for R2, which previously caused
+    // `CodecConsensusCaller` to drop R2 via a per-record `isFrPair` filter and then throw
+    // `scala.MatchError` on the resulting singleton read-name bucket.
+    val builder = new SamBuilder(readLength=129, baseQuality=35)
+    val raw = builder.addPair(
+      contig = 0, start1 = 96, start2 = 49,
+      cigar1 = "68S53M8S", cigar2 = "28S48M53S",
+      attrs  = Map(("RX", "ACC-TGA"), ("MI", "hi"))
+    ).tapEach(setReadSequence)
+
+    val caller = new CodecConsensusCaller(readNamePrefix="codec", minReadsPerStrand=1, minDuplexLength=1)
+    noException should be thrownBy caller.consensusReadsFromSamRecords(raw)
+  }
+
+  it should "not throw on a template whose primary read-name bucket has only one record" in {
+    // Exercise the defensive path for a degenerate template that reaches the CODEC caller with
+    // only one primary record per read name (e.g. an orphan introduced upstream). Previously
+    // the `Seq(rec, mate)` pattern match at line 176 crashed with MatchError.
+    val builder = new SamBuilder(readLength=30, baseQuality=35)
+    val pair    = builder.addPair(
+      contig=0, start1=1, start2=11, attrs=Map(("RX", "ACC-TGA"), ("MI", "hi"))
+    ).tapEach(setReadSequence)
+    val onlyR1 = pair.filter(_.firstOfPair)
+
+    val caller = new CodecConsensusCaller(readNamePrefix="codec", minReadsPerStrand=1, minDuplexLength=1)
+    caller.consensusReadsFromSamRecords(onlyR1) shouldBe Seq()
+  }
+
   it should "not emit a consensus when the reads are a cross-chromosomal chimeric pair" in {
     val builder = new SamBuilder(readLength=30, baseQuality=35)
     val caller = new CodecConsensusCaller(readNamePrefix="codec", minReadsPerStrand=1, minDuplexLength=1)
@@ -290,6 +322,86 @@ class CodecConsensusCallerTest extends UnitSpec with OptionValues {
       .consensusReadsFromSamRecords(raw) should have length 0
   }
 
+  //////////////////////////////////////////////////////////////////////////////
+  // Tests for orderedPrimaryPairs
+  //////////////////////////////////////////////////////////////////////////////
+
+  "CodecConsensusCaller.orderedPrimaryPairs" should "order templates by first-occurrence BAM-iteration order" in {
+    // Names chosen so that Scala's `HashMap` iteration order on `groupBy(_.name).toSeq` is
+    // extremely unlikely to coincide with insertion order; the helper must return them in the
+    // order they were observed in the input regardless of hash layout.
+    val builder = new SamBuilder(readLength=30, baseQuality=35)
+    val names   = Seq("zulu", "alpha", "mike", "bravo", "yankee")
+    names.foreach { n =>
+      builder.addPair(name=n, contig=0, start1=1, start2=11).tapEach(setReadSequence)
+    }
+    val ordered = CodecConsensusCaller.orderedPrimaryPairs(builder.toSeq)
+    ordered.map(_._1) shouldBe names
+    ordered.foreach { case (_, recs) => recs should have size 2 }
+  }
+
+  it should "ignore secondary/supplementary records only when the caller pre-filters them" in {
+    // The helper itself does not filter; callers pass pre-filtered primaries. This test documents
+    // that contract by passing records that include a supplementary alignment and verifying it is
+    // grouped alongside the primary under the same read name.
+    val builder = new SamBuilder(readLength=30, baseQuality=35)
+    val pair    = builder.addPair(name="t1", contig=0, start1=1, start2=11).tapEach(setReadSequence)
+    pair.head.supplementary = true
+    val ordered = CodecConsensusCaller.orderedPrimaryPairs(builder.toSeq)
+    ordered.map(_._1) shouldBe Seq("t1")
+    ordered.head._2 should have size 2
+  }
+
+  //////////////////////////////////////////////////////////////////////////////
+  // Tests for isPrimaryFrPair
+  //////////////////////////////////////////////////////////////////////////////
+
+  "CodecConsensusCaller.isPrimaryFrPair" should "return the same answer regardless of argument order" in {
+    val builder = new SamBuilder(readLength=129, baseQuality=35)
+    val pair = builder.addPair(
+      contig=0, start1=96, start2=49, cigar1="68S53M8S", cigar2="28S48M53S"
+    ).tapEach(setReadSequence)
+    val r1 = pair.head
+    val r2 = pair.last
+    CodecConsensusCaller.isPrimaryFrPair(r1, r2) shouldBe CodecConsensusCaller.isPrimaryFrPair(r2, r1)
+  }
+
+  it should "classify a typical FR pair as FR" in {
+    val builder = new SamBuilder(readLength=30, baseQuality=35)
+    val Seq(r1, r2) = builder.addPair(contig=0, start1=10, start2=50).toSeq
+    CodecConsensusCaller.isPrimaryFrPair(r1, r2) shouldBe true
+  }
+
+  it should "classify an RF pair as not FR" in {
+    val builder = new SamBuilder(readLength=30, baseQuality=35)
+    val Seq(r1, r2) = builder.addPair(
+      contig=0, start1=50, start2=100, strand1=Minus, strand2=Plus
+    ).toSeq
+    CodecConsensusCaller.isPrimaryFrPair(r1, r2) shouldBe false
+  }
+
+  it should "reject cross-chromosomal pairs" in {
+    val builder = new SamBuilder(readLength=30, baseQuality=35)
+    val Seq(r1, r2) = builder.addPair(contig=0, start1=10, start2=50).toSeq
+    r1.refIndex = 0
+    r2.refIndex = 1
+    CodecConsensusCaller.isPrimaryFrPair(r1, r2) shouldBe false
+  }
+
+  it should "reject a pair with one mate unmapped" in {
+    val builder = new SamBuilder(readLength=30, baseQuality=35)
+    val Seq(r1, r2) = builder.addPair(contig=0, start1=10, start2=10, unmapped2=true).toSeq
+    CodecConsensusCaller.isPrimaryFrPair(r1, r2) shouldBe false
+  }
+
+  it should "reject a tandem pair" in {
+    val builder = new SamBuilder(readLength=30, baseQuality=35)
+    val Seq(r1, r2) = builder.addPair(
+      contig=0, start1=10, start2=50, strand1=Plus, strand2=Plus
+    ).toSeq
+    CodecConsensusCaller.isPrimaryFrPair(r1, r2) shouldBe false
+  }
+
   //////////////////////////////////////////////////////////////////////////////
   // Tests for quality masking
   //////////////////////////////////////////////////////////////////////////////