refactor(pipeline): relocate reusable types out of unified_pipeline (R6a)

src/lib/batch_weight.rs

@@ -0,0 +1,26 @@
+//! Batch weight trait for template-based batching.
+//!
+//! Moved out of `unified_pipeline::base` (C5/R6a) so `grouper`/`mi_group` can
+//! depend on it without going through the (now-legacy) `unified_pipeline`
+//! module. Stays in the root crate rather than `fgumi-bam-io`: its surviving
+//! implementations target foreign types (`noodles::sam::alignment::RecordBuf`,
+//! `fgumi_raw_bam::RawRecord`, `Vec<u8>`), which are only legal `impl`s while
+//! the trait itself is defined in this crate (orphan rule).
+
+/// Trait for groups that can report their "weight" for batching purposes.
+///
+/// The weight is typically the number of templates in the group, allowing
+/// the pipeline to batch groups based on total templates rather than group count.
+/// This provides more consistent batch sizes across datasets with varying
+/// templates-per-group ratios.
+///
+/// # Example
+///
+/// For a position group with 10 templates, `batch_weight()` returns 10.
+/// The pipeline accumulates groups until the total weight reaches a threshold
+/// (e.g., 500 templates), then flushes the batch.
+pub trait BatchWeight {
+    /// Returns the weight of this group for batching purposes.
+    /// For position groups, this is typically the number of templates.
+    fn batch_weight(&self) -> usize;
+}

src/lib/commands/common.rs

@@ -12,10 +12,11 @@ use std::sync::Arc;
 use crate::assigner::Strategy;
 #[cfg(feature = "consensus")]
 use crate::logging::OperationTimer;
