fix: address final dual-review findings

Reject non-finite downsample fraction; fix parallel-assigner invalid-UMI parity
(invalid reads -> None in both paths); log aligner try_wait errors; mark the
terminal extract chain tail as SerializedBytes; strengthen oracles for dedup,
coalesce, sort, runall, streaming, and the driver round-robin path.

Author: nh13

benches/core_functions.rs

@@ -343,10 +343,12 @@ fn bench_vanilla_consensus_caller(c: &mut Criterion) {
             .map(|i| {
                 let mut read_seq = seq.clone();
                 // Introduce a single-base error on every 10th read. NOTE: this
-                // only fires for `i` a positive multiple of 10, so the smaller
-                // molecule sizes here (num_reads in {2, 3, 5}) get NO injected
-                // error and exercise the unanimous fast path; only num_reads
-                // 10 and 20 actually introduce a mismatch.
+                // only fires for `i` a positive multiple of 10. Because the loop
+                // is `0..num_reads`, `i` only reaches 10 when `num_reads == 20`;
+                // for `num_reads` in {2, 3, 5, 10} the index never hits 10, so
+                // those sizes get NO injected error and exercise the unanimous
+                // fast path. Only the `num_reads == 20` case introduces a
+                // mismatch (at i = 10).
                 if i > 0 && i % 10 == 0 {
                     read_seq[i % read_len] = b"TGCA"[(read_seq[i % read_len] as usize) % 4];
                 }

crates/fgumi-consensus/src/base_builder.rs

@@ -821,6 +821,15 @@ mod tests {
     }
 
     // ==================== Observation-count saturation (S7-001 / FU-002) ====================
+    //
+    // NOTE on fgbio parity: clamping `observations_for_base` / `contributions` at
+    // `u16::MAX` is an INTENTIONAL divergence from fgbio once a family's counts
+    // exceed the `u16` ceiling — `ConsensusBaseBuilder` uses `u16` counters, so it
+    // saturates where fgbio (wider counters) keeps counting. This only matters for
+    // pathologically large families (>65535 observations of a single base, or a
+    // cross-base contribution sum past the ceiling), well beyond any realistic UMI
+    // family. The tests below pin the clamp (no wrap toward zero); they are NOT a
+    // parity bug, so do not "fix" the saturation to match fgbio.
 
     #[test]
     fn test_observations_saturate_at_u16_max() {

crates/fgumi-consensus/src/codec_caller.rs

@@ -4299,6 +4299,13 @@ mod tests {
         }
     }
 
+    // NOTE (no fgbio oracle here, by design): the cD/cM/cE and ad/ae/bd/be
+    // saturation behavior is pinned by the helper-clamp tests below plus the
+    // public BAM-tag surface exercised elsewhere; we do NOT add an fgbio
+    // dependency to cross-check serialized tags. fgbio's codec consensus
+    // operates on raw bytes (established in S7-002-bitenc-perf-analysis.md), and
+    // the saturation at u16::MAX is an intentional u16-counter divergence for
+    // pathological families — not a parity bug to validate against fgbio.
     #[test]
     fn test_duplex_depth_error_saturate_to_u16_at_high_counts() {
         let options = CodecConsensusOptions::default();

crates/fgumi-pipeline-core/src/runtime/driver.rs

@@ -670,12 +670,20 @@ mod tests {
         // disable the sticky fast-path so the loop terminates rather than hangs.
         run_worker_loop(&mut worker, &mut entries, &contexts, &drain_counters, &signal, None);
 
-        // The source must have been called at least twice: once idle (sticky),
-        // then again to Finished (round-robin).
-        assert!(
-            calls.load(Ordering::Relaxed) >= 2,
-            "source should idle once then finish via round-robin, got {} call(s)",
-            calls.load(Ordering::Relaxed)
+        // The source must have been called EXACTLY twice: call 1 = idle in the
+        // sticky fast-path (`NoProgress`, which does NOT remove it there — that
+        // block only reaps a `Finished`), call 2 = `Finished` during the
+        // round-robin pass. An exact count is the branch-specific signal for the
+        // `removed_sticky_owner` path: if that round-robin branch had failed to
+        // clear `sticky_live`, the next outer iteration's sticky fast-path would
+        // re-invoke the (already-removed-from-`live`) source, producing a third
+        // call (or the loop would spin / hang). `== 2` therefore proves removal
+        // happened via round-robin AND that it correctly disabled the fast path.
+        assert_eq!(
+            calls.load(Ordering::Relaxed),
+            2,
+            "source must be called exactly twice (sticky idle, then round-robin finish); \
+             a different count means the removed_sticky_owner branch did not gate the fast path",
         );
     }
 }

src/lib/aligner.rs

@@ -203,19 +203,34 @@ impl AlignerProcess {
         // Wait for the process to actually exit so its resources are cleaned up.
         let deadline = std::time::Instant::now() + Duration::from_secs(1);
         // Tracks whether the loop gave up because the 1s deadline elapsed while
-        // the process was still alive. Only that case warrants a warning; a
-        // successful reap or a non-leak `try_wait` error (e.g. ECHILD) does not.
+        // the process was still alive. Only that case warrants the leak warning.
         let mut deadline_exceeded = false;
-        // `Ok(None)` means still alive — keep polling. Any other result
-        // (`Ok(Some)` = reaped, or `Err` such as ECHILD = already reaped) is a
-        // non-leak: exit without warning. Only hitting the deadline while still
-        // alive sets `deadline_exceeded`.
-        while let Ok(None) = self.child.try_wait() {
-            if std::time::Instant::now() >= deadline {
-                deadline_exceeded = true;
-                break;
+        loop {
+            match self.child.try_wait() {
+                // Reaped cleanly: resources cleaned up, no warning.
+                Ok(Some(_)) => break,
+                // Still alive: keep polling until the deadline.
+                Ok(None) => {
+                    if std::time::Instant::now() >= deadline {
+                        deadline_exceeded = true;
+                        break;
+                    }
+                    thread::sleep(Duration::from_millis(50));
+                }
+                // `try_wait` failed. The most common case here is `ECHILD`
+                // (the child was already reaped, e.g. by a SIGCHLD handler),
+                // which is a benign non-leak. But we do NOT silently bypass
+                // the leak path for *any* error — an unexpected errno could
+                // mean the process is still around. Log it with context and
+                // stop polling (we cannot make progress on a `try_wait` error).
+                Err(err) => {
+                    log::warn!(
+                        "aligner process (pid {pid}) try_wait failed after SIGKILL: {err}; \
+                         assuming already reaped (e.g. ECHILD) but unable to confirm exit"
+                    );
+                    break;
+                }
             }
-            thread::sleep(Duration::from_millis(50));
         }
         if deadline_exceeded {
             log::warn!(

src/lib/commands/downsample.rs

@@ -95,9 +95,11 @@ pub struct Downsample {
 ///
 /// # Errors
 ///
-/// Returns an error if `fraction` is `<= 0.0` or `> 1.0`.
+/// Returns an error if `fraction` is non-finite (NaN/±inf), `<= 0.0`, or `> 1.0`.
 fn validate_fraction(fraction: f64) -> Result<()> {
-    if fraction <= 0.0 || fraction > 1.0 {
+    // Reject non-finite values (NaN/±inf) first, since NaN comparisons are always
+    // false and would otherwise slip past the range test below.
+    if !fraction.is_finite() || fraction <= 0.0 || fraction > 1.0 {
         bail!("--fraction must be between 0.0 (exclusive) and 1.0 (inclusive), got {}", fraction);
     }
     Ok(())
@@ -522,6 +524,15 @@ mod tests {
         assert!(validate_fraction(1.0).is_ok());
     }
 
+    #[test]
+    fn test_validate_fraction_non_finite() {
+        // NaN and ±inf must be rejected: NaN comparisons are always false and
+        // would otherwise slip past the `<= 0.0 || > 1.0` range test.
+        assert!(validate_fraction(f64::NAN).is_err());
+        assert!(validate_fraction(f64::INFINITY).is_err());
+        assert!(validate_fraction(f64::NEG_INFINITY).is_err());
+    }
+
     #[test]
     fn test_write_histogram() {
         use tempfile::NamedTempFile;

src/lib/commands/runall.rs

@@ -456,20 +456,25 @@ pub struct RunAll {
     #[command(flatten)]
     pub correct_opts: crate::commands::correct::MultiCorrectOptions,
 
-    // ───────── aligner-side options (used only when --start-from=align) ─────────
+    // ───────── aligner-side options (used whenever the chain includes Stage::Align) ─────────
     /// Per-stage aligner tuning, exposed as `--aligner::preset`,
     /// `--aligner::command`, `--aligner::threads`, `--aligner::chunk-size`
     /// via the `MultiAlignerOptions` companion struct (generated by
     /// `#[multi_options]` on `AlignerOptions` in `crate::aligner`).
-    /// Ignored when `--start-from` is not `align`.
+    /// Used whenever the derived stage chain includes `Stage::Align` — not
+    /// only on `--start-from=align`, but also on fused runs that reach align
+    /// from an upstream start (e.g. `--start-from extract` or
+    /// `--start-from correct` with a `--stop-after` past zipper). Ignored only
+    /// when the chain has no align stage.
     #[command(flatten)]
     pub aligner_opts: crate::aligner::MultiAlignerOptions,
 
     /// Override path for the aligner binary (preset mode only). When
     /// unset, the preset's binary (`bwa-mem3` / `bwa`) is found via
     /// `which::which()` on `PATH`. Rejected with a clear error if
     /// `--aligner::command` is used (command mode owns its own
-    /// binary). Ignored when `--start-from` is not `align`.
+    /// binary). Used whenever the chain includes `Stage::Align` (see
+    /// `--aligner::*` above); ignored only when the chain has no align stage.
     ///
     /// Note: this flag is at the runall top level (`--aligner-bin`),
     /// NOT inside the `--aligner::*` family. The `--aligner::*` flags
@@ -569,12 +574,16 @@ pub struct RunAll {
     ///
     /// **Input-state hazard:** the validator cannot check whether
     /// the on-disk BAM actually matches the declared `start_from`.
-    /// Passing a raw-unsorted BAM to `--start-from consensus`, or a
-    /// sorted-but-not-grouped BAM to `--start-from consensus`, will
-    /// not produce a clear error today; the chain will run on
-    /// whatever it finds. Make sure the input matches the declared
-    /// start stage. (The consensus algorithm itself is chosen by
-    /// `--consensus {simplex,duplex,codec}`, not by the start value.)
+    /// Because `--start-from consensus` reuses the `group → consensus`
+    /// chain (see the Executor honesty note above), a *sorted-but-not-grouped*
+    /// BAM is handled correctly — it is grouped in-pipeline before the
+    /// consensus caller. The remaining hazard is a *raw-unsorted* BAM:
+    /// `--start-from consensus` does not sort, so the position-grouping
+    /// step will see unsorted input and silently produce wrong groups
+    /// rather than a clear error. Make sure a `consensus` start is given
+    /// at least template-coordinate-sorted input. (The consensus algorithm
+    /// itself is chosen by `--consensus {simplex,duplex,codec}`, not by the
+    /// start value.)
     ///
     /// Case-insensitive. Must satisfy `start_from.ord() <=
     /// stop_after.ord()` and the consensus-terminal rules

src/lib/pipeline/chains/builder.rs

@@ -1407,7 +1407,11 @@ impl<'a> ChainBuilder<'a> {
                 .pipeline
                 .append_step(SerializeBamRecords::new(self.tuning.per_step_byte_limit), tail);
             self.current_tail = Some(tail);
-            self.chain_tail_kind = ChainTailKind::DecodedRecordBatch;
+            // SerializeBamRecords emits DecompressedBlock (serialized bytes), not
+            // DecodedRecordBatch. Mark the tail honestly as SerializedBytes, matching
+            // every other terminal serialize path (e.g. add_dedup); nothing downstream
+            // of a Terminal stage reads this, but the kind must never lie.
+            self.chain_tail_kind = ChainTailKind::SerializedBytes;
         } else {
             self.current_tail = Some(tail);
             self.chain_tail_kind = ChainTailKind::BamTemplateBatch;

src/lib/pipeline/steps/coalesce.rs

@@ -258,10 +258,33 @@ mod tests {
     /// ~512 KB block (caught by the per-block size bound below).
     const COALESCE_INPUT_BLOCKS: usize = 64 * 8;
 
+    /// Deterministic, per-block-unique fill for the block with the given
+    /// `batch_serial`. The bytes vary by position (a tiny LCG seeded by the
+    /// serial) so two distinct blocks never share a byte pattern — this is what
+    /// lets the full-stream oracle catch reorder/drop/duplicate bugs, not just
+    /// total-byte conservation. Both the source and the independently-built
+    /// expected stream call this, so the only way the streams match is if
+    /// coalesce concatenates every block exactly once, in order.
+    fn block_fill(batch_serial: u64) -> Vec<u8> {
+        // Seed off the serial; +1 avoids a degenerate all-zero seed for serial 0.
+        let mut state = batch_serial.wrapping_mul(0x9E37_79B9_7F4A_7C15).wrapping_add(1);
+        (0..COALESCE_INPUT_BLOCK_BYTES)
+            .map(|_| {
+                // SplitMix64-style step for a well-mixed per-byte sequence.
+                state = state.wrapping_add(0x9E37_79B9_7F4A_7C15);
+                let mut z = state;
+                z = (z ^ (z >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9);
+                z = (z ^ (z >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB);
+                // Keep the low byte; truncation is intentional (we want a u8 fill).
+                ((z ^ (z >> 31)) & 0xFF) as u8
+            })
+            .collect()
+    }
+
     /// Source emitting `remaining` fixed-size `DecompressedBlock`s via a shared
     /// atomic counter (safe for Serial single-worker execution). Each block's
-    /// `bytes` is `COALESCE_INPUT_BLOCK_BYTES` of a deterministic fill so the
-    /// sink can verify byte preservation without an ordering assumption.
+    /// `bytes` is a per-block-unique `block_fill(batch_serial)` so the sink can
+    /// reconstruct and verify the exact concatenated byte stream.
     #[derive(Clone)]
     struct BlockSource {
         remaining: Arc<AtomicU64>,
@@ -283,10 +306,7 @@ mod tests {
             if n == 0 {
                 return Ok(StepOutcome::Finished);
             }
-            let block = DecompressedBlock {
-                batch_serial: n,
-                bytes: vec![0xCD; COALESCE_INPUT_BLOCK_BYTES],
-            };
+            let block = DecompressedBlock { batch_serial: n, bytes: block_fill(n) };
             match ctx.outputs.push(block) {
                 Ok(()) => {
                     self.remaining.fetch_sub(1, AtomicOrd::AcqRel);
@@ -302,10 +322,12 @@ mod tests {
     }
 
     /// Sink recording the byte length of every emitted block (so the test can
-    /// bound the per-block size) plus the running total.
+    /// bound the per-block size) plus the exact concatenated byte stream (so the
+    /// test can compare it against an independently-built expected stream).
     #[derive(Clone)]
     struct SizeRecordingSink {
         sizes: Arc<Mutex<Vec<usize>>>,
+        stream: Arc<Mutex<Vec<u8>>>,
     }
     impl Step for SizeRecordingSink {
         type Input = DecompressedBlock;
@@ -322,13 +344,12 @@ mod tests {
         fn try_run(&mut self, ctx: &mut StepCtx<'_, Self>) -> io::Result<StepOutcome> {
             match ctx.input.pop() {
                 Some(block) => {
-                    // Every byte must be the source's fill — proving the
-                    // concatenation neither drops nor corrupts bytes.
-                    assert!(
-                        block.bytes.iter().all(|&b| b == 0xCD),
-                        "coalesced block carries unexpected bytes"
-                    );
+                    // Record the per-block size for the memory bound, and append
+                    // the bytes verbatim so the test can compare the full stream
+                    // against the expected concatenation (catches reorder /
+                    // drop / duplicate, not just total-byte conservation).
                     self.sizes.lock().expect("sink mutex").push(block.bytes.len());
+                    self.stream.lock().expect("stream mutex").extend_from_slice(&block.bytes);
                     Ok(StepOutcome::Progress)
                 }
                 None if ctx.input.is_drained() => Ok(StepOutcome::Finished),
@@ -342,16 +363,18 @@ mod tests {
 
     #[test]
     fn coalesce_flushes_at_threshold_and_preserves_bytes() {
-        let remaining = Arc::new(AtomicU64::new(u64::try_from(COALESCE_INPUT_BLOCKS).unwrap()));
+        let n_blocks = u64::try_from(COALESCE_INPUT_BLOCKS).unwrap();
+        let remaining = Arc::new(AtomicU64::new(n_blocks));
         let sizes = Arc::new(Mutex::new(Vec::new()));
+        let stream = Arc::new(Mutex::new(Vec::new()));
 
         let coalesce = CoalesceBytes::new(COALESCE_THRESHOLD_BYTES, 4 * 1024);
 
         let builder = PipelineBuilder::new();
         builder
             .chain(BlockSource { remaining: Arc::clone(&remaining) })
             .chain(coalesce)
-            .chain(SizeRecordingSink { sizes: Arc::clone(&sizes) })
+            .chain(SizeRecordingSink { sizes: Arc::clone(&sizes), stream: Arc::clone(&stream) })
             .into_sink_marker();
 
         let pipeline = builder.build().unwrap();
@@ -365,6 +388,26 @@ mod tests {
         // Byte conservation: every input byte reaches the sink exactly once.
         assert_eq!(total_out, total_in, "coalesce dropped or duplicated bytes");
 
+        // Byte-stream identity: independently build the expected concatenation in
+        // the source's emission order (`BlockSource` counts the shared atomic
+        // DOWN from `n_blocks` to 1, so it emits `batch_serial = N, N-1, …, 1`)
+        // and require the sink's flat byte stream to match it exactly. Because
+        // each block's fill is per-block-unique (`block_fill`), this fails if any
+        // block is reordered, dropped, or duplicated — a far stronger oracle than
+        // the aggregate-count check above.
+        let expected_stream: Vec<u8> = (1..=n_blocks).rev().flat_map(block_fill).collect();
+        let actual_stream = stream.lock().expect("stream mutex").clone();
+        assert_eq!(
+            actual_stream.len(),
+            expected_stream.len(),
+            "coalesced stream length differs from expected"
+        );
+        assert!(
+            actual_stream == expected_stream,
+            "coalesced byte stream diverges from the expected in-order concatenation \
+             (reorder/drop/duplicate)"
+        );
+
         // Memory bound: the step flushes once `pending` reaches the threshold,
         // then resets. Each emitted block therefore carries between
         // `threshold` and `threshold + (MAX_BATCHES_PER_LOCK - 1) * input` —