diff --git a/src/commands/clip.rs b/src/commands/clip.rs
index b5229f2d6..3854583ee 100644
--- a/src/commands/clip.rs
+++ b/src/commands/clip.rs
@@ -175,7 +175,8 @@ struct ClipProcessedBatch {
 
 impl MemoryEstimate for ClipProcessedBatch {
     fn estimate_heap_size(&self) -> usize {
-        self.clipped_records.iter().map(MemoryEstimate::estimate_heap_size).sum()
+        self.clipped_records.iter().map(MemoryEstimate::estimate_heap_size).sum::<usize>()
+            + self.clipped_records.capacity() * std::mem::size_of::<RecordBuf>()
     }
 }
 
@@ -2462,4 +2463,29 @@ mod tests {
 
         Ok(())
     }
+
+    #[test]
+    fn test_clip_processed_batch_memory_estimate() {
+        use fgumi_lib::sam::builder::RecordBuilder;
+        use noodles::sam::alignment::record_buf::RecordBuf;
+
+        let record = RecordBuilder::new().sequence("ACGT").qualities(&[30, 30, 30, 30]).build();
+        let mut records = Vec::with_capacity(10);
+        records.push(record);
+
+        let batch = ClipProcessedBatch {
+            clipped_records: records,
+            templates_count: 1,
+            overlap_clipped_count: 0,
+            extend_clipped_count: 0,
+        };
+
+        let estimate = batch.estimate_heap_size();
+        // Should include Vec overhead for capacity * size_of::<RecordBuf>()
+        let vec_overhead = 10 * std::mem::size_of::<RecordBuf>();
+        assert!(
+            estimate >= vec_overhead,
+            "estimate {estimate} should include Vec<RecordBuf> overhead {vec_overhead}"
+        );
+    }
 }
diff --git a/src/commands/codec.rs b/src/commands/codec.rs
index a5fc71299..9b9ac3ef6 100644
--- a/src/commands/codec.rs
+++ b/src/commands/codec.rs
@@ -60,7 +60,9 @@ struct CodecProcessedBatch {
 
 impl MemoryEstimate for CodecProcessedBatch {
     fn estimate_heap_size(&self) -> usize {
-        self.consensus_output.data.len() + self.rejects.iter().map(Vec::len).sum::<usize>()
+        self.consensus_output.data.capacity()
+            + self.rejects.iter().map(Vec::capacity).sum::<usize>()
+            + self.rejects.capacity() * std::mem::size_of::<Vec<u8>>()
     }
 }
 
@@ -1185,4 +1187,23 @@ mod tests {
 
         Ok(())
     }
+
+    #[test]
+    fn test_codec_processed_batch_memory_estimate() {
+        let mut data = Vec::with_capacity(1024);
+        data.extend_from_slice(&[0u8; 100]);
+        let mut reject = Vec::with_capacity(512);
+        reject.extend_from_slice(&[0u8; 50]);
+
+        let batch = CodecProcessedBatch {
+            consensus_output: ConsensusOutput { data, count: 1 },
+            rejects: vec![reject],
+            groups_count: 1,
+            stats: CodecConsensusStats::default(),
+        };
+
+        let estimate = batch.estimate_heap_size();
+        assert!(estimate >= 1024 + 512, "estimate {estimate} should be >= 1536 (capacities)");
+        assert!(estimate > 1024 + 512, "estimate {estimate} should include Vec overhead");
+    }
 }