-use crate::unified_pipeline::{
-    BACKPRESSURE_THRESHOLD_BYTES, BamPipelineConfig, Q5_BACKPRESSURE_THRESHOLD_BYTES,
-    SchedulerStrategy, stage_high_water_mark,
+use crate::pipeline::backpressure::{
+    BACKPRESSURE_THRESHOLD_BYTES, Q5_BACKPRESSURE_THRESHOLD_BYTES, stage_high_water_mark,
 };
+use crate::scheduler_strategy::SchedulerStrategy;
+use crate::unified_pipeline::BamPipelineConfig;
 use crate::validation::validate_input_exists;
 use bytesize::ByteSize;
 use clap::Args;
@@ -1777,7 +1778,7 @@ pub(crate) fn warn_unwired_pipeline_flags(scheduler_opts: &SchedulerOptions) {
     // `info!`) to match the sibling "flag has no effect on the chain path"
     // diagnostics in `group::execute_chain` (--debug-memory, FGUMI_SHORT_CIRCUIT).
     let requested_scheduler = scheduler_opts.strategy();
-    if requested_scheduler != crate::unified_pipeline::scheduler::SchedulerStrategy::default() {
+    if requested_scheduler != SchedulerStrategy::default() {
         log::warn!(
             "--scheduler has no effect in the typed-step pipeline: the chain engine \
              does not use a pluggable scheduler strategy, so the requested strategy \

src/lib/fastq_parse.rs

@@ -6,7 +6,7 @@
 
 use std::io;
 
-use crate::unified_pipeline::MemoryEstimate;
+use fgumi_bam_io::MemoryEstimate;
 
 /// A parsed FASTQ record stored as a single heap allocation.
 ///

src/lib/grouper.rs

@@ -11,8 +11,9 @@ use std::io;
 
 use noodles::sam::alignment::RecordBuf;
 
+use crate::batch_weight::BatchWeight;
 use crate::template::{Template, TemplateBatch};
-use crate::unified_pipeline::{BatchWeight, DecodedRecord, Grouper, MemoryEstimate};
+use fgumi_bam_io::{DecodedRecord, Grouper, MemoryEstimate};
 use fgumi_metrics::TemplateFilterCounts;
 use fgumi_raw_bam;
 use fgumi_raw_bam::{RawRecord, raw_record_to_record_buf};
@@ -268,7 +269,7 @@ impl Grouper for TemplateGrouper {
     }
 }
 
-use crate::unified_pipeline::GroupKey;
+use fgumi_bam_io::GroupKey;
 
 impl MemoryEstimate for ProcessedPositionGroup {
     fn estimate_heap_size(&self) -> usize {
@@ -1089,7 +1090,7 @@ mod tests {
 
     #[test]
     fn test_single_raw_record_grouper_emits_each_record() {
-        use crate::unified_pipeline::{DecodedRecord, GroupKey};
+        use fgumi_bam_io::{DecodedRecord, GroupKey};
 
         let mut grouper = SingleRawRecordGrouper::new();
         let raw1 = RawRecord::from(vec![1u8; 36]);
@@ -1901,10 +1902,10 @@ mod tests {
     /// group contract. Every MC-validation test below builds its record this way.
     fn decoded_via_decode(flags: u16, pos: i32, mc: Option<&str>, name: &[u8]) -> DecodedRecord {
         use crate::read_info::LibraryIndex;
-        use crate::unified_pipeline::compute_group_key_from_raw;
+        use fgumi_bam_io::compute_group_key_from_raw;
 
         let raw = build_raw(flags, pos, mc, name, false);
-        let (key, _) = compute_group_key_from_raw(&raw, &LibraryIndex::default(), None, None);
+        let key = compute_group_key_from_raw(&raw, &LibraryIndex::default(), None);
         DecodedRecord::from_raw_bytes(raw, key)
     }
 
@@ -1937,10 +1938,10 @@ mod tests {
         #[case] byte_scan_finds_mc: bool,
     ) {
         use crate::read_info::LibraryIndex;
-        use crate::unified_pipeline::compute_group_key_from_raw;
+        use fgumi_bam_io::compute_group_key_from_raw;
 
         let raw = build_raw(flags, 99, mc, b"read1", prepend_non_z_mc);
-        let (key, _) = compute_group_key_from_raw(&raw, &LibraryIndex::default(), None, None);
+        let key = compute_group_key_from_raw(&raw, &LibraryIndex::default(), None);
 
         assert_eq!(
             fgumi_raw_bam::find_mc_tag_in_record(&raw).is_some(),
@@ -2186,7 +2187,7 @@ mod tests {
 
     #[test]
     fn test_build_templates_from_raw_bytes() {
-        use crate::unified_pipeline::{DecodedRecord, GroupKey};
+        use fgumi_bam_io::{DecodedRecord, GroupKey};
         use fgumi_raw_bam;
 
         let key = GroupKey::single(0, 100, 0, 0, 0, 12345);
@@ -2210,7 +2211,7 @@ mod tests {
 
     #[test]
     fn test_build_templates_from_raw_bytes_paired() {
-        use crate::unified_pipeline::{DecodedRecord, GroupKey};
+        use fgumi_bam_io::{DecodedRecord, GroupKey};
         use fgumi_raw_bam;
 
         let key1 = GroupKey::paired(0, 100, 0, 0, 200, 1, 0, 0, 12345);

src/lib/mi_group.rs

@@ -8,9 +8,10 @@ use anyhow::Result;
 use fgumi_raw_bam::RawRecord;
 use std::io;
 
-use crate::unified_pipeline::{BatchWeight, DecodedRecord, MemoryEstimate};
+use crate::batch_weight::BatchWeight;
+use fgumi_bam_io::{DecodedRecord, MemoryEstimate};
 
-use crate::unified_pipeline::Grouper;
+use fgumi_bam_io::Grouper;
 use std::collections::VecDeque;
 
 // ============================================================================
@@ -1049,14 +1050,14 @@ mod tests {
     /// Helper: create a `DecodedRecord` from raw bytes with a dummy group key.
     fn make_raw_decoded_record(tag_name: &str, tag_value: &str) -> DecodedRecord {
         let raw = make_raw_bam_with_tag(tag_name, tag_value);
-        let key = crate::unified_pipeline::GroupKey::single(0, 0, 0, 0, 0, 0);
+        let key = fgumi_bam_io::GroupKey::single(0, 0, 0, 0, 0, 0);
         DecodedRecord::from_raw_bytes(raw, key)
     }
 
     /// Helper: create a `DecodedRecord` from raw bytes with no tags.
     fn make_raw_decoded_record_no_tag() -> DecodedRecord {
         let raw = make_raw_bam_without_tag();
-        let key = crate::unified_pipeline::GroupKey::single(0, 0, 0, 0, 0, 0);
+        let key = fgumi_bam_io::GroupKey::single(0, 0, 0, 0, 0, 0);
         DecodedRecord::from_raw_bytes(raw, key)
     }
 
@@ -1233,7 +1234,7 @@ mod tests {
 
         let rec_b = make_raw_bam_with_tag("MI", "1/B");
 
-        let key = crate::unified_pipeline::GroupKey::single(0, 0, 0, 0, 0, 0);
+        let key = fgumi_bam_io::GroupKey::single(0, 0, 0, 0, 0, 0);
         let records = vec![
             DecodedRecord::from_raw_bytes(rec_primary, key),
             DecodedRecord::from_raw_bytes(rec_secondary, key),
@@ -1258,7 +1259,7 @@ mod tests {
         // `MiGroup::mi` for logging but never written back to the output records.
         fn make_two_tag_decoded(mi: &str, cb: &str) -> DecodedRecord {
             let raw = make_raw_bam_with_two_tags("MI", mi, "CB", cb);
-            let key = crate::unified_pipeline::GroupKey::single(0, 0, 0, 0, 0, 0);
+            let key = fgumi_bam_io::GroupKey::single(0, 0, 0, 0, 0, 0);
             DecodedRecord::from_raw_bytes(raw, key)
         }
 

src/lib/mod.rs

@@ -127,6 +127,7 @@
 //! - [noodles](https://github.com/zaeleus/noodles) - Rust bioinformatics I/O
 
 pub mod aligner;
+pub mod batch_weight;
 pub mod commands;
 pub mod version;
 pub use fgumi_bgzf::reader as bgzf_reader;
@@ -152,6 +153,7 @@ pub mod read_structure;
 pub mod reference;
 pub use fgumi_metrics::rejection;
 pub mod sam;
+pub mod scheduler_strategy;
 pub mod system;
 pub mod tag_reversal;
 pub mod template;

src/lib/pipeline/backpressure.rs

@@ -0,0 +1,156 @@
+//! Per-stage backpressure thresholds and the queue-memory-budget/high-water
+//! mark relationship.
+//!
+//! Moved out of `unified_pipeline::base` (C5/R6a) — the definitions are pure
+//! (a `u64` constant and a `fn(u64, u64) -> u64`) with no dependency on the
+//! `unified_pipeline` scheduler/queue machinery, so they relocate cleanly to
+//! the root crate's `pipeline` module. `unified_pipeline::base` keeps a
+//! `pub use` shim so its own internal call sites and the still-present legacy
+//! command paths keep resolving these names unchanged.
+
+/// Default high-water mark for a reorder-buffered stage (512 MiB).
+///
+/// This is the point at which one stage stops taking on new work and the
+/// scheduler starts prioritizing downstream steps (Process, Serialize,
+/// Compress, Write) so the stage can drain.
+///
+/// It is reached only through `ReorderBufferState::effective_limit`, so it
+/// applies to exactly three stages: Q2 and Q3 (whose counters span the queue
+/// and its reorder buffer both) and the post-Group write reorder buffer. The
+/// remaining queues — Q1, Q2b, Q4, Q6, Q7 — carry byte counters but no byte
+/// mark of their own; they are bounded by their slot count and by the
+/// pipeline-wide total gated at the Read step.
+///
+/// # Architecture — where the scheduler reads its drain signal
+///
+/// Only one of the three feeds the scheduler's `memory_high` / `memory_drained`
+/// pair, and the two pipelines pick different ones:
+/// - **BAM pipeline**: `q3_reorder_state`, the reorder buffer after Decode and
+///   before Group (`BamPipelineState::is_memory_high`)
+/// - **FASTQ pipeline**: `output.write_reorder_state`, the post-Compress write
+///   reorder buffer (`FastqStepContext::get_backpressure`)
+///
+/// For the BAM pipeline that placement is deliberate: memory is tracked before
+/// the exclusive Group step rather than after it, because tracking after Group
+/// releases the pre-Group buffer's bytes before knowing whether the post-Group
+/// queue can accept them, leaving data in an untracked intermediate buffer.
+///
+/// # Threshold behaviour
+///
+/// - When tracked memory >= the mark, `is_memory_high()` returns true: the
+///   producing step declines new work and the scheduler enters "drain mode"
+/// - When tracked memory < half the mark, `is_memory_drained()` returns true
+///   and the scheduler exits drain mode (hysteresis prevents thrashing)
+///
+/// # Relationship to `--max-memory`
+///
+/// This is a *per-stage trigger*, not the user's capacity budget; see
+/// [`stage_high_water_mark`] for why the two are separate and why a larger
+/// budget does not raise this. The budget's capacity role is enforced on the
+/// pipeline's total in-flight bytes at the Read step
+/// (`BamPipelineState::read_admission_allowed`).
+pub const BACKPRESSURE_THRESHOLD_BYTES: u64 = 512 * 1024 * 1024; // 512 MiB
+
+/// Default high-water mark for Q5, the processed queue (256 MiB).
+///
+/// This is set lower than the Q3 mark (256 MiB vs 512 MiB) because items in Q5
+/// are typically larger (e.g., `SimplexProcessedBatch` with `RecordBuf` vectors).
+/// When Q5 memory reaches this mark, the Process step pauses to let
+/// downstream steps (Serialize, Compress, Write) catch up.
+///
+/// Like [`BACKPRESSURE_THRESHOLD_BYTES`] it is applied through
+/// [`stage_high_water_mark`], so a `--max-memory` below it tightens it and one
+/// above it leaves it alone.
+pub const Q5_BACKPRESSURE_THRESHOLD_BYTES: u64 = 256 * 1024 * 1024; // 256 MiB
+
+/// The high-water mark one pipeline stage backs off at, given the declared
+/// queue memory budget and that stage's default mark.
+///
+/// Returns `default_mark` when `queue_memory_limit` is 0 (meaning "unset"),
+/// and `min(queue_memory_limit, default_mark)` otherwise.
+///
+/// # Why the budget cannot raise the mark (issue #765)
+///
+/// `--max-memory` is a *capacity* budget: how many bytes the pipeline may hold
+/// in total. A stage's high-water mark is a *trigger*: the depth at which that
+/// one stage stops pulling in work so the rest of the pipeline can catch up.
+/// Historically the mark was the only place the budget reached, which fused the
+/// two ideas into one number and made `--max-memory` above 512 MiB inert. The
+/// capacity half now lives where it belongs — on the pipeline's total in-flight
+/// bytes, gated at the Read step — and what is left here is only the trigger.
+///
+/// A trigger should not scale with the total budget, for three reasons:
+///
+/// - **It would buy no throughput.** The scheduler's response to `memory_high`
+///   is a priority reordering, not a stop. Raising the mark only postpones the
+///   moment output steps are favoured, so the stage parks more bytes and the
+///   pipeline runs no faster.
+/// - **Reorder capacity is bounded by serial skew, not by bytes.** The
+///   pre-Group reorder buffers exist to absorb out-of-order completion between
+///   workers. That skew is bounded by the number of batches in flight, which is
+///   bounded by the queue slot count, so bytes past the skew window are dead
+///   weight.
+/// - **It would let one stage monopolize the budget.** These marks are the only
+///   byte bound on Q2, Q3 and Q5 individually. Uncapped, a single stage could
+///   park the entire budget and the total gate would fire before any other
+///   stage buffered anything — converting a pipeline into a queue.
+///
+/// The downward direction is the useful one and is preserved: a budget smaller
+/// than the default mark pulls the mark down with it, so no single stage can
+/// hold more than the whole declared budget.
+#[inline]
+#[must_use]
+pub fn stage_high_water_mark(queue_memory_limit: u64, default_mark: u64) -> u64 {
+    if queue_memory_limit == 0 { default_mark } else { queue_memory_limit.min(default_mark) }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    // ========================================================================
+    // Stage high-water mark tests (issue #765)
+    // ========================================================================
+
+    #[test]
+    fn test_stage_high_water_mark_treats_zero_as_unset() {
+        assert_eq!(
+            stage_high_water_mark(0, BACKPRESSURE_THRESHOLD_BYTES),
+            BACKPRESSURE_THRESHOLD_BYTES
+        );
+        assert_eq!(
+            stage_high_water_mark(0, Q5_BACKPRESSURE_THRESHOLD_BYTES),
+            Q5_BACKPRESSURE_THRESHOLD_BYTES
+        );
+    }
+
+    #[test]
+    fn test_stage_high_water_mark_tightens_under_a_small_budget() {
+        // A budget below the default mark pulls the mark down with it, so a
+        // single stage can never park more than the whole declared budget.
+        assert_eq!(
+            stage_high_water_mark(64 * 1024 * 1024, BACKPRESSURE_THRESHOLD_BYTES),
+            64 * 1024 * 1024
+        );
+        assert_eq!(
+            stage_high_water_mark(64 * 1024 * 1024, Q5_BACKPRESSURE_THRESHOLD_BYTES),
+            64 * 1024 * 1024
+        );
+    }
+
+    #[test]
+    fn test_stage_high_water_mark_does_not_scale_above_the_default() {
+        // Deliberate: the mark is a per-stage trigger, not the user's capacity
+        // budget. See the `stage_high_water_mark` docs for why raising it buys
+        // nothing. The budget's capacity role lives at the Read admission gate.
+        let huge = 32 * 1024 * 1024 * 1024;
+        assert_eq!(
+            stage_high_water_mark(huge, BACKPRESSURE_THRESHOLD_BYTES),
+            BACKPRESSURE_THRESHOLD_BYTES
+        );
+        assert_eq!(
+            stage_high_water_mark(huge, Q5_BACKPRESSURE_THRESHOLD_BYTES),
+            Q5_BACKPRESSURE_THRESHOLD_BYTES
+        );
+    }
+}

src/lib/pipeline/mod.rs

@@ -24,5 +24,6 @@
 /// `crate::pipeline::core::…` path resolves — which is what lets the ported
 /// step sources compile unmodified.
 pub use fgumi_pipeline_core as core;
+pub mod backpressure;
 pub mod chains;
 pub mod steps;

src/lib/pipeline/steps/tuning.rs

@@ -34,9 +34,9 @@ pub struct BamPipelineTuning {
 
 impl BamPipelineTuning {
     /// Auto-tuned defaults for `threads` worker threads, mirroring the legacy
-    /// `PipelineConfig::auto_tuned` (`unified_pipeline::base`). Deliberately not
-    /// an intra-doc link: that module is deleted once the commands migrate off
-    /// it, and a link would break again then.
+    /// `PipelineConfig::auto_tuned` from the now-deprecated unified pipeline.
+    /// Deliberately not an intra-doc link: that code is deleted once the
+    /// commands migrate off it, and a link would break again then.
     #[must_use]
     pub fn auto_tuned(threads: usize) -> Self {
         let threads = threads.max(1);