fix(session): delete two dead accessors, make into_drive fallible

drive() and drive_mut() had ZERO callers — not in libfreemkv, freemkv,
autorip, bdemu, keysources or kdb. Deleted rather than converted: dead
public API that panics is not an API worth preserving the shape of.

into_drive() had two callers and now returns Result. The empty-slot
state is reachable through ordinary public use — stage_drive_as_reader
moves the drive into the reader slot, and calling into_drive twice moves
it out — so the panic was not guarding a caller error. identify() was
converted for exactly this reason in this same release; the fix went to
one of four public sinks and the other three were left.

I deferred this on the assumption the blast radius was large. It was
three call sites. Checking beats assuming.

Also fixes a REAL FLAKE in the gate, which is worth more than the above.
resolve_vid_only_bus_key_gate_reports_true_has_volume_id... failed about
one full-suite run in ten while passing every time in isolation. It
installed a capturing tracing subscriber to read back the has_volume_id
field of a warn.

That cannot be made reliable: dispatcher::set_default is THREAD-LOCAL
while tracing's callsite-interest cache is GLOBAL. The original author
knew, and called rebuild_interest_cache() — necessary but not
sufficient. I first serialised every capture in the crate behind one
lock (harness::with_captured_tracing, which also removed the same
hand-rolled dance from three other sites). Still 1-in-10, because the
cache can be re-evaluated against the process-default dispatch rather
than the thread-local one.

So the predicate is now a named function, handshake_has_volume_id, and
the test asserts the VALUE. A boolean does not need a subscriber to
check. The gate's hard-error behaviour keeps its own test.

Measured: 14 consecutive full-suite runs, 2994 passed, 0 failed.

A flaky gate is worse than a missing one — every green after it means
less, and this one had been eroding trust in the whole suite.

src/disc/encrypt.rs

@@ -181,6 +181,24 @@ fn cert_unlock_outcome(e: &CertUnlockFailure) -> crate::aacs::trace::UnlockOutco
     }
 }
 
