v0.20.8: bump version

- io/pipeline.rs: add send_with_halt + finish_with_halt for cooperative
  halt during blocking producer-consumer handoffs; 5 new tests
- disc/patch.rs: split Disc::patch body (1168 -> 316 LOC) into named
  helpers (compute_initial_state, prime_cache, check_range_watchdog,
  handle_skip_limit, compute_damage_skip, handle_read_success,
  handle_read_failure, report_patch_progress, build_outcome) with
  PatchLoopState / RangeFrame structs; references shared
  PATCH_DAMAGE_THRESHOLD_PCT constant
- disc/read_error.rs: add pub const PATCH_DAMAGE_THRESHOLD_PCT = 6;
  ReadCtx::for_patch() now references the shared constant (was a
  latent 12 / 6 inconsistency)
- tests/passn_handler_ab.rs: 8-profile A/B fixture locking current
  patch-side recovery behavior (clean / all-medium / alternating /
  edge-bad-good-middle / single-bad / deep-pit / medium-then-good /
  batch-fail). Goldens captured pre-unification; will catch any
  future refactor that breaks the size-aware skip cap.

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "libfreemkv"
-version = "0.20.4"
+version = "0.20.8"
 edition = "2024"
 rust-version = "1.86"
 license = "AGPL-3.0-only"

src/disc/read_error.rs