diff --git a/src/commands/dedup.rs b/src/commands/dedup.rs
index 3ec488777..26e70e452 100644
--- a/src/commands/dedup.rs
+++ b/src/commands/dedup.rs
@@ -186,6 +186,7 @@ impl BatchWeight for ProcessedDedupGroup {
 impl MemoryEstimate for ProcessedDedupGroup {
     fn estimate_heap_size(&self) -> usize {
         self.templates.iter().map(MemoryEstimate::estimate_heap_size).sum::<usize>()
+            + self.templates.capacity() * std::mem::size_of::<Template>()
             + self.family_sizes.capacity() * std::mem::size_of::<(usize, u64)>()
             + std::mem::size_of::<DedupMetrics>()
     }
@@ -2805,4 +2806,26 @@ mod tests {
         bam_fields::set_flags(&mut rec, flags_with_dup & !bam_fields::flags::DUPLICATE);
         assert_eq!(bam_fields::flags(&rec) & bam_fields::flags::DUPLICATE, 0);
     }
+
+    #[test]
+    fn test_processed_dedup_group_memory_estimate() {
+        let template = create_test_template("test", &[30, 30, 30, 30]);
+        let mut templates = Vec::with_capacity(10);
+        templates.push(template);
+
+        let batch = ProcessedDedupGroup {
+            templates,
+            family_sizes: AHashMap::new(),
+            dedup_metrics: DedupMetrics::default(),
+            input_record_count: 1,
+        };
+
+        let estimate = batch.estimate_heap_size();
+        // Should include Vec<Template> overhead for capacity * size_of::<Template>()
+        let vec_overhead = 10 * std::mem::size_of::<Template>();
+        assert!(
+            estimate >= vec_overhead,
+            "estimate {estimate} should include Vec<Template> overhead {vec_overhead}"
+        );
+    }
 }
diff --git a/src/commands/duplex.rs b/src/commands/duplex.rs
index 499da65e3..e3724cf40 100644
--- a/src/commands/duplex.rs
+++ b/src/commands/duplex.rs
@@ -68,7 +68,9 @@ struct DuplexProcessedBatch {
 
 impl MemoryEstimate for DuplexProcessedBatch {
     fn estimate_heap_size(&self) -> usize {
-        self.consensus_output.data.len() + self.rejects.iter().map(Vec::len).sum::<usize>()
+        self.consensus_output.data.capacity()
+            + self.rejects.iter().map(Vec::capacity).sum::<usize>()
+            + self.rejects.capacity() * std::mem::size_of::<Vec<u8>>()
     }
 }
 
@@ -1621,4 +1623,24 @@ mod tests {
 
         Ok(())
     }
+
+    #[test]
+    fn test_duplex_processed_batch_memory_estimate() {
+        let mut data = Vec::with_capacity(1024);
+        data.extend_from_slice(&[0u8; 100]);
+        let mut reject = Vec::with_capacity(512);
+        reject.extend_from_slice(&[0u8; 50]);
+
+        let batch = DuplexProcessedBatch {
+            consensus_output: ConsensusOutput { data, count: 1 },
+            rejects: vec![reject],
+            groups_count: 1,
+            stats: ConsensusCallingStats::default(),
+            overlapping_stats: None,
+        };
+
+        let estimate = batch.estimate_heap_size();
+        assert!(estimate >= 1024 + 512, "estimate {estimate} should be >= 1536 (capacities)");
+        assert!(estimate > 1024 + 512, "estimate {estimate} should include Vec overhead");
+    }
 }
diff --git a/src/commands/filter.rs b/src/commands/filter.rs
index d1e9a4241..faa4e7676 100644
--- a/src/commands/filter.rs
+++ b/src/commands/filter.rs
@@ -13,26 +13,23 @@ use anyhow::{Result, bail};
 use clap::Parser;
 use crossbeam_queue::SegQueue;
 use fgumi_lib::alignment_tags::regenerate_alignment_tags_raw;
-use fgumi_lib::bam_io::{
-    create_bam_reader, create_bam_reader_for_pipeline, create_raw_bam_reader, create_raw_bam_writer,
-};
+use fgumi_lib::bam_io::create_bam_reader_for_pipeline;
 use fgumi_lib::consensus_filter::{
     FilterConfig, FilterResult, compute_read_stats_raw, filter_duplex_read_raw, filter_read_raw,
     is_duplex_consensus_raw, mask_bases_raw, mask_duplex_bases_raw, template_passes_raw,
 };
 use fgumi_lib::grouper::{SingleRawRecordGrouper, TemplateGrouper};
 use fgumi_lib::logging::OperationTimer;
-use fgumi_lib::progress::ProgressTracker;
 use fgumi_lib::read_info::LibraryIndex;
 use fgumi_lib::reference::ReferenceReader;
 use fgumi_lib::sort::bam_fields;
 use fgumi_lib::tag_reversal::reverse_per_base_tags_raw;
-use fgumi_lib::template::{TemplateBatch, TemplateIterator};
+use fgumi_lib::template::TemplateBatch;
 use fgumi_lib::unified_pipeline::{
-    GroupKeyConfig, Grouper, MemoryEstimate, run_bam_pipeline_from_reader,
+    BamPipelineConfig, BatchWeight, GroupKeyConfig, Grouper, MemoryEstimate,
+    run_bam_pipeline_from_reader, run_bam_pipeline_from_reader_with_secondary,
 };
 use fgumi_lib::validation::validate_file_exists;
-use fgumi_lib::vendored::raw_bam_record::RawRecord;
 use log::info;
 use noodles::sam::Header;
 use std::io;
@@ -152,10 +149,6 @@ pub struct Filter {
     #[arg(short = 'R', long = "reverse-per-base-tags", default_value = "false")]
     pub reverse_per_base_tags: bool,
 
-    /// Sort order for output BAM
-    #[arg(short = 'S', long = "sort-order")]
-    pub sort_order: Option<String>,
-
     /// Threading options for parallel processing
     #[command(flatten)]
     pub threading: ThreadingOptions,
@@ -210,10 +203,11 @@ struct FilterProcessedBatchRaw {
 impl MemoryEstimate for FilterProcessedBatchRaw {
     fn estimate_heap_size(&self) -> usize {
         let vec_overhead = std::mem::size_of::<Vec<u8>>();
-        let kept: usize = self.kept_records.iter().map(|v| v.capacity() + vec_overhead).sum();
-        let rejected: usize =
-            self.rejected_records.iter().map(|v| v.capacity() + vec_overhead).sum();
-        kept + rejected
+        let kept_outer = self.kept_records.capacity() * vec_overhead;
+        let kept_inner: usize = self.kept_records.iter().map(|v| v.capacity()).sum();
+        let rejected_outer = self.rejected_records.capacity() * vec_overhead;
+        let rejected_inner: usize = self.rejected_records.iter().map(|v| v.capacity()).sum();
+        kept_outer + kept_inner + rejected_outer + rejected_inner
     }
 }
 
@@ -243,8 +237,28 @@ struct CollectedFilterMetrics {
     failed_records: u64,
     /// Total bases masked.
     total_bases_masked: u64,
-    /// Rejected records (if tracking).
-    rejects: Vec<Vec<u8>>,
+}
+
+/// Shared state for a filter pipeline run, built by `setup_pipeline`.
+struct FilterPipelineSetup {
+    pipeline_config: BamPipelineConfig,
+    config: Arc<FilterConfig>,
+    reference: Option<Arc<ReferenceReader>>,
+    collected_metrics: Arc<SegQueue<CollectedFilterMetrics>>,
+    progress_counter: Arc<AtomicU64>,
+}
+
+/// Captures needed by the process closure, extracted from `Filter` and `FilterPipelineSetup`.
+struct FilterProcessCaptures {
+    config: Arc<FilterConfig>,
+    reference: Option<Arc<ReferenceReader>>,
+    min_base_quality: Option<u8>,
+    should_reverse_tags: bool,
+    min_mean_base_quality: Option<f64>,
+    max_no_call_fraction: f64,
+    require_single_strand_agreement: bool,
+    progress: Arc<AtomicU64>,
+    header: Header,
 }
 
 impl Command for Filter {
@@ -274,30 +288,20 @@ impl Command for Filter {
         }
         info!("Max no-call fraction: {}", self.max_no_call_fraction);
 
-        // ========================================================================
-        // CRITICAL: Check --threads mode BEFORE creating any file handles
-        // ========================================================================
-        // Route to 7-step unified pipeline when --threads is specified.
-        // Use single-threaded fast path when --threads is not provided.
-        // This must be checked first to avoid creating file handles that the pipeline
-        // will need to manage itself.
-        let total_reads = if let Some(threads) = self.threading.threads {
-            // Open input using streaming-capable reader for pipeline use
-            let (reader, header) = create_bam_reader_for_pipeline(&self.io.input)?;
-
-            // Add @PG record with PP chaining to input's last program
-            let header = crate::commands::common::add_pg_record(header, command_line)?;
-
-            let track_rejects = self.rejects.is_some();
-
-            // Route to appropriate 7-step pipeline mode
-            if self.filter_by_template {
-                self.execute_threads_mode_template(threads, reader, header, track_rejects)?
-            } else {
-                self.execute_threads_mode_single_read(threads, reader, header, track_rejects)?
-            }
+        // Open input using streaming-capable reader for pipeline use
+        let (reader, header) = create_bam_reader_for_pipeline(&self.io.input)?;
+
+        // Add @PG record with PP chaining to input's last program
+        let header = crate::commands::common::add_pg_record(header, command_line)?;
+
+        let track_rejects = self.rejects.is_some();
+        let threads = self.threading.threads.unwrap_or(1);
+
+        // Route to appropriate 7-step pipeline mode
+        let total_reads = if self.filter_by_template {
+            self.execute_threads_mode_template(threads, reader, header, track_rejects)?
         } else {
-            self.execute_single_threaded(command_line)?
+            self.execute_threads_mode_single_read(threads, reader, header, track_rejects)?
         };
 
         timer.log_completion(total_reads);
@@ -306,219 +310,165 @@ impl Command for Filter {
 }
 
 impl Filter {
-    /// Execute in single-threaded mode.
-    ///
-    /// Returns the total number of reads processed.
-    fn execute_single_threaded(&self, command_line: &str) -> Result<u64> {
-        // ========================================================================
-        // Single-threaded mode: use simple sequential implementation
-        // ========================================================================
-
-        // Create filter configuration
-        let config = FilterConfig::new(
+    // ========================================================================
+    // 7-Step Unified Pipeline Implementation
+    // ========================================================================
+
+    /// Build the shared pipeline configuration, filter config, reference, metrics
+    /// queue, and progress counter used by both pipeline modes.
+    fn setup_pipeline(&self, num_threads: usize, header: &Header) -> Result<FilterPipelineSetup> {
+        let mut pipeline_config = build_pipeline_config(
+            &self.scheduler_opts,
+            &self.compression,
+            &self.queue_memory,
+            num_threads,
+        )?;
+
+        let library_index = LibraryIndex::from_header(header);
+        pipeline_config.group_key_config = Some(GroupKeyConfig::new_raw_no_cell(library_index));
+
+        let config = Arc::new(FilterConfig::new(
             &self.min_reads,
             &self.max_read_error_rate,
             &self.max_base_error_rate,
             self.min_base_quality,
             self.min_mean_base_quality,
             self.max_no_call_fraction,
-        );
-
-        // Read header from input BAM (we need it before branching to set up writers)
-        let reader_threads = self.threading.num_threads();
-        let (_, header) = create_bam_reader(&self.io.input, reader_threads)?;
-
-        // Add @PG record with PP chaining to input's last program
-        let header = crate::commands::common::add_pg_record(header, command_line)?;
-
-        // Create output BAM with multi-threaded BGZF writer (raw byte writer)
-        let writer_threads = self.threading.num_threads();
-        let mut writer = create_raw_bam_writer(
-            &self.io.output,
-            &header,
-            writer_threads,
-            self.compression.compression_level,
-        )?;
-
-        // Create optional rejects writer
-        let mut rejects_writer = match self.rejects.as_ref() {
-            Some(path) => Some(create_raw_bam_writer(
-                path,
-                &header,
-                writer_threads,
-                self.compression.compression_level,
-            )?),
-            None => None,
-        };
+        ));
 
-        // Create reference reader for tag regeneration (always enabled, matching fgbio behavior)
         info!("Loading reference genome into memory...");
         let ref_load_start = Instant::now();
-        let reference_reader = Some(ReferenceReader::new(&self.reference)?);
-        let ref_load_elapsed = ref_load_start.elapsed();
-        info!("Reference loaded in {:.1}s", ref_load_elapsed.as_secs_f64());
-
-        // Process reads
-        let mut total_reads: u64 = 0;
-        let mut passed_reads: u64 = 0;
-        let mut failed_reads: u64 = 0;
-        let mut total_bases_masked: u64 = 0;
-        let progress = ProgressTracker::new("Processed records").with_interval(1_000_000);
-
-        info!("Processing consensus reads...");
-
-        if self.filter_by_template {
-            // Template-aware filtering using streaming TemplateIterator
-            // TODO: Use raw-byte template grouper to avoid RecordBuf decode/re-encode overhead
-            info!("Using template-aware filtering (streaming)");
-
-            let (mut reader, _) = create_bam_reader(&self.io.input, reader_threads)?;
-            let record_iter = reader.record_bufs(&header).map(|r| r.map_err(Into::into));
-            let template_iter = TemplateIterator::new(record_iter);
-            let mut total_templates = 0u64;
-
-            for template_result in template_iter {
-                let template = template_result?;
-                // Convert template records to raw bytes
-                let mut template_records: Vec<Vec<u8>> = template
-                    .records
-                    .iter()
-                    .map(|r| {
-                        let mut buf = Vec::new();
-                        fgumi_lib::vendored::bam_codec::encode_record_buf(&mut buf, &header, r)?;
-                        Ok(buf)
-                    })
-                    .collect::<Result<Vec<_>>>()?;
-                total_reads += template_records.len() as u64;
-                total_templates += 1;
+        let reference: Option<Arc<ReferenceReader>> =
+            Some(Arc::new(ReferenceReader::new(&self.reference)?));
+        info!("Reference loaded in {:.1}s", ref_load_start.elapsed().as_secs_f64());
 
-                let mut pass_map: AHashMap<usize, bool> = AHashMap::new();
+        let collected_metrics: Arc<SegQueue<CollectedFilterMetrics>> = Arc::new(SegQueue::new());
+        let progress_counter = Arc::new(AtomicU64::new(0));
 
-                for (idx, record) in template_records.iter_mut().enumerate() {
-                    let (masked, pass) = Self::process_record_raw(
-                        record,
-                        &config,
-                        reference_reader.as_ref(),
-                        &header,
-                        self.reverse_per_base_tags,
-                        self.min_base_quality,
-                        self.require_single_strand_agreement,
-                        self.min_mean_base_quality,
-                        self.max_no_call_fraction,
-                    )?;
-                    total_bases_masked += masked;
-                    pass_map.insert(idx, pass);
-                }
+        Ok(FilterPipelineSetup {
+            pipeline_config,
+            config,
+            reference,
+            collected_metrics,
+            progress_counter,
+        })
+    }
 
-                let template_pass = template_passes_raw(&template_records, &pass_map);
+    /// Build the process closure captures from self and setup.
+    fn process_captures(
+        &self,
+        setup: &FilterPipelineSetup,
+        header: &Header,
+    ) -> FilterProcessCaptures {
+        FilterProcessCaptures {
+            config: Arc::clone(&setup.config),
+            reference: setup.reference.clone(),
+            min_base_quality: self.min_base_quality,
+            should_reverse_tags: self.reverse_per_base_tags,
+            min_mean_base_quality: self.min_mean_base_quality,
+            max_no_call_fraction: self.max_no_call_fraction,
+            require_single_strand_agreement: self.require_single_strand_agreement,
+            progress: Arc::clone(&setup.progress_counter),
+            header: header.clone(),
+        }
+    }
 
-                for (idx, record) in template_records.iter().enumerate() {
-                    let flags = bam_fields::flags(record);
-                    let is_primary = (flags & bam_fields::flags::SECONDARY) == 0
-                        && (flags & bam_fields::flags::SUPPLEMENTARY) == 0;
+    /// Run the filter pipeline with the given grouper and process function.
+    ///
+    /// This is the common pipeline executor shared by single-read and template modes.
+    /// It builds the serialize function, runs the pipeline (with optional secondary
+    /// output for rejects), and aggregates metrics.
+    fn run_filter_pipeline<G, GrouperFn, ProcessFn>(
+        &self,
+        setup: FilterPipelineSetup,
+        reader: Box<dyn std::io::Read + Send>,
+        header: Header,
+        grouper_fn: GrouperFn,
+        process_fn: ProcessFn,
+    ) -> Result<u64>
+    where
+        G: Send + BatchWeight + MemoryEstimate + 'static,
+        GrouperFn: FnOnce(&Header) -> Box<dyn Grouper<Group = G> + Send>,
+        ProcessFn: Fn(G) -> io::Result<FilterProcessedBatchRaw> + Send + Sync + 'static,
+    {
+        let collected_for_serialize = Arc::clone(&setup.collected_metrics);
+
+        // Primary serialize: write kept records
+        let serialize_fn = move |processed: FilterProcessedBatchRaw,
+                                 _header: &Header,
+                                 output: &mut Vec<u8>|
+              -> io::Result<u64> {
+            collected_for_serialize.push(CollectedFilterMetrics {
+                total_records: processed.records_count,
+                passed_records: processed.passed_count,
+                failed_records: processed.records_count - processed.passed_count,
+                total_bases_masked: processed.bases_masked,
+            });
 
-                    if is_primary {
-                        if template_pass {
-                            writer.write_raw_record(record)?;
-                            passed_reads += 1;
-                        } else {
-                            if let Some(ref mut rw) = rejects_writer {
-                                rw.write_raw_record(record)?;
-                            }
-                            failed_reads += 1;
-                        }
-                    } else {
-                        let record_pass = pass_map.get(&idx).copied().unwrap_or(false);
-                        if template_pass && record_pass {
-                            writer.write_raw_record(record)?;
-                            passed_reads += 1;
-                        } else {
-                            if let Some(ref mut rw) = rejects_writer {
-                                rw.write_raw_record(record)?;
-                            }
-                            failed_reads += 1;
-                        }
-                    }
-                }
+            serialize_raw_records(&processed.kept_records, output)
+        };
 
-                if total_templates.is_multiple_of(100_000) {
-                    info!("Processed {total_templates} templates");
-                }
-            }
+        if let Some(rejects_path) = &self.rejects {
+            // Secondary serialize: write rejected records
+            let secondary_serialize_fn =
+                |batch: &FilterProcessedBatchRaw, buf: &mut Vec<u8>| -> io::Result<u64> {
+                    serialize_raw_records(&batch.rejected_records, buf)
+                };
+
+            run_bam_pipeline_from_reader_with_secondary(
+                setup.pipeline_config,
+                reader,
+                header,
+                &self.io.output,
+                None,
+                rejects_path,
+                grouper_fn,
+                process_fn,
+                serialize_fn,
+                secondary_serialize_fn,
+            )?;
         } else {
-            // Single-read filtering: process reads independently using raw bytes
-            info!("Using single-read filtering");
-
-            let (mut raw_reader, _) = create_raw_bam_reader(&self.io.input, reader_threads)?;
-            let mut raw_record = RawRecord::new();
-
-            while raw_reader.read_record(&mut raw_record)? > 0 {
-                let mut record = raw_record.as_ref().to_vec();
-                total_reads += 1;
-
-                let (masked, pass) = Self::process_record_raw(
-                    &mut record,
-                    &config,
-                    reference_reader.as_ref(),
-                    &header,
-                    self.reverse_per_base_tags,
-                    self.min_base_quality,
-                    self.require_single_strand_agreement,
-                    self.min_mean_base_quality,
-                    self.max_no_call_fraction,
-                )?;
-                total_bases_masked += masked;
-
-                if pass {
-                    writer.write_raw_record(&record)?;
-                    passed_reads += 1;
-                } else {
-                    if let Some(ref mut rw) = rejects_writer {
-                        rw.write_raw_record(&record)?;
-                    }
-                    failed_reads += 1;
-                }
-
-                progress.log_if_needed(1);
-            }
+            run_bam_pipeline_from_reader(
+                setup.pipeline_config,
+                reader,
+                header,
+                &self.io.output,
+                None,
+                grouper_fn,
+                process_fn,
+                serialize_fn,
+            )?;
         }
 
-        progress.log_final();
-        info!("Filtering complete");
-        info!("Total reads: {total_reads}");
-        info!("Passed reads: {passed_reads}");
-        info!("Failed reads: {failed_reads}");
-        info!("Total bases masked: {total_bases_masked}");
+        // Aggregate metrics
+        let mut total_reads = 0u64;
+        let mut passed_reads = 0u64;
+        let mut failed_reads = 0u64;
+        let mut total_bases_masked = 0u64;
+
+        while let Some(metrics) = setup.collected_metrics.pop() {
+            total_reads += metrics.total_records;
+            passed_reads += metrics.passed_records;
+            failed_reads += metrics.failed_records;
+            total_bases_masked += metrics.total_bases_masked;
+        }
 
-        // Write statistics file if requested
         if let Some(stats_path) = &self.stats {
-            self.write_statistics(
-                stats_path,
-                total_reads as usize,
-                passed_reads as usize,
-                failed_reads as usize,
-                total_bases_masked as usize,
-            )?;
+            self.write_filter_stats(stats_path, total_reads, passed_reads, failed_reads)?;
         }
 
-        // Flush and finish the writers
-        writer.finish()?;
-        if let Some(rw) = rejects_writer {
-            rw.finish()?;
+        info!("Processed {total_reads} reads; kept {passed_reads} and rejected {failed_reads}");
+        if self.rejects.is_some() && failed_reads > 0 {
+            info!("Wrote {failed_reads} rejected records to rejects file");
         }
+        info!("Total bases masked: {total_bases_masked}");
 
         Ok(total_reads)
     }
 
-    // ========================================================================
-    // 7-Step Unified Pipeline Implementation
-    // ========================================================================
-
     /// Execute using the 7-step unified pipeline (single-read mode, raw bytes).
     ///
     /// Each record is filtered independently without template awareness.
-    /// This mode is used when `--filter-by-template` is false.
-    #[allow(clippy::too_many_lines)]
     fn execute_threads_mode_single_read(
         &self,
         num_threads: usize,
@@ -526,60 +476,13 @@ impl Filter {
         header: Header,
         track_rejects: bool,
     ) -> Result<u64> {
-        // Configure pipeline - filter is Balanced workload
-        let mut pipeline_config = build_pipeline_config(
-            &self.scheduler_opts,
-            &self.compression,
-            &self.queue_memory,
-            num_threads,
-        )?;
+        let setup = self.setup_pipeline(num_threads, &header)?;
+        let ctx = self.process_captures(&setup, &header);
 
-        // Enable raw-byte mode: skip noodles decode
-        let library_index = LibraryIndex::from_header(&header);
-        pipeline_config.group_key_config = Some(GroupKeyConfig::new_raw_no_cell(library_index));
-
-        // Create filter configuration
-        let config = Arc::new(FilterConfig::new(
-            &self.min_reads,
-            &self.max_read_error_rate,
-            &self.max_base_error_rate,
-            self.min_base_quality,
-            self.min_mean_base_quality,
-            self.max_no_call_fraction,
-        ));
-
-        // Load reference for tag regeneration (always enabled, matching fgbio behavior)
-        info!("Loading reference genome into memory...");
-        let ref_load_start = Instant::now();
-        let reference: Option<Arc<ReferenceReader>> =
-            Some(Arc::new(ReferenceReader::new(&self.reference)?));
-        let ref_load_elapsed = ref_load_start.elapsed();
-        info!("Reference loaded in {:.1}s", ref_load_elapsed.as_secs_f64());
-
-        // Lock-free metrics collection
-        let collected_metrics: Arc<SegQueue<CollectedFilterMetrics>> = Arc::new(SegQueue::new());
-        let collected_for_serialize = Arc::clone(&collected_metrics);
-
-        // Configuration for closures
-        let config_for_process = Arc::clone(&config);
-        let min_base_quality = self.min_base_quality;
-        let should_reverse_tags = self.reverse_per_base_tags;
-        let min_mean_base_quality = self.min_mean_base_quality;
-        let max_no_call_fraction = self.max_no_call_fraction;
-        let require_single_strand_agreement = self.require_single_strand_agreement;
-        let reference_for_process = reference;
-
-        // Progress tracking
-        let progress_counter = Arc::new(AtomicU64::new(0));
-        let progress_for_process = Arc::clone(&progress_counter);
-
-        // Grouper: process individual raw-byte records
         let grouper_fn = move |_header: &Header| {
             Box::new(SingleRawRecordGrouper::new()) as Box<dyn Grouper<Group = Vec<u8>> + Send>
         };
 
-        // Process function: filter each raw record
-        let header_for_process = header.clone();
         let process_fn = move |mut record: Vec<u8>| -> io::Result<FilterProcessedBatchRaw> {
             let mut kept_records = Vec::new();
             let mut rejected_records = Vec::new();
@@ -587,14 +490,14 @@ impl Filter {
 
             let (bases_masked, pass) = Self::process_record_raw(
                 &mut record,
-                &config_for_process,
-                reference_for_process.as_deref(),
-                &header_for_process,
-                should_reverse_tags,
-                min_base_quality,
-                require_single_strand_agreement,
-                min_mean_base_quality,
-                max_no_call_fraction,
+                &ctx.config,
+                ctx.reference.as_deref(),
+                &ctx.header,
+                ctx.should_reverse_tags,
+                ctx.min_base_quality,
+                ctx.require_single_strand_agreement,
+                ctx.min_mean_base_quality,
+                ctx.max_no_call_fraction,
             )
             .map_err(io::Error::other)?;
 
@@ -605,8 +508,7 @@ impl Filter {
                 rejected_records.push(record);
             }
 
-            // Progress logging
-            let count = progress_for_process.fetch_add(1, Ordering::Relaxed);
+            let count = ctx.progress.fetch_add(1, Ordering::Relaxed);
             if (count + 1).is_multiple_of(1_000_000) {
                 info!("Processed {} records", count + 1);
             }
@@ -620,48 +522,12 @@ impl Filter {
             })
         };
 
-        // Serialize function: raw bytes directly (bypasses bam_codec)
-        let serialize_fn = move |processed: FilterProcessedBatchRaw,
-                                 _header: &Header,
-                                 output: &mut Vec<u8>|
-              -> io::Result<u64> {
-            // Push metrics to lock-free queue
-            collected_for_serialize.push(CollectedFilterMetrics {
-                total_records: processed.records_count,
-                passed_records: processed.passed_count,
-                failed_records: processed.records_count - processed.passed_count,
-                total_bases_masked: processed.bases_masked,
-                rejects: processed.rejected_records,
-            });
-
-            // Serialize kept records directly as raw BAM bytes
-            serialize_raw_records(&processed.kept_records, output)
-        };
-
-        // Clone header before pipeline (needed for post-pipeline aggregation)
-        let header_for_metrics = header.clone();
-
-        // Run the 7-step pipeline with the already-opened reader (supports streaming)
-        let _records_written = run_bam_pipeline_from_reader(
-            pipeline_config,
-            reader,
-            header,
-            &self.io.output,
-            None, // Use input header for output
-            grouper_fn,
-            process_fn,
-            serialize_fn,
-        )?;
-
-        // ========== Post-pipeline: Aggregate metrics ==========
-        self.aggregate_and_finalize_metrics(collected_metrics, track_rejects, &header_for_metrics)
+        self.run_filter_pipeline(setup, reader, header, grouper_fn, process_fn)
     }
 
     /// Execute using the 7-step unified pipeline (template-aware mode).
     ///
     /// All primary reads in a template must pass for the template to pass.
-    /// This mode is used when `--filter-by-template` is true (the default).
-    #[allow(clippy::too_many_lines)]
     fn execute_threads_mode_template(
         &self,
         num_threads: usize,
@@ -669,66 +535,19 @@ impl Filter {
         header: Header,
         track_rejects: bool,
     ) -> Result<u64> {
-        // Configure pipeline - filter is Balanced workload
-        let mut pipeline_config = build_pipeline_config(
-            &self.scheduler_opts,
-            &self.compression,
-            &self.queue_memory,
-            num_threads,
-        )?;
-
-        // Enable raw-byte mode: skip noodles decode
-        let library_index = LibraryIndex::from_header(&header);
-        pipeline_config.group_key_config = Some(GroupKeyConfig::new_raw_no_cell(library_index));
-
-        // Create filter configuration
-        let config = Arc::new(FilterConfig::new(
-            &self.min_reads,
-            &self.max_read_error_rate,
-            &self.max_base_error_rate,
-            self.min_base_quality,
-            self.min_mean_base_quality,
-            self.max_no_call_fraction,
-        ));
+        let setup = self.setup_pipeline(num_threads, &header)?;
+        let ctx = self.process_captures(&setup, &header);
 
-        // Load reference for tag regeneration (always enabled, matching fgbio behavior)
-        info!("Loading reference genome into memory...");
-        let ref_load_start = Instant::now();
-        let reference: Option<Arc<ReferenceReader>> =
-            Some(Arc::new(ReferenceReader::new(&self.reference)?));
-        let ref_load_elapsed = ref_load_start.elapsed();
-        info!("Reference loaded in {:.1}s", ref_load_elapsed.as_secs_f64());
-
-        // Lock-free metrics collection
-        let collected_metrics: Arc<SegQueue<CollectedFilterMetrics>> = Arc::new(SegQueue::new());
-        let collected_for_serialize = Arc::clone(&collected_metrics);
-
-        // Configuration for closures
         #[cfg(test)]
         const BATCH_SIZE: usize = 50;
         #[cfg(not(test))]
         const BATCH_SIZE: usize = 1000;
 
-        let config_for_process = Arc::clone(&config);
-        let min_base_quality = self.min_base_quality;
-        let should_reverse_tags = self.reverse_per_base_tags;
-        let min_mean_base_quality = self.min_mean_base_quality;
-        let max_no_call_fraction = self.max_no_call_fraction;
-        let require_single_strand_agreement = self.require_single_strand_agreement;
-        let reference_for_process = reference;
-
-        // Progress tracking
-        let progress_counter = Arc::new(AtomicU64::new(0));
-        let progress_for_process = Arc::clone(&progress_counter);
-
-        // Grouper: batch templates by QNAME
         let grouper_fn = move |_header: &Header| {
             Box::new(TemplateGrouper::new(BATCH_SIZE))
                 as Box<dyn Grouper<Group = TemplateBatch> + Send>
         };
 
-        // Process function: filter each template batch
-        let header_for_process = header.clone();
         let process_fn = move |batch: TemplateBatch| -> io::Result<FilterProcessedBatchRaw> {
             let mut kept_records: Vec<Vec<u8>> = Vec::new();
             let mut rejected_records: Vec<Vec<u8>> = Vec::new();
@@ -742,30 +561,27 @@ impl Filter {
                     .ok_or_else(|| io::Error::other("template has no raw records"))?;
                 let mut pass_map: AHashMap<usize, bool> = AHashMap::new();
 
-                // Process each record in the template
                 for (idx, record) in template_records.iter_mut().enumerate() {
                     total_records += 1;
 
                     let (masked, pass) = Self::process_record_raw(
                         record,
-                        &config_for_process,
-                        reference_for_process.as_deref(),
-                        &header_for_process,
-                        should_reverse_tags,
-                        min_base_quality,
-                        require_single_strand_agreement,
-                        min_mean_base_quality,
-                        max_no_call_fraction,
+                        &ctx.config,
+                        ctx.reference.as_deref(),
+                        &ctx.header,
+                        ctx.should_reverse_tags,
+                        ctx.min_base_quality,
+                        ctx.require_single_strand_agreement,
+                        ctx.min_mean_base_quality,
+                        ctx.max_no_call_fraction,
                     )
                     .map_err(io::Error::other)?;
                     bases_masked += masked;
                     pass_map.insert(idx, pass);
                 }
 
-                // Check if template passes (all primary reads must pass)
                 let template_pass = template_passes_raw(&template_records, &pass_map);
 
-                // Collect kept/rejected records
                 for (idx, record) in template_records.into_iter().enumerate() {
                     let flags = bam_fields::flags(&record);
                     let is_primary = (flags & bam_fields::flags::SECONDARY) == 0
@@ -779,7 +595,6 @@ impl Filter {
                             rejected_records.push(record);
                         }
                     } else {
-                        // Supplementary/secondary: only write if template passes AND individual record passes
                         let record_pass = pass_map.get(&idx).copied().unwrap_or(false);
                         if template_pass && record_pass {
                             passed_count += 1;
@@ -791,8 +606,7 @@ impl Filter {
                 }
             }
 
-            // Progress logging
-            let count = progress_for_process.fetch_add(total_records, Ordering::Relaxed);
+            let count = ctx.progress.fetch_add(total_records, Ordering::Relaxed);
             if (count + total_records) / 1_000_000 > count / 1_000_000 {
                 info!("Processed {} records", count + total_records);
             }
@@ -806,108 +620,7 @@ impl Filter {
             })
         };
 
-        // Serialize function: convert records to bytes and collect metrics
-        let serialize_fn = move |processed: FilterProcessedBatchRaw,
-                                 _header: &Header,
-                                 output: &mut Vec<u8>|
-              -> io::Result<u64> {
-            // Push metrics to lock-free queue
-            collected_for_serialize.push(CollectedFilterMetrics {
-                total_records: processed.records_count,
-                passed_records: processed.passed_count,
-                failed_records: processed.records_count - processed.passed_count,
-                total_bases_masked: processed.bases_masked,
-                rejects: processed.rejected_records,
-            });
-
-            // Serialize kept records as raw bytes
-            serialize_raw_records(&processed.kept_records, output)
-        };
-
-        // Clone header before pipeline (needed for post-pipeline aggregation)
-        let header_for_metrics = header.clone();
-
-        // Run the 7-step pipeline with the already-opened reader (supports streaming)
-        let _records_written = run_bam_pipeline_from_reader(
-            pipeline_config,
-            reader,
-            header,
-            &self.io.output,
-            None, // Use input header for output
-            grouper_fn,
-            process_fn,
-            serialize_fn,
-        )?;
-
-        // ========== Post-pipeline: Aggregate metrics ==========
-        self.aggregate_and_finalize_metrics(collected_metrics, track_rejects, &header_for_metrics)
-    }
-
-    /// Aggregate metrics from the pipeline and finalize output.
-    ///
-    /// Returns the total number of reads processed.
-    fn aggregate_and_finalize_metrics(
-        &self,
-        collected_metrics: Arc<SegQueue<CollectedFilterMetrics>>,
-        track_rejects: bool,
-        header: &Header,
-    ) -> Result<u64> {
-        let mut total_reads = 0u64;
-        let mut passed_reads = 0u64;
-        let mut failed_reads = 0u64;
-        let mut total_bases_masked = 0u64;
-        let mut total_rejects_written = 0u64;
-
-        // Create rejects writer up front so we can stream rejects incrementally
-        let mut rejects_writer = if track_rejects {
-            self.rejects
-                .as_ref()
-                .map(|rejects_path| {
-                    let writer_threads = self.threading.num_threads();
-                    create_raw_bam_writer(
-                        rejects_path,
-                        header,
-                        writer_threads,
-                        self.compression.compression_level,
-                    )
-                })
-                .transpose()?
-        } else {
-            None
-        };
-
-        while let Some(metrics) = collected_metrics.pop() {
-            total_reads += metrics.total_records;
-            passed_reads += metrics.passed_records;
-            failed_reads += metrics.failed_records;
-            total_bases_masked += metrics.total_bases_masked;
-
-            if let Some(ref mut rw) = rejects_writer {
-                for record in &metrics.rejects {
-                    rw.write_raw_record(record)?;
-                    total_rejects_written += 1;
-                }
-            }
-        }
-
-        // Finalize rejects writer
-        if let Some(rw) = rejects_writer {
-            rw.finish()?;
-            if total_rejects_written > 0 {
-                info!("Wrote {total_rejects_written} rejected records to rejects file");
-            }
-        }
-
-        // Write stats file if requested
-        if let Some(stats_path) = &self.stats {
-            self.write_filter_stats(stats_path, total_reads, passed_reads, failed_reads)?;
-        }
-
-        // Log summary
-        info!("Processed {total_reads} reads; kept {passed_reads} and rejected {failed_reads}");
-        info!("Total bases masked: {total_bases_masked}");
-
-        Ok(total_reads)
+        self.run_filter_pipeline(setup, reader, header, grouper_fn, process_fn)
     }
 
     /// Write filtering statistics to a file.
@@ -1171,100 +884,6 @@ impl Filter {
 
         Ok(())
     }
-
-    /// Writes filtering statistics to a TSV file.
-    ///
-    /// Creates a human-readable statistics file containing filtering parameters,
-    /// read counts, pass/fail rates, and total bases masked during filtering.
-    ///
-    /// # Arguments
-    ///
-    /// * `stats_path` - Path where the statistics file should be written
-    /// * `total_reads` - Total number of reads processed
-    /// * `passed_reads` - Number of reads that passed filtering
-    /// * `failed_reads` - Number of reads that failed filtering
-    /// * `total_bases_masked` - Total number of bases masked across all reads
-    ///
-    /// # Returns
-    ///
-    /// `Ok(())` on success.
-    ///
-    /// # Errors
-    ///
-    /// Returns an error if the file cannot be created or written to.
-    fn write_statistics(
-        &self,
-        stats_path: &PathBuf,
-        total_reads: usize,
-        passed_reads: usize,
-        failed_reads: usize,
-        total_bases_masked: usize,
-    ) -> Result<()> {
-        use std::io::Write;
-
-        let mut file = std::fs::File::create(stats_path)?;
-
-        writeln!(file, "# Filter Statistics")?;
-        writeln!(file, "#")?;
-        writeln!(file, "# Input: {}", self.io.input.display())?;
-        writeln!(file, "# Output: {}", self.io.output.display())?;
-        writeln!(file, "#")?;
-        writeln!(file, "# Filtering Parameters:")?;
-        writeln!(file, "#   Min reads: {:?}", self.min_reads)?;
-        writeln!(file, "#   Max read error rate: {:?}", self.max_read_error_rate)?;
-        writeln!(file, "#   Max base error rate: {:?}", self.max_base_error_rate)?;
-        if let Some(q) = self.min_base_quality {
-            writeln!(file, "#   Min base quality: {q}")?;
-        }
-        if let Some(q) = self.min_mean_base_quality {
-            writeln!(file, "#   Min mean base quality: {q}")?;
-        }
-        writeln!(file, "#   Max no-call fraction: {}", self.max_no_call_fraction)?;
-        writeln!(file, "#   Filter by template: {}", self.filter_by_template)?;
-        writeln!(file, "#")?;
-        writeln!(file)?;
-
-        writeln!(file, "METRIC\tVALUE")?;
-        writeln!(file, "total_reads\t{total_reads}")?;
-        writeln!(file, "passed_reads\t{passed_reads}")?;
-        writeln!(file, "failed_reads\t{failed_reads}")?;
-        writeln!(
-            file,
-            "pass_rate\t{:.4}",
-            if total_reads > 0 { passed_reads as f64 / total_reads as f64 } else { 0.0 }
-        )?;
-        writeln!(
-            file,
-            "fail_rate\t{:.4}",
-            if total_reads > 0 { failed_reads as f64 / total_reads as f64 } else { 0.0 }
-        )?;
-        writeln!(file, "total_bases_masked\t{total_bases_masked}")?;
-
-        info!("Statistics written to {}", stats_path.display());
-
-        Ok(())
-    }
-
-    /// Gets the threshold for a specific consensus level (duplex/AB/BA)
-    ///
-    /// If only 1 value provided: applies to all levels
-    /// If 2 values provided: [duplex, single-strand]
-    /// If 3 values provided: [duplex, AB, BA]
-    #[allow(dead_code)]
-    fn get_threshold<T: Copy>(values: &[T], index: usize) -> T {
-        match values.len() {
-            1 => values[0],
-            2 => {
-                if index == 0 {
-                    values[0] // duplex
-                } else {
-                    values[1] // AB or BA
-                }
-            }
-            3 => values[index],
-            _ => unreachable!("Validation should catch this"),
-        }
-    }
 }
 
 #[cfg(test)]
@@ -1287,7 +906,7 @@ mod tests {
             min_mean_base_quality: None,
             max_no_call_fraction: 0.2,
             reverse_per_base_tags: false,
-            sort_order: None,
+
             threading: ThreadingOptions::none(),
             compression: CompressionOptions { compression_level: 1 },
             filter_by_template: true,
@@ -1335,27 +954,6 @@ mod tests {
         assert!(cmd.validate_parameters().is_err());
     }
 
-    #[test]
-    fn test_threshold_selection() {
-        // Test with 1 value
-        let values = vec![10];
-        assert_eq!(Filter::get_threshold(&values, 0), 10);
-        assert_eq!(Filter::get_threshold(&values, 1), 10);
-        assert_eq!(Filter::get_threshold(&values, 2), 10);
-
-        // Test with 2 values
-        let values = vec![5, 10];
-        assert_eq!(Filter::get_threshold(&values, 0), 5); // duplex
-        assert_eq!(Filter::get_threshold(&values, 1), 10); // AB
-        assert_eq!(Filter::get_threshold(&values, 2), 10); // BA
-
-        // Test with 3 values
-        let values = vec![5, 10, 15];
-        assert_eq!(Filter::get_threshold(&values, 0), 5); // duplex
-        assert_eq!(Filter::get_threshold(&values, 1), 10); // AB
-        assert_eq!(Filter::get_threshold(&values, 2), 15); // BA
-    }
-
     #[test]
     fn test_default_filter_parameters() {
         let filter = Filter {
@@ -1371,7 +969,7 @@ mod tests {
             min_mean_base_quality: None,
             max_no_call_fraction: 0.1,
             reverse_per_base_tags: true,
-            sort_order: None,
+
             threading: ThreadingOptions::none(),
             compression: CompressionOptions { compression_level: 1 },
             filter_by_template: true,
@@ -1402,7 +1000,7 @@ mod tests {
             min_mean_base_quality: None,
             max_no_call_fraction: 0.1,
             reverse_per_base_tags: true,
-            sort_order: None,
+
             threading: ThreadingOptions::new(8),
             compression: CompressionOptions { compression_level: 1 },
             filter_by_template: true,
@@ -1431,7 +1029,6 @@ mod tests {
             min_mean_base_quality: Some(20.0),
             max_no_call_fraction: 0.05,
             reverse_per_base_tags: true,
-            sort_order: Some("coordinate".to_string()),
             threading: ThreadingOptions::none(),
             compression: CompressionOptions { compression_level: 1 },
             filter_by_template: true,
@@ -1463,7 +1060,7 @@ mod tests {
             min_mean_base_quality: None,
             max_no_call_fraction: 0.1,
             reverse_per_base_tags: true,
-            sort_order: None,
+
             threading: ThreadingOptions::none(),
             compression: CompressionOptions { compression_level: 1 },
             filter_by_template: true,
@@ -1492,7 +1089,7 @@ mod tests {
             min_mean_base_quality: None,
             max_no_call_fraction: 0.1,
             reverse_per_base_tags: true,
-            sort_order: None,
+
             threading: ThreadingOptions::none(),
             compression: CompressionOptions { compression_level: 1 },
             filter_by_template: true,
@@ -1841,33 +1438,6 @@ mod tests {
         assert!(is_duplex_consensus(&record));
     }
 
-    #[test]
-    fn test_get_threshold_single_value() {
-        // Single value should apply to all levels
-        let values = vec![10];
-        assert_eq!(Filter::get_threshold(&values, 0), 10);
-        assert_eq!(Filter::get_threshold(&values, 1), 10);
-        assert_eq!(Filter::get_threshold(&values, 2), 10);
-    }
-
-    #[test]
-    fn test_get_threshold_two_values() {
-        // Two values: first for duplex, second for both AB and BA
-        let values = vec![5, 10];
-        assert_eq!(Filter::get_threshold(&values, 0), 5); // Duplex
-        assert_eq!(Filter::get_threshold(&values, 1), 10); // AB
-        assert_eq!(Filter::get_threshold(&values, 2), 10); // BA
-    }
-
-    #[test]
-    fn test_get_threshold_three_values() {
-        // Three values: duplex, AB, BA
-        let values = vec![5, 10, 15];
-        assert_eq!(Filter::get_threshold(&values, 0), 5); // Duplex
-        assert_eq!(Filter::get_threshold(&values, 1), 10); // AB
-        assert_eq!(Filter::get_threshold(&values, 2), 15); // BA
-    }
-
     #[test]
     fn test_validate_max_no_call_fraction_integer() {
         // When max_no_call_fraction >= 1.0, it should be an integer
@@ -1884,7 +1454,7 @@ mod tests {
             min_mean_base_quality: None,
             max_no_call_fraction: 5.5, // >= 1.0 but not an integer
             reverse_per_base_tags: true,
-            sort_order: None,
+
             threading: ThreadingOptions::none(),
             compression: CompressionOptions { compression_level: 1 },
             filter_by_template: true,
@@ -1919,7 +1489,7 @@ mod tests {
             min_mean_base_quality: None,
             max_no_call_fraction: 5.0, // >= 1.0 and IS an integer
             reverse_per_base_tags: true,
-            sort_order: None,
+
             threading: ThreadingOptions::none(),
             compression: CompressionOptions { compression_level: 1 },
             filter_by_template: true,
@@ -1952,7 +1522,7 @@ mod tests {
             min_mean_base_quality: None,
             max_no_call_fraction: 0.1,
             reverse_per_base_tags: true,
-            sort_order: None,
+
             threading: ThreadingOptions::none(),
             compression: CompressionOptions { compression_level: 1 },
             filter_by_template: true,
@@ -1982,7 +1552,7 @@ mod tests {
             min_mean_base_quality: None,
             max_no_call_fraction: 0.1,
             reverse_per_base_tags: true,
-            sort_order: None,
+
             threading: ThreadingOptions::none(),
             compression: CompressionOptions { compression_level: 1 },
             filter_by_template: true,
@@ -2012,7 +1582,7 @@ mod tests {
             min_mean_base_quality: None,
             max_no_call_fraction: 0.1,
             reverse_per_base_tags: true,
-            sort_order: None,
+
             threading: ThreadingOptions::new(8),
             compression: CompressionOptions { compression_level: 1 },
             filter_by_template: true,
@@ -2041,7 +1611,7 @@ mod tests {
             min_mean_base_quality: None,
             max_no_call_fraction: 0.1,
             reverse_per_base_tags: true,
-            sort_order: None,
+
             threading: ThreadingOptions::none(),
             compression: CompressionOptions { compression_level: 1 },
             filter_by_template: true,
@@ -2070,7 +1640,7 @@ mod tests {
             min_mean_base_quality: None,
             max_no_call_fraction: 0.1,
             reverse_per_base_tags: true,
-            sort_order: None,
+
             threading: ThreadingOptions::none(),
             compression: CompressionOptions { compression_level: 1 },
             filter_by_template: false,
@@ -2084,64 +1654,6 @@ mod tests {
         assert!(!filter.filter_by_template);
     }
 
-    #[test]
-    fn test_filter_with_sort_order_coordinate() {
-        let filter = Filter {
-            io: BamIoOptions {
-                input: PathBuf::from("input.bam"),
-                output: PathBuf::from("output.bam"),
-            },
-            reference: PathBuf::from("ref.fa"),
-            min_reads: vec![3],
-            max_read_error_rate: vec![0.1],
-            max_base_error_rate: vec![0.2],
-            min_base_quality: Some(13),
-            min_mean_base_quality: None,
-            max_no_call_fraction: 0.1,
-            reverse_per_base_tags: true,
-            sort_order: Some("coordinate".to_string()),
-            threading: ThreadingOptions::none(),
-            compression: CompressionOptions { compression_level: 1 },
-            filter_by_template: true,
-            rejects: None,
-            stats: None,
-            require_single_strand_agreement: false,
-            scheduler_opts: SchedulerOptions::default(),
-            queue_memory: QueueMemoryOptions::default(),
-        };
-
-        assert_eq!(filter.sort_order, Some("coordinate".to_string()));
-    }
-
-    #[test]
-    fn test_filter_with_sort_order_queryname() {
-        let filter = Filter {
-            io: BamIoOptions {
-                input: PathBuf::from("input.bam"),
-                output: PathBuf::from("output.bam"),
-            },
-            reference: PathBuf::from("ref.fa"),
-            min_reads: vec![3],
-            max_read_error_rate: vec![0.1],
-            max_base_error_rate: vec![0.2],
-            min_base_quality: Some(13),
-            min_mean_base_quality: None,
-            max_no_call_fraction: 0.1,
-            reverse_per_base_tags: true,
-            sort_order: Some("queryname".to_string()),
-            threading: ThreadingOptions::none(),
-            compression: CompressionOptions { compression_level: 1 },
-            filter_by_template: true,
-            rejects: None,
-            stats: None,
-            require_single_strand_agreement: false,
-            scheduler_opts: SchedulerOptions::default(),
-            queue_memory: QueueMemoryOptions::default(),
-        };
-
-        assert_eq!(filter.sort_order, Some("queryname".to_string()));
-    }
-
     #[test]
     fn test_filter_reverse_per_base_tags_disabled() {
         let filter = Filter {
@@ -2157,7 +1669,7 @@ mod tests {
             min_mean_base_quality: None,
             max_no_call_fraction: 0.1,
             reverse_per_base_tags: false,
-            sort_order: None,
+
             threading: ThreadingOptions::none(),
             compression: CompressionOptions { compression_level: 1 },
             filter_by_template: true,
@@ -2186,7 +1698,7 @@ mod tests {
             min_mean_base_quality: None,
             max_no_call_fraction: 0.1,
             reverse_per_base_tags: true,
-            sort_order: None,
+
             threading: ThreadingOptions::none(),
             compression: CompressionOptions { compression_level: 1 },
             filter_by_template: true,
@@ -2215,7 +1727,7 @@ mod tests {
             min_mean_base_quality: Some(25.0),
             max_no_call_fraction: 0.1,
             reverse_per_base_tags: true,
-            sort_order: None,
+
             threading: ThreadingOptions::none(),
             compression: CompressionOptions { compression_level: 1 },
             filter_by_template: true,
@@ -2244,7 +1756,7 @@ mod tests {
             min_mean_base_quality: None,
             max_no_call_fraction: 0.01,
             reverse_per_base_tags: true,
-            sort_order: None,
+
             threading: ThreadingOptions::none(),
             compression: CompressionOptions { compression_level: 1 },
             filter_by_template: true,
@@ -2275,7 +1787,7 @@ mod tests {
             min_mean_base_quality: None,
             max_no_call_fraction: 0.5,
             reverse_per_base_tags: true,
-            sort_order: None,
+
             threading: ThreadingOptions::none(),
             compression: CompressionOptions { compression_level: 1 },
             filter_by_template: true,
@@ -2306,7 +1818,7 @@ mod tests {
             min_mean_base_quality: None,
             max_no_call_fraction: 0.1,
             reverse_per_base_tags: true,
-            sort_order: None,
+
             threading: ThreadingOptions::none(),
             compression: CompressionOptions { compression_level: 1 },
             filter_by_template: true,
@@ -2336,7 +1848,7 @@ mod tests {
             min_mean_base_quality: None,
             max_no_call_fraction: 0.1,
             reverse_per_base_tags: true,
-            sort_order: None,
+
             threading: ThreadingOptions::none(),
             compression: CompressionOptions { compression_level: 1 },
             filter_by_template: true,
@@ -2368,7 +1880,6 @@ mod tests {
             min_mean_base_quality: Some(20.0),
             max_no_call_fraction: 0.1,
             reverse_per_base_tags: true,
-            sort_order: Some("coordinate".to_string()),
             threading: ThreadingOptions::new(4),
             compression: CompressionOptions { compression_level: 1 },
             filter_by_template: true,
@@ -2382,7 +1893,6 @@ mod tests {
         assert!(filter.rejects.is_some());
         assert!(filter.stats.is_some());
         assert!(filter.require_single_strand_agreement);
-        assert!(filter.sort_order.is_some());
         assert!(filter.min_mean_base_quality.is_some());
     }
 
@@ -2401,7 +1911,7 @@ mod tests {
             min_mean_base_quality: None,
             max_no_call_fraction: 0.0,
             reverse_per_base_tags: true,
-            sort_order: None,
+
             threading: ThreadingOptions::none(),
             compression: CompressionOptions { compression_level: 1 },
             filter_by_template: true,
@@ -2430,7 +1940,7 @@ mod tests {
             min_mean_base_quality: None,
             max_no_call_fraction: 0.1,
             reverse_per_base_tags: true,
-            sort_order: None,
+
             threading: ThreadingOptions::none(),
             compression: CompressionOptions { compression_level: 1 },
             filter_by_template: true,
@@ -2533,7 +2043,7 @@ mod tests {
             min_mean_base_quality: None,
             max_no_call_fraction: 0.2,
             reverse_per_base_tags: false,
-            sort_order: None,
+
             threading: ThreadingOptions::new(threads),
             compression: CompressionOptions { compression_level: 1 },
             filter_by_template: true,
@@ -2653,7 +2163,7 @@ mod tests {
             min_mean_base_quality: None,
             max_no_call_fraction: 0.2, // 20% max Ns allowed
             reverse_per_base_tags: false,
-            sort_order: None,
+
             threading: ThreadingOptions::none(),
             compression: CompressionOptions { compression_level: 1 },
             filter_by_template: true,
@@ -2806,7 +2316,7 @@ mod tests {
             min_mean_base_quality: None,
             max_no_call_fraction: 0.2,
             reverse_per_base_tags: false,
-            sort_order: None,
+
             threading: ThreadingOptions::none(),
             compression: CompressionOptions { compression_level: 1 },
             filter_by_template: true,
@@ -2878,7 +2388,7 @@ mod tests {
             min_mean_base_quality: None,
             max_no_call_fraction: 0.2, // Max 20% Ns allowed
             reverse_per_base_tags: false,
-            sort_order: None,
+
             threading: ThreadingOptions::none(),
             compression: CompressionOptions { compression_level: 1 },
             filter_by_template: true,
@@ -2944,7 +2454,7 @@ mod tests {
             min_mean_base_quality: Some(20.0), // Filter reads with mean quality < 20
             max_no_call_fraction: 1.0,
             reverse_per_base_tags: false,
-            sort_order: None,
+
             threading: ThreadingOptions::none(),
             compression: CompressionOptions { compression_level: 1 },
             filter_by_template: true,
@@ -3067,7 +2577,7 @@ mod tests {
             min_mean_base_quality: None,
             max_no_call_fraction: 0.5,
             reverse_per_base_tags: false,
-            sort_order: None,
+
             threading: ThreadingOptions::none(),
             compression: CompressionOptions { compression_level: 1 },
             filter_by_template: true,
@@ -3135,7 +2645,7 @@ mod tests {
             min_mean_base_quality: None,
             max_no_call_fraction: 0.5,
             reverse_per_base_tags: false,
-            sort_order: None,
+
             threading: ThreadingOptions::none(),
             compression: CompressionOptions { compression_level: 1 },
             filter_by_template: false, // Independent filtering
@@ -3201,7 +2711,7 @@ mod tests {
             min_mean_base_quality: None,
             max_no_call_fraction: 0.5,
             reverse_per_base_tags: false,
-            sort_order: None,
+
             threading: ThreadingOptions::none(),
             compression: CompressionOptions { compression_level: 1 },
             filter_by_template: true,
@@ -3271,7 +2781,7 @@ mod tests {
             min_mean_base_quality: None,
             max_no_call_fraction: 0.5,
             reverse_per_base_tags: false,
-            sort_order: None,
+
             threading: ThreadingOptions::none(),
             compression: CompressionOptions { compression_level: 1 },
             filter_by_template: true,
@@ -3294,7 +2804,7 @@ mod tests {
             min_mean_base_quality: None,
             max_no_call_fraction: 0.5,
             reverse_per_base_tags: false,
-            sort_order: None,
+
             threading: ThreadingOptions::new(4),
             compression: CompressionOptions { compression_level: 1 },
             filter_by_template: true,
@@ -3357,7 +2867,7 @@ mod tests {
             min_mean_base_quality: None,
             max_no_call_fraction: 1.0,
             reverse_per_base_tags: false,
-            sort_order: None,
+
             threading: ThreadingOptions::none(),
             compression: CompressionOptions { compression_level: 1 },
             filter_by_template: true,
@@ -3411,7 +2921,7 @@ mod tests {
             min_mean_base_quality: None,
             max_no_call_fraction: 1.0,
             reverse_per_base_tags: true, // Enable tag reversal
-            sort_order: None,
+
             threading: ThreadingOptions::none(),
             compression: CompressionOptions { compression_level: 1 },
             filter_by_template: true,
@@ -3468,7 +2978,7 @@ mod tests {
             min_mean_base_quality: None,
             max_no_call_fraction: 0.5,
             reverse_per_base_tags: false,
-            sort_order: None,
+
             threading: ThreadingOptions::new(4), // Multi-threaded
             compression: CompressionOptions { compression_level: 1 },
             filter_by_template: true,
@@ -3534,7 +3044,7 @@ mod tests {
             min_mean_base_quality: None,
             max_no_call_fraction: 0.5,
             reverse_per_base_tags: false,
-            sort_order: None,
+
             threading: ThreadingOptions::none(),
             compression: CompressionOptions { compression_level: 1 },
             filter_by_template: true,
@@ -3557,7 +3067,7 @@ mod tests {
             min_mean_base_quality: None,
             max_no_call_fraction: 0.5,
             reverse_per_base_tags: false,
-            sort_order: None,
+
             threading: ThreadingOptions::new(4),
             compression: CompressionOptions { compression_level: 1 },
             filter_by_template: true,
@@ -3638,7 +3148,7 @@ mod tests {
             min_mean_base_quality: None,
             max_no_call_fraction: 0.5,
             reverse_per_base_tags: true, // Also test reverse tags in this mode
-            sort_order: None,
+
             threading: ThreadingOptions::none(),
             compression: CompressionOptions { compression_level: 1 },
             filter_by_template: false, // Non-template mode
@@ -3747,7 +3257,7 @@ mod tests {
             min_mean_base_quality: None,
             max_no_call_fraction: 0.5,
             reverse_per_base_tags: false,
-            sort_order: None,
+
             threading: ThreadingOptions::none(),
             compression: CompressionOptions { compression_level: 1 },
             filter_by_template: true,
@@ -3862,7 +3372,7 @@ mod tests {
             min_mean_base_quality: None,
             max_no_call_fraction: 0.5,
             reverse_per_base_tags: false,
-            sort_order: None,
+
             threading: ThreadingOptions::new(4),
             compression: CompressionOptions { compression_level: 1 },
             filter_by_template: true,
@@ -3898,7 +3408,7 @@ mod tests {
             min_mean_base_quality: None,
             max_no_call_fraction: 0.5,
             reverse_per_base_tags: false,
-            sort_order: None,
+
             threading: ThreadingOptions::none(),
             compression: CompressionOptions { compression_level: 1 },
             filter_by_template: true,
@@ -3929,7 +3439,7 @@ mod tests {
             min_mean_base_quality: None,
             max_no_call_fraction: 0.5,
             reverse_per_base_tags: false,
-            sort_order: None,
+
             threading: ThreadingOptions::none(),
             compression: CompressionOptions { compression_level: 1 },
             filter_by_template: true,
@@ -3984,7 +3494,7 @@ mod tests {
             min_mean_base_quality: None,
             max_no_call_fraction: 0.5,
             reverse_per_base_tags: true, // Test reverse tags in parallel mode
-            sort_order: None,
+
             threading: ThreadingOptions::new(4),
             compression: CompressionOptions { compression_level: 1 },
             filter_by_template: true,
@@ -4044,7 +3554,7 @@ mod tests {
             min_mean_base_quality: None,
             max_no_call_fraction: 0.5,
             reverse_per_base_tags: false,
-            sort_order: None,
+
             threading: ThreadingOptions::new(4), // Multi-threaded
             compression: CompressionOptions { compression_level: 1 },
             filter_by_template: true,
@@ -4069,11 +3579,11 @@ mod tests {
     /// Parameterized test for all threading modes.
     ///
     /// Tests:
-    /// - `None`: Single-threaded fast path, no pipeline
+    /// - `None`: Pipeline with default (1 thread)
     /// - `Some(1)`: Pipeline with 1 thread
     /// - `Some(2)`: Pipeline with 2 threads
     #[rstest]
-    #[case::fast_path(ThreadingOptions::none())]
+    #[case::default(ThreadingOptions::none())]
     #[case::pipeline_1(ThreadingOptions::new(1))]
     #[case::pipeline_2(ThreadingOptions::new(2))]
     fn test_threading_modes(#[case] threading: ThreadingOptions) -> Result<()> {
@@ -4107,7 +3617,7 @@ mod tests {
             min_mean_base_quality: None,
             max_no_call_fraction: 0.5,
             reverse_per_base_tags: false,
-            sort_order: None,
+
             threading,
             compression: CompressionOptions { compression_level: 1 },
             filter_by_template: true,
diff --git a/src/commands/simplex.rs b/src/commands/simplex.rs
index bd8bba667..25d85b167 100644
--- a/src/commands/simplex.rs
+++ b/src/commands/simplex.rs
@@ -70,7 +70,9 @@ struct SimplexProcessedBatch {
 
 impl MemoryEstimate for SimplexProcessedBatch {
     fn estimate_heap_size(&self) -> usize {
-        self.consensus_output.data.len() + self.rejects.iter().map(Vec::len).sum::<usize>()
+        self.consensus_output.data.capacity()
+            + self.rejects.iter().map(Vec::capacity).sum::<usize>()
+            + self.rejects.capacity() * std::mem::size_of::<Vec<u8>>()
     }
 }
 
@@ -1585,4 +1587,26 @@ mod tests {
 
         Ok(())
     }
+
+    #[test]
+    fn test_simplex_processed_batch_memory_estimate() {
+        let mut data = Vec::with_capacity(1024);
+        data.extend_from_slice(&[0u8; 100]);
+        let mut reject = Vec::with_capacity(512);
+        reject.extend_from_slice(&[0u8; 50]);
+
+        let batch = SimplexProcessedBatch {
+            consensus_output: ConsensusOutput { data, count: 1 },
+            rejects: vec![reject],
+            groups_count: 1,
+            stats: ConsensusCallingStats::default(),
+            overlapping_stats: None,
+        };
+
+        let estimate = batch.estimate_heap_size();
+        // Should use capacity (1024 + 512) not len (100 + 50)
+        assert!(estimate >= 1024 + 512, "estimate {estimate} should be >= 1536 (capacities)");
+        // Should also include Vec<Vec<u8>> overhead
+        assert!(estimate > 1024 + 512, "estimate {estimate} should include Vec overhead");
+    }
 }
diff --git a/src/lib/bam_io.rs b/src/lib/bam_io.rs
index fb05f1207..e3d8b2909 100644
--- a/src/lib/bam_io.rs
+++ b/src/lib/bam_io.rs
@@ -244,6 +244,19 @@ impl RawBamWriter {
         self.inner.write_all(record_bytes)
     }
 
+    /// Write pre-formatted BAM record bytes directly to the BGZF stream.
+    ///
+    /// Unlike `write_raw_record`, this does NOT add a length prefix — the data
+    /// must already contain properly formatted BAM records (4-byte LE size + record bytes).
+    /// Used by the pipeline for bulk secondary output.
+    ///
+    /// # Errors
+    /// Returns an error if writing to the underlying writer fails.
+    pub fn write_raw_bytes(&mut self, data: &[u8]) -> io::Result<()> {
+        use std::io::Write;
+        self.inner.write_all(data)
+    }
+
     /// Finish writing and close the writer.
     ///
     /// # Errors
@@ -1505,4 +1518,26 @@ mod tests {
 
         Ok(())
     }
+
+    #[test]
+    fn test_write_raw_bytes() -> Result<()> {
+        let temp = tempfile::NamedTempFile::new()?;
+        let header = noodles::sam::Header::default();
+        let mut writer = create_raw_bam_writer(temp.path(), &header, 1, 1)?;
+
+        // Write pre-formatted BAM record bytes (4-byte LE length + record)
+        let record_bytes = vec![1, 2, 3, 4, 5];
+        let mut formatted = Vec::new();
+        #[allow(clippy::cast_possible_truncation)]
+        let len = record_bytes.len() as u32;
+        formatted.extend_from_slice(&len.to_le_bytes());
+        formatted.extend_from_slice(&record_bytes);
+
+        writer.write_raw_bytes(&formatted)?;
+        writer.finish()?;
+
+        // Verify file is non-empty (contains header + data + EOF)
+        assert!(temp.path().metadata()?.len() > 0);
+        Ok(())
+    }
 }
diff --git a/src/lib/template.rs b/src/lib/template.rs
index 23e825048..c2b89d93f 100644
--- a/src/lib/template.rs
+++ b/src/lib/template.rs
@@ -1573,6 +1573,8 @@ impl MemoryEstimate for TemplateBatch {
 #[must_use]
 pub fn estimate_record_buf_heap_size(record: &RecordBuf) -> usize {
     // Name: Option<BString> which is a Vec<u8>
+    // NB: noodles API returns &BStr, so we can only measure len(), not capacity().
+    // Same constraint applies to sequence and quality_scores below.
     let name_size = record.name().map_or(0, |n| n.len());
 
     // Sequence: noodles stores bases as unpacked ASCII (1 byte per base in Vec<u8>)
diff --git a/src/lib/unified_pipeline/bam.rs b/src/lib/unified_pipeline/bam.rs
index 8faf08403..297869a68 100644
--- a/src/lib/unified_pipeline/bam.rs
+++ b/src/lib/unified_pipeline/bam.rs
@@ -1429,6 +1429,11 @@ pub struct PipelineFunctions<G: Send, P: Send> {
     /// Appends serialized BAM bytes to the buffer and returns the record count.
     /// This enables buffer reuse in single-threaded mode to avoid allocations.
     pub serialize_fn: Box<dyn Fn(P, &mut Vec<u8>) -> io::Result<u64> + Send + Sync>,
+
+    /// Optional secondary serialization (e.g., rejected records).
+    /// Called with a borrow of P BEFORE the primary `serialize_fn` consumes it.
+    pub secondary_serialize_fn:
+        Option<Box<dyn Fn(&P, &mut Vec<u8>) -> io::Result<u64> + Send + Sync>>,
 }
 
 impl<G: Send, P: Send> PipelineFunctions<G, P> {
@@ -1438,7 +1443,21 @@ impl<G: Send, P: Send> PipelineFunctions<G, P> {
         ProcessFn: Fn(G) -> io::Result<P> + Send + Sync + 'static,
         SerializeFn: Fn(P, &mut Vec<u8>) -> io::Result<u64> + Send + Sync + 'static,
     {
-        Self { process_fn: Box::new(process_fn), serialize_fn: Box::new(serialize_fn) }
+        Self {
+            process_fn: Box::new(process_fn),
+            serialize_fn: Box::new(serialize_fn),
+            secondary_serialize_fn: None,
+        }
+    }
+
+    /// Attach a secondary serialize function for dual-output pipelines.
+    #[must_use]
+    pub fn with_secondary_serialize<F>(mut self, f: F) -> Self
+    where
+        F: Fn(&P, &mut Vec<u8>) -> io::Result<u64> + Send + Sync + 'static,
+    {
+        self.secondary_serialize_fn = Some(Box::new(f));
+        self
     }
 }
 
@@ -2690,12 +2709,21 @@ fn try_step_serialize<G: Send + 'static, P: Send + MemoryEstimate + 'static>(
     // =========================================================================
     // Priority 4: Serialize all items
     // =========================================================================
-    // Prepare worker's serialization buffer
+    // Prepare worker's serialization buffers
     worker.core.serialization_buffer.clear();
+    worker.core.secondary_serialization_buffer.clear();
 
     // Serialize all items into worker's buffer
     let mut total_record_count: u64 = 0;
     for item in batch {
+        // Secondary serialize (borrows item) — must run before primary consumes it
+        if let Some(ref secondary_fn) = fns.secondary_serialize_fn {
+            if let Err(e) = (secondary_fn)(&item, &mut worker.core.secondary_serialization_buffer) {
+                state.set_error(e);
+                return false;
+            }
+        }
+        // Primary serialize (consumes item)
         match (fns.serialize_fn)(item, &mut worker.core.serialization_buffer) {
             Ok(record_count) => {
                 total_record_count += record_count;
@@ -2713,6 +2741,16 @@ fn try_step_serialize<G: Send + 'static, P: Send + MemoryEstimate + 'static>(
         Vec::with_capacity(SERIALIZATION_BUFFER_CAPACITY),
     );
 
+    // Build secondary data if any was serialized
+    let secondary_data = if worker.core.secondary_serialization_buffer.is_empty() {
+        None
+    } else {
+        Some(std::mem::replace(
+            &mut worker.core.secondary_serialization_buffer,
+            Vec::with_capacity(SERIALIZATION_BUFFER_CAPACITY),
+        ))
+    };
+
     // Record serialized bytes for throughput metrics
     if let Some(stats) = state.stats() {
         stats.serialized_bytes.fetch_add(combined_data.len() as u64, Ordering::Relaxed);
@@ -2721,7 +2759,8 @@ fn try_step_serialize<G: Send + 'static, P: Send + MemoryEstimate + 'static>(
     // =========================================================================
     // Priority 5: Try to push result (non-blocking)
     // =========================================================================
-    let batch = SerializedBatch { data: combined_data, record_count: total_record_count };
+    let batch =
+        SerializedBatch { data: combined_data, record_count: total_record_count, secondary_data };
     let heap_size = batch.estimate_heap_size();
     match state.output.serialized.push((serial, batch)) {
         Ok(()) => {
@@ -2794,6 +2833,21 @@ fn try_step_write<G: Send + 'static, P: Send + MemoryEstimate + 'static>(
                 batch_bytes += block.data.len() as u64;
             }
 
+            // Write secondary data (e.g., rejects) in the same serial order
+            if let Some(ref secondary_data) = batch.secondary_data {
+                if !secondary_data.is_empty() {
+                    if let Some(ref secondary_mutex) = state.output.secondary_output {
+                        let mut sw_guard = secondary_mutex.lock();
+                        if let Some(ref mut sw) = *sw_guard {
+                            if let Err(e) = sw.write_raw_bytes(secondary_data) {
+                                state.set_error(e);
+                                return (false, false);
+                            }
+                        }
+                    }
+                }
+            }
+
             // Record bytes written for throughput metrics
             if let Some(stats) = state.stats() {
                 stats.bytes_written.fetch_add(batch_bytes, Ordering::Relaxed);
@@ -2974,6 +3028,9 @@ struct SingleThreadedBuffers {
     /// Buffer for serialized BAM record data.
     /// Cleared and reused each group to avoid allocation.
     serialized: Vec<u8>,
+    /// Buffer for secondary serialized data (e.g., rejected records).
+    /// Only used when a secondary serialize function is set.
+    secondary: Vec<u8>,
 }
 
 impl SingleThreadedBuffers {
@@ -2984,6 +3041,7 @@ impl SingleThreadedBuffers {
             decompressed: Vec::with_capacity(256 * 1024),
             // Typical group serializes to ~64KB
             serialized: Vec::with_capacity(64 * 1024),
+            secondary: Vec::new(),
         }
     }
 }
@@ -3001,6 +3059,7 @@ fn run_bam_pipeline_single_threaded<G, P>(
     mut grouper: Box<dyn Grouper<Group = G> + Send>,
     fns: PipelineFunctions<G, P>,
     group_key_config: GroupKeyConfig,
+    mut secondary_writer: Option<crate::bam_io::RawBamWriter>,
 ) -> io::Result<u64>
 where
     G: Send + 'static,
@@ -3060,7 +3119,13 @@ where
                 // Step 6: Process
                 let processed = (fns.process_fn)(group)?;
 
-                // Step 7: Serialize (reuse buffer)
+                // Step 7a: Secondary serialize (borrows processed)
+                buffers.secondary.clear();
+                if let Some(ref secondary_fn) = fns.secondary_serialize_fn {
+                    (secondary_fn)(&processed, &mut buffers.secondary)?;
+                }
+
+                // Step 7b: Primary serialize (consumes processed, reuse buffer)
                 buffers.serialized.clear();
                 let record_count = (fns.serialize_fn)(processed, &mut buffers.serialized)?;
 
@@ -3071,6 +3136,13 @@ where
                 // Step 9: Write any completed blocks to output
                 compressor.write_blocks_to(output.as_mut())?;
 
+                // Write secondary data
+                if !buffers.secondary.is_empty() {
+                    if let Some(ref mut sw) = secondary_writer {
+                        sw.write_raw_bytes(&buffers.secondary)?;
+                    }
+                }
+
                 progress.log_if_needed(record_count);
             }
         }
@@ -3084,11 +3156,24 @@ where
 
             for group in groups {
                 let processed = (fns.process_fn)(group)?;
+
+                buffers.secondary.clear();
+                if let Some(ref secondary_fn) = fns.secondary_serialize_fn {
+                    (secondary_fn)(&processed, &mut buffers.secondary)?;
+                }
+
                 buffers.serialized.clear();
                 let record_count = (fns.serialize_fn)(processed, &mut buffers.serialized)?;
                 compressor.write_all(&buffers.serialized)?;
                 compressor.maybe_compress()?;
                 compressor.write_blocks_to(output.as_mut())?;
+
+                if !buffers.secondary.is_empty() {
+                    if let Some(ref mut sw) = secondary_writer {
+                        sw.write_raw_bytes(&buffers.secondary)?;
+                    }
+                }
+
                 progress.log_if_needed(record_count);
             }
         }
@@ -3099,7 +3184,13 @@ where
         // Step 6: Process
         let processed = (fns.process_fn)(final_group)?;
 
-        // Step 7: Serialize (reuse buffer)
+        // Step 7a: Secondary serialize (borrows processed)
+        buffers.secondary.clear();
+        if let Some(ref secondary_fn) = fns.secondary_serialize_fn {
+            (secondary_fn)(&processed, &mut buffers.secondary)?;
+        }
+
+        // Step 7b: Primary serialize (consumes processed, reuse buffer)
         buffers.serialized.clear();
         let record_count = (fns.serialize_fn)(processed, &mut buffers.serialized)?;
 
@@ -3110,6 +3201,13 @@ where
         // Step 9: Write any completed blocks to output
         compressor.write_blocks_to(output.as_mut())?;
 
+        // Write secondary data
+        if !buffers.secondary.is_empty() {
+            if let Some(ref mut sw) = secondary_writer {
+                sw.write_raw_bytes(&buffers.secondary)?;
+            }
+        }
+
         progress.log_if_needed(record_count);
     }
 
@@ -3120,6 +3218,13 @@ where
     // Flush output
     output.flush()?;
 
+    // Finalize secondary output writer
+    if let Some(writer) = secondary_writer {
+        writer.finish().map_err(|e| {
+            io::Error::new(e.kind(), format!("Failed to finalize secondary output: {e}"))
+        })?;
+    }
+
     Ok(progress.count())
 }
 
@@ -3157,6 +3262,7 @@ pub fn run_bam_pipeline<G, P>(
     grouper: Box<dyn Grouper<Group = G> + Send>,
     fns: PipelineFunctions<G, P>,
     group_key_config: GroupKeyConfig,
+    secondary_writer: Option<crate::bam_io::RawBamWriter>,
 ) -> io::Result<u64>
 where
     G: Send + BatchWeight + MemoryEstimate + 'static,
@@ -3177,10 +3283,15 @@ where
             grouper,
             fns,
             group_key_config,
+            secondary_writer,
         );
     }
 
-    let state = Arc::new(BamPipelineState::<G, P>::new(config, input, output, group_key_config));
+    let mut state = BamPipelineState::<G, P>::new(config, input, output, group_key_config);
+    if let Some(sw) = secondary_writer {
+        state.output.set_secondary_output(sw);
+    }
+    let state = Arc::new(state);
 
     // Set num_threads for stats display
     if let Some(stats) = state.stats() {
@@ -3353,7 +3464,25 @@ where
     join_monitor_thread(monitor_handle);
 
     // Finalize: check errors, flush output, log stats
-    finalize_pipeline(&*state)
+    let result = finalize_pipeline(&*state);
+
+    // Finalize secondary output writer (if present)
+    if let Some(ref secondary_mutex) = state.output.secondary_output {
+        let mut guard = secondary_mutex.lock();
+        if let Some(writer) = guard.take() {
+            if let Err(e) = writer.finish().map_err(|e| {
+                io::Error::new(e.kind(), format!("Failed to finalize secondary output: {e}"))
+            }) {
+                if result.is_err() {
+                    log::error!("Secondary output finalization also failed: {e}");
+                } else {
+                    return Err(e);
+                }
+            }
+        }
+    }
+
+    result
 }
 
 // ============================================================================
@@ -3438,7 +3567,7 @@ pub fn serialize_bam_records(
 ) -> io::Result<SerializedBatch> {
     let mut data = Vec::with_capacity(records.len() * 256);
     let record_count = serialize_bam_records_into(records, header, &mut data)?;
-    Ok(SerializedBatch { data, record_count })
+    Ok(SerializedBatch { data, record_count, secondary_data: None })
 }
 
 /// Serialize a single BAM record to bytes.
@@ -3452,7 +3581,7 @@ pub fn serialize_bam_records(
 pub fn serialize_bam_record(record: &RecordBuf, header: &Header) -> io::Result<SerializedBatch> {
     let mut data = Vec::with_capacity(256);
     let record_count = serialize_bam_record_into(record, header, &mut data)?;
-    Ok(SerializedBatch { data, record_count })
+    Ok(SerializedBatch { data, record_count, secondary_data: None })
 }
 
 /// Serialize a single BAM record to bytes, appending to the provided buffer.
@@ -3701,6 +3830,7 @@ where
         grouper,
         fns,
         group_key_config,
+        None,
     );
 
     // After pipeline completes, write BGZF EOF block to finalize the BAM file
@@ -3829,6 +3959,7 @@ where
         grouper,
         fns,
         group_key_config,
+        None,
     );
 
     // After pipeline completes, write BGZF EOF block to finalize the BAM file
@@ -3953,6 +4084,7 @@ where
         grouper,
         fns,
         group_key_config,
+        None,
     );
 
     // After pipeline completes, write BGZF EOF block to finalize the BAM file
@@ -3973,6 +4105,119 @@ where
     result
 }
 
+/// Run a BAM pipeline from an already-opened reader, with a secondary output file.
+///
+/// This variant routes rejected/secondary records through the pipeline's ordering
+/// infrastructure so both primary and secondary output files maintain input order.
+///
+/// The secondary serialize function is called with a borrow of the processed batch
+/// BEFORE the primary serialize function consumes it.
+///
+/// # Errors
+///
+/// Returns an I/O error if any pipeline step or file I/O fails.
+#[allow(clippy::too_many_arguments, clippy::needless_pass_by_value)]
+pub fn run_bam_pipeline_from_reader_with_secondary<
+    G,
+    P,
+    R,
+    GrouperFn,
+    ProcessFn,
+    SerializeFn,
+    SecondaryFn,
+>(
+    config: BamPipelineConfig,
+    input: R,
+    input_header: Header,
+    output_path: &Path,
+    output_header: Option<Header>,
+    secondary_output_path: &Path,
+    grouper_fn: GrouperFn,
+    process_fn: ProcessFn,
+    serialize_fn: SerializeFn,
+    secondary_serialize_fn: SecondaryFn,
+) -> io::Result<u64>
+where
+    G: Send + BatchWeight + MemoryEstimate + 'static,
+    P: Send + MemoryEstimate + 'static,
+    R: Read + Send + 'static,
+    GrouperFn: FnOnce(&Header) -> Box<dyn Grouper<Group = G> + Send>,
+    ProcessFn: Fn(G) -> io::Result<P> + Send + Sync + 'static,
+    SerializeFn: Fn(P, &Header, &mut Vec<u8>) -> io::Result<u64> + Send + Sync + 'static,
+    SecondaryFn: Fn(&P, &mut Vec<u8>) -> io::Result<u64> + Send + Sync + 'static,
+{
+    // Use output_header if provided, otherwise clone input_header
+    let output_header = output_header.unwrap_or_else(|| input_header.clone());
+
+    // Create primary output BAM and write the output header
+    let output_file = File::create(output_path)
+        .map_err(|e| io::Error::new(e.kind(), format!("Failed to create output: {e}")))?;
+
+    let mut header_writer = bam::io::Writer::new(output_file);
+    header_writer
+        .write_header(&output_header)
+        .map_err(|e| io::Error::other(format!("Failed to write BAM header: {e}")))?;
+
+    let mut bgzf_writer = header_writer.into_inner();
+    bgzf_writer
+        .try_finish()
+        .map_err(|e| io::Error::other(format!("Failed to finish BGZF header: {e}")))?;
+    let output = bgzf_writer.into_inner();
+    let output = BufWriter::with_capacity(IO_BUFFER_SIZE, output);
+
+    // Create secondary output (reject BAM) with its own BGZF compression
+    let secondary_writer = crate::bam_io::create_raw_bam_writer(
+        secondary_output_path,
+        &output_header,
+        1, // single-threaded BGZF for secondary
+        config.compression_level,
+    )
+    .map_err(|e| io::Error::other(format!("Failed to create secondary output: {e}")))?;
+
+    // Build GroupKeyConfig
+    let group_key_config = config.group_key_config.unwrap_or_else(|| {
+        use noodles::sam::alignment::record::data::field::Tag;
+        let library_index = LibraryIndex::from_header(&input_header);
+        let cell_tag = Tag::from([b'C', b'B']);
+        GroupKeyConfig::new(library_index, cell_tag)
+    });
+
+    let grouper = grouper_fn(&input_header);
+
+    let fns = PipelineFunctions::new(process_fn, move |p: P, buf: &mut Vec<u8>| {
+        serialize_fn(p, &output_header, buf)
+    })
+    .with_secondary_serialize(secondary_serialize_fn);
+
+    let pipeline_config = config.pipeline;
+
+    let result = run_bam_pipeline(
+        pipeline_config,
+        Box::new(input),
+        Box::new(output),
+        grouper,
+        fns,
+        group_key_config,
+        Some(secondary_writer),
+    );
+
+    // After pipeline completes, write BGZF EOF block to finalize the primary BAM file
+    if result.is_ok() {
+        use std::io::Write as _;
+
+        let mut output_file = std::fs::OpenOptions::new()
+            .append(true)
+            .open(output_path)
+            .map_err(|e| io::Error::new(e.kind(), format!("Failed to open output for EOF: {e}")))?;
+
+        output_file
+            .write_all(&BGZF_EOF)
+            .map_err(|e| io::Error::new(e.kind(), format!("Failed to write BGZF EOF: {e}")))?;
+    }
+
+    result
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -4257,7 +4502,8 @@ mod tests {
         // Add item to write_reorder buffer
         {
             let mut write_reorder = state.output.write_reorder.lock();
-            let batch = CompressedBlockBatch { blocks: vec![], record_count: 0 };
+            let batch =
+                CompressedBlockBatch { blocks: vec![], record_count: 0, secondary_data: None };
             write_reorder.insert(0, batch);
         }
 
@@ -4266,4 +4512,26 @@ mod tests {
         let err = result.unwrap_err();
         assert!(err.non_empty_queues.iter().any(|s| s.contains("write_reorder")));
     }
+
+    #[test]
+    fn test_pipeline_functions_secondary_serialize() {
+        let fns = PipelineFunctions::<Vec<u8>, Vec<u8>>::new(Ok, |data, buf| {
+            buf.extend_from_slice(&data);
+            Ok(1)
+        });
+        assert!(fns.secondary_serialize_fn.is_none());
+
+        let fns = fns.with_secondary_serialize(|data: &Vec<u8>, buf: &mut Vec<u8>| {
+            buf.extend_from_slice(data);
+            Ok(1)
+        });
+        assert!(fns.secondary_serialize_fn.is_some());
+
+        // Verify the secondary serialize function works
+        let test_data = vec![1u8, 2, 3, 4];
+        let mut buf = Vec::new();
+        let count = (fns.secondary_serialize_fn.as_ref().unwrap())(&test_data, &mut buf).unwrap();
+        assert_eq!(count, 1);
+        assert_eq!(buf, vec![1, 2, 3, 4]);
+    }
 }
diff --git a/src/lib/unified_pipeline/base.rs b/src/lib/unified_pipeline/base.rs
index 248ff56f7..38f74a54e 100644
--- a/src/lib/unified_pipeline/base.rs
+++ b/src/lib/unified_pipeline/base.rs
@@ -238,8 +238,8 @@ pub fn log_comprehensive_memory_stats(stats: &PipelineStats) {
         stats.update_system_rss(rss);
     }
 
-    // TODO: Queue memory integration deferred - requires pipeline reference to read actual queues
-    // For now, queue memory is tracked through estimates only
+    // Queue memory integration deferred — reading actual queue sizes requires a pipeline
+    // reference that is not available here.  Queue memory is tracked through estimates only.
 
     let breakdown = stats.get_memory_breakdown();
 
@@ -890,19 +890,22 @@ pub struct CompressedBlockBatch {
     pub blocks: Vec<CompressedBlock>,
     /// Number of records/templates in this batch (for progress logging).
     pub record_count: u64,
+    /// Optional secondary data passed through without compression.
+    /// Used for dual-output pipelines (e.g., rejected records).
+    pub secondary_data: Option<Vec<u8>>,
 }
 
 impl CompressedBlockBatch {
     /// Create a new empty batch.
     #[must_use]
     pub fn new() -> Self {
-        Self { blocks: Vec::new(), record_count: 0 }
+        Self { blocks: Vec::new(), record_count: 0, secondary_data: None }
     }
 
     /// Create a batch with pre-allocated capacity.
     #[must_use]
     pub fn with_capacity(capacity: usize) -> Self {
-        Self { blocks: Vec::with_capacity(capacity), record_count: 0 }
+        Self { blocks: Vec::with_capacity(capacity), record_count: 0, secondary_data: None }
     }
 
     /// Number of blocks in this batch.
@@ -927,6 +930,7 @@ impl CompressedBlockBatch {
     pub fn clear(&mut self) {
         self.blocks.clear();
         self.record_count = 0;
+        self.secondary_data = None;
     }
 }
 
@@ -935,6 +939,7 @@ impl MemoryEstimate for CompressedBlockBatch {
         // Vec overhead + sum of block data sizes
         self.blocks.iter().map(|b| b.data.capacity()).sum::<usize>()
             + self.blocks.capacity() * std::mem::size_of::<CompressedBlock>()
+            + self.secondary_data.as_ref().map_or(0, Vec::capacity)
     }
 }
 
@@ -1504,7 +1509,8 @@ impl MemoryEstimate for RecordBuf {
 
 impl MemoryEstimate for Vec<RecordBuf> {
     fn estimate_heap_size(&self) -> usize {
-        self.iter().map(MemoryEstimate::estimate_heap_size).sum()
+        self.iter().map(MemoryEstimate::estimate_heap_size).sum::<usize>()
+            + self.capacity() * std::mem::size_of::<RecordBuf>()
     }
 }
 
@@ -1617,13 +1623,16 @@ pub struct SerializedBatch {
     pub data: Vec<u8>,
     /// Number of records/templates in this batch (for progress logging).
     pub record_count: u64,
+    /// Optional secondary output data (e.g., rejected records).
+    /// Passed through compression unchanged and written to a secondary output file.
+    pub secondary_data: Option<Vec<u8>>,
 }
 
 impl SerializedBatch {
     /// Create a new empty batch.
     #[must_use]
     pub fn new() -> Self {
-        Self { data: Vec::new(), record_count: 0 }
+        Self { data: Vec::new(), record_count: 0, secondary_data: None }
     }
 
     /// Returns true if the batch contains no data.
@@ -1636,12 +1645,13 @@ impl SerializedBatch {
     pub fn clear(&mut self) {
         self.data.clear();
         self.record_count = 0;
+        self.secondary_data = None;
     }
 }
 
 impl MemoryEstimate for SerializedBatch {
     fn estimate_heap_size(&self) -> usize {
-        self.data.capacity()
+        self.data.capacity() + self.secondary_data.as_ref().map_or(0, Vec::capacity)
     }
 }
 
@@ -1688,6 +1698,8 @@ pub struct OutputPipelineQueues<G, P: MemoryEstimate> {
     // ========== Output ==========
     /// Output file, mutex-protected for exclusive access.
     pub output: Mutex<Option<Box<dyn Write + Send>>>,
+    /// Optional secondary output writer for dual-output pipelines (e.g., reject BAM).
+    pub secondary_output: Option<Mutex<Option<crate::bam_io::RawBamWriter>>>,
 
     // ========== Completion and Error Tracking ==========
     /// Flag indicating an error occurred.
@@ -1732,6 +1744,7 @@ impl<G: Send, P: Send + MemoryEstimate> OutputPipelineQueues<G, P> {
             compressed_heap_bytes: AtomicU64::new(0),
             write_reorder: Mutex::new(ReorderBuffer::new()),
             output: Mutex::new(Some(output)),
+            secondary_output: None,
             error_flag: AtomicBool::new(false),
             error: Mutex::new(None),
             items_written: AtomicU64::new(0),
@@ -1741,6 +1754,13 @@ impl<G: Send, P: Send + MemoryEstimate> OutputPipelineQueues<G, P> {
         }
     }
 
+    /// Set a secondary output writer for dual-output pipelines.
+    ///
+    /// Must be called before the pipeline is started.
+    pub fn set_secondary_output(&mut self, writer: crate::bam_io::RawBamWriter) {
+        self.secondary_output = Some(Mutex::new(Some(writer)));
+    }
+
     // ========== Error Handling ==========
 
     /// Record an error and signal threads to stop.
@@ -1857,6 +1877,9 @@ pub struct WorkerCoreState {
     /// Reusable buffer for serialization.
     /// Swapped out each batch to avoid per-batch allocation.
     pub serialization_buffer: Vec<u8>,
+    /// Reusable buffer for secondary serialization (e.g., reject output).
+    /// Only allocated when a secondary serialize function is used.
+    pub secondary_serialization_buffer: Vec<u8>,
     /// Current backoff duration in microseconds (for adaptive backoff).
     pub backoff_us: u64,
     /// Recycled serialization buffers to avoid repeated allocation.
@@ -1886,6 +1909,7 @@ impl WorkerCoreState {
             compressor: InlineBgzfCompressor::new(compression_level),
             scheduler: create_scheduler(scheduler_strategy, thread_id, num_threads, active_steps),
             serialization_buffer: Vec::with_capacity(SERIALIZATION_BUFFER_CAPACITY),
+            secondary_serialization_buffer: Vec::new(),
             backoff_us: MIN_BACKOFF_US,
             recycled_buffers: Vec::with_capacity(2),
         }
@@ -4535,7 +4559,7 @@ where
     // =========================================================================
     // Priority 4: Compress the data
     // =========================================================================
-    let SerializedBatch { data, record_count } = serialized;
+    let SerializedBatch { data, record_count, secondary_data } = serialized;
 
     // Scope the compressor borrow so we can recycle the buffer afterwards
     let blocks = {
@@ -4561,7 +4585,7 @@ where
     let compressed_bytes: u64 = blocks.iter().map(|b| b.data.len() as u64).sum();
     state.record_compressed_bytes_out(compressed_bytes);
 
-    let batch = CompressedBlockBatch { blocks, record_count };
+    let batch = CompressedBlockBatch { blocks, record_count, secondary_data };
 
     // =========================================================================
     // Priority 5: Try to push result
@@ -4925,7 +4949,7 @@ where
     state.record_serialized_bytes(data.len() as u64);
     state.record_serialized_records(total_records);
 
-    let result_batch = SerializedBatch { data, record_count: total_records };
+    let result_batch = SerializedBatch { data, record_count: total_records, secondary_data: None };
 
     // Priority 6: Push result
     match state.serialize_output_push((serial, result_batch)) {
@@ -5512,9 +5536,11 @@ mod tests {
     fn test_compressed_block_batch_clear() {
         let mut batch = CompressedBlockBatch::new();
         batch.record_count = 42;
+        batch.secondary_data = Some(vec![1, 2, 3]);
         batch.clear();
         assert!(batch.is_empty());
         assert_eq!(batch.record_count, 0);
+        assert!(batch.secondary_data.is_none());
     }
 
     #[test]
@@ -5544,9 +5570,11 @@ mod tests {
         let mut batch = SerializedBatch::new();
         batch.data.extend_from_slice(&[1, 2, 3]);
         batch.record_count = 10;
+        batch.secondary_data = Some(vec![4, 5, 6]);
         batch.clear();
         assert!(batch.is_empty());
         assert_eq!(batch.record_count, 0);
+        assert!(batch.secondary_data.is_none());
     }
 
     // ========================================================================
@@ -5820,4 +5848,59 @@ mod tests {
         batch.data.reserve(2048);
         assert!(batch.estimate_heap_size() >= 2048);
     }
+
+    #[test]
+    fn test_memory_estimate_vec_record_buf() {
+        use crate::sam::builder::RecordBuilder;
+        use noodles::sam::alignment::record_buf::RecordBuf;
+
+        let record = RecordBuilder::new().sequence("ACGT").qualities(&[30, 30, 30, 30]).build();
+        let mut records: Vec<RecordBuf> = Vec::with_capacity(10);
+        records.push(record);
+
+        let estimate = records.estimate_heap_size();
+        // Should include Vec overhead for capacity * size_of::<RecordBuf>()
+        let vec_overhead = 10 * std::mem::size_of::<RecordBuf>();
+        assert!(
+            estimate >= vec_overhead,
+            "estimate {estimate} should include Vec<RecordBuf> overhead {vec_overhead}"
+        );
+    }
+
+    #[test]
+    fn test_serialized_batch_memory_estimate_with_secondary() {
+        let batch = SerializedBatch {
+            data: vec![0u8; 100],
+            record_count: 5,
+            secondary_data: Some(vec![0u8; 50]),
+        };
+        let estimate = batch.estimate_heap_size();
+        assert!(
+            estimate >= 150,
+            "Should include both primary ({}) and secondary data, got {estimate}",
+            batch.data.capacity()
+        );
+    }
+
+    #[test]
+    fn test_serialized_batch_memory_estimate_without_secondary() {
+        let batch = SerializedBatch { data: vec![0u8; 100], record_count: 5, secondary_data: None };
+        let estimate = batch.estimate_heap_size();
+        assert!(estimate >= 100);
+        assert!(estimate < 150, "Should not include phantom secondary data, got {estimate}");
+    }
+
+    #[test]
+    fn test_compressed_block_batch_memory_with_secondary() {
+        let batch = CompressedBlockBatch {
+            blocks: vec![],
+            record_count: 0,
+            secondary_data: Some(vec![0u8; 200]),
+        };
+        assert!(
+            batch.estimate_heap_size() >= 200,
+            "Should include secondary data, got {}",
+            batch.estimate_heap_size()
+        );
+    }
 }
diff --git a/src/lib/unified_pipeline/fastq.rs b/src/lib/unified_pipeline/fastq.rs
index 265753883..7e5274ec9 100644
--- a/src/lib/unified_pipeline/fastq.rs
+++ b/src/lib/unified_pipeline/fastq.rs
@@ -176,7 +176,8 @@ pub struct FastqDecompressedChunk {
 
 impl MemoryEstimate for FastqDecompressedBatch {
     fn estimate_heap_size(&self) -> usize {
-        self.chunks.iter().map(|c| c.data.capacity()).sum()
+        self.chunks.iter().map(|c| c.data.capacity()).sum::<usize>()
+            + self.chunks.capacity() * std::mem::size_of::<FastqDecompressedChunk>()
     }
 }
 
@@ -207,7 +208,8 @@ impl MemoryEstimate for FastqBoundaryBatch {
         self.streams
             .iter()
             .map(|s| s.data.capacity() + s.offsets.capacity() * std::mem::size_of::<usize>())
-            .sum()
+            .sum::<usize>()
+            + self.streams.capacity() * std::mem::size_of::<FastqStreamBoundaries>()
     }
 }
 
@@ -239,8 +241,10 @@ impl MemoryEstimate for FastqParsedBatch {
                     .iter()
                     .map(|r| r.name.capacity() + r.sequence.capacity() + r.quality.capacity())
                     .sum::<usize>()
+                    + stream.records.capacity() * std::mem::size_of::<FastqRecord>()
             })
-            .sum()
+            .sum::<usize>()
+            + self.streams.capacity() * std::mem::size_of::<FastqParsedStream>()
     }
 }
 
@@ -2167,7 +2171,8 @@ fn fastq_try_step_parse<R: BufRead + Send, P: Send + MemoryEstimate>(
                 Ok(()) => {
                     #[cfg(feature = "memory-debug")]
                     {
-                        // TODO: compute heap size if needed for memory-debug
+                        // Heap size tracking for pushed templates is not yet implemented;
+                        // queue memory is tracked through estimates only for now.
                         let q4_heap: u64 = 0;
                         state.output.groups_heap_bytes.fetch_add(q4_heap, Ordering::AcqRel);
                     }
@@ -2480,7 +2485,11 @@ where
     // =========================================================================
     // Priority 6: Try to push result (non-blocking)
     // =========================================================================
-    let batch = SerializedBatch { data: combined_data, record_count: total_record_count };
+    let batch = SerializedBatch {
+        data: combined_data,
+        record_count: total_record_count,
+        secondary_data: None,
+    };
     let heap_size = batch.estimate_heap_size();
     match state.output.serialized.push((serial, batch)) {
         Ok(()) => {
@@ -3939,4 +3948,77 @@ mod tests {
         );
         assert!(!steps.is_active(Group));
     }
+
+    #[test]
+    fn test_fastq_decompressed_batch_memory_estimate() {
+        let mut data = Vec::with_capacity(2048);
+        data.extend_from_slice(&[0u8; 100]);
+        let mut chunks = Vec::with_capacity(4);
+        chunks.push(FastqDecompressedChunk { stream_idx: 0, data });
+
+        let batch = FastqDecompressedBatch { chunks, serial: 0 };
+        let estimate = batch.estimate_heap_size();
+
+        // Should use data capacity (2048) not len (100)
+        assert!(estimate >= 2048, "estimate {estimate} should be >= 2048 (data capacity)");
+        // Should include Vec<FastqDecompressedChunk> overhead
+        let vec_overhead = 4 * std::mem::size_of::<FastqDecompressedChunk>();
+        assert!(
+            estimate >= 2048 + vec_overhead,
+            "estimate {estimate} should include chunk Vec overhead"
+        );
+    }
+
+    #[test]
+    fn test_fastq_boundary_batch_memory_estimate() {
+        let mut data = Vec::with_capacity(1024);
+        data.extend_from_slice(&[0u8; 100]);
+        let mut offsets = Vec::with_capacity(16);
+        offsets.push(0usize);
+
+        let mut streams = Vec::with_capacity(4);
+        streams.push(FastqStreamBoundaries { stream_idx: 0, data, offsets });
+
+        let batch = FastqBoundaryBatch { streams, serial: 0 };
+        let estimate = batch.estimate_heap_size();
+
+        let expected_min = 1024
+            + 16 * std::mem::size_of::<usize>()
+            + 4 * std::mem::size_of::<FastqStreamBoundaries>();
+        assert!(
+            estimate >= expected_min,
+            "estimate {estimate} should be >= {expected_min} (capacities + overhead)"
+        );
+    }
+
+    #[test]
+    fn test_fastq_parsed_batch_memory_estimate() {
+        use crate::grouper::FastqRecord;
+
+        let mut name = Vec::with_capacity(64);
+        name.extend_from_slice(b"read1");
+        let mut seq = Vec::with_capacity(256);
+        seq.extend_from_slice(b"ACGT");
+        let mut qual = Vec::with_capacity(256);
+        qual.extend_from_slice(b"IIII");
+
+        let record = FastqRecord { name, sequence: seq, quality: qual };
+        let mut records = Vec::with_capacity(8);
+        records.push(record);
+
+        let mut streams = Vec::with_capacity(4);
+        streams.push(FastqParsedStream { stream_idx: 0, records });
+
+        let batch = FastqParsedBatch { streams, serial: 0 };
+        let estimate = batch.estimate_heap_size();
+
+        let record_data = 64 + 256 + 256; // capacities of name, seq, qual
+        let records_overhead = 8 * std::mem::size_of::<FastqRecord>();
+        let streams_overhead = 4 * std::mem::size_of::<FastqParsedStream>();
+        let expected_min = record_data + records_overhead + streams_overhead;
+        assert!(
+            estimate >= expected_min,
+            "estimate {estimate} should be >= {expected_min} (capacities + overhead)"
+        );
+    }
 }