mux/driver: split on_progress; stage live drive as session reader

Split the ambiguous MuxEvents::on_progress(bytes, total) into two
callbacks — on_read_progress(bytes_read, total) and
on_write_progress(bytes_written, total) — so a consumer no longer has
to guess which side of the pipeline a progress figure came from. The
CLI drives its bar from the write side; autorip from the read side.
The reader EventFn's BytesRead now maps to on_read_progress; the
per-frame emit in drive_mux to on_write_progress. Both keep empty
defaults; NoopEvents and the driver tests are updated to match.

Add DiscSession::stage_drive_as_reader so the live single-pass path can
run through MuxInput::Session: the owned Drive (itself a SectorSource)
moves into the reader slot for mux_stream to take. The drive is now
held as Option<Drive> with the device path cached up front, so the
mux driver can still name the device after the drive has been staged;
the driver's Session arm uses the new device_path() accessor.

src/mux/driver.rs

@@ -108,17 +108,27 @@ pub struct MuxOptions {
 /// [`Arc<dyn MuxEvents>`] and clones it into the constructors' `'static`
 /// [`EventFn`] (via [`reader_event_fn`]) so the reader-side events fire from the
 /// highway's producer thread / the live `DiscStream`'s read loop. The driver
-/// fires [`Self::on_output_opened`] and the write-side [`Self::on_progress`]
+/// fires [`Self::on_output_opened`] and the write-side [`Self::on_write_progress`]
 /// from the driving thread; the reader-side events
 /// ([`Self::on_sector_skipped`] / [`Self::on_batch_size_changed`] /
-/// [`Self::on_read_error`], plus a read-side [`Self::on_progress`]) are fired
+/// [`Self::on_read_error`], plus [`Self::on_read_progress`]) are fired
 /// by that cloned `EventFn`.
+///
+/// Progress is split into two callbacks because consumers drive their progress
+/// UI from different sides of the pipeline: the CLI renders from the WRITE side
+/// (bytes finalised to the sink), autorip from the READ side (bytes pulled off
+/// the disc). Keeping them distinct avoids the old ambiguous single
+/// `on_progress(bytes, total)` where the caller couldn't tell which number it
+/// was handed.
 pub trait MuxEvents: Send + Sync + 'static {
     /// Fired once, immediately after the output sink is created.
     fn on_output_opened(&self, _title: &DiscTitle) {}
+    /// Fired periodically from the reader side with the running read-byte count
+    /// and the source extents' total byte estimate.
+    fn on_read_progress(&self, _bytes_read: u64, _bytes_total: u64) {}
     /// Fired periodically during the frame pump with the running written-byte
     /// count and the title's total byte estimate.
-    fn on_progress(&self, _bytes_written: u64, _bytes_total: u64) {}
+    fn on_write_progress(&self, _bytes_written: u64, _bytes_total: u64) {}
     /// A bad sector was skipped (zero-filled) at `lba`.
     fn on_sector_skipped(&self, _lba: u32) {}
     /// The adaptive read batch size changed.
@@ -216,7 +226,7 @@ pub fn mux_stream(
             // `take_reader` below.
             let (title, format, keys, playlist) = {
                 let disc = session.disc().ok_or_else(|| Error::DeviceNotReady {
-                    path: session.drive().device_path().to_string(),
+                    path: session.device_path().to_string(),
                 })?;
                 let title = disc
                     .titles
@@ -235,7 +245,7 @@ pub fn mux_stream(
             // A missing staged reader ("already consumed" / never staged) is a
             // clean error, not a panic (contract Q2).
             let reader = session.take_reader().ok_or_else(|| Error::DeviceNotReady {
-                path: session.drive().device_path().to_string(),
+                path: session.device_path().to_string(),
             })?;
             let mut stream = crate::mux::DiscStream::new(
                 reader,
@@ -276,8 +286,8 @@ pub fn mux_stream(
 /// borrow cannot satisfy the `'static` bound.
 ///
 /// Mapping (real [`EventKind`] variants):
-/// - `BytesRead { bytes, total }`   → [`MuxEvents::on_progress`] (read-side; the
+/// - `BytesRead { bytes, total }`   → [`MuxEvents::on_read_progress`] (read-side;
-///   file highway's only reader event — `total` is the extents' byte total)
+///   the file highway's only reader event — `total` is the extents' byte total)
 /// - `SectorSkipped { sector }`     → [`MuxEvents::on_sector_skipped`] (live only)
 /// - `BatchSizeChanged { new_size, reason }` → [`MuxEvents::on_batch_size_changed`]
 ///   (live only)
@@ -287,7 +297,7 @@ pub fn mux_stream(
 /// (the disc's LBA space), so they are narrowed with `as u32`.
 fn reader_event_fn(events: Arc<dyn MuxEvents>) -> crate::sector::prefetched::EventFn {
     Box::new(move |e: Event| match e.kind {
-        EventKind::BytesRead { bytes, total } => events.on_progress(bytes, total),
+        EventKind::BytesRead { bytes, total } => events.on_read_progress(bytes, total),
         EventKind::SectorSkipped { sector } => events.on_sector_skipped(sector as u32),
         EventKind::BatchSizeChanged { new_size, reason } => {
             events.on_batch_size_changed(new_size, reason)
@@ -387,7 +397,7 @@ fn drive_mux(
 
     // The write consumer runs on its own thread so the latency-bound sink write
     // overlaps the next `stream.read()`. `bytes` mirrors the consumer's running
-    // written-byte count out to the driving thread for `on_progress`.
+    // written-byte count out to the driving thread for `on_write_progress`.
     let bytes = Arc::new(AtomicU64::new(0));
     let sink = WriteSink {
         output: output_stream,
@@ -421,7 +431,7 @@ fn drive_mux(
                         interrupted = true;
                         break;
                     }
-                    events.on_progress(bytes.load(Ordering::Relaxed), total_bytes);
+                    events.on_write_progress(bytes.load(Ordering::Relaxed), total_bytes);
                 }
                 Ok(None) => break,
                 Err(e) => {
@@ -717,10 +727,11 @@ mod tests {
     struct CountingEvents {
         opened: AtomicBool,
         progress_calls: AtomicU64,
-        /// Set once `on_progress` is called with a `total` equal to the ISO
+        /// Set once `on_read_progress` is called with a `total` equal to the ISO
         /// extents' byte total — the fingerprint of the *read-side* `BytesRead`
-        /// event (the write-side `on_progress` carries the title's `size_bytes`,
+        /// event (the write-side `on_write_progress` carries the title's
-        /// a different number), so it isolates the `EventFn` translation.
+        /// `size_bytes`, a different number), so it isolates the `EventFn`
+        /// translation.
         saw_read_total: AtomicBool,
         read_total: u64,
         skipped: AtomicU64,
@@ -744,7 +755,7 @@ mod tests {
         fn on_output_opened(&self, _title: &DiscTitle) {
             self.opened.store(true, Ordering::SeqCst);
         }
-        fn on_progress(&self, _bytes_written: u64, bytes_total: u64) {
+        fn on_read_progress(&self, _bytes_read: u64, bytes_total: u64) {
             self.progress_calls.fetch_add(1, Ordering::SeqCst);
             if bytes_total == self.read_total {
                 self.saw_read_total.store(true, Ordering::SeqCst);
@@ -793,7 +804,7 @@ mod tests {
         assert_eq!(
             events.progress_calls.load(Ordering::SeqCst),
             1,
-            "BytesRead → on_progress"
+            "BytesRead → on_read_progress"
         );
         assert!(
             events.saw_read_total.load(Ordering::SeqCst),
@@ -866,8 +877,8 @@ mod tests {
     ///
     /// Mutation: passing `None` (instead of `Some(reader_event_fn(...))`) to
     /// `build_iso_pipeline` in the ISO arm leaves `saw_read_total` false — the
-    /// write-side `on_progress` carries `size_bytes` (0 here), never the 6144
+    /// write-side `on_write_progress` carries `size_bytes` (0 here), never the
-    /// extents total — and this test fails.
+    /// 6144 extents total — and this test fails.
     #[test]
     fn mux_stream_iso_forwards_reader_progress_through_arc() {
         let es = [0xDE, 0xAD, 0xBE, 0xEF, 0x11, 0x22];
@@ -919,7 +930,7 @@ mod tests {
         );
         assert!(
             events.progress_calls.load(Ordering::SeqCst) > 0,
-            "at least one on_progress call observed"
+            "at least one on_read_progress call observed"
         );
     }
 }

src/session.rs

@@ -152,11 +152,19 @@ pub struct KeySpec {
 /// `DiscStream`) reach it via [`Self::drive_mut`] / [`Self::into_drive`]; the
 /// scanned [`Disc`] comes out via [`Self::disc`] / [`Self::take_disc`].
 pub struct DiscSession {
-    drive: Drive,
+    /// The opened drive. `Some` from [`Self::open`] until
+    /// [`Self::stage_drive_as_reader`] (live-drive mux) or [`Self::into_drive`]
+    /// moves it out. The cached [`Self::device_path`] survives that move so the
+    /// mux driver can still name the device in an error without the drive.
+    drive: Option<Drive>,
+    /// The drive's device path, cached at [`Self::open`] so it outlives a
+    /// [`Self::stage_drive_as_reader`] that moves the drive into `reader`.
+    device: String,
     spec: KeySpec,
     disc: Option<Disc>,
-    /// Sector source for a later file/live mux to `.take()` (steps 3–4).
+    /// Sector source for a later file/live mux to `.take()` (steps 3–4). The
-    /// Unpopulated in the current step; shapes the struct for the mux hoist.
+    /// file path stages a `FileSectorSource`; the live-drive path stages the
+    /// drive itself via [`Self::stage_drive_as_reader`].
     reader: Option<Box<dyn SectorSource>>,
     /// The read-time AACS fetch closure, built by [`Self::resolve_keys`] and
     /// retained so a later mux (step 4) can install it into the decrypt
@@ -214,8 +222,10 @@ impl DiscSession {
             tracing::warn!(target: "freemkv::session", error = %e, "probe_disc advisory failed (continuing)");
         }
 
+        let device = drive.device_path().to_string();
         Ok(DiscSession {
-            drive,
+            drive: Some(drive),
+            device,
             spec,
             disc: None,
             reader: None,
@@ -226,7 +236,7 @@ impl DiscSession {
     /// Fast disc identification — name/format only, no playlist parse. Wraps
     /// [`Disc::identify`].
     pub fn identify(&mut self) -> Result<DiscId> {
-        Disc::identify(&mut self.drive)
+        Disc::identify(self.drive_mut())
     }
 
     /// Full structure scan. Forwards the session's [`KeySpec`] credentials /
@@ -234,7 +244,7 @@ impl DiscSession {
     /// set), runs [`Disc::scan`], stores the result, and returns a borrow.
     pub fn scan(&mut self, opts: ScanOptions) -> Result<&Disc> {
         let opts = forward_key_material(&mut self.spec, opts);
-        let disc = Disc::scan(&mut self.drive, &opts)?;
+        let disc = Disc::scan(self.drive.as_mut().expect("drive present for scan"), &opts)?;
         self.disc = Some(disc);
         Ok(self.disc.as_ref().expect("disc just stored"))
     }
@@ -252,7 +262,7 @@ impl DiscSession {
         // The disc must have been scanned so its AACS inputs are captured.
         if self.disc.is_none() {
             return Err(Error::DeviceNotReady {
-                path: self.drive.device_path().to_string(),
+                path: self.device.clone(),
             });
         }
         // Sample through the staged reader when present (file-backed), else the
@@ -263,7 +273,11 @@ impl DiscSession {
             resolve_keys_for(reader.as_mut(), disc, sources)
         } else {
             let disc = self.disc.as_mut().expect("disc present (checked above)");
-            resolve_keys_for(&mut self.drive, disc, sources)
+            resolve_keys_for(
+                self.drive.as_mut().expect("drive present for key sampling"),
+                disc,
+                sources,
+            )
         };
         self.key_fetch = resolved.key_fetch;
         Ok(resolved.trace)
@@ -293,27 +307,51 @@ impl DiscSession {
         self.disc.take()
     }
 
-    /// Shared access to the opened drive (identity, profile, path).
+    /// Shared access to the opened drive (identity, profile, path). Panics if the
+    /// drive has already been staged into the reader slot
+    /// ([`Self::stage_drive_as_reader`]) or moved out via [`Self::into_drive`] —
+    /// use [`Self::device_path`] for a name that survives those moves.
     pub fn drive(&self) -> &Drive {
-        &self.drive
+        self.drive.as_ref().expect("drive present")
+    }
+
+    /// The opened drive's device path. Cached at [`Self::open`], so it remains
+    /// available after [`Self::stage_drive_as_reader`] moves the drive into the
+    /// reader slot (the mux driver names the device here without the drive).
+    pub fn device_path(&self) -> &str {
+        &self.device
     }
 
     /// Mutable access to the opened drive — for ciphertext sampling and other
     /// direct reads consumers still perform.
     pub fn drive_mut(&mut self) -> &mut Drive {
-        &mut self.drive
+        self.drive.as_mut().expect("drive present")
     }
 
     /// Lock the tray so the disc cannot eject mid-rip. Unlock is guaranteed by
-    /// `Drive::drop`.
+    /// `Drive::drop`. A no-op if the drive is no longer held by the session.
     pub fn lock_tray(&mut self) {
-        self.drive.lock_tray();
+        if let Some(drive) = self.drive.as_mut() {
+            drive.lock_tray();
+        }
     }
 
     /// Consume the session, returning the owned drive (e.g. to move into a
     /// `DiscStream` for a live-drive mux).
     pub fn into_drive(self) -> Drive {
-        self.drive
+        self.drive.expect("drive present")
+    }
+
+    /// Stage the owned drive as the session's boxed sector source so a live
+    /// single-pass mux can drive it through
+    /// [`MuxInput::Session`](crate::mux::MuxInput::Session). Moves the `Drive`
+    /// (itself a [`SectorSource`]) into the `reader` slot; the cached
+    /// [`Self::device_path`] keeps the device name available afterward. A no-op
+    /// if the drive was already staged or moved out.
+    pub fn stage_drive_as_reader(&mut self) {
+        if let Some(drive) = self.drive.take() {
+            self.reader = Some(Box::new(drive));
+        }
     }
 
     /// Consume the session, returning the sector source staged for a later mux