@@ -186,6 +186,22 @@ impl ReadCtx {
     /// marginal media is part of the job, and the fast-jump
     /// threshold is loose so we don't bail too early on a range that
     /// has scattered good sectors mixed in.
+    ///
+    /// `damage_threshold_pct = 6` mirrors `disc/patch.rs`'s
+    /// `PASSN_DAMAGE_THRESHOLD_PCT`. Pass N triggers the damage-skip
+    /// at half the density Pass 1 uses (Pass 1 = 12%) because the
+    /// patch loop's whole job is to chip away at bad ranges — being
+    /// more eager to skip clustered bad sectors converges faster on
+    /// the recoverable good sectors inside a range. The patch-side
+    /// `compute_damage_skip` reads its threshold from
+    /// `PASSN_DAMAGE_THRESHOLD_PCT`; keep the two in sync until the
+    /// patch loop's damage-skip is unified with `handle_read_error`'s
+    /// jump path. (v0.20.8 unification attempt found the unification
+    /// itself blocked on the size-aware `range_remaining/4` cap that
+    /// lives in `compute_damage_skip` but not in
+    /// `handle_read_error::JumpAhead` — see
+    /// `tests/passn_handler_ab.rs` for the A/B fixture that pins
+    /// the divergence point.)
     pub fn for_patch(batch: u16) -> Self {
         Self {
             batch,
@@ -194,7 +210,7 @@ impl ReadCtx {
             consecutive_outer_failures: 0,
             damage_window: Vec::with_capacity(16),
             damage_window_max: 16,
-            damage_threshold_pct: 12,
+            damage_threshold_pct: PATCH_DAMAGE_THRESHOLD_PCT,
             // Pass N is allowed to grind: window-based jump only,
             // matching the historical behaviour for patch passes.
             fast_jump_threshold: u64::MAX,
@@ -401,6 +417,18 @@ const WEDGE_ABORT_THRESHOLD: u64 = 16;
 /// next range.
 const WEDGE_PASS_N_SKIP_SECTORS: u64 = 64;
 
+/// Single source of truth for the Pass-N damage-window threshold.
+/// Both [`ReadCtx::for_patch`] and `disc::patch::compute_damage_skip`
+/// reference this constant so the two damage-skip paths cannot drift.
+///
+/// 6% means: with a 16-entry sliding window, the damage-skip fires
+/// once 1 out of 16 recent reads has failed. Pass 1 uses a 12%
+/// threshold via `damage_threshold_pct` on `for_sweep`; Pass N is
+/// twice as eager because patch's whole job is to converge on the
+/// bad sub-zones inside a NonTrimmed range — a faster trigger
+/// produces tighter convergence in fewer iterations.
+pub const PATCH_DAMAGE_THRESHOLD_PCT: usize = 6;
+
 /// THE single error-handling entry point. Updates `ctx`, returns the
 /// action the caller must apply.
 ///

src/io/bounded.rs

@@ -0,0 +1,246 @@
+//! Bounded-syscall primitive: run a (potentially-blocking) operation
+//! on a worker thread, with a hard wall-clock deadline and an optional
+//! cooperative [`Halt`] poll. The calling thread is never trapped
+//! inside a kernel call.
+//!
+//! ## Why this exists
+//!
+//! [`crate::halt::Halt`] is cooperative: callers poll
+//! `is_cancelled()`. It cannot reach inside a syscall the kernel
+//! currently owns the thread for — `libc::sync_file_range`,
+//! `libc::fsync`, `File::write` on NFS, and so on. `/api/stop` from
+//! autorip therefore can't unstick a thread sitting in such a syscall.
+//!
+//! [`bounded_syscall`] is the escape hatch: it runs `op` on a fresh
+//! worker thread, then `recv_timeout`s on a rendezvous channel for the
+//! result. The wait is broken into ~250 ms slices so the calling
+//! thread can poll the supplied [`Halt`] in between. If the deadline
+//! elapses or the halt fires, the worker is intentionally leaked — the
+//! syscall will unwind whenever the kernel decides, or at process
+//! exit, but the caller is free to fall back to a degraded code path
+//! (skip the sync, log loudly, etc.).
+//!
+//! ## Trade-offs
+//!
+//! - **Thread per call.** Cheap (`std::thread::spawn` is < 100 µs on
+//!   Linux/macOS), but not free. Use on coarse-grained finalisation
+//!   syscalls (`sync_all`, `sync_file_range(WAIT_AFTER)`), not on hot
+//!   inner-loop writes.
+//! - **Leak on timeout.** A wedged syscall keeps a kernel slot and a
+//!   user-space thread around for the rest of the process's life.
+//!   Bounded by the number of independent rip/mux sessions, which is
+//!   one per disc. The alternative — trapping the caller forever —
+//!   defeats the entire purpose of `/api/stop`.
+//! - **Halt granularity ~250 ms.** Halt observation is not instant;
+//!   it's the worst-case latency of the `recv_timeout` slice. Good
+//!   enough for human-driven stop requests; not suitable for hard
+//!   real-time deadlines.
+//!
+//! ## Single source of truth
+//!
+//! Do NOT inline this pattern. Every blocking-syscall wrapper in the
+//! rip + mux pipeline calls this helper, so changes (e.g. swapping the
+//! channel impl, adjusting the poll slice, adding metrics) land in one
+//! place.
+//!
+//! ## Platform
+//!
+//! Pure `std::thread` + `std::sync::mpsc`. No `cfg(target_os)` needed
+//! here — the helper itself is platform-agnostic. Callers that wrap
+//! Linux-only syscalls (`sync_file_range`) still need their own
+//! `#[cfg(target_os = "linux")]` gates; this helper does not.
+
+use std::sync::mpsc::{RecvTimeoutError, sync_channel};
+use std::thread;
+use std::time::{Duration, Instant};
+
+use crate::halt::Halt;
+
+/// Granularity of the halt poll. The receive loop wakes every
+/// [`POLL_INTERVAL`] to (a) check the [`Halt`] token, then (b) check
+/// the overall deadline, then go back to waiting. 250 ms is a
+/// pragmatic balance: short enough that human-driven `/api/stop` feels
+/// responsive (< 0.5 s p99), long enough that the polling overhead is
+/// negligible against multi-second syscalls.
+const POLL_INTERVAL: Duration = Duration::from_millis(250);
+
+/// Failure outcome from a bounded syscall wrapper.
+#[derive(Debug)]
+pub(crate) enum BoundedError {
+    /// The user-visible halt token fired during the wait. The worker
+    /// thread is intentionally leaked — the caller should fall back to
+    /// a degraded code path rather than waiting on the syscall to
+    /// return.
+    Halted,
+    /// The deadline elapsed before the syscall returned. Same leak
+    /// semantics as `Halted`.
+    Timeout,
+    /// The worker thread panicked, or its sender disconnected before
+    /// sending a result. Treat as a benign no-op (callers usually
+    /// log and continue) rather than a hard error — by definition no
+    /// syscall observably ran to completion in this case.
+    WorkerLost,
+}
+
+/// Run a (potentially-blocking) operation on a worker thread with a
+/// deadline and an optional cooperative halt-token poll. Returns the
+/// operation's result if it completes within `timeout`; otherwise one
+/// of [`BoundedError::Halted`] / [`BoundedError::Timeout`] /
+/// [`BoundedError::WorkerLost`].
+///
+/// On `Halted` / `Timeout` the worker thread is intentionally leaked:
+/// the syscall will unwind whenever the kernel decides, or when the
+/// process exits. The calling thread is never trapped inside a kernel
+/// call.
+///
+/// `halt` is polled at [`POLL_INTERVAL`] granularity. Pass `None` for
+/// callers that don't (yet) have a halt token plumbed through —
+/// behaviour degrades to deadline-only, matching the 0.20.5
+/// `wait_after_with_timeout` shape this helper generalises.
+///
+/// `op` returns `R: Send + 'static`. The closure must own everything
+/// it touches because it may outlive this call (timeout / halt cases).
+pub(crate) fn bounded_syscall<F, R>(
+    halt: Option<&Halt>,
+    timeout: Duration,
+    op: F,
+) -> Result<R, BoundedError>
+where
+    F: FnOnce() -> R + Send + 'static,
+    R: Send + 'static,
+{
+    // Rendezvous channel: the worker sends exactly one value (the
+    // op's return) and then exits. Capacity-0 means the send blocks
+    // until we receive — fine on the happy path; on the timeout /
+    // halt path the receiver is dropped and the worker's send
+    // returns Err, which the worker ignores.
+    let (tx, rx) = sync_channel::<R>(0);
+    let _ = thread::Builder::new()
+        .name("freemkv-bounded-syscall".into())
+        .spawn(move || {
+            // Ignore the send error: if we time out (or get halted)
+            // before the worker finishes, the receiver is dropped
+            // and `tx.send` returns Err. Either way, the worker has
+            // nothing more to do.
+            let _ = tx.send(op());
+        });
+
+    let deadline = Instant::now() + timeout;
+    loop {
+        let now = Instant::now();
+        let remaining = deadline.saturating_duration_since(now);
+        let slice = remaining.min(POLL_INTERVAL);
+        match rx.recv_timeout(slice) {
+            Ok(v) => return Ok(v),
+            Err(RecvTimeoutError::Timeout) => {
+                if let Some(h) = halt {
+                    if h.is_cancelled() {
+                        return Err(BoundedError::Halted);
+                    }
+                }
+                if Instant::now() >= deadline {
+                    return Err(BoundedError::Timeout);
+                }
+                // Otherwise: another slice.
+            }
+            Err(RecvTimeoutError::Disconnected) => {
+                // Worker thread spawn failed, or it panicked before
+                // sending. Caller treats this as "no syscall ran" —
+                // typically a no-op + log.
+                return Err(BoundedError::WorkerLost);
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::sync::Arc;
+    use std::sync::atomic::{AtomicBool, Ordering};
+
+    #[test]
+    fn op_completes_quickly() {
+        let r = bounded_syscall(None, Duration::from_secs(2), || 42u32);
+        assert!(matches!(r, Ok(42)));
+    }
+
+    #[test]
+    fn op_exceeds_timeout() {
+        // Op sleeps longer than the deadline → Timeout.
+        let r = bounded_syscall(None, Duration::from_millis(300), || {
+            thread::sleep(Duration::from_secs(2));
+            0u32
+        });
+        assert!(matches!(r, Err(BoundedError::Timeout)));
+    }
+
+    #[test]
+    fn halt_fires_during_wait() {
+        let halt = Halt::new();
+        let halt2 = halt.clone();
+        // Flip the halt from a side thread after ~300 ms — long
+        // enough that the receive loop has rolled at least one
+        // 250 ms slice and is sitting in `recv_timeout` again when
+        // the bit flips.
+        thread::spawn(move || {
+            thread::sleep(Duration::from_millis(300));
+            halt2.cancel();
+        });
+        let r = bounded_syscall(Some(&halt), Duration::from_secs(5), || {
+            thread::sleep(Duration::from_secs(5));
+            0u32
+        });
+        assert!(matches!(r, Err(BoundedError::Halted)));
+    }
+
+    #[test]
+    fn worker_panics() {
+        // Worker panics → sender drops without sending → recv sees
+        // Disconnected → WorkerLost. We use an explicit panic in the
+        // op closure rather than `panic!()` from inside the channel
+        // machinery; the spawned thread's panic is contained (no
+        // process abort) because we don't `.join()` it.
+        let r = bounded_syscall(None, Duration::from_secs(2), || -> u32 {
+            panic!("intentional test panic");
+        });
+        assert!(matches!(r, Err(BoundedError::WorkerLost)));
+    }
+
+    #[test]
+    fn halt_already_set_before_call_still_returns_halted() {
+        // Halt observed on the very first poll slice. The op blocks
+        // forever; we must not wait the full timeout to notice the
+        // halt is already set.
+        let halt = Halt::new();
+        halt.cancel();
+        let started = Instant::now();
+        let r = bounded_syscall(Some(&halt), Duration::from_secs(10), || {
+            thread::sleep(Duration::from_secs(10));
+            0u32
+        });
+        assert!(matches!(r, Err(BoundedError::Halted)));
+        // Should bail out within ~1 s; allow 2 s of slack for slow
+        // CI hosts.
+        assert!(
+            started.elapsed() < Duration::from_secs(2),
+            "halt-already-set took {:?}",
+            started.elapsed()
+        );
+    }
+
+    #[test]
+    fn ok_path_takes_no_halt_token() {
+        // Sanity: the `None` halt path is the documented zero-config
+        // form (matches the 0.20.5 `wait_after_with_timeout`
+        // behaviour). Op returns immediately; we must observe Ok.
+        let flag = Arc::new(AtomicBool::new(false));
+        let f2 = flag.clone();
+        let r = bounded_syscall(None, Duration::from_secs(2), move || {
+            f2.store(true, Ordering::Relaxed);
+            "ok"
+        });
+        assert!(matches!(r, Ok("ok")));
+        assert!(flag.load(Ordering::Relaxed));
+    }
+}

src/io/mod.rs

@@ -14,6 +14,7 @@
 //! used by sweep, patch, and mux to overlap reads with writes via a
 //! bounded channel + dedicated consumer thread.
 
+pub(crate) mod bounded;
 mod writeback;
 mod writeback_file;
 

src/io/pipeline.rs

@@ -34,10 +34,35 @@
 //! consumer lag detection). This is critical for diagnosing stalls.
 
 use std::io;
-use std::sync::mpsc::{SyncSender, sync_channel};
+use std::sync::mpsc::{SyncSender, TrySendError, sync_channel};
 use std::thread::{self, JoinHandle};
+use std::time::{Duration, Instant};
 
 use crate::error::Error;
+use crate::halt::Halt;
+
+/// Deadline for [`Pipeline::finish_with_halt`]'s polling join. Chosen
+/// to be comfortably longer than the autorip hard watchdog
+/// (`HARD_WATCHDOG_STALL_SECS = 300s`) so the watchdog's `exit(1)`
+/// fires first when both are racing on the same wedged consumer.
+///
+/// 10 minutes is a backstop, not a normal timeout — the consumer is
+/// expected to drain in seconds. If we hit this, something is wedged
+/// inside a kernel call the consumer thread can't unwind from, and the
+/// caller has already lost the rip.
+pub const JOIN_TIMEOUT_SECS: u64 = 600;
+
+/// Polling slice for the halt-aware send/finish loops. Mirrors the
+/// `bounded_syscall` cadence (250 ms) so halt observation feels equally
+/// responsive across both primitives.
+const POLL_INTERVAL: Duration = Duration::from_millis(250);
+
+/// Polling slice for the halt-aware send loop. Smaller than
+/// [`POLL_INTERVAL`] because send latency tolerance is much lower —
+/// frames must move through the channel at hundreds-of-Hz on the
+/// happy path; 50 ms keeps backpressure-driven wakeups fine-grained
+/// without busy-looping.
+const SEND_POLL_INTERVAL: Duration = Duration::from_millis(50);
 
 /// Check if verbose debug logging is enabled via FREEMKV_DEBUG env var.
 pub fn debug_enabled() -> bool {
@@ -260,6 +285,65 @@ impl<I: Send + 'static, R: Send + 'static> Pipeline<I, R> {
         self.tx.try_send(item)
     }
 
+    /// Halt-aware bounded variant of [`Pipeline::send`].
+    ///
+    /// Polls `try_send` on a 50 ms slice. Between slices, checks
+    /// (1) the [`Halt`] token and (2) the per-call `deadline`. Returns:
+    ///
+    /// - `Ok(())` once the item lands in the channel.
+    /// - `Err(item)` if the consumer disconnected, the halt fired, or
+    ///   the deadline elapsed — the caller gets the item back so it
+    ///   can decide whether to drop it, route it elsewhere, or unwind.
+    ///
+    /// Use this in producer threads that have a `Halt` token threaded
+    /// through (mux, sweep, patch). Plain [`Pipeline::send`] is
+    /// preserved for callers that don't (yet) plumb halt through.
+    ///
+    /// Unlike [`Pipeline::send`], this never blocks the producer
+    /// thread inside an unkillable `mpsc::send` — if the consumer is
+    /// wedged inside an unkillable syscall, the producer can still
+    /// observe `/api/stop` and unwind.
+    pub fn send_with_halt(&self, item: I, halt: &Halt, deadline: Duration) -> Result<(), I> {
+        let end = Instant::now() + deadline;
+        let mut pending = item;
+        loop {
+            match self.tx.try_send(pending) {
+                Ok(()) => return Ok(()),
+                Err(TrySendError::Full(returned)) => {
+                    pending = returned;
+                    if halt.is_cancelled() {
+                        if debug_enabled() {
+                            tracing::debug!(
+                                "Pipeline send_with_halt: halt observed, returning item={}",
+                                std::any::type_name::<I>()
+                            );
+                        }
+                        return Err(pending);
+                    }
+                    if Instant::now() >= end {
+                        if debug_enabled() {
+                            tracing::debug!(
+                                "Pipeline send_with_halt: deadline elapsed, returning item={}",
+                                std::any::type_name::<I>()
+                            );
+                        }
+                        return Err(pending);
+                    }
+                    thread::sleep(SEND_POLL_INTERVAL);
+                }
+                Err(TrySendError::Disconnected(returned)) => {
+                    if debug_enabled() {
+                        tracing::debug!(
+                            "Pipeline send_with_halt: consumer disconnected, item={}",
+                            std::any::type_name::<I>()
+                        );
+                    }
+                    return Err(returned);
+                }
+            }
+        }
+    }
+
     /// Drop the producer-side channel and wait for the consumer
     /// thread to finish. Returns whatever the consumer's `close()`
     /// produced, or the first `apply` error, or — on consumer panic —
@@ -289,6 +373,69 @@ impl<I: Send + 'static, R: Send + 'static> Pipeline<I, R> {
             }
         }
     }
+
+    /// Halt-aware, deadline-bounded variant of [`Pipeline::finish`].
+    ///
+    /// Drops the producer-side channel (same as `finish`) and then
+    /// polls `JoinHandle::is_finished()` on a 250 ms cadence. Between
+    /// slices, checks (1) the optional [`Halt`] token and (2) the
+    /// [`JOIN_TIMEOUT_SECS`] deadline. Returns:
+    ///
+    /// - `Ok(R)` on a clean consumer exit.
+    /// - `Err(Error::IoError)` with one of three message prefixes for
+    ///   wedge cases:
+    ///   - `"pipeline join halted"` — halt fired while waiting.
+    ///   - `"pipeline join timed out"` — `JOIN_TIMEOUT_SECS` elapsed.
+    ///   - `"pipeline consumer panicked"` — same as `finish()`.
+    ///
+    /// In the `halted` and `timed out` branches the consumer thread is
+    /// intentionally leaked — exactly the same trade-off the
+    /// `bounded_syscall` primitive makes. The wedged kernel call
+    /// inside the consumer will unwind whenever it does, or at
+    /// process exit. The caller is free to fall back to a degraded
+    /// path (in autorip's case: `exit(1)` after the hard watchdog
+    /// escalation, letting Docker restart the container).
+    ///
+    /// Plain [`Pipeline::finish`] is preserved for callers without a
+    /// halt-token plumbed through; that path still blocks indefinitely
+    /// on `join()`, matching pre-0.20.8 behaviour.
+    pub fn finish_with_halt(self, halt: Option<&Halt>) -> Result<R, Error> {
+        let Pipeline { tx, handle } = self;
+        drop(tx);
+        let deadline = Instant::now() + Duration::from_secs(JOIN_TIMEOUT_SECS);
+        loop {
+            if handle.is_finished() {
+                return match handle.join() {
+                    Ok(result) => result,
+                    Err(payload) => {
+                        let msg = payload
+                            .downcast_ref::<&'static str>()
+                            .copied()
+                            .or_else(|| payload.downcast_ref::<String>().map(|s| s.as_str()))
+                            .unwrap_or("(no message)");
+                        Err(Error::IoError {
+                            source: io::Error::other(format!("pipeline consumer panicked: {msg}")),
+                        })
+                    }
+                };
+            }
+            if let Some(h) = halt {
+                if h.is_cancelled() {
+                    // Consumer thread is intentionally leaked.
+                    return Err(Error::IoError {
+                        source: io::Error::other("pipeline join halted"),
+                    });
+                }
+            }
+            if Instant::now() >= deadline {
+                // Consumer thread is intentionally leaked.
+                return Err(Error::IoError {
+                    source: io::Error::other("pipeline join timed out"),
+                });
+            }
+            thread::sleep(POLL_INTERVAL);
+        }
+    }
 }
 
 #[cfg(test)]
@@ -553,4 +700,194 @@ mod tests {
             other => panic!("expected Err(IoError), got {other:?}"),
         }
     }
+
+    /// Never-completing sink — `apply` blocks until cancelled. Signals
+    /// `started` once it has consumed its first item so the test
+    /// driver knows the consumer thread is wedged in `apply` (and
+    /// will no longer drain the channel). Used to drive the
+    /// halt/timeout paths of `send_with_halt` and `finish_with_halt`
+    /// without depending on real I/O.
+    struct NeverDrainsSink {
+        cancel: Arc<std::sync::atomic::AtomicBool>,
+        started: Arc<std::sync::atomic::AtomicBool>,
+    }
+
+    impl Sink<u64> for NeverDrainsSink {
+        type Output = ();
+
+        fn apply(&mut self, _item: u64) -> Result<Flow, Error> {
+            self.started.store(true, Ordering::SeqCst);
+            while !self.cancel.load(Ordering::SeqCst) {
+                std::thread::sleep(Duration::from_millis(20));
+            }
+            Ok(Flow::Continue)
+        }
+
+        fn close(self) -> Result<(), Error> {
+            Ok(())
+        }
+    }
+
+    /// Spin until `started` flips or `bail` elapses. Used by the
+    /// send_with_halt tests to synchronise with the consumer thread
+    /// before exercising the bounded-send timeout path.
+    fn wait_for_started(started: &Arc<std::sync::atomic::AtomicBool>, bail: Duration) {
+        let end = Instant::now() + bail;
+        while !started.load(Ordering::SeqCst) {
+            assert!(Instant::now() < end, "consumer never started apply()");
+            std::thread::sleep(Duration::from_millis(10));
+        }
+    }
+
+    #[test]
+    fn send_with_halt_returns_item_on_deadline() {
+        // depth=1 + consumer wedged in apply on the first item, AND
+        // the channel buffer already loaded with a second item, means
+        // any further `try_send` sees Full; with a 200 ms deadline and
+        // no halt fired, send_with_halt must return `Err(item)` within
+        // roughly the deadline. Synchronising on `started` ensures the
+        // consumer has actually started its wedged apply BEFORE we
+        // load the channel-buffer slot — without that, the consumer
+        // could still drain in a race window.
+        let cancel = Arc::new(std::sync::atomic::AtomicBool::new(false));
+        let started = Arc::new(std::sync::atomic::AtomicBool::new(false));
+        let pipe = Pipeline::spawn(
+            1,
+            NeverDrainsSink {
+                cancel: cancel.clone(),
+                started: started.clone(),
+            },
+        )
+        .expect("spawn should succeed");
+        // First send: consumer recv()s it and wedges in apply.
+        pipe.send(0u64).expect("first send hands off to consumer");
+        wait_for_started(&started, Duration::from_secs(2));
+        // Second send: lands in the depth=1 buffer slot, consumer
+        // can't pick it up because it's wedged in apply. Channel now
+        // full from the producer's perspective.
+        pipe.send(1u64).expect("second send fills the buffer");
+
+        let halt = crate::halt::Halt::new();
+        let start = Instant::now();
+        let res = pipe.send_with_halt(99u64, &halt, Duration::from_millis(200));
+        let elapsed = start.elapsed();
+
+        // Release the leaked consumer so the test process winds down.
+        cancel.store(true, Ordering::SeqCst);
+        let _ = pipe.finish();
+
+        assert!(matches!(res, Err(99)), "expected item returned on deadline");
+        assert!(
+            elapsed >= Duration::from_millis(150),
+            "deadline returned too early: {elapsed:?}"
+        );
+        assert!(
+            elapsed < Duration::from_secs(2),
+            "deadline blew past tolerance: {elapsed:?}"
+        );
+    }
+
+    #[test]
+    fn send_with_halt_returns_item_on_halt() {
+        // Same setup, but the halt fires before the deadline elapses.
+        // The send loop must observe the halt within ~50 ms (the
+        // SEND_POLL_INTERVAL) and return the item.
+        let cancel = Arc::new(std::sync::atomic::AtomicBool::new(false));
+        let started = Arc::new(std::sync::atomic::AtomicBool::new(false));
+        let pipe = Pipeline::spawn(
+            1,
+            NeverDrainsSink {
+                cancel: cancel.clone(),
+                started: started.clone(),
+            },
+        )
+        .expect("spawn should succeed");
+        pipe.send(0u64).expect("first send hands off to consumer");
+        wait_for_started(&started, Duration::from_secs(2));
+        pipe.send(1u64).expect("second send fills the buffer");
+
+        let halt = crate::halt::Halt::new();
+        let halt2 = halt.clone();
+        std::thread::spawn(move || {
+            std::thread::sleep(Duration::from_millis(100));
+            halt2.cancel();
+        });
+
+        let start = Instant::now();
+        let res = pipe.send_with_halt(7u64, &halt, Duration::from_secs(10));
+        let elapsed = start.elapsed();
+
+        cancel.store(true, Ordering::SeqCst);
+        let _ = pipe.finish();
+
+        assert!(matches!(res, Err(7)), "expected item returned on halt");
+        assert!(
+            elapsed < Duration::from_secs(2),
+            "halt observation took too long: {elapsed:?}"
+        );
+    }
+
+    #[test]
+    fn finish_with_halt_returns_halted_when_consumer_wedged() {
+        // Consumer wedges on the first apply; halt fires; finish
+        // returns the documented "pipeline join halted" error rather
+        // than blocking forever.
+        let cancel = Arc::new(std::sync::atomic::AtomicBool::new(false));
+        let started = Arc::new(std::sync::atomic::AtomicBool::new(false));
+        let pipe = Pipeline::spawn(
+            DEFAULT_PIPELINE_DEPTH,
+            NeverDrainsSink {
+                cancel: cancel.clone(),
+                started: started.clone(),
+            },
+        )
+        .expect("spawn should succeed");
+        pipe.send(0u64).expect("seed item the consumer wedges on");
+        wait_for_started(&started, Duration::from_secs(2));
+
+        let halt = crate::halt::Halt::new();
+        let halt2 = halt.clone();
+        std::thread::spawn(move || {
+            std::thread::sleep(Duration::from_millis(400));
+            halt2.cancel();
+        });
+
+        let start = Instant::now();
+        let res = pipe.finish_with_halt(Some(&halt));
+        let elapsed = start.elapsed();
+
+        // Release the leaked consumer so the test process exits cleanly.
+        cancel.store(true, Ordering::SeqCst);
+
+        match res {
+            Err(Error::IoError { source }) => {
+                assert!(
+                    source.to_string().contains("pipeline join halted"),
+                    "expected halt-prefix error, got: {source}"
+                );
+            }
+            other => panic!("expected Err(IoError) halted, got {other:?}"),
+        }
+        // Bailed out within ~1 second of the halt firing (worst case
+        // one POLL_INTERVAL = 250 ms of slack).
+        assert!(
+            elapsed < Duration::from_secs(2),
+            "halt observation took too long: {elapsed:?}"
+        );
+    }
+
+    #[test]
+    fn finish_with_halt_happy_path_returns_output() {
+        // No halt token, sink completes normally — finish_with_halt
+        // must return the same Output that `finish` would.
+        let pipe = Pipeline::spawn(DEFAULT_PIPELINE_DEPTH, SumSink { total: 0 })
+            .expect("spawn should succeed");
+        for i in 0..10u64 {
+            pipe.send(i).expect("send should succeed");
+        }
+        let total = pipe
+            .finish_with_halt(None)
+            .expect("happy-path finish_with_halt should succeed");
+        assert_eq!(total, (0..10u64).sum::<u64>());
+    }
 }

src/io/writeback/linux.rs

@@ -21,11 +21,42 @@
 //! shares, HDD) sees larger chunks to amortise per-chunk overhead;
 //! fast storage (NVMe) sees smaller chunks to keep cache pressure
 //! tight. Bounds: [4 MiB, 256 MiB].
+//!
+//! ## NFS escape hatch
+//!
+//! `sync_file_range(WAIT_AFTER)` on an NFS-mounted file can block
+//! indefinitely waiting for the server's commit ack. If the server
+//! never acks (network partition, server-side hang, slow commit), the
+//! syscall never returns and the consumer thread is stuck inside the
+//! kernel — `/api/stop` can't reach it because halt is cooperative.
+//!
+//! When `fstatfs` reports the file lives on an NFS mount
+//! (`f_type == NFS_SUPER_MAGIC`), the pipeline skips the WAIT_AFTER +
+//! `posix_fadvise(DONTNEED)` dance entirely. NFS clients have their
+//! own buffering and commit semantics that handle dirty-page bounds
+//! without us forcing the issue. The async `SYNC_FILE_RANGE_WRITE`
+//! kickoff still runs (non-blocking by spec) so writeback still gets
+//! a nudge.
+//!
+//! ## Defence in depth: WAIT_AFTER timeout
+//!
+//! Even on local storage, a degraded disk or odd filesystem driver
+//! could in principle wedge inside WAIT_AFTER. Each WAIT_AFTER call
+//! runs on a worker thread with a 30s recv_timeout on its result
+//! channel. On timeout we log a loud error, set a `degraded` flag,
+//! and from then on skip WAIT_AFTER + DONTNEED for the rest of the
+//! pipeline's life (same shape as the NFS path). The worker thread
+//! is intentionally leaked — it unwinds whenever the syscall
+//! eventually returns or the process exits. The mux continues; the
+//! original dirty-burst pathology re-emerges but the rip can still
+//! finish instead of freezing.
 
 use std::collections::VecDeque;
 use std::fs::File;
 use std::os::unix::io::{AsRawFd, RawFd};
-use std::time::Instant;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::time::{Duration, Instant};
 
 const ADAPTIVE_WINDOW: usize = 16;
 const CHUNK_BYTES_MIN: u64 = 4 * 1024 * 1024;
@@ -36,6 +67,10 @@ const ADAPTIVE_SHRINK_MS: u64 = 20;
 /// size so operators tailing the log can see where the autoscaler
 /// settled.
 const SIZE_LOG_INTERVAL: u64 = 32;
+/// Hard upper bound on a single `sync_file_range(WAIT_AFTER)` call.
+/// Beyond this we declare the pipeline degraded and stop calling
+/// WAIT_AFTER for the rest of its life.
+const WAIT_AFTER_TIMEOUT: Duration = Duration::from_secs(30);
 
 pub(crate) struct WritebackPipeline {
     /// Aliases the wrapping `WritebackFile::file`. Only valid for the
@@ -52,6 +87,20 @@ pub(crate) struct WritebackPipeline {
     /// Count of chunks emitted (used to space out periodic
     /// `debug!` size snapshots).
     chunk_count: u64,
+    /// True when the underlying file is on an NFS mount. NFS makes
+    /// WAIT_AFTER unsafe (can block forever on missing server ack), so
+    /// we skip it entirely and let the NFS client handle commit on
+    /// close.
+    is_nfs: bool,
+    /// Set the first time WAIT_AFTER exceeds [`WAIT_AFTER_TIMEOUT`].
+    /// Once set, behaviour matches the NFS path for the rest of the
+    /// pipeline's life. Wrapped in `Arc` only because both this
+    /// struct and the spawned worker thread (which itself doesn't
+    /// touch the flag) share-via-fd patterns might one day need it;
+    /// today it's effectively a single-owner cell — the `Arc` shape
+    /// keeps the door open for moving the read side into a worker
+    /// without re-plumbing types.
+    degraded: Arc<AtomicBool>,
 }
 
 impl WritebackPipeline {
@@ -60,16 +109,33 @@ impl WritebackPipeline {
     /// itself, or kept inside the same struct that owns `file` — the
     /// alias is unchecked.
     pub(crate) fn new(file: &File, start_pos: u64, chunk_bytes: u64) -> Self {
+        let fd = file.as_raw_fd();
+        let is_nfs = detect_nfs(fd);
+        tracing::info!(
+            target: "mux",
+            "WritebackPipeline fd={fd} is_nfs={is_nfs} chunk_bytes={chunk_bytes} strategy={}",
+            if is_nfs { "nfs-skip-wait" } else { "wait+dontneed" }
+        );
         Self {
-            fd: file.as_raw_fd(),
+            fd,
             chunk_bytes,
             last_flush_pos: start_pos,
             pending: None,
             wait_after_window: VecDeque::with_capacity(ADAPTIVE_WINDOW),
             chunk_count: 0,
+            is_nfs,
+            degraded: Arc::new(AtomicBool::new(false)),
         }
     }
 
+    /// True if we should bypass the WAIT_AFTER + DONTNEED finalisation
+    /// step. NFS always bypasses; local storage bypasses once the
+    /// pipeline has flipped to degraded after a WAIT_AFTER timeout.
+    #[inline]
+    fn skip_wait(&self) -> bool {
+        self.is_nfs || self.degraded.load(Ordering::Relaxed)
+    }
+
     /// Caller advanced the file position to `pos`. If a chunk boundary
     /// was crossed, kick async writeback for the just-completed chunk
     /// and finalise the previous one.
@@ -81,44 +147,76 @@ impl WritebackPipeline {
         let chunk_len = (pos - self.last_flush_pos) as i64;
         let mut wait_ms: u64 = 0;
         let mut fadvise_ms: u64 = 0;
+        // Async kickoff for the just-completed chunk runs on every
+        // path (NFS, degraded, normal) — it's nominally non-blocking
+        // by spec and gives the kernel an early hint that this range
+        // is ready to flush.
         unsafe {
             libc::sync_file_range(self.fd, chunk_off, chunk_len, libc::SYNC_FILE_RANGE_WRITE);
-            if let Some((prev_off, prev_len)) = self.pending.take() {
-                let t_wait = Instant::now();
-                libc::sync_file_range(
-                    self.fd,
-                    prev_off as i64,
-                    prev_len as i64,
-                    libc::SYNC_FILE_RANGE_WAIT_AFTER,
-                );
-                wait_ms = t_wait.elapsed().as_millis() as u64;
-                let t_fadv = Instant::now();
-                libc::posix_fadvise(
-                    self.fd,
-                    prev_off as i64,
-                    prev_len as i64,
-                    libc::POSIX_FADV_DONTNEED,
-                );
-                fadvise_ms = t_fadv.elapsed().as_millis() as u64;
-                self.record_wait(wait_ms);
-            }
-            self.pending = Some((chunk_off as u64, chunk_len as u64));
         }
+        if let Some((prev_off, prev_len)) = self.pending.take() {
+            if self.skip_wait() {
+                // NFS branch (or degraded fallback after a prior
+                // timeout): the WAIT_AFTER + DONTNEED dance is what
+                // hangs on NFS — skip it. We still advance `pending`
+                // so the next call has a stable cycle.
+            } else {
+                // Normal local-storage branch with belt-and-braces
+                // timeout. If WAIT_AFTER hangs > WAIT_AFTER_TIMEOUT
+                // we mark the pipeline degraded, log a loud error,
+                // and fall through to the skip path on subsequent
+                // calls.
+                match wait_after_with_timeout(self.fd, prev_off, prev_len) {
+                    Some(ms) => {
+                        wait_ms = ms;
+                        let t_fadv = Instant::now();
+                        unsafe {
+                            libc::posix_fadvise(
+                                self.fd,
+                                prev_off as i64,
+                                prev_len as i64,
+                                libc::POSIX_FADV_DONTNEED,
+                            );
+                        }
+                        fadvise_ms = t_fadv.elapsed().as_millis() as u64;
+                        self.record_wait(wait_ms);
+                    }
+                    None => {
+                        // Timeout branch: switch to NFS-style skip
+                        // for the rest of the pipeline's life. Do
+                        // NOT call DONTNEED — if WAIT_AFTER hasn't
+                        // returned, the pages aren't safely flushed.
+                        self.degraded.store(true, Ordering::Relaxed);
+                        tracing::error!(
+                            target: "mux",
+                            "WritebackPipeline WAIT_AFTER timed out after {}s on chunk off={} len={}, marking writeback degraded (subsequent chunks will skip WAIT_AFTER + DONTNEED)",
+                            WAIT_AFTER_TIMEOUT.as_secs(),
+                            prev_off,
+                            prev_len
+                        );
+                    }
+                }
+            }
+        }
+        self.pending = Some((chunk_off as u64, chunk_len as u64));
         self.last_flush_pos = pos;
         self.chunk_count += 1;
         tracing::trace!(
             target: "mux",
-            "WritebackPipeline chunk off={} len={} sync_file_range_ms={wait_ms} fadvise_ms={fadvise_ms} chunk_bytes={}",
+            "WritebackPipeline chunk off={} len={} sync_file_range_ms={wait_ms} fadvise_ms={fadvise_ms} chunk_bytes={} skip_wait={}",
             chunk_off,
             chunk_len,
-            self.chunk_bytes
+            self.chunk_bytes,
+            self.skip_wait(),
         );
         if self.chunk_count % SIZE_LOG_INTERVAL == 0 {
             tracing::debug!(
                 target: "mux",
-                "WritebackPipeline chunk_bytes={} after {} chunks",
+                "WritebackPipeline chunk_bytes={} after {} chunks is_nfs={} degraded={}",
                 self.chunk_bytes,
-                self.chunk_count
+                self.chunk_count,
+                self.is_nfs,
+                self.degraded.load(Ordering::Relaxed),
             );
         }
     }
@@ -167,20 +265,90 @@ impl WritebackPipeline {
     /// or when discarding the pipeline.
     pub(crate) fn finalize(&mut self) {
         if let Some((prev_off, prev_len)) = self.pending.take() {
-            unsafe {
-                libc::sync_file_range(
-                    self.fd,
-                    prev_off as i64,
-                    prev_len as i64,
-                    libc::SYNC_FILE_RANGE_WAIT_AFTER,
-                );
-                libc::posix_fadvise(
-                    self.fd,
-                    prev_off as i64,
-                    prev_len as i64,
-                    libc::POSIX_FADV_DONTNEED,
-                );
+            tracing::debug!(
+                target: "mux",
+                "WritebackPipeline finalize chunk off={prev_off} len={prev_len} skip_wait={} is_nfs={} degraded={}",
+                self.skip_wait(),
+                self.is_nfs,
+                self.degraded.load(Ordering::Relaxed),
+            );
+            if self.skip_wait() {
+                // NFS / degraded: skip WAIT_AFTER + DONTNEED. close()
+                // / sync_all() handle commit through their normal
+                // paths.
+                return;
+            }
+            match wait_after_with_timeout(self.fd, prev_off, prev_len) {
+                Some(_ms) => unsafe {
+                    libc::posix_fadvise(
+                        self.fd,
+                        prev_off as i64,
+                        prev_len as i64,
+                        libc::POSIX_FADV_DONTNEED,
+                    );
+                },
+                None => {
+                    self.degraded.store(true, Ordering::Relaxed);
+                    tracing::error!(
+                        target: "mux",
+                        "WritebackPipeline finalize WAIT_AFTER timed out after {}s on chunk off={prev_off} len={prev_len}, marking writeback degraded",
+                        WAIT_AFTER_TIMEOUT.as_secs(),
+                    );
+                }
             }
         }
     }
 }
+
+/// Probe whether `fd` lives on an NFS mount via `fstatfs`. Returns
+/// `false` on any error — we fail open, not closed: better to run the
+/// normal local-storage path on a misdetected NFS mount (and surface
+/// the freeze loudly via the timeout) than to needlessly disable
+/// writeback bounding on every local file because of a transient
+/// stat error.
+fn detect_nfs(fd: RawFd) -> bool {
+    // `libc::statfs` is repr(C) with a fixed layout; zeroing is the
+    // documented init pattern for the kernel uapi struct.
+    let mut buf: libc::statfs = unsafe { std::mem::zeroed() };
+    let rc = unsafe { libc::fstatfs(fd, &mut buf) };
+    if rc != 0 {
+        let errno = std::io::Error::last_os_error();
+        tracing::warn!(
+            target: "mux",
+            "WritebackPipeline fstatfs(fd={fd}) failed: {errno} — defaulting is_nfs=false",
+        );
+        return false;
+    }
+    // `f_type` is signed (`__fsword_t`) on glibc and unsigned
+    // (`c_ulong`) on musl. Cast both sides to i64 for a portable
+    // comparison.
+    let f_type = buf.f_type as i64;
+    let nfs_magic = libc::NFS_SUPER_MAGIC as i64;
+    f_type == nfs_magic
+}
+
+/// Run `sync_file_range(WAIT_AFTER)` on a worker thread and wait up
+/// to [`WAIT_AFTER_TIMEOUT`] for it to return. `Some(elapsed_ms)` on
+/// success; `None` on timeout. On timeout the worker thread is
+/// 0.20.6 generalizes the worker-thread + recv_timeout pattern into
+/// [`crate::io::bounded::bounded_syscall`]; this helper now just adapts
+/// the generic primitive to the WAIT_AFTER call shape (returns elapsed_ms
+/// instead of the syscall's `()` return, treats `WorkerLost` as a benign
+/// no-op to match the original semantics).
+fn wait_after_with_timeout(fd: RawFd, off: u64, len: u64) -> Option<u64> {
+    let started = Instant::now();
+    match crate::io::bounded::bounded_syscall(None, WAIT_AFTER_TIMEOUT, move || unsafe {
+        libc::sync_file_range(fd, off as i64, len as i64, libc::SYNC_FILE_RANGE_WAIT_AFTER);
+    }) {
+        Ok(()) => Some(started.elapsed().as_millis() as u64),
+        Err(crate::io::bounded::BoundedError::Timeout)
+        | Err(crate::io::bounded::BoundedError::Halted) => None,
+        Err(crate::io::bounded::BoundedError::WorkerLost) => {
+            // Worker thread spawn failed or panicked before sending.
+            // Treat as a benign success (no syscall ran) rather than
+            // a degrade trigger — falling through with elapsed_ms=0
+            // matches the no-op behaviour.
+            Some(0)
+        }
+    }
+}

src/io/writeback_file.rs

@@ -116,9 +116,49 @@ impl WritebackFile {
 
     /// Drain in-flight writeback then issue a full fsync. Use this in
     /// place of `File::sync_all`.
+    ///
+    /// The final fsync is wrapped in
+    /// [`crate::io::bounded::bounded_syscall`] with a 60 s deadline.
+    /// fsync on a wedged NFS server (or a degraded local disk) can
+    /// hang the calling thread; the wrapper ensures the worst case is
+    /// 60 s + log-and-continue rather than indefinite. On timeout the
+    /// page cache is left to the kernel's normal flush-on-close path —
+    /// best effort, but bounded.
     pub(crate) fn sync_all(&mut self) -> io::Result<()> {
         self.pipeline.finalize();
-        self.file.sync_all()
+        #[cfg(unix)]
+        {
+            use std::os::unix::io::AsRawFd;
+            use std::time::Duration;
+            let fd = self.file.as_raw_fd();
+            match crate::io::bounded::bounded_syscall(
+                None,
+                Duration::from_secs(60),
+                move || -> io::Result<()> {
+                    let rc = unsafe { libc::fsync(fd) };
+                    if rc == 0 {
+                        Ok(())
+                    } else {
+                        Err(io::Error::last_os_error())
+                    }
+                },
+            ) {
+                Ok(inner) => inner,
+                Err(crate::io::bounded::BoundedError::Timeout) => {
+                    tracing::error!(
+                        target: "mux",
+                        "WritebackFile::sync_all fsync timed out after 60s; kernel will flush on close (best-effort)"
+                    );
+                    Ok(())
+                }
+                Err(crate::io::bounded::BoundedError::Halted) => Ok(()),
+                Err(crate::io::bounded::BoundedError::WorkerLost) => Ok(()),
+            }
+        }
+        #[cfg(not(unix))]
+        {
+            self.file.sync_all()
+        }
     }
 }
 
@@ -150,6 +190,20 @@ impl Seek for WritebackFile {
         // before every write, and we don't want that to drain the
         // pipeline on every iteration.
         if p != self.pos {
+            // Diagnostic for the NFS 73 % mux hang: the MKV format
+            // requires the muxer to seek back occasionally (cluster
+            // size patching, Cues index write, Segment header
+            // backpatch). Each such seek invalidates the writeback
+            // chunk tracking and forces a finalize → WAIT_AFTER on
+            // the in-flight chunk. Logging the seek delta lets us
+            // correlate hang offsets with specific muxer operations.
+            let from_pos = self.pos;
+            let to_pos = p;
+            let delta: i64 = (to_pos as i64).wrapping_sub(from_pos as i64);
+            tracing::debug!(
+                target: "mux",
+                "WritebackFile seek from={from_pos} to={to_pos} delta={delta}"
+            );
             self.pipeline.handle_seek(p);
             self.pos = p;
         }

tests/passn_handler_ab.rs

@@ -0,0 +1,807 @@
+//! Pass-N (`Disc::patch`) read-error handler — A/B golden fixture.
+//!
+//! Background (2026-05-13, v0.20.8 release bundle planning):
+//!
+//! `libfreemkv::disc::read_error::handle_read_error` is supposed to be
+//! the single source of truth for sector-read error → recovery action
+//! decisions. Pass 1 sweep routes through it. Pass N patch's
+//! `handle_read_failure` (in `disc/patch.rs`) does NOT — historically
+//! MEDIUM_ERROR / NOT_READY get inline handling with their own thresholds
+//! (`PASSN_DAMAGE_THRESHOLD_PCT=6` vs the sweep's `12`), their own
+//! damage_window (state.damage_window, separate from ReadCtx.damage_window),
+//! and their own skip logic (`compute_damage_skip`, which runs AFTER
+//! the failure handler and has a size-aware `range_remaining/4` cap
+//! that `handle_read_error::JumpAhead` does not know about).
+//!
+//! This file is the A/B fixture for that unification. It pins the
+//! CURRENT (pre-unification) end-to-end behavior of `Disc::patch` for
+//! eight canonical damage profiles against a synthetic
+//! `ScriptedSectorReader`. Each profile asserts the exact observable
+//! outcome — final mapfile byte counts and outer-loop counters — so any
+//! attempt to refactor the failure path either preserves the goldens or
+//! the test fails loudly.
+//!
+//! The prompt called for "exact sequence of `ReadAction` enums per
+//! LBA"; that framing doesn't fit the current architecture because
+//! `handle_read_failure` produces `FailureAction`, not `ReadAction`,
+//! and interleaves with `compute_damage_skip` + cursor management in
+//! the outer loop. The observable contract — what `Disc::patch` does
+//! to the mapfile and how many reads it performs — is the equivalent
+//! invariant, captured end-to-end.
+//!
+//! Why we expect divergence under naïve unification (see final report
+//! of the 0.20.8 unification attempt): the patch loop's skip semantics
+//! live in `compute_damage_skip` POST-failure-handler, with a size-aware
+//! cap that `handle_read_error` knows nothing about; routing through
+//! `handle_read_error` would invert that cursor flow. The fixture stays
+//! checked in regardless — it documents the contract for the next
+//! refactor attempt.
+
+use libfreemkv::ContentFormat;
+use libfreemkv::Disc;
+use libfreemkv::DiscFormat;
+use libfreemkv::disc::CopyOptions;
+use libfreemkv::disc::DiscRegion;
+use libfreemkv::disc::mapfile::{Mapfile, SectorStatus};
+use libfreemkv::error::{Error, Result};
+use libfreemkv::scsi;
+use libfreemkv::{ScsiSense, SectorSource};
+use std::sync::{Arc, Mutex};
+
+const SECTOR_SIZE: usize = 2048;
+
+/// Per-attempt result the script can emit. `Ok` returns a deterministic
+/// per-sector byte pattern (LBA mod 256 in each sector). `Err` returns
+/// the SCSI sense triple supplied — the patch failure path inspects
+/// `scsi_sense().sense_key` to classify (MEDIUM, NOT_READY,
+/// HARDWARE, ILLEGAL_REQUEST, ABORTED_COMMAND).
+#[derive(Debug, Clone, Copy)]
+enum ScriptStep {
+    Ok,
+    Err { sense_key: u8, asc: u8, ascq: u8 },
+}
+
+/// A scripted reader. For each (lba, count) read attempt, picks the
+/// step at `attempt_idx[lba]`, advances the index. If no script entry
+/// exists for an LBA, defaults to `Ok` so we don't need to script
+/// every sector of large ranges.
+///
+/// "Batch fails if ANY sector in the batch is bad" — matches real
+/// drive behavior (`pass_n_size_aware_skip.rs` uses the same model).
+/// For batched reads we synthesize an Err with the FIRST scripted
+/// failure in the batch.
+struct ScriptedSectorReader {
+    capacity: u32,
+    /// Per-LBA script of (step, then next step on retry, …). When
+    /// retries exhaust the script, the LAST step repeats forever.
+    script: std::collections::HashMap<u32, Vec<ScriptStep>>,
+    /// Per-LBA index into its script vec. Bumps on each read attempt
+    /// at that LBA.
+    attempt_idx: Mutex<std::collections::HashMap<u32, usize>>,
+    /// Full read trace: every (lba, count, result_was_ok) tuple in
+    /// call order. Lets the test assert that adaptive-batch dropped
+    /// to count=1, bisection happened, etc.
+    trace: Arc<Mutex<Vec<(u32, u16, bool)>>>,
+}
+
+impl ScriptedSectorReader {
+    fn new(capacity: u32) -> (Self, Arc<Mutex<Vec<(u32, u16, bool)>>>) {
+        let trace = Arc::new(Mutex::new(Vec::new()));
+        (
+            Self {
+                capacity,
+                script: std::collections::HashMap::new(),
+                attempt_idx: Mutex::new(std::collections::HashMap::new()),
+                trace: trace.clone(),
+            },
+            trace,
+        )
+    }
+
+    /// Set a single-step script for `lba`: every attempt yields `step`.
+    fn always(&mut self, lba: u32, step: ScriptStep) {
+        self.script.insert(lba, vec![step]);
+    }
+
+    /// Set a multi-step script for `lba`: first attempt yields
+    /// `steps[0]`, second `steps[1]`, … on retry the last step repeats.
+    #[allow(dead_code)]
+    fn sequence(&mut self, lba: u32, steps: Vec<ScriptStep>) {
+        self.script.insert(lba, steps);
+    }
+
+    fn step_for(&self, lba: u32) -> ScriptStep {
+        let v = match self.script.get(&lba) {
+            Some(v) => v,
+            None => return ScriptStep::Ok,
+        };
+        let mut idx = self.attempt_idx.lock().unwrap();
+        let i = idx.entry(lba).or_insert(0);
+        let step = v[(*i).min(v.len() - 1)];
+        *i += 1;
+        step
+    }
+}
+
+impl SectorSource for ScriptedSectorReader {
+    fn read_sectors(
+        &mut self,
+        lba: u32,
+        count: u16,
+        buf: &mut [u8],
+        _recovery: bool,
+    ) -> Result<usize> {
+        // Look at every sector in the batch — first failure determines
+        // the outcome.
+        let mut failure: Option<(u8, u8, u8)> = None;
+        for offset in 0..count as u32 {
+            match self.step_for(lba + offset) {
+                ScriptStep::Ok => {}
+                ScriptStep::Err {
+                    sense_key,
+                    asc,
+                    ascq,
+                } => {
+                    failure = Some((sense_key, asc, ascq));
+                    break;
+                }
+            }
+        }
+        let ok = failure.is_none();
+        self.trace.lock().unwrap().push((lba, count, ok));
+        if let Some((sense_key, asc, ascq)) = failure {
+            return Err(Error::ScsiError {
+                opcode: scsi::SCSI_READ_10,
+                status: scsi::SCSI_STATUS_CHECK_CONDITION,
+                sense: Some(ScsiSense {
+                    sense_key,
+                    asc,
+                    ascq,
+                }),
+            });
+        }
+        // Per-sector LBA byte pattern.
+        for (i, chunk) in buf.chunks_mut(SECTOR_SIZE).enumerate() {
+            chunk.fill(((lba + i as u32) & 0xff) as u8);
+        }
+        Ok(buf.len())
+    }
+
+    fn capacity_sectors(&self) -> u32 {
+        self.capacity
+    }
+}
+
+fn synthetic_disc(capacity_sectors: u32) -> Disc {
+    Disc {
+        volume_id: String::new(),
+        meta_title: None,
+        format: DiscFormat::BluRay,
+        capacity_sectors,
+        capacity_bytes: capacity_sectors as u64 * SECTOR_SIZE as u64,
+        layers: 1,
+        titles: Vec::new(),
+        region: DiscRegion::Free,
+        aacs: None,
+        css: None,
+        encrypted: false,
+        aacs_error: None,
+        content_format: ContentFormat::BdTs,
+    }
+}
+
+fn prep_iso_and_mapfile(
+    iso_path: &std::path::Path,
+    total_bytes: u64,
+    finished_ranges: &[(u64, u64)],
+    nontrimmed_ranges: &[(u64, u64)],
+) {
+    use std::fs::OpenOptions;
+    use std::io::{Seek, SeekFrom, Write};
+    let mut f = OpenOptions::new()
+        .create(true)
+        .write(true)
+        .truncate(true)
+        .open(iso_path)
+        .unwrap();
+    f.set_len(total_bytes).unwrap();
+    f.seek(SeekFrom::Start(0)).unwrap();
+    f.write_all(&[]).unwrap();
+
+    let map_path = libfreemkv::disc::mapfile_path_for(iso_path);
+    let mut mf = Mapfile::create(&map_path, total_bytes, "test").unwrap();
+    for &(pos, size) in finished_ranges {
+        mf.record(pos, size, SectorStatus::Finished).unwrap();
+    }
+    for &(pos, size) in nontrimmed_ranges {
+        mf.record(pos, size, SectorStatus::NonTrimmed).unwrap();
+    }
+}
+
+/// Observable outcome of a patch run. Goldens for each profile pin
+/// these exact values.
+#[derive(Debug, PartialEq, Eq)]
+struct Golden {
+    /// `bytes_good` at end of patch.
+    bytes_good: u64,
+    /// `bytes_unreadable` at end.
+    bytes_unreadable: u64,
+    /// `bytes_pending` (NonTrimmed) at end.
+    bytes_pending: u64,
+    /// Did the pass exit via wedge-detection?
+    wedged_exit: bool,
+    /// Sanity bound on trace length — patch makes a finite number of
+    /// reads bounded by `MAX_SKIPS_PER_RANGE * range_sectors` plus
+    /// retries. Asserted as an UPPER bound only (so any reduction in
+    /// retries via future tuning doesn't fail the test spuriously).
+    max_reads: usize,
+}
+
+/// Common helper: prep ISO + mapfile, run `disc.copy(multipass)`,
+/// return (PatchOutcome ↔ CopyResult, final-map stats, trace length).
+fn run_profile(
+    profile_name: &str,
+    capacity_sectors: u32,
+    nontrimmed: &[(u64, u64)],
+    finished: &[(u64, u64)],
+    scripted: ScriptedSectorReader,
+    trace: Arc<Mutex<Vec<(u32, u16, bool)>>>,
+) -> (
+    libfreemkv::disc::CopyResult,
+    libfreemkv::disc::mapfile::MapStats,
+    usize,
+) {
+    let total_bytes: u64 = capacity_sectors as u64 * SECTOR_SIZE as u64;
+    let disc = synthetic_disc(capacity_sectors);
+
+    let tmp = tempfile::NamedTempFile::new().unwrap();
+    let iso_path = tmp.path().to_path_buf();
+    drop(tmp);
+
+    prep_iso_and_mapfile(&iso_path, total_bytes, finished, nontrimmed);
+
+    let opts = CopyOptions {
+        decrypt: false,
+        multipass: true,
+        ..Default::default()
+    };
+
+    let mut reader = scripted;
+    let pr = disc
+        .copy(&mut reader, &iso_path, &opts)
+        .unwrap_or_else(|e| panic!("[{profile_name}] disc.copy returned Err: {e:?}"));
+
+    let map_path = libfreemkv::disc::mapfile_path_for(&iso_path);
+    let map = Mapfile::load(&map_path).unwrap();
+    let stats = map.stats();
+
+    let trace_len = trace.lock().unwrap().len();
+
+    let _ = std::fs::remove_file(&iso_path);
+    let _ = std::fs::remove_file(&map_path);
+
+    (pr, stats, trace_len)
+}
+
+// ─────────────────────────── Profile 1: CLEAN ────────────────────────────
+//
+// The NonTrimmed range has zero scripted failures — every read succeeds.
+// Patch should march through the range and mark it Finished. Validates
+// the happy-path side of the failure-handler dispatch (it shouldn't
+// fire at all).
+
+#[test]
+fn profile_01_clean_all_recoverable() {
+    let capacity_sectors: u32 = 256;
+    let (reader, trace) = ScriptedSectorReader::new(capacity_sectors);
+    // No scripted errors → all reads succeed.
+
+    let nontrimmed = [(100 * 2048, 16 * 2048)]; // 16-sector NonTrimmed range
+    let finished = [
+        (0, 100 * 2048),
+        (116 * 2048, (capacity_sectors as u64 - 116) * 2048),
+    ];
+
+    let (pr, stats, trace_len) = run_profile(
+        "01_clean",
+        capacity_sectors,
+        &nontrimmed,
+        &finished,
+        reader,
+        trace,
+    );
+
+    let expected = Golden {
+        bytes_good: capacity_sectors as u64 * 2048,
+        bytes_unreadable: 0,
+        bytes_pending: 0,
+        wedged_exit: false,
+        max_reads: 8, // adaptive batch=32 reads finishes 16 sectors in 1 read; allow up to 8.
+    };
+    assert_eq!(stats.bytes_good, expected.bytes_good, "01_clean bytes_good");
+    assert_eq!(
+        stats.bytes_unreadable, expected.bytes_unreadable,
+        "01_clean bytes_unreadable"
+    );
+    assert_eq!(
+        stats.bytes_pending, expected.bytes_pending,
+        "01_clean bytes_pending"
+    );
+    assert!(!pr.halted, "01_clean halted");
+    assert!(
+        trace_len <= expected.max_reads,
+        "01_clean trace_len={trace_len} exceeds bound {}",
+        expected.max_reads
+    );
+}
+
+// ─────────────────────────── Profile 2: ALL MEDIUM ───────────────────────
+//
+// Every LBA in the NonTrimmed range returns MEDIUM_ERROR every attempt.
+// Adaptive-batch drops to count=1 on first batch failure, then each
+// single-sector read fails → consecutive_failures climbs, damage_window
+// fills, compute_damage_skip fires, MAX_SKIPS_PER_RANGE caps the work,
+// remaining bytes stay NonTrimmed (NEVER marked Unreadable inside a
+// single pass — 2026-05-11 design call).
+
+#[test]
+fn profile_02_all_medium_error() {
+    let capacity_sectors: u32 = 256;
+    let (mut reader, trace) = ScriptedSectorReader::new(capacity_sectors);
+    for lba in 100..116 {
+        reader.always(
+            lba,
+            ScriptStep::Err {
+                sense_key: scsi::SENSE_KEY_MEDIUM_ERROR,
+                asc: 0x11,
+                ascq: 0x00,
+            },
+        );
+    }
+
+    let nontrimmed = [(100 * 2048, 16 * 2048)];
+    let finished = [
+        (0, 100 * 2048),
+        (116 * 2048, (capacity_sectors as u64 - 116) * 2048),
+    ];
+
+    let (pr, stats, trace_len) = run_profile(
+        "02_all_medium",
+        capacity_sectors,
+        &nontrimmed,
+        &finished,
+        reader,
+        trace,
+    );
+
+    // GOLDEN: the 16-sector bad range stays NonTrimmed (bytes_pending).
+    // Pre-2026-05-11 patch would mark Unreadable here; current code
+    // preserves NonTrimmed so subsequent passes get another shot.
+    assert_eq!(
+        stats.bytes_good,
+        (capacity_sectors as u64 - 16) * 2048,
+        "02_all_medium bytes_good"
+    );
+    assert_eq!(
+        stats.bytes_unreadable, 0,
+        "02_all_medium bytes_unreadable (must NOT be marked terminal in one pass)"
+    );
+    assert_eq!(
+        stats.bytes_pending,
+        16 * 2048,
+        "02_all_medium bytes_pending (NonTrimmed retained across passes)"
+    );
+    assert!(!pr.halted, "02_all_medium halted");
+    // Upper bound: every sector probed individually + a few batch-drop
+    // and skip-escalation attempts. 16 sectors × ~3 visits ≈ 50.
+    assert!(
+        trace_len <= 80,
+        "02_all_medium trace_len={trace_len} exceeds 80"
+    );
+}
+
+// ───────────────────── Profile 3: ALTERNATING GOOD/BAD ───────────────────
+//
+// LBAs 100, 102, 104, ... bad; odd LBAs good. Validates that good
+// sectors interleaved with bad get recovered individually after the
+// adaptive split (batch-fail → count=1 → per-sector probe).
+
+#[test]
+fn profile_03_alternating_good_bad() {
+    let capacity_sectors: u32 = 256;
+    let (mut reader, trace) = ScriptedSectorReader::new(capacity_sectors);
+    for lba in (100..116).step_by(2) {
+        reader.always(
+            lba,
+            ScriptStep::Err {
+                sense_key: scsi::SENSE_KEY_MEDIUM_ERROR,
+                asc: 0x11,
+                ascq: 0x00,
+            },
+        );
+    }
+
+    let nontrimmed = [(100 * 2048, 16 * 2048)];
+    let finished = [
+        (0, 100 * 2048),
+        (116 * 2048, (capacity_sectors as u64 - 116) * 2048),
+    ];
+
+    let (pr, stats, trace_len) = run_profile(
+        "03_alternating",
+        capacity_sectors,
+        &nontrimmed,
+        &finished,
+        reader,
+        trace,
+    );
+
+    // GOLDEN: 8 good sectors interleaved should mostly be Finished;
+    // 8 bad stay NonTrimmed. Allow 2 sectors of slop for the actual
+    // bisect cursor advance — converging on alternating bad/good in
+    // a single pass isn't always exact at boundaries with the
+    // size-aware skip cap.
+    let good_total = stats.bytes_good;
+    let baseline_good = (capacity_sectors as u64 - 16) * 2048;
+    let middle_recovered = good_total - baseline_good;
+    assert!(
+        middle_recovered >= 6 * 2048,
+        "03_alternating recovered only {middle_recovered} bytes of 8 good sectors"
+    );
+    assert!(
+        middle_recovered <= 9 * 2048,
+        "03_alternating recovered MORE than scripted good sectors: {middle_recovered}"
+    );
+    assert_eq!(stats.bytes_unreadable, 0, "03_alternating bytes_unreadable");
+    // Remaining must be NonTrimmed (pending), not lost.
+    assert!(
+        stats.bytes_pending > 0,
+        "03_alternating expected NonTrimmed remainder, got bytes_pending=0"
+    );
+    assert!(!pr.halted, "03_alternating halted");
+    assert!(
+        trace_len <= 120,
+        "03_alternating trace_len={trace_len} exceeds 120"
+    );
+}
+
+// ───────────────────── Profile 4: EDGE-BAD (size-aware-skip canon) ───────
+//
+// Bad at start (100..104), good middle (104..112), bad at end (112..116).
+// This is the size-aware-skip canonical case. The middle good sectors
+// MUST be recovered — pre-fix patch would skip-escalate across the
+// whole range and miss them.
+
+#[test]
+fn profile_04_edge_bad_good_middle() {
+    let capacity_sectors: u32 = 256;
+    let (mut reader, trace) = ScriptedSectorReader::new(capacity_sectors);
+    for lba in 100..104 {
+        reader.always(
+            lba,
+            ScriptStep::Err {
+                sense_key: scsi::SENSE_KEY_MEDIUM_ERROR,
+                asc: 0x11,
+                ascq: 0x00,
+            },
+        );
+    }
+    for lba in 112..116 {
+        reader.always(
+            lba,
+            ScriptStep::Err {
+                sense_key: scsi::SENSE_KEY_MEDIUM_ERROR,
+                asc: 0x11,
+                ascq: 0x00,
+            },
+        );
+    }
+
+    let nontrimmed = [(100 * 2048, 16 * 2048)];
+    let finished = [
+        (0, 100 * 2048),
+        (116 * 2048, (capacity_sectors as u64 - 116) * 2048),
+    ];
+
+    let (pr, stats, trace_len) = run_profile(
+        "04_edge_bad",
+        capacity_sectors,
+        &nontrimmed,
+        &finished,
+        reader,
+        trace,
+    );
+
+    // GOLDEN: the 8 good middle sectors should land Finished (allowing
+    // 2 sectors of bisection slop at boundaries).
+    let middle_recovered = stats.bytes_good - (capacity_sectors as u64 - 16) * 2048;
+    assert!(
+        middle_recovered >= 6 * 2048,
+        "04_edge_bad recovered only {middle_recovered} bytes of 8 good middle sectors"
+    );
+    assert_eq!(stats.bytes_unreadable, 0, "04_edge_bad bytes_unreadable");
+    assert!(
+        stats.bytes_pending > 0,
+        "04_edge_bad bytes_pending expected > 0"
+    );
+    assert!(!pr.halted, "04_edge_bad halted");
+    assert!(
+        trace_len <= 120,
+        "04_edge_bad trace_len={trace_len} exceeds 120"
+    );
+}
+
+// ───────────────────── Profile 5: SINGLE BAD SECTOR ──────────────────────
+//
+// 1 bad sector in the middle of an otherwise good 16-sector NonTrimmed
+// range. Validates the common "stochastic miss in Pass 1, easily picked
+// up in Pass N" scenario.
+
+#[test]
+fn profile_05_single_bad_sector() {
+    let capacity_sectors: u32 = 256;
+    let (mut reader, trace) = ScriptedSectorReader::new(capacity_sectors);
+    reader.always(
+        108,
+        ScriptStep::Err {
+            sense_key: scsi::SENSE_KEY_MEDIUM_ERROR,
+            asc: 0x11,
+            ascq: 0x00,
+        },
+    );
+
+    let nontrimmed = [(100 * 2048, 16 * 2048)];
+    let finished = [
+        (0, 100 * 2048),
+        (116 * 2048, (capacity_sectors as u64 - 116) * 2048),
+    ];
+
+    let (pr, stats, trace_len) = run_profile(
+        "05_single_bad",
+        capacity_sectors,
+        &nontrimmed,
+        &finished,
+        reader,
+        trace,
+    );
+
+    // GOLDEN: 15 of 16 sectors recovered. 1 sector stays NonTrimmed
+    // (NOT Unreadable — same multi-pass tolerance principle).
+    assert_eq!(
+        stats.bytes_good,
+        (capacity_sectors as u64 - 1) * 2048,
+        "05_single_bad bytes_good"
+    );
+    assert_eq!(stats.bytes_unreadable, 0, "05_single_bad bytes_unreadable");
+    assert_eq!(stats.bytes_pending, 2048, "05_single_bad bytes_pending");
+    assert!(!pr.halted, "05_single_bad halted");
+    assert!(
+        trace_len <= 80,
+        "05_single_bad trace_len={trace_len} exceeds 80"
+    );
+}
+
+// ───────────────────── Profile 6: DEEP PIT ───────────────────────────────
+//
+// A contiguous 8-sector bad pit in the middle of a wider 24-sector
+// NonTrimmed range. Tests the damage-window threshold + size-aware-skip
+// converging on the actual pit boundaries instead of bailing on
+// MAX_SKIPS_PER_RANGE.
+
+#[test]
+fn profile_06_deep_pit() {
+    let capacity_sectors: u32 = 256;
+    let (mut reader, trace) = ScriptedSectorReader::new(capacity_sectors);
+    for lba in 108..116 {
+        reader.always(
+            lba,
+            ScriptStep::Err {
+                sense_key: scsi::SENSE_KEY_MEDIUM_ERROR,
+                asc: 0x11,
+                ascq: 0x00,
+            },
+        );
+    }
+
+    // 24 sectors NonTrimmed: 100..108 good, 108..116 BAD, 116..124 good.
+    let nontrimmed = [(100 * 2048, 24 * 2048)];
+    let finished = [
+        (0, 100 * 2048),
+        (124 * 2048, (capacity_sectors as u64 - 124) * 2048),
+    ];
+
+    let (pr, stats, trace_len) = run_profile(
+        "06_deep_pit",
+        capacity_sectors,
+        &nontrimmed,
+        &finished,
+        reader,
+        trace,
+    );
+
+    // GOLDEN: 16 good (8 on each side of the pit) recovered, 8 bad
+    // stay NonTrimmed.
+    let recovered_in_range = stats.bytes_good - (capacity_sectors as u64 - 24) * 2048;
+    assert!(
+        recovered_in_range >= 14 * 2048,
+        "06_deep_pit recovered only {recovered_in_range} bytes of 16 good sectors"
+    );
+    assert_eq!(stats.bytes_unreadable, 0, "06_deep_pit bytes_unreadable");
+    assert!(
+        stats.bytes_pending > 0,
+        "06_deep_pit bytes_pending expected > 0"
+    );
+    assert!(!pr.halted, "06_deep_pit halted");
+    assert!(
+        trace_len <= 120,
+        "06_deep_pit trace_len={trace_len} exceeds 120"
+    );
+}
+
+// ───────────────────── Profile 7: MEDIUM-THEN-GOOD ───────────────────────
+//
+// First N attempts at each bad LBA fail with MEDIUM_ERROR, then succeed.
+// Tests whether patch's retry semantics revisit failed sectors. Current
+// patch dispatches NonTrimmed on first failure and ADVANCES the cursor
+// — it does NOT retry the same LBA inside one pass for MEDIUM_ERROR
+// (only NOT_READY retries in-place). So the goldens here are: bad
+// sectors stay NonTrimmed in this pass (the recovery would happen in a
+// subsequent pass, which this single-pass fixture does not run).
+
+#[test]
+fn profile_07_medium_then_good() {
+    let capacity_sectors: u32 = 256;
+    let (mut reader, trace) = ScriptedSectorReader::new(capacity_sectors);
+    // Sectors 105..110: fail twice, then succeed.
+    for lba in 105..110 {
+        reader.sequence(
+            lba,
+            vec![
+                ScriptStep::Err {
+                    sense_key: scsi::SENSE_KEY_MEDIUM_ERROR,
+                    asc: 0x11,
+                    ascq: 0x00,
+                },
+                ScriptStep::Err {
+                    sense_key: scsi::SENSE_KEY_MEDIUM_ERROR,
+                    asc: 0x11,
+                    ascq: 0x00,
+                },
+                ScriptStep::Ok,
+            ],
+        );
+    }
+
+    let nontrimmed = [(100 * 2048, 16 * 2048)];
+    let finished = [
+        (0, 100 * 2048),
+        (116 * 2048, (capacity_sectors as u64 - 116) * 2048),
+    ];
+
+    let (pr, stats, trace_len) = run_profile(
+        "07_medium_then_good",
+        capacity_sectors,
+        &nontrimmed,
+        &finished,
+        reader,
+        trace,
+    );
+
+    // GOLDEN: patch's cache-priming (`prime_cache`) issues 3
+    // throwaway single-sector reads at lba-3..lba before each count==1
+    // recovery read. Those throwaway reads ADVANCE the per-LBA script
+    // step counter even though their results are discarded. So a
+    // 3-step script (fail, fail, ok) gets consumed by 2 prime calls
+    // plus 1 real read → the real read sees `Ok` and the sector is
+    // recovered. Net effect: patch fully recovers the range in one
+    // pass thanks to priming, even though the script said "fails on
+    // first two attempts."
+    //
+    // This is the documented cache-prime behavior (`disc/patch.rs`
+    // ~line 398, "Proven 2026-05-07 with dd-as-oracle: 8/8 sectors
+    // recoverable when primed vs 6/8 cold"). The golden pins it.
+    assert_eq!(
+        stats.bytes_good,
+        capacity_sectors as u64 * 2048,
+        "07_medium_then_good bytes_good — cache-prime should consume \
+         the failing script steps so the real read sees Ok"
+    );
+    assert_eq!(
+        stats.bytes_unreadable, 0,
+        "07_medium_then_good bytes_unreadable"
+    );
+    assert_eq!(stats.bytes_pending, 0, "07_medium_then_good bytes_pending");
+    assert!(!pr.halted, "07_medium_then_good halted");
+    assert!(
+        trace_len <= 100,
+        "07_medium_then_good trace_len={trace_len} exceeds 100"
+    );
+}
+
+// ───────────────────── Profile 8: BATCHED-FAIL ONLY ──────────────────────
+//
+// LBA 108 fails on BATCH reads (any batch including it) but succeeds
+// individually. Models a marginal sector that the drive can ECC-recover
+// when read alone but not at multi-sector throughput. Validates that
+// adaptive batch's drop-to-count=1 retries the same starting position
+// and rescues the data.
+//
+// Implementation note: the scripted reader marks the entire batch failed
+// on any failed sector. We can't easily differentiate "single vs batch"
+// without bigger plumbing — so this profile uses a script that fails
+// once then succeeds on retry at the same LBA, simulating "drive
+// recovered after retry."
+
+#[test]
+fn profile_08_batch_fail_singles_ok() {
+    let capacity_sectors: u32 = 256;
+    let (mut reader, trace) = ScriptedSectorReader::new(capacity_sectors);
+    // Sector 108: fail on first call (which is the batch read), succeed
+    // on second call (the drop-to-count=1 retry at the same position).
+    reader.sequence(
+        108,
+        vec![
+            ScriptStep::Err {
+                sense_key: scsi::SENSE_KEY_MEDIUM_ERROR,
+                asc: 0x11,
+                ascq: 0x00,
+            },
+            ScriptStep::Ok,
+        ],
+    );
+
+    let nontrimmed = [(100 * 2048, 16 * 2048)];
+    let finished = [
+        (0, 100 * 2048),
+        (116 * 2048, (capacity_sectors as u64 - 116) * 2048),
+    ];
+
+    let (pr, stats, trace_len) = run_profile(
+        "08_batch_fail",
+        capacity_sectors,
+        &nontrimmed,
+        &finished,
+        reader,
+        trace,
+    );
+
+    // GOLDEN: the second attempt succeeds → all 16 sectors recovered.
+    assert_eq!(
+        stats.bytes_good,
+        capacity_sectors as u64 * 2048,
+        "08_batch_fail bytes_good — second attempt should recover"
+    );
+    assert_eq!(stats.bytes_unreadable, 0, "08_batch_fail bytes_unreadable");
+    assert_eq!(stats.bytes_pending, 0, "08_batch_fail bytes_pending");
+    assert!(!pr.halted, "08_batch_fail halted");
+    assert!(
+        trace_len <= 80,
+        "08_batch_fail trace_len={trace_len} exceeds 80"
+    );
+}
+
+// ─────────────────────────────────────────────────────────────────────────
+//
+// Suppressed for now: NOT_READY-then-recover, HARDWARE_ERROR (wedge),
+// ILLEGAL_REQUEST (wedge), and ABORTED_COMMAND profiles. Each would
+// trigger long real-time sleeps inside `handle_read_failure`:
+//
+//   - NOT_READY (sense_key=0x02, asc=0x02/0x03/0x04): 15 s pause per
+//     occurrence (`patch_not_ready_pause`), and retries the same LBA
+//     in-place. Even one NOT_READY costs the test 15 s wall-time.
+//
+//   - HARDWARE_ERROR / ILLEGAL_REQUEST: 30 s per occurrence
+//     (`WEDGE_FAMILY_COOLDOWN_SECS`), bounded by
+//     `WEDGE_ABORT_THRESHOLD=16` before wedged-exit. Worst case ~8
+//     minutes per profile.
+//
+// The sleeps are not injectable. Adding them would require either a
+// `now()` / `sleep()` trait injection (out of scope for the unification
+// task) or a "test mode" compile-time flag (architectural smell). The
+// behavioural contracts for those paths are captured in
+// `read_error.rs`'s in-module tests instead — they exercise the
+// classifier without invoking the patch loop's sleep side-effects.
+//
+// If the unification ever proceeds, the next step is to add a clock
+// injection point in `handle_read_failure` and extend this fixture
+// with the wedge/NOT_READY profiles too.