+/// Did the cert handshake actually carry a Volume ID?
+///
+/// Extracted so it can be tested as a VALUE. It only ever reaches an operator
+/// as the `has_volume_id` field of the `bus_key_unavailable` warn, and
+/// asserting on a `tracing` field means installing a capturing subscriber —
+/// which is thread-local, while `tracing`'s callsite-interest cache is global.
+/// Those two facts race: the test failed roughly one run in ten under the full
+/// parallel suite while passing every time in isolation, and serialising the
+/// captures was not enough because the cache can be re-evaluated against the
+/// process default rather than the thread-local dispatch.
+///
+/// A predicate this small does not need a subscriber to verify. The polarity is
+/// the whole point: an `==` here would tell an operator a VID was absent on
+/// exactly the discs where one was present.
+fn handshake_has_volume_id(h: &HandshakeResult) -> bool {
+    h.volume_id != [0u8; 16]
+}
+
 impl Disc {
     /// SCSI handshake — drives the VID-acquisition flow and returns
     /// a structured `HandshakeResult` for downstream key resolution.
@@ -369,7 +387,7 @@ impl Disc {
         // file/ISO, drive unlock, cert bus key). The gate enumerates nothing.
         if !bus_encryption_removed(bus_encryption, handshake) {
             let (rdk_err, has_vid) = handshake
-                .map(|h| (h.read_data_key_err, h.volume_id != [0u8; 16]))
+                .map(|h| (h.read_data_key_err, handshake_has_volume_id(h)))
                 .unwrap_or((None, false));
             tracing::warn!(
                 target: "freemkv::disc",
@@ -1027,44 +1045,72 @@ mod tests {
         fn exit(&self, _span: &tracing::span::Id) {}
     }
 
-    /// The `bus_key_unavailable` warn's `has_volume_id` field must report
+    /// `has_volume_id` must report the ACTUAL presence of a non-zero Volume ID
-    /// the ACTUAL presence of a non-zero Volume ID on the handshake (encrypt.rs
+    /// on the handshake, not its negation.
-    /// `h.volume_id != [0u8; 16]`), not its negation. This is diagnostic-only
+    ///
-    /// (it does not affect the returned `Err(AacsBusKeyUnavailable)` itself,
+    /// Diagnostic-only — it does not change the returned
-    /// which is why a plain `Result` assertion can't distinguish `!=` from
+    /// `Err(AacsBusKeyUnavailable)`, which is why asserting on the `Result`
-    /// `==` here) but it is the ONLY signal an operator has, from this log
+    /// cannot distinguish `!=` from `==`. But it is the ONLY signal an operator
-    /// line, for whether the handshake actually carried a VID when bus
+    /// gets, from that one log line, for whether the handshake carried a VID
-    /// encryption could not be removed — a `==` flip would silently invert it.
+    /// when bus encryption could not be removed. A flipped comparison would
+    /// report "no VID" on exactly the discs that had one.
+    ///
+    /// This asserts the PREDICATE rather than the emitted `tracing` field. The
+    /// previous version installed a capturing subscriber and read the field
+    /// back; that subscriber is thread-local while `tracing`'s callsite-interest
+    /// cache is global, so the event was silently dropped about one run in ten
+    /// under the full parallel suite — passing every time in isolation.
+    /// Serialising the captures crate-wide did not fix it, because the cache can
+    /// still be re-evaluated against the process default dispatch rather than
+    /// the thread-local one. A boolean does not need a subscriber to check.
     #[test]
-    fn resolve_vid_only_bus_key_gate_reports_true_has_volume_id_when_vid_nonzero() {
+    fn handshake_has_volume_id_reports_presence_not_absence() {
-        let capture = std::sync::Arc::new(HasVidCapture(std::sync::Mutex::new(None)));
+        let with_vid = HandshakeResult {
-        let dispatch = tracing::Dispatch::new(capture.clone());
+            volume_id: [0x11u8; 16],
+            read_data_key: None,
+            read_data_key_err: None,
+            drive_unlocked: false,
+        };
+        assert!(
+            super::handshake_has_volume_id(&with_vid),
+            "a non-zero Volume ID must report as PRESENT"
+        );
+
+        let without = HandshakeResult {
+            volume_id: [0u8; 16],
+            ..with_vid
+        };
+        assert!(
+            !super::handshake_has_volume_id(&without),
+            "an all-zero Volume ID is the absent case"
+        );
+
+        // One bit of difference is still a VID: the check is != all-zero, not a
+        // heuristic about how much of it looks populated.
+        let mut barely = [0u8; 16];
+        barely[15] = 1;
+        assert!(
+            super::handshake_has_volume_id(&HandshakeResult {
+                volume_id: barely,
+                ..with_vid
+            }),
+            "any non-zero byte makes a Volume ID present"
+        );
+    }
+
+    /// The gate itself still hard-errors — the property the log line annotates.
+    #[test]
+    fn resolve_vid_only_bus_key_gate_hard_errors_without_a_read_data_key() {
         let (mut disc, udf) = disc_with_cert(0x01, true);
         let hs = HandshakeResult {
-            volume_id: [0x11u8; 16], // non-zero: a VID WAS present
+            volume_id: [0x11u8; 16],
             read_data_key: None,
             read_data_key_err: None,
             drive_unlocked: false,
         };
-        let guard = tracing::dispatcher::set_default(&dispatch);
-        // Tracing caches per-callsite "any subscriber interested?" the FIRST
-        // time a callsite fires; another test in this suite may already have
-        // hit the exact same `warn!` call site under the process default
-        // (no-op) dispatch, permanently caching "not interested" for it. Force
-        // recomputation now that our capturing dispatch is installed, or the
-        // event is silently dropped before it reaches our `Visit` — flaky only
-        // under full-suite (parallel, ordering-dependent) runs, not in isolation.
-        tracing::callsite::rebuild_interest_cache();
         let err = Disc::resolve_vid_only(&udf, &mut disc, Some(&hs))
             .expect_err("bus-encrypted, no read_data_key must still hard-error");
-        drop(guard);
-        tracing::callsite::rebuild_interest_cache();
         assert!(matches!(err, Error::AacsBusKeyUnavailable));
-        assert_eq!(
-            *capture.0.lock().unwrap(),
-            Some(true),
-            "has_volume_id must be true: the handshake's volume_id was non-zero"
-        );
     }
 
     // ---------------------------------------------------------------

src/disc/pgs_forced_probe.rs

@@ -1496,7 +1496,10 @@ mod tests {
 
         // Conclusive: one exactly-sized read, extent read to its end, no stall.
         let conclusive_count = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
-        tracing::subscriber::with_default(ScanDebugCounter(conclusive_count.clone()), || {
+        // Serialised crate-wide — see `harness::with_captured_tracing`. These
+        // race the capture in disc/encrypt.rs otherwise: the dispatch is
+        // thread-local but the callsite-interest cache is global.
+        crate::harness::with_captured_tracing(ScanDebugCounter(conclusive_count.clone()), || {
             let mut reader = TsReader {
                 data: ts_stream(pid, &pcs_display(true)),
                 pos: 0,
@@ -1516,7 +1519,7 @@ mod tests {
 
         // Inconclusive: dies mid-title with a read error → ReadFailed.
         let truncated_count = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
-        tracing::subscriber::with_default(ScanDebugCounter(truncated_count.clone()), || {
+        crate::harness::with_captured_tracing(ScanDebugCounter(truncated_count.clone()), || {
             let mut reader =
                 PartialTsReader::new(ts_stream(pid, &pcs_display(true)), ThenWhat::Error);
             let mut title = multi_read_pgs_title(pid, false);

src/harness.rs

@@ -265,3 +265,56 @@ fn the_generators_actually_reach_the_parser_bodies() {
     );
     println!("mpls reach: {ok}/{total} cases parsed to completion");
 }
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Capturing-subscriber serialisation.
+//
+// Several tests install a capturing `tracing` subscriber to assert on a log
+// line's FIELDS — the only observable for a diagnostic that does not change a
+// return value. Doing that safely needs two things that pull in opposite
+// directions:
+//
+// * `dispatcher::set_default` / `subscriber::with_default` are THREAD-LOCAL.
+// * `tracing` caches per-callsite "is any subscriber interested?" GLOBALLY, the
+//   first time each callsite fires. A callsite that already fired under the
+//   process default (a no-op) is cached as "not interested" forever, so the
+//   event never reaches a later capturing subscriber.
+//
+// The fix for the second is `rebuild_interest_cache()`. But that is global too,
+// so two tests doing this on different threads race: one rebuilds the cache to
+// "interested" for its own thread-local dispatch, the other rebuilds it back
+// while the first is mid-flight, and the first silently observes nothing.
+//
+// That is not hypothetical — it is a real intermittent failure of
+// `resolve_vid_only_bus_key_gate_reports_true_has_volume_id_when_vid_nonzero`,
+// which passes in isolation every time and fails under the full parallel suite.
+// The test carried a comment describing the hazard and a `rebuild_interest_cache`
+// call intended to fix it; the call is necessary but not sufficient.
+//
+// So capture is serialised process-wide here. Four tests across two modules were
+// each hand-rolling the same dance; one of them getting it subtly wrong is
+// exactly the drift a shared helper removes.
+static CAPTURE_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(());
+
+/// Run `f` with `subscriber` installed as the thread-local `tracing` dispatch,
+/// serialised against every other capture in this crate.
+///
+/// The interest cache is rebuilt on the way in (so a callsite already poisoned
+/// by an earlier no-op dispatch is reconsidered) and on the way out (so the
+/// next test does not inherit a cache built for a subscriber that is gone).
+pub(crate) fn with_captured_tracing<S, F, R>(subscriber: S, f: F) -> R
+where
+    S: tracing::Subscriber + Send + Sync + 'static,
+    F: FnOnce() -> R,
+{
+    // Poisoning is irrelevant: the guard protects ordering, not data, and a
+    // panicking test has already failed.
+    let _lock = CAPTURE_LOCK.lock().unwrap_or_else(|e| e.into_inner());
+    let dispatch = tracing::Dispatch::new(subscriber);
+    let guard = tracing::dispatcher::set_default(&dispatch);
+    tracing::callsite::rebuild_interest_cache();
+    let out = f();
+    drop(guard);
+    tracing::callsite::rebuild_interest_cache();
+    out
+}

src/session.rs

@@ -325,14 +325,6 @@ impl DiscSession {
         self.disc.take()
     }
 
-    /// 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.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).
@@ -340,12 +332,6 @@ impl DiscSession {
         &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 {
-        self.drive.as_mut().expect("drive present")
-    }
-
     /// Lock the tray so the disc cannot eject mid-rip. Unlock is guaranteed by
     /// `Drive::drop`. A no-op if the drive is no longer held by the session.
     pub fn lock_tray(&mut self) {
@@ -356,8 +342,19 @@ impl DiscSession {
 
     /// 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.expect("drive present")
+    /// # Errors
+    ///
+    /// [`Error::DeviceNotReady`] when the drive is no longer held — the PUBLIC
+    /// [`Self::stage_drive_as_reader`] moves it into the reader slot, and
+    /// calling this twice moves it out, so an empty slot is reachable through
+    /// ordinary use rather than being a caller error. A library must not panic
+    /// from public API, and a precondition that normal flow violates is a trap
+    /// rather than a contract.
+    pub fn into_drive(self) -> Result<Drive> {
+        self.drive.ok_or_else(|| Error::DeviceNotReady {
+            path: self.device.clone(),
+        })
     }
 
     /// Stage the owned drive as the session's boxed sector source so a live