mux: StreamSelection primitive + apply sites for per-title stream selection

The demux pipeline is declaration-driven off DiscTitle.streams (build_demux_state,
DiscStream::new, and the MKV writer all key off that list), so 'which streams to
keep' is already a pipeline capability with no public knob. This adds the knob:

- mux/select.rs: StreamSelection { audio, subtitle: PidFilter::All | Only(Vec<u16>) }
  + apply(&mut DiscTitle): keep Video always, keep Audio/Subtitle whose PID the
  filter lists, prune the rest (and the parallel codec_privates in lockstep);
  error SelectionPidUnknown on a listed PID absent from the title (fail loud, not
  a silently-missing track). Pure; 6 unit tests. Re-exported at crate root.
- Error::SelectionPidUnknown (E6014).
- MuxOptions gains  (+ derives Default now) applied in mux_stream's
  Iso/Session arms before the highway/DiscStream builds demux state (and before
  probe_and_remap's DVD AC-3 PID rewrite). InputOptions gains  applied
  in input()'s iso arm right after the title-index bounds check.

PIDs not languages -- language->PID is caller/engine policy. Default All/All is a
no-op (apply gated on !is_all()), so the no-selection path is byte-identical:
nothing below the title-finalization line changes (ts/ps/demux_thread/
pipelined_stream/mkv/disc untouched). All 2488 lib tests pass on 1.86.

Design: freemkv-private/audit/engine-split/STREAM-SELECTION-DESIGN.md (Fable).

docs/rip-recovery.md

@@ -19,10 +19,11 @@ reach into the others.
 The library exposes flat verbs; the caller drives the multipass loop. Autorip
 runs `Disc::sweep` once, then loops `Disc::patch` until either the mapfile is
 clean or the configured retry budget is exhausted, then hands the ISO off to
-the mux pipeline. The `freemkv` CLI does the same shape with a
+the mux pipeline. The `freemkv` CLI does the same shape, but as of 1.6.0 the
-terminal-output progress sink. Layer 3 runs inside any consumer of
+loop itself (including the multi-title rip loop) lives one layer up, in the
-`DiscStream` (direct PES pipeline, ISO playback, etc.) without caller
+shared `freemkv-engine` crate, with a terminal-output progress sink plugged
-involvement.
+into it as the `Sink`. Layer 3 runs inside any consumer of `DiscStream`
+(direct PES pipeline, ISO playback, etc.) without caller involvement.
 
 Three primitives compose the disc-side flow:
 

src/error.rs

@@ -59,6 +59,7 @@ pub const E_MKV_INVALID: u16 = 6008;
 pub const E_NO_STREAMS: u16 = 6009;
 pub const E_HALTED: u16 = 6010;
 pub const E_MAPFILE_INVALID: u16 = 6011;
+pub const E_SELECTION_PID_UNKNOWN: u16 = 6014;
 pub const E_UDF_BUFFER_TOO_SMALL: u16 = 6012;
 pub const E_UDF_NOT_FILESYSTEM: u16 = 6013;
 
@@ -292,6 +293,12 @@ pub enum Error {
     IfoParse,
     MkvInvalid,
     NoStreams,
+    /// A [`crate::StreamSelection`] listed a PID that does not exist in the
+    /// title's declared streams — a caller bug (e.g. a stale scan), reported
+    /// loudly rather than silently producing an MKV missing a requested track.
+    SelectionPidUnknown {
+        pid: u16,
+    },
     /// ddrescue mapfile parse failed. `kind` is a stable, language-neutral
     /// identifier (e.g. `"status_char"`, `"hex"`); not a translatable
     /// English message.
@@ -579,6 +586,7 @@ impl Error {
             Error::IfoParse => E_IFO_PARSE,
             Error::MkvInvalid => E_MKV_INVALID,
             Error::NoStreams => E_NO_STREAMS,
+            Error::SelectionPidUnknown { .. } => E_SELECTION_PID_UNKNOWN,
             Error::MapfileInvalid { .. } => E_MAPFILE_INVALID,
             Error::AacsNoKeys => E_AACS_NO_KEYS,
             Error::AacsCertShort => E_AACS_CERT_SHORT,
@@ -786,6 +794,9 @@ impl std::fmt::Display for Error {
             Error::InvalidCdbLength { len, max } => {
                 write!(f, "E{}: {}/{}", self.code(), len, max)
             }
+            Error::SelectionPidUnknown { pid } => {
+                write!(f, "E{}: 0x{:04x}", self.code(), pid)
+            }
             _ => write!(f, "E{}", self.code()),
         }
     }

src/lib.rs

@@ -272,6 +272,7 @@ pub use mux::{Mp4FitReport, Mp4SkipReason, mp4_fit_report};
 // `SectorSource` to get plaintext sectors out.
 pub use mux::build_iso_pipeline;
 pub use mux::resolve_mux_key_map;
+pub use mux::select::{PidFilter, StreamSelection};
 pub use mux::{MuxEvents, MuxInput, MuxOptions, MuxOutcome, NoopEvents, mux_stream};
 pub use scsi::{DriveInfo, ScsiSense, ScsiTransport, SenseFamily, drive_has_disc, list_drives};
 pub use sector::{

src/mux/driver.rs

@@ -153,6 +153,11 @@ pub enum MuxInput<'a> {
 }
 
 /// Tuning / behaviour knobs for a mux run.
+///
+/// `Default` = keep-everything, no-skip, decrypt, no send deadline — the
+/// archival default. Added so callers set only the fields they care about
+/// (and so the additive `selection` field doesn't churn every constructor).
+#[derive(Default)]
 pub struct MuxOptions {
     /// Skip past read errors (zero-fill + continue) on the live-drive path
     /// instead of aborting. Wired onto `DiscStream::skip_errors`.
@@ -161,6 +166,11 @@ pub struct MuxOptions {
     pub batch_sectors: u16,
     /// Ciphertext passthrough — skip decryption / CSS self-crack.
     pub raw: bool,
+    /// Which audio/subtitle streams to keep in the muxed title. Default keeps
+    /// every stream (video is always kept). Applied to the title before the
+    /// demux pipeline is built, so track headers, `codec_privates`, and frame
+    /// routing all follow the pruned list. See [`crate::StreamSelection`].
+    pub selection: crate::StreamSelection,
     /// Per-frame write-pipeline send deadline.
     ///
     /// - `Some(d)` — a hard `d` timeout: a sink that back-pressures a single
@@ -278,6 +288,14 @@ pub fn mux_stream(
             keys,
             key_fetch,
         } => {
+            // Prune to the selected audio/subtitle streams BEFORE the highway
+            // builds its demux state from `title.streams` (and before
+            // `build_iso_pipeline`'s `probe_and_remap` may rewrite DVD AC-3
+            // PIDs). Video is always kept; a no-op for the default All/All.
+            let mut title = title;
+            opts.selection
+                .apply(&mut title)
+                .map_err(std::io::Error::from)?;
             let reader = FileSectorSource::open(path)?;
             let stream = build_iso_pipeline(
                 reader,
@@ -299,7 +317,7 @@ pub fn mux_stream(
             // Pull everything we need out of the disc as owned values so the
             // immutable disc borrow is released before the mutable
             // `take_reader` below.
-            let (title, format, mut keys, playlist) = {
+            let (mut title, format, mut keys, playlist) = {
                 let disc = session.disc().ok_or_else(|| Error::DeviceNotReady {
                     path: session.device_path().to_string(),
                 })?;
@@ -319,6 +337,14 @@ pub fn mux_stream(
                 // (see `session_mux_keys`), never the whole-disc `decrypt_keys()`.
                 (title, disc.content_format, session_mux_keys(disc), playlist)
             };
+            // Prune to the selected streams before `DiscStream::new` builds its
+            // demux tables from `title.streams` (and before its inline
+            // `probe_and_remap`). Only touches the stream list, so the
+            // ciphertext sampling in `resolve_inline_base_map` (keyed on extents)
+            // is unaffected. No-op for the default All/All.
+            opts.selection
+                .apply(&mut title)
+                .map_err(std::io::Error::from)?;
             // A missing staged reader ("already consumed" / never staged) is a
             // clean error, not a panic (contract Q2).
             let mut reader = session.take_reader().ok_or_else(|| Error::DeviceNotReady {
@@ -1308,6 +1334,7 @@ mod tests {
             batch_sectors: 8192,
             raw: false,
             send_deadline: Some(Duration::from_secs(60)),
+            selection: Default::default(),
         };
         let halt = Halt::new();
         let out = mux_stream(
@@ -1416,6 +1443,7 @@ mod tests {
             batch_sectors: us as u16,
             raw: false,
             send_deadline: Some(Duration::from_secs(60)),
+            selection: Default::default(),
         };
         let halt = Halt::new();
         // Drains to a NoStreams refusal (zeroed data resolves no headers); we
@@ -1534,6 +1562,7 @@ mod tests {
             batch_sectors: 3,   // one aligned unit per read
             raw: false,
             send_deadline: Some(Duration::from_secs(60)),
+            selection: Default::default(),
         };
         let halt = Halt::new();
         let out = mux_stream(
@@ -1635,6 +1664,7 @@ mod tests {
             batch_sectors: 3,   // one aligned unit per read
             raw: false,
             send_deadline: Some(Duration::from_secs(60)),
+            selection: Default::default(),
         };
         let halt = Halt::new();
         let out = mux_stream(
@@ -1685,6 +1715,7 @@ mod tests {
             batch_sectors: 3,
             raw: false,
             send_deadline: Some(Duration::from_secs(60)),
+            selection: Default::default(),
         };
         let halt = Halt::new();
         let err = mux_stream(
@@ -1769,6 +1800,7 @@ mod tests {
             batch_sectors: 3,
             raw: false,
             send_deadline: Some(Duration::from_secs(60)),
+            selection: Default::default(),
         };
         let halt = Halt::new();
         let err = mux_stream(
@@ -2060,6 +2092,7 @@ mod tests {
             batch_sectors: 0,
             raw: false,
             send_deadline: None,
+            selection: Default::default(),
         };
         assert_eq!(effective_send_deadline(cli.send_deadline), NO_SEND_DEADLINE);
         let autorip = MuxOptions {
@@ -2067,6 +2100,7 @@ mod tests {
             batch_sectors: 8192,
             raw: false,
             send_deadline: Some(Duration::from_secs(60)),
+            selection: Default::default(),
         };
         assert_eq!(
             effective_send_deadline(autorip.send_deadline),

src/mux/mod.rs

@@ -28,6 +28,7 @@ pub mod disc;
 pub mod driver;
 pub mod pipelined_stream;
 pub mod resolve;
+pub mod select;
 
 // Internal-only modules. Every reference is via `crate::mux::…` /
 // `super::…` from inside the crate; nothing in the downstream crates or

src/mux/resolve.rs

@@ -330,6 +330,12 @@ pub struct InputOptions {
     /// still-scrambled unit is re-tried via the application's key source.
     /// Application seam only; the library makes no network call.
     pub key_fetch: Option<crate::sector::KeyFetch>,
+    /// Which audio/subtitle streams to keep. `input()` scans the source and
+    /// picks the title internally, so the caller can't prune the `DiscTitle`
+    /// itself — it passes the selection here and `input()` applies it right
+    /// after the title-index bounds check. Default keeps every stream (video is
+    /// always kept). See [`crate::StreamSelection`].
+    pub selection: crate::StreamSelection,
 }
 
 // `KeyFetchFactory` holds a trait object that is not `Debug`; hand-roll the
@@ -342,6 +348,7 @@ impl std::fmt::Debug for InputOptions {
             .field("title_index", &self.title_index)
             .field("raw", &self.raw)
             .field("key_fetch", &self.key_fetch.is_some())
+            .field("selection", &self.selection)
             .finish()
     }
 }
@@ -403,6 +410,14 @@ pub fn input(url: &str, opts: &InputOptions) -> io::Result<Box<dyn crate::pes::S
                 }
                 .into());
             }
+            // Prune to the selected audio/subtitle streams now, on the scanned
+            // (pre-`probe_and_remap`) title, so everything downstream — the
+            // TrueHD channel-correction probe, the final title clone, and
+            // `build_iso_pipeline`'s demux/track construction — sees the pruned
+            // list. Video is always kept; a no-op for the default All/All.
+            opts.selection
+                .apply(&mut disc.titles[idx])
+                .map_err(|e| -> io::Error { e.into() })?;
             // Per-title key resolution. DVD CSS is resolved at exactly ONE site —
             // `build_iso_pipeline`'s per-title crack (below), which decrypts a
             // crackable title, passes a genuinely-clear one through, and

src/mux/select.rs

@@ -0,0 +1,299 @@
+//! Per-title audio/subtitle stream selection — the pure primitive.
+//!
+//! The demux pipeline is **declaration-driven**: every demux table
+//! (`build_demux_state` in `mux/resolve.rs`, `DiscStream::new` in `mux/disc.rs`)
+//! is built from the input [`DiscTitle`]'s `streams` list, and the MKV writer
+//! builds its track headers + `codec_privates` from that same list. A PID not
+//! declared there is never tracked, extracted, or written. So "which streams to
+//! keep" is already a capability of the pipeline — it just has no public knob.
+//!
+//! [`StreamSelection::apply`] is that knob: prune the `DiscTitle.streams` list
+//! (video always kept) BEFORE the mux path finalizes the title, and everything
+//! downstream — track headers, `codec_privates`, PID routing, frame emission —
+//! follows from the pruned list by construction, with zero scattered PID
+//! checks. This is language-agnostic: PIDs, not languages (the language→PID
+//! mapping is the caller's/engine's policy).
+
+use crate::disc::{DiscTitle, Stream};
+use crate::error::{Error, Result};
+
+/// Which PIDs to keep for one stream class (audio or subtitle). Video is always
+/// kept, so it has no filter.
+#[derive(Clone, Debug, Default, PartialEq, Eq)]
+pub enum PidFilter {
+    /// Keep every stream of this class. The default; [`StreamSelection::apply`]
+    /// is a no-op for an All/All selection, so the no-selection path is
+    /// byte-identical to no selection at all.
+    #[default]
+    All,
+    /// Keep only the streams whose PID is listed. `Only(vec![])` is legal and
+    /// means keep none (a video-only output when both classes are `Only([])`).
+    Only(Vec<u16>),
+}
+
+/// A per-title stream selection: which audio and which subtitle PIDs to keep.
+/// Video is always retained (it is implicit and never pruned).
+#[derive(Clone, Debug, Default, PartialEq, Eq)]
+pub struct StreamSelection {
+    pub audio: PidFilter,
+    pub subtitle: PidFilter,
+}
+
+impl StreamSelection {
+    /// True for the All/All default. Apply sites gate on `!is_all()` so the
+    /// no-selection path never even clones the title.
+    pub fn is_all(&self) -> bool {
+        matches!(self.audio, PidFilter::All) && matches!(self.subtitle, PidFilter::All)
+    }
+
+    /// Prune `title.streams` in place: keep every [`Stream::Video`]
+    /// unconditionally; keep an [`Stream::Audio`]/[`Stream::Subtitle`] iff its
+    /// PID passes the corresponding [`PidFilter`]; drop the rest. Declared order
+    /// is preserved. The parallel `codec_privates` vec is pruned in lockstep
+    /// when it is populated (it is empty on a freshly-scanned title, non-empty
+    /// only if a caller pre-filled it).
+    ///
+    /// Errors [`Error::SelectionPidUnknown`] if a filter lists a PID that does
+    /// not exist in `title.streams` — a caller bug (e.g. a stale scan). Fail
+    /// loud rather than silently emit an MKV missing a requested track. On
+    /// error the title is left unmodified.
+    pub fn apply(&self, title: &mut DiscTitle) -> Result<()> {
+        if self.is_all() {
+            return Ok(());
+        }
+
+        // Validate every listed PID exists in the title before mutating, so an
+        // unknown PID leaves the title untouched (no partial prune).
+        for pid in self.listed_pids() {
+            let present = title.streams.iter().any(|s| stream_pid(s) == Some(pid));
+            if !present {
+                return Err(Error::SelectionPidUnknown { pid });
+            }
+        }
+
+        let codec_privates_aligned = title.codec_privates.len() == title.streams.len();
+
+        // Retain by index so we can prune the parallel codec_privates in lockstep.
+        let keep: Vec<bool> = title
+            .streams
+            .iter()
+            .map(|s| self.keeps(s))
+            .collect::<Vec<_>>();
+
+        let mut i = 0;
+        title.streams.retain(|_| {
+            let k = keep[i];
+            i += 1;
+            k
+        });
+        if codec_privates_aligned {
+            let mut j = 0;
+            title.codec_privates.retain(|_| {
+                let k = keep[j];
+                j += 1;
+                k
+            });
+        }
+        Ok(())
+    }
+
+    /// Whether this selection keeps `stream`.
+    fn keeps(&self, stream: &Stream) -> bool {
+        match stream {
+            Stream::Video(_) => true,
+            Stream::Audio(a) => filter_keeps(&self.audio, a.pid),
+            Stream::Subtitle(s) => filter_keeps(&self.subtitle, s.pid),
+        }
+    }
+
+    /// Every PID explicitly listed across both filters (for existence checking).
+    fn listed_pids(&self) -> Vec<u16> {
+        let mut v = Vec::new();
+        if let PidFilter::Only(pids) = &self.audio {
+            v.extend_from_slice(pids);
+        }
+        if let PidFilter::Only(pids) = &self.subtitle {
+            v.extend_from_slice(pids);
+        }
+        v
+    }
+}
+
+fn filter_keeps(filter: &PidFilter, pid: u16) -> bool {
+    match filter {
+        PidFilter::All => true,
+        PidFilter::Only(pids) => pids.contains(&pid),
+    }
+}
+
+/// The PID of an audio/subtitle stream; `None` for video (which is never
+/// filtered, so its PID is irrelevant to selection).
+fn stream_pid(stream: &Stream) -> Option<u16> {
+    match stream {
+        Stream::Audio(a) => Some(a.pid),
+        Stream::Subtitle(s) => Some(s.pid),
+        Stream::Video(_) => None,
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::disc::{
+        AudioChannels, AudioStream, Codec, ColorSpace, FrameRate, HdrFormat, LabelPurpose,
+        LabelQualifier, Resolution, SampleRate, SubtitleStream, VideoStream,
+    };
+
+    fn video(pid: u16) -> Stream {
+        Stream::Video(VideoStream {
+            pid,
+            codec: Codec::Hevc,
+            resolution: Resolution::R2160p,
+            frame_rate: FrameRate::F23_976,
+            hdr: HdrFormat::Hdr10,
+            color_space: ColorSpace::Bt2020,
+            display_aspect: None,
+            secondary: false,
+            label: String::new(),
+            measured_cicp: None,
+        })
+    }
+    fn audio(pid: u16, lang: &str) -> Stream {
+        Stream::Audio(AudioStream {
+            pid,
+            codec: Codec::TrueHd,
+            channels: AudioChannels::Stereo,
+            language: lang.into(),
+            sample_rate: SampleRate::S48,
+            secondary: false,
+            purpose: LabelPurpose::Normal,
+            label: String::new(),
+        })
+    }
+    fn subtitle(pid: u16, lang: &str) -> Stream {
+        Stream::Subtitle(SubtitleStream {
+            pid,
+            codec: Codec::Pgs,
+            language: lang.into(),
+            forced: false,
+            qualifier: LabelQualifier::None,
+            codec_data: None,
+        })
+    }
+
+    // video + 3 audio (eng/spa/fra) + 2 subs (eng/spa).
+    fn title() -> DiscTitle {
+        let mut t = DiscTitle::empty();
+        t.streams = vec![
+            video(0x1011),
+            audio(0x1100, "eng"),
+            audio(0x1101, "spa"),
+            audio(0x1102, "fra"),
+            subtitle(0x1200, "eng"),
+            subtitle(0x1201, "spa"),
+        ];
+        t
+    }
+
+    fn pids(t: &DiscTitle) -> Vec<u16> {
+        t.streams
+            .iter()
+            .filter_map(|s| match s {
+                Stream::Video(v) => Some(v.pid),
+                Stream::Audio(a) => Some(a.pid),
+                Stream::Subtitle(s) => Some(s.pid),
+            })
+            .collect()
+    }
+
+    #[test]
+    fn apply_all_is_identity_and_untouched() {
+        let sel = StreamSelection::default();
+        assert!(sel.is_all());
+        let mut t = title();
+        let before = pids(&t);
+        sel.apply(&mut t).unwrap();
+        assert_eq!(pids(&t), before, "All/All must not change the stream list");
+    }
+
+    #[test]
+    fn apply_only_retains_listed_audio_pids_in_declared_order() {
+        // Keep eng+fra audio (skip spa); leave subtitles alone.
+        let sel = StreamSelection {
+            audio: PidFilter::Only(vec![0x1100, 0x1102]),
+            subtitle: PidFilter::All,
+        };
+        let mut t = title();
+        sel.apply(&mut t).unwrap();
+        assert_eq!(
+            pids(&t),
+            vec![0x1011, 0x1100, 0x1102, 0x1200, 0x1201],
+            "video + eng/fra audio (order preserved) + both subs"
+        );
+    }
+
+    #[test]
+    fn apply_only_empty_yields_video_only() {
+        let sel = StreamSelection {
+            audio: PidFilter::Only(vec![]),
+            subtitle: PidFilter::Only(vec![]),
+        };
+        let mut t = title();
+        sel.apply(&mut t).unwrap();
+        assert_eq!(pids(&t), vec![0x1011], "only the video stream survives");
+    }
+
+    #[test]
+    fn apply_subtitle_filter_does_not_touch_audio() {
+        let sel = StreamSelection {
+            audio: PidFilter::All,
+            subtitle: PidFilter::Only(vec![0x1200]),
+        };
+        let mut t = title();
+        sel.apply(&mut t).unwrap();
+        assert_eq!(
+            pids(&t),
+            vec![0x1011, 0x1100, 0x1101, 0x1102, 0x1200],
+            "all audio kept, only eng subtitle kept"
+        );
+    }
+
+    #[test]
+    fn apply_unknown_pid_errors_and_leaves_title_untouched() {
+        let sel = StreamSelection {
+            audio: PidFilter::Only(vec![0x9999]),
+            subtitle: PidFilter::All,
+        };
+        let mut t = title();
+        let before = pids(&t);
+        let err = sel.apply(&mut t).unwrap_err();
+        assert!(matches!(err, Error::SelectionPidUnknown { pid: 0x9999 }));
+        assert_eq!(pids(&t), before, "title unmodified on error");
+    }
+
+    #[test]
+    fn apply_prunes_codec_privates_in_lockstep_when_populated() {
+        // A caller that pre-filled codec_privates parallel to streams: pruning
+        // must keep the two vecs aligned.
+        let mut t = title();
+        t.codec_privates = vec![
+            Some(vec![0xAA]), // video 0x1011
+            Some(vec![0x11]), // audio 0x1100 eng
+            Some(vec![0x22]), // audio 0x1101 spa
+            Some(vec![0x33]), // audio 0x1102 fra
+            None,             // sub 0x1200
+            None,             // sub 0x1201
+        ];
+        let sel = StreamSelection {
+            audio: PidFilter::Only(vec![0x1100]),
+            subtitle: PidFilter::Only(vec![]),
+        };
+        sel.apply(&mut t).unwrap();
+        assert_eq!(pids(&t), vec![0x1011, 0x1100]);
+        assert_eq!(
+            t.codec_privates,
+            vec![Some(vec![0xAA]), Some(vec![0x11])],
+            "codec_privates pruned to match the retained streams, in order"
+        );
+    }
+}