keys: one resolved key set per rip, proven on first read

ResolvedKeySet resolves keys up front, forensic keys included, and every
read surface decrypts through it. A keyless session never muxes
ciphertext; a single declared key keys only the pieces it opens; held
keys are tried before side reads. Resolve reports whether the disc's
VID would help, returns its trace on refusal, and never codes an
unreachable key service as Missing.

Cargo.lock

@@ -332,6 +332,7 @@ dependencies = [
  "serde",
  "serde_json",
  "sha1",
+ "sha2",
  "socket2",
  "tempfile",
  "tracing",

Cargo.toml

@@ -40,6 +40,7 @@ rip = [
     "dep:serde",
     "dep:serde_json",
     "dep:sha1",
+    "dep:sha2",
     "dep:aes",
     "dep:freemkv-unlock",
     "dep:zip",
@@ -59,6 +60,8 @@ test-util = ["rip"]
 serde = { version = "1", features = ["derive"], optional = true }
 serde_json = { version = "1", optional = true }
 sha1 = { version = "0.11", optional = true }
+# SHA-256 for the key-set fingerprints (VID and legacy key identity, keys-upfront §4.1, J18).
+sha2 = { version = "0.11", optional = true }
 aes = { version = "0.9", optional = true }
 # Interim path dep for local cross-repo dev; `release-orchestrate.yml` re-pins
 # this to `{ git = ".../freemkv-unlock", tag = "vX.Y.Z" }` in the tagged commit

src/dirimage/mod.rs

@@ -307,4 +307,4 @@ impl SectorSource for DirImage {
 }
 
 #[cfg(test)]
-mod tests;
+pub(crate) mod tests;

src/dirimage/tests.rs

@@ -541,7 +541,7 @@ fn scan_image_enumerates_a_bdmv_folder() {
 
 /// A one-PlayItem MPLS long enough that `parse_playlist` keeps it (it drops
 /// anything under 30 s as a menu stub).
-fn one_item_mpls(clip_id: &[u8; 5]) -> Vec<u8> {
+pub(crate) fn one_item_mpls(clip_id: &[u8; 5]) -> Vec<u8> {
     let mut buf = Vec::new();
     buf.extend_from_slice(b"MPLS0200");
     buf.extend_from_slice(&40u32.to_be_bytes()); // playlist_start
@@ -582,7 +582,7 @@ fn one_item_mpls(clip_id: &[u8; 5]) -> Vec<u8> {
 
 /// A CLPI with only the fields `clpi::parse` needs: magic, zeroed section
 /// starts, and the source packet count at 56.
-fn minimal_clpi(source_packets: u32) -> Vec<u8> {
+pub(crate) fn minimal_clpi(source_packets: u32) -> Vec<u8> {
     let mut d = vec![0u8; 60];
     d[0..4].copy_from_slice(b"HDMV");
     d[4..8].copy_from_slice(b"0200");

src/disc/extract.rs

@@ -42,6 +42,11 @@ pub struct ExtractOptions<'a> {
     /// the in-flight file is left as `.partial` — never a half-written file
     /// that looks complete. `None` disables cancellation.
     pub halt: Option<crate::halt::Halt>,
+    /// The rip's up-front key set (KU §3.1), scope `WholeDisc`. For an AACS disc every file
+    /// is read through the set's reader: proven files by its map, the rest proven on
+    /// arrival, and a readable unit no held key opens stops the run (E7032). `None` keeps
+    /// the legacy disc-banked keys (until KU-X2).
+    pub keys: Option<&'a crate::keys::ResolvedKeySet>,
 }
 
 impl ExtractOptions<'_> {
@@ -190,11 +195,23 @@ impl Disc {
         // when a scrambled VOB group needs it. AACS / None discs keep the
         // disc-wide keys for every file.
         let mut base_keys = self.decrypt_keys();
+        // The rip's key set replaces the disc-banked keys and the count rule below (K-2).
+        let keyed = opts.keys.filter(|s| s.is_aacs());
+        if let Some(set) = keyed {
+            crate::keys::check_decryptable(
+                self,
+                false,
+                Some(set),
+                &crate::keys::KeyScope::WholeDisc,
+            )?;
+            base_keys = set.decrypt_keys();
+        }
 
         // AACS key map, by CPS-unit count. Single-CPS: one Unit Key opens
         // everything, so a blanket key-0 map covers even orphan clips. Multi-CPS
         // builds an exact per-title map instead (a blanket map would mis-decrypt).
         let key_map = match &base_keys {
+            _ if keyed.is_some() => None,
             DecryptKeys::Aacs { unit_keys, .. } if unit_keys.len() <= 1 => {
                 Some(std::sync::Arc::new(
                     crate::decrypt::AacsKeyMap::from_ranges(vec![(0, u32::MAX, 0)]),
@@ -220,7 +237,12 @@ impl Disc {
         // Phase 2: stream each file through the decrypting decorator, which owns a
         // borrowing wrapper so the caller keeps `reader`. Keys swap per CSS VTS
         // group via `set_keys`; AACS/None keep `base_keys` throughout.
-        let mut dec = DecryptingSectorSource::new(Borrowed(reader), base_keys.clone());
+        let mut dec = match keyed {
+            Some(set) => {
+                set.decrypting(Borrowed(reader), None, crate::keys::StopKind::Image, false)?
+            }
+            None => DecryptingSectorSource::new(Borrowed(reader), base_keys.clone()),
+        };
         if let Some(map) = key_map {
             dec = dec.with_key_map(map);
         }
@@ -396,6 +418,9 @@ impl SectorSource for Borrowed<'_> {
     fn unmapped_stream_files(&self) -> &[crate::sector::bus_removal::UnmappedStreamFile] {
         self.0.unmapped_stream_files()
     }
+    fn random_access(&self) -> bool {
+        self.0.random_access()
+    }
 }
 
 /// Whether `dir` lives on a case-INSENSITIVE filesystem (macOS APFS/HFS+ and
@@ -758,6 +783,8 @@ fn read_batch<S: SectorSource>(
             Ok(_) if attempt < READ_RETRIES => continue,
             Ok(_) => return Ok(false),
             Err(Error::Halted) => return Err(Error::Halted),
+            // The key set's loud stop (KU §2.4): never a hole, the run stops `.partial`.
+            Err(e @ (Error::WholeDiscKeyMissing | Error::NoDiscKey { .. })) => return Err(e),
             Err(Error::DecryptFailed) => return Ok(false),
             Err(_) if attempt < READ_RETRIES => continue,
             Err(_) => return Ok(false),

src/disc/mod.rs

@@ -664,6 +664,9 @@ impl SectorSource for RecoveryReads<'_> {
     fn unmapped_stream_files(&self) -> &[crate::sector::bus_removal::UnmappedStreamFile] {
         self.inner.unmapped_stream_files()
     }
+    fn random_access(&self) -> bool {
+        self.inner.random_access()
+    }
 }
 
 // A read fault a re-read may clear: not a dead transport, a gone source or an image's end.
@@ -3097,6 +3100,10 @@ impl SectorSource for FileReadAhead<'_> {
     fn unmapped_stream_files(&self) -> &[crate::sector::bus_removal::UnmappedStreamFile] {
         self.inner.unmapped_stream_files()
     }
+    fn random_access(&self) -> bool {
+        self.inner.random_access()
+    }
+
     fn read_sectors(
         &mut self,
         lba: u32,

src/keys/arrival.rs

@@ -0,0 +1,256 @@
+//! Proof on first read (KU §2.4): the decrypting readers prove a held key on a piece that
+//! `resolve` could not prove up front, the first time one of its encrypted units is read.
+//!
+//! - A partner is another readable, non-segment `Enc` unit of the same piece: first from the
+//!   current batch, then from at most one backward and one forward side read (`recovery =
+//!   false`, never retried, straight to the random-access inner source). A failed side read
+//!   only means "no partner there": it is swallowed, never classified, never surfaced.
+//! - A held key that opens the unit and a partner is `Proven`. With no readable partner, a
+//!   key that opens the unit is `Provisional`; the piece's next readable `Enc` unit confirms
+//!   it or stops the rip (a key conflict).
+//! - A readable unit is never reported as a read error or withheld. One no held key opens is
+//!   a loud stop (E7022 / E7032). FMTS segment units are never part of this proof.
+
+use super::{Proof, ProofCache, ResolvedKeySet, StopKind};
+use crate::aacs::content::{
+    ALIGNED_UNIT_LEN, aacs_unit_encrypted, clear_copy_permission_indicator, decrypt_unit, is_clean,
+};
+use crate::disc::ContentFormat;
+use crate::error::{Error, Result};
+use crate::sector::SectorSource;
+use crate::whole_disc::UNIT;
+
+/// Units a side read covers at most, each way (KU §2.4: "up to 32 units").
+const SIDE_UNITS: u64 = 32;
+
+/// The on-arrival proof a set's reader carries (KU §2.4). Holds the set's lazy and clear
+/// pieces, its base-key count, the FMTS segment ranges and the set's [`ProofCache`].
+pub(crate) struct Arrival {
+    // (start, end, anchor, piece index), sorted by start, disjoint.
+    spans: Vec<(u32, u32, u64, usize)>,
+    // (piece id, candidate slot).
+    pieces: Vec<(u32, Option<usize>)>,
+    base: usize,
+    segments: Vec<(u32, u32)>,
+    cache: ProofCache,
+    format: ContentFormat,
+    stop: StopKind,
+}
+
+impl Arrival {
+    pub(crate) fn new(set: &ResolvedKeySet, stop: StopKind) -> Self {
+        let i = &set.0;
+        let mut spans = Vec::new();
+        let mut pieces = Vec::with_capacity(i.arrival.len());
+        for (p, piece) in i.arrival.iter().enumerate() {
+            pieces.push((piece.id, piece.candidate));
+            for &(s, n, anchor) in &piece.spans {
+                spans.push((s, s.saturating_add(n), anchor, p));
+            }
+        }
+        spans.sort_unstable_by_key(|s| s.0);
+        let mut segments = i.segments.clone();
+        segments.sort_unstable();
+        Arrival {
+            spans,
+            pieces,
+            base: i.pool.len(),
+            segments,
+            cache: i.proofs.clone(),
+            format: i.content_format,
+            stop,
+        }
+    }
+
+    // The arrival span holding `lba`: (start, end, anchor, piece).
+    fn span_at(&self, lba: u32) -> Option<(u32, u32, u64, usize)> {
+        let i = self.spans.partition_point(|s| s.0 <= lba).checked_sub(1)?;
+        let s = self.spans[i];
+        (lba < s.1).then_some(s)
+    }
+
+    fn in_segment(&self, lba: u32) -> bool {
+        let i = self.segments.partition_point(|s| s.0 <= lba);
+        i > 0 && lba < self.segments[i - 1].1
+    }
+
+    fn opens(&self, unit: &[u8], key: &[u8; 16]) -> bool {
+        let mut u = unit.to_vec();
+        decrypt_unit(&mut u, key);
+        is_clean(&u, self.format)
+    }
+
+    // A non-segment encrypted unit of piece `piece` (a partner candidate).
+    fn is_partner(&self, lba: u32, unit: &[u8], piece: usize) -> bool {
+        unit.len() == ALIGNED_UNIT_LEN
+            && self.span_at(lba).is_some_and(|s| s.3 == piece)
+            && !self.in_segment(lba)
+            && aacs_unit_encrypted(unit, self.format)
+    }
+
+    fn stop(&self, lba: u32, why: &'static str) -> Error {
+        let err = self.stop.error();
+        tracing::error!(
+            target: "freemkv::keys",
+            lba,
+            code = err.code(),
+            why,
+            "a readable encrypted unit opens with no held key; stopping (output stays .partial)"
+        );
+        err
+    }
+
+    /// Prove and decrypt the lazy-piece units of `buf`, read at `lba` on the unit grid.
+    /// Keyed pieces are left to the key map; clear and segment units are untouched.
+    pub(crate) fn process(
+        &self,
+        inner: &mut dyn SectorSource,
+        lba: u32,
+        buf: &mut [u8],
+        unit_keys: &[(u32, [u8; 16])],
+    ) -> Result<()> {
+        if self.spans.is_empty() {
+            return Ok(());
+        }
+        let n_units = buf.len() / ALIGNED_UNIT_LEN;
+        for u in 0..n_units {
+            let at = lba.saturating_add(u as u32 * UNIT as u32);
+            let Some(span) = self.span_at(at) else {
+                continue;
+            };
+            let range = u * ALIGNED_UNIT_LEN..(u + 1) * ALIGNED_UNIT_LEN;
+            if self.in_segment(at) || !aacs_unit_encrypted(&buf[range.clone()], self.format) {
+                continue;
+            }
+            let (id, candidate) = self.pieces[span.3];
+            let slot = match self.cache.get(id) {
+                Some(Proof::Proven(s)) => s,
+                Some(Proof::Provisional(s)) => {
+                    // KU §2.4: the next readable unit confirms a provisional key or stops.
+                    if !self.opens(&buf[range.clone()], &unit_keys[s].1) {
+                        return Err(self.stop(at, "provisional key contradicted"));
+                    }
+                    self.cache.set(id, Proof::Proven(s));
+                    s
+                }
+                None => self.prove(inner, lba, buf, u, span, candidate, unit_keys)?,
+            };
+            self.decrypt_proven(&mut buf[range], &unit_keys[slot].1);
+        }
+        Ok(())
+    }
+
+    // Decrypt one unit with its proven key and mark it plaintext.
+    fn decrypt_proven(&self, unit: &mut [u8], key: &[u8; 16]) {
+        decrypt_unit(unit, key);
+        // KS-5 [BD] §3.10.2: CPI "shall be set to 00₂ if the data is not encrypted" — a unit
+        // freemkv decrypted, on-arrival proven included (KU §5.4).
+        clear_copy_permission_indicator(unit, self.format);
+    }
+
+    // Find the held key for unit `u` of `buf` (KU §2.4 steps 1–2) and record it.
+    #[allow(clippy::too_many_arguments)]
+    fn prove(
+        &self,
+        inner: &mut dyn SectorSource,
+        lba: u32,
+        buf: &[u8],
+        u: usize,
+        span: (u32, u32, u64, usize),
+        candidate: Option<usize>,
+        unit_keys: &[(u32, [u8; 16])],
+    ) -> Result<usize> {
+        let at = lba.saturating_add(u as u32 * UNIT as u32);
+        let unit = &buf[u * ALIGNED_UNIT_LEN..(u + 1) * ALIGNED_UNIT_LEN];
+        // The held keys that open U, the candidate first. None: stop now, no side read.
+        let openers: Vec<usize> = candidate
+            .into_iter()
+            .chain((0..self.base).filter(|&s| Some(s) != candidate))
+            .filter(|&s| self.opens(unit, &unit_keys[s].1))
+            .collect();
+        if openers.is_empty() {
+            return Err(self.stop(at, "no held key opens the unit"));
+        }
+        // Step 1: partners already in the batch, both ways; then the side reads.
+        let mut partners: Vec<Vec<u8>> = buf
+            .chunks(ALIGNED_UNIT_LEN)
+            .enumerate()
+            .filter(|&(j, c)| {
+                j != u && self.is_partner(lba.saturating_add(j as u32 * UNIT as u32), c, span.3)
+            })
+            .map(|(_, c)| c.to_vec())
+            .collect();
+        if partners.is_empty() {
+            partners = self.side_partners(inner, at, span);
+        }
+        let id = self.pieces[span.3].0;
+        if partners.is_empty() {
+            // No readable partner: provisional one-unit proof, false-accept ≈ pool × 1e-5.
+            tracing::debug!(target: "freemkv::keys", lba = at, "provisional one-unit proof");
+            self.cache.set(id, Proof::Provisional(openers[0]));
+            return Ok(openers[0]);
+        }
+        match openers
+            .into_iter()
+            .find(|&s| partners.iter().any(|p| self.opens(p, &unit_keys[s].1)))
+        {
+            Some(s) => {
+                self.cache.set(id, Proof::Proven(s));
+                Ok(s)
+            }
+            None => Err(self.stop(at, "no held key opens the unit and a partner")),
+        }
+    }
+
+    // Step 1.2: at most one backward and one forward side read inside U's span, each up to
+    // 32 units, `recovery = false`, never retried. A failure only means "no partner there".
+    fn side_partners(
+        &self,
+        inner: &mut dyn SectorSource,
+        at: u32,
+        (start, end, anchor, piece): (u32, u32, u64, usize),
+    ) -> Vec<Vec<u8>> {
+        // KS-1 [BD] §3.10.1: "encryption is applied to every Aligned Unit in the file"; units
+        // on the file's own grid (KS-2, KS-7 Informative: sectors of a unit are contiguous).
+        let first = crate::whole_disc::unit_head(start, anchor);
+        let last_end = anchor + (end as u64).saturating_sub(anchor) / UNIT * UNIT;
+        let at = at as u64;
+        let back = (first.max(at.saturating_sub(SIDE_UNITS * UNIT)), at);
+        let fwd = (at + UNIT, last_end.min(at + UNIT + SIDE_UNITS * UNIT));
+        for (from, to) in [back, fwd] {
+            if to <= from {
+                continue;
+            }
+            let found = self.side_read(inner, from, to, piece);
+            if !found.is_empty() {
+                return found;
+            }
+        }
+        Vec::new()
+    }
+
+    fn side_read(
+        &self,
+        inner: &mut dyn SectorSource,
+        from: u64,
+        to: u64,
+        piece: usize,
+    ) -> Vec<Vec<u8>> {
+        let (Ok(lba), Ok(count)) = (u32::try_from(from), u16::try_from(to - from)) else {
+            return Vec::new();
+        };
+        let mut side = vec![0u8; count as usize * 2048];
+        match inner.read_sectors(lba, count, &mut side, false) {
+            Ok(n) => side[..n.min(side.len())]
+                .chunks(ALIGNED_UNIT_LEN)
+                .enumerate()
+                .filter(|&(j, c)| self.is_partner(lba + j as u32 * UNIT as u32, c, piece))
+                .map(|(_, c)| c.to_vec())
+                .collect(),
+            Err(e) => {
+                tracing::debug!(target: "freemkv::keys", lba, error = %e, "side read found no partner");
+                Vec::new()
+            }
+        }
+    }
+}

src/keys/fmts.rs

@@ -0,0 +1,183 @@
+//! FMTS (AACS 2.1) forensic keys up front (KU §5). No public spec: the `IndividualSegment.tbl`
+//! / `.fmts` model and the index-1 anchor contract are evidence (KS-25, KS-26).
+
+use crate::aacs::content::ALIGNED_UNIT_LEN;
+use crate::aacs::segment::{Segment, clip_byte_to_lba, parse_individual_segments};
+use crate::decrypt::Phase;
+use crate::disc::{ContentFormat, Extent};
+use crate::error::{Error, Result};
+use crate::halt::Halt;
+use crate::sector::SectorSource;
+use std::collections::HashMap;
+
+/// Units per anchor or phase batch: the key service's sample floor.
+const BATCH_UNITS: usize = crate::keysource::MIN_SAMPLE_UNITS;
+/// Index-1 segments tried as the anchor, and same-index segments per phase probe (KU §2.3
+/// step 11: "≤ 2 phases × the readable index-1 segments (≤ 16)").
+const MAX_ANCHOR_ATTEMPTS: usize = 16;
+
+/// The disc's forensic layout: its segments, the forensic clip and each segment's sectors.
+pub(crate) struct Layout {
+    pub(crate) segments: Vec<Segment>,
+    pub(crate) clip: Vec<Extent>,
+    /// `[start, end)` sectors and index of each addressable segment.
+    pub(crate) ranges: Vec<(u32, u32, u16)>,
+    /// A segment that straddles a clip extent: no defensible sector range.
+    pub(crate) unresolved: bool,
+}
+
+/// Read the disc's forensic layout. `Ok(None)`: not FMTS (no or empty segment table).
+pub(crate) fn layout(
+    fs: &crate::udf::UdfFs,
+    reader: &mut dyn SectorSource,
+) -> Result<Option<Layout>> {
+    let tbl = match fs.read_file(reader, "/AACS/IndividualSegment.tbl") {
+        Ok(t) => t,
+        Err(Error::UdfNotFound { .. }) => return Ok(None),
+        Err(e) => return Err(e),
+    };
+    let Some(segments) = parse_individual_segments(&tbl).filter(|s| !s.is_empty()) else {
+        return Ok(None);
+    };
+    let Some(clip) = crate::mux::resolve::forensic_clip_extents(fs, reader)? else {
+        // No defensible anchor for the segment byte space: refuse rather than guess.
+        tracing::warn!(target: "freemkv::keys", "fmts: forensic clip not identifiable");
+        return Err(Error::FmtsKeyMissing);
+    };
+    let segments = crate::mux::resolve::filter_addressable_segments(segments, &clip);
+    let mut ranges = Vec::with_capacity(segments.len());
+    let mut unresolved = false;
+    for seg in &segments {
+        let (start, end) = (seg.start_spn as u64 * 192, (seg.end_spn as u64 + 1) * 192);
+        let (Some(a), Some(b)) = (
+            clip_byte_to_lba(&clip, start),
+            clip_byte_to_lba(&clip, end.saturating_sub(1)),
+        ) else {
+            unresolved = true;
+            continue;
+        };
+        if seg.start_spn <= seg.end_spn && b >= a && (b - a) as u64 == (end - 1 - start) / 2048 {
+            ranges.push((a, b + 1, seg.index));
+        } else {
+            unresolved = true;
+        }
+    }
+    Ok(Some(Layout {
+        segments,
+        clip,
+        ranges,
+        unresolved,
+    }))
+}
+
+/// What the index-1 anchor produced.
+pub(crate) enum Anchor {
+    /// The source's complete ordered forensic set (element i = index i + 1).
+    Keys(Vec<[u8; 16]>),
+    /// Every anchor batch read faulted: nothing to ask with (KU §5.4 Pending).
+    Pending,
+    /// Anchors were read and asked with, and no source had the set; the first source
+    /// failure, if any.
+    Missing(Option<Error>),
+}
+
+// Aligned unit `index` of `seg`: clip byte `start_spn * 192 + index * 6144`.
+fn read_unit(
+    reader: &mut dyn SectorSource,
+    clip: &[Extent],
+    seg: &Segment,
+    index: usize,
+) -> Option<Vec<u8>> {
+    let byte = seg.start_spn as u64 * 192 + index as u64 * ALIGNED_UNIT_LEN as u64;
+    let lba = clip_byte_to_lba(clip, byte)?;
+    let mut unit = vec![0u8; ALIGNED_UNIT_LEN];
+    match reader.read_sectors(lba, 3, &mut unit, false) {
+        Ok(n) if n == ALIGNED_UNIT_LEN => Some(unit),
+        _ => None,
+    }
+}
+
+/// Sends one anchor batch to the sources: `Ok(None)` for an empty answer.
+pub(crate) type AskFn<'a> = dyn FnMut(&[Vec<u8>]) -> Result<Option<Vec<[u8; 16]>>> + 'a;
+
+/// Anchor the forensic set from one index-1 batch per phase (KU §5.1; evidence KS-26: the
+/// decode server "returns the whole set only for an index-1 anchor"). `ask` sends one batch
+/// and returns `Ok(None)` for an empty answer.
+pub(crate) fn anchor(
+    reader: &mut dyn SectorSource,
+    layout: &Layout,
+    halt: Option<&Halt>,
+    ask: &mut AskFn,
+) -> Result<Anchor> {
+    let mut asked = false;
+    let mut failure: Option<Error> = None;
+    let anchors = layout.segments.iter().filter(|s| s.index == 1);
+    for seg in anchors.take(MAX_ANCHOR_ATTEMPTS) {
+        for phase in [0usize, 1] {
+            if halt.is_some_and(|h| h.is_cancelled()) {
+                return Err(Error::Halted);
+            }
+            let batch: Option<Vec<Vec<u8>>> = (0..BATCH_UNITS)
+                .map(|p| read_unit(reader, &layout.clip, seg, p * 2 + phase))
+                .collect();
+            let Some(batch) = batch else {
+                continue;
+            };
+            asked = true;
+            match ask(&batch) {
+                // KS-26 (evidence): a non-empty reply to an index-1 batch is the whole set.
+                Ok(Some(keys)) if !keys.is_empty() => return Ok(Anchor::Keys(keys)),
+                Ok(_) => {}
+                Err(Error::Halted) => return Err(Error::Halted),
+                Err(e) => {
+                    failure.get_or_insert(e);
+                }
+            }
+        }
+    }
+    Ok(if asked {
+        Anchor::Missing(failure)
+    } else {
+        Anchor::Pending
+    })
+}
+
+/// Each index's decrypt phase under its key (KU §5.1, §5.3). `Ok(None)`: an index whose
+/// units read but decrypted clean in neither parity (a wrong key). An index whose every
+/// probe faulted is `Phase::Verify`.
+pub(crate) fn phases(
+    reader: &mut dyn SectorSource,
+    layout: &Layout,
+    keys: &[[u8; 16]],
+    format: ContentFormat,
+    halt: Option<&Halt>,
+) -> Result<Option<HashMap<u16, Phase>>> {
+    use crate::mux::resolve::{IndexProbe, probe_index_phase};
+    let mut out = HashMap::new();
+    for (i, key) in keys.iter().enumerate() {
+        if halt.is_some_and(|h| h.is_cancelled()) {
+            return Err(Error::Halted);
+        }
+        let tag = (i + 1) as u16;
+        let probe = probe_index_phase(
+            &layout.segments,
+            tag,
+            BATCH_UNITS,
+            MAX_ANCHOR_ATTEMPTS,
+            format,
+            key,
+            |seg, unit| read_unit(reader, &layout.clip, seg, unit),
+        );
+        match probe {
+            IndexProbe::Phase(p) => {
+                out.insert(tag, p);
+            }
+            IndexProbe::WrongKey => return Ok(None),
+            // KS-25/KS-26 (evidence, no public spec): unknown phase → `Phase::Verify` (§5.3).
+            IndexProbe::ReadFault => {
+                out.insert(tag, Phase::Verify);
+            }
+        }
+    }
+    Ok(Some(out))
+}

src/keys/mod.rs

@@ -0,0 +1,784 @@
+//! Keys up front, memory only (keys-upfront design, KU).
+//!
+//! [`ResolvedKeySet::resolve`] is the one place that asks a [`crate::KeySource`] for keys.
+//! It runs once per rip, before any output, over the rip's [`KeyScope`]. It proves each
+//! held key against real ciphertext of each stream file (a *piece*), counting CPS units by
+//! the DECLARED `Num_of_CPS_Unit` (KS-14), never by the keys held. The result is an
+//! immutable, in-memory set: keys and the disc's Volume ID live in it for this rip only,
+//! behind a redacting `Debug`, with no byte or VID accessor and no `Serialize`.
+//!
+//! A piece that could not be proven up front is *Lazy*: its readers prove a held key on
+//! it the first time it is read (the on-arrival proof), looking at
+//! neighbouring units both ways, then falling back to a provisional one-unit proof. A
+//! readable unit is never withheld; one that no held key opens is a loud stop.
+
+mod arrival;
+mod fmts;
+mod resolve;
+#[cfg(test)]
+mod tests;
+
+pub(crate) use arrival::Arrival;
+
+use crate::aacs::trace::ResolutionTrace;
+use crate::decrypt::{AacsKeyMap, DecryptKeys};
+use crate::disc::{ContentFormat, Disc, DiscFormat};
+use crate::error::{Error, Result};
+use crate::halt::Halt;
+use crate::sector::{DecryptingSectorSource, SectorSource};
+use crate::session::KeySourceFactory;
+use crate::whole_disc::{UnitSpan, WholeDiscReader};
+use std::collections::HashMap;
+use std::sync::{Arc, Mutex};
+
+/// What a rip decrypts, and so what `resolve` must key (KU §2.5).
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub enum KeyScope {
+    /// Nothing is decrypted (a raw copy): no key-source call at all.
+    None,
+    /// These titles (indices into `Disc::titles`): an MKV/M2TS/MP4/demux rip.
+    Titles(Vec<usize>),
+    /// Every stream file: a decrypted ISO or folder.
+    WholeDisc,
+}
+
+/// Options for [`ResolvedKeySet::resolve`].
+#[derive(Default)]
+pub struct ResolveKeysOptions<'a> {
+    /// Stop token: checked before each piece, probe and request; interrupts retry waits.
+    pub halt: Option<&'a Halt>,
+    /// A set this rip already holds (e.g. from `info` or the scan): its keys join the pool
+    /// first, and its in-memory VID is used when the disc has none (KU §5.4).
+    pub seed: Option<&'a ResolvedKeySet>,
+    /// An in-memory VID the caller re-read from the drive (KU §4.2). Never written.
+    pub vid: Option<[u8; 16]>,
+    /// Set by `resolve` when no VID was in hand and one would help (KU J23): a source
+    /// reported a Media Key it could not finish without the VID (`KeyNode::NoVid`), or a
+    /// configured source [`uses_vid`](crate::KeySource::uses_vid). Never cleared.
+    pub vid_would_help: Option<&'a std::sync::atomic::AtomicBool>,
+    /// Filled with the per-source walk on `Ok` and on `Err` alike, so a refusal still says
+    /// which source answered what. It holds labels, node enums and counts, never key bytes.
+    pub trace: Option<&'a Mutex<ResolutionTrace>>,
+}
+
+/// The outcome of [`ResolvedKeySet::resolve`]: the set, plus the per-source walk.
+pub struct KeyResolution {
+    pub keys: ResolvedKeySet,
+    pub trace: ResolutionTrace,
+}
+
+/// Which held key opened a piece on arrival.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub(crate) enum Proof {
+    /// The key opened the piece's unit and a partner unit.
+    Proven(usize),
+    /// The key opened one unit with no readable partner; the next readable unit confirms it.
+    Provisional(usize),
+}
+
+/// The rip's in-memory record of on-arrival proofs (KU §2.4, J2, J16): which HELD key
+/// opened a piece. It never adds a key and is never written. The set owns one, shared by
+/// every clone of the set and handed to each reader the set builds, so later passes and
+/// the mux reuse a proof instead of side-reading again.
+#[derive(Clone, Default)]
+pub struct ProofCache(Arc<Mutex<HashMap<u32, Proof>>>);
+
+impl ProofCache {
+    /// Pieces with a recorded proof.
+    pub fn len(&self) -> usize {
+        self.lock().len()
+    }
+
+    /// No proof recorded yet.
+    pub fn is_empty(&self) -> bool {
+        self.lock().is_empty()
+    }
+
+    fn lock(&self) -> std::sync::MutexGuard<'_, HashMap<u32, Proof>> {
+        self.0.lock().unwrap_or_else(|e| e.into_inner())
+    }
+
+    pub(crate) fn get(&self, piece: u32) -> Option<Proof> {
+        self.lock().get(&piece).copied()
+    }
+
+    pub(crate) fn set(&self, piece: u32, proof: Proof) {
+        self.lock().insert(piece, proof);
+    }
+}
+
+impl std::fmt::Debug for ProofCache {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("ProofCache")
+            .field("pieces", &self.len())
+            .finish()
+    }
+}
+
+/// The forensic (AACS 2.1 FMTS) state of a set (KU §5).
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum ForensicState {
+    /// No forensic clip in scope.
+    None,
+    /// The index keys and each index's phase are held.
+    Resolved,
+    /// Every index-1 anchor read faulted: the forensic keys are asked for later, once, from
+    /// the recovered image (KU §5.4).
+    Pending,
+}
+
+/// A set's shape, for the front end's status line and the qa log (KU §7.6). No key bytes.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct KeySetStatus {
+    /// Distinct held keys proven on a piece.
+    pub proven: usize,
+    /// `Num_of_CPS_Unit` of the disc's title-key file (KS-14); `None` if unparseable.
+    pub declared: Option<usize>,
+    /// Pieces keyed by a proven key.
+    pub keyed: usize,
+    /// Pieces whose every probe read clear.
+    pub clear: usize,
+    /// Pieces left to the on-arrival proof.
+    pub lazy: usize,
+    pub forensic: ForensicState,
+    /// The `label()` of the source whose key keyed the first keyed piece (`"seed"` for a
+    /// key the seed set held).
+    pub origin: Option<&'static str>,
+    /// HD DVD: the single declared key was applied without proof (KU §2.6).
+    pub best_effort: bool,
+    /// Key-source requests `resolve` made (retries of one request count once).
+    pub requests: u32,
+}
+
+/// Whether a disc can be decrypted with a set (KU §3.5).
+#[derive(Debug)]
+pub enum DecryptStatus {
+    NotEncrypted,
+    Ready,
+    AacsKeysMissing(Error),
+    CssNotCracked(Error),
+    ForensicPending,
+}
+
+// Which loud stop the on-arrival proof raises: a title rip (E7022) or an image or folder
+// (E7032), KU §6.
+#[derive(Clone, Debug)]
+pub(crate) enum StopKind {
+    Title { disc_hash: String },
+    Image,
+}
+
+impl StopKind {
+    pub(crate) fn error(&self) -> Error {
+        match self {
+            StopKind::Title { disc_hash } => Error::NoDiscKey {
+                disc_hash: disc_hash.clone(),
+            },
+            StopKind::Image => Error::WholeDiscKeyMissing,
+        }
+    }
+}
+
+// A piece the readers prove on arrival: its id (first sector), its unit grid, and the key
+// to try first (KU §2.3 step 9.3, J21).
+#[derive(Clone, Debug)]
+pub(crate) struct ArrivalPiece {
+    pub(crate) id: u32,
+    pub(crate) spans: Vec<UnitSpan>,
+    pub(crate) candidate: Option<usize>,
+}
+
+// Everything a set holds. Key bytes and the VID never leave the crate.
+pub(crate) struct Inner {
+    pub(crate) aacs: bool,
+    pub(crate) disc_hash: String,
+    pub(crate) capacity: u32,
+    pub(crate) format: DiscFormat,
+    pub(crate) content_format: ContentFormat,
+    pub(crate) vid: Option<[u8; 16]>,
+    pub(crate) n_decl: Option<usize>,
+    pub(crate) scope: KeyScope,
+    // Base keys (pool slot = index); forensic index keys follow them in `decrypt_keys`.
+    pub(crate) pool: Vec<[u8; 16]>,
+    pub(crate) fmts_keys: Vec<[u8; 16]>,
+    pub(crate) fmts_phases: HashMap<u16, crate::decrypt::Phase>,
+    pub(crate) map: Arc<AacsKeyMap>,
+    pub(crate) spans: Vec<UnitSpan>,
+    pub(crate) arrival: Vec<ArrivalPiece>,
+    pub(crate) lazy: Vec<(u32, u32)>,
+    pub(crate) proven: Vec<usize>,
+    pub(crate) keyed: usize,
+    pub(crate) clear: usize,
+    pub(crate) forensic: ForensicState,
+    pub(crate) forensic_clip: Vec<(u32, u32)>,
+    pub(crate) segments: Vec<(u32, u32)>,
+    pub(crate) requests: u32,
+    pub(crate) best_effort: bool,
+    pub(crate) origin: Option<&'static str>,
+    pub(crate) proofs: ProofCache,
+}
+
+impl Inner {
+    fn empty() -> Self {
+        Inner {
+            aacs: false,
+            disc_hash: String::new(),
+            capacity: 0,
+            format: DiscFormat::BluRay,
+            content_format: ContentFormat::BdTs,
+            vid: None,
+            n_decl: None,
+            scope: KeyScope::None,
+            pool: Vec::new(),
+            fmts_keys: Vec::new(),
+            fmts_phases: HashMap::new(),
+            map: Arc::new(AacsKeyMap::from_ranges(Vec::new())),
+            spans: Vec::new(),
+            arrival: Vec::new(),
+            lazy: Vec::new(),
+            proven: Vec::new(),
+            keyed: 0,
+            clear: 0,
+            forensic: ForensicState::None,
+            forensic_clip: Vec::new(),
+            segments: Vec::new(),
+            requests: 0,
+            best_effort: false,
+            origin: None,
+            proofs: ProofCache::default(),
+        }
+    }
+}
+
+/// The rip's keys, resolved up front and held in memory only (KU §2.1).
+///
+/// Immutable and cheap to clone (an `Arc`); `Send + Sync`. Bound to one disc
+/// ([`is_for`](Self::is_for)) and one [`KeyScope`]. Holds no key-source factory, so nothing
+/// can ask a source after [`resolve`](Self::resolve) returns. No `Serialize`, no accessor
+/// for key bytes or the Volume ID, and a redacting `Debug`.
+#[derive(Clone)]
+pub struct ResolvedKeySet(pub(crate) Arc<Inner>);
+
+impl std::fmt::Debug for ResolvedKeySet {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        let i = &self.0;
+        f.debug_struct("ResolvedKeySet")
+            .field("aacs", &i.aacs)
+            .field("disc_hash", &i.disc_hash)
+            .field("scope", &i.scope)
+            .field("keys", &"<redacted>")
+            .field("keys_len", &i.pool.len())
+            .field("vid", &i.vid.map(|_| "<redacted>"))
+            .field("status", &self.status())
+            .finish()
+    }
+}
+
+fn norm_hash(h: &str) -> String {
+    crate::hex::strip_hex_prefix(h).to_ascii_lowercase()
+}
+
+fn overlaps(a: &[(u32, u32)], s: u32, e: u32) -> bool {
+    a.iter().any(|&(x, y)| x < e && s < y)
+}
+
+// SHA-256 over a domain tag and bytes (KU §4.1 fingerprints).
+fn sha256(tag: &[u8], bytes: &[u8]) -> [u8; 32] {
+    use sha2::Digest;
+    let mut h = sha2::Sha256::new();
+    h.update(tag);
+    h.update(bytes);
+    h.finalize().into()
+}
+
+impl ResolvedKeySet {
+    /// Resolve the keys `scope` needs, once, before any output (KU §2.3). The only code in
+    /// the product that asks a [`crate::KeySource`] for keys; `sources` is called once and
+    /// every source it builds is dropped before this returns (LK7).
+    ///
+    /// `Err` refuses the rip before any output: E7022 (a title piece is Missing), E7032 (an
+    /// image or folder piece is Missing), E7013 (a single declared key opens none of the
+    /// ciphertext), E7026 (forensic keys Missing), E7028–30 (a source failed), `Halted`.
+    pub fn resolve(
+        disc: &Disc,
+        reader: &mut dyn SectorSource,
+        scope: KeyScope,
+        sources: &KeySourceFactory,
+        opts: ResolveKeysOptions,
+    ) -> Result<KeyResolution> {
+        resolve::resolve(
+            disc,
+            reader,
+            scope,
+            sources,
+            opts,
+            &resolve::RealClock::new(),
+        )
+    }
+
+    /// A set holding no keys: for a raw copy, or a disc with no AACS.
+    pub fn none() -> ResolvedKeySet {
+        ResolvedKeySet(Arc::new(Inner::empty()))
+    }
+
+    /// Whether this set keys AACS content: `false` for [`none`](Self::none), which a caller
+    /// must not take as covering an AACS disc's titles ([`covers`](Self::covers) has no disc).
+    pub fn is_aacs(&self) -> bool {
+        self.0.aacs
+    }
+
+    /// Whether this set was resolved for `disc`: the same disc hash, capacity and format,
+    /// and the same VID fingerprint when both are known. [`none`](Self::none) holds nothing
+    /// and fits any disc.
+    pub fn is_for(&self, disc: &Disc) -> bool {
+        let i = &self.0;
+        if !i.aacs {
+            return true;
+        }
+        let Some(aacs) = disc.aacs.as_ref() else {
+            return false;
+        };
+        if norm_hash(&aacs.disc_hash) != norm_hash(&i.disc_hash)
+            || disc.format != i.format
+            || (disc.capacity_sectors != 0
+                && i.capacity != 0
+                && disc.capacity_sectors != i.capacity)
+        {
+            return false;
+        }
+        match (i.vid, aacs.volume_id) {
+            (Some(held), vid) if vid != [0u8; 16] => held == vid,
+            _ => true,
+        }
+    }
+
+    /// Whether the set was resolved over `scope`: structural coverage only. It says nothing
+    /// about whether an AACS title is keyed: a set with no AACS key ([`none`](Self::none))
+    /// covers every scope. Use the engine's disc-aware check, or [`is_aacs`](Self::is_aacs).
+    pub fn covers(&self, scope: &KeyScope) -> bool {
+        if !self.0.aacs {
+            return true;
+        }
+        match (&self.0.scope, scope) {
+            (_, KeyScope::None) => true,
+            (KeyScope::WholeDisc, _) => true,
+            (KeyScope::Titles(have), KeyScope::Titles(want)) => {
+                want.iter().all(|t| have.contains(t))
+            }
+            _ => false,
+        }
+    }
+
+    /// Counts and states for the front end (KU §7.6). No key bytes.
+    pub fn status(&self) -> KeySetStatus {
+        let i = &self.0;
+        KeySetStatus {
+            proven: i.proven.len(),
+            declared: i.n_decl,
+            keyed: i.keyed,
+            clear: i.clear,
+            lazy: i.arrival.len() - i.clear.min(i.arrival.len()),
+            forensic: i.forensic,
+            origin: i.origin,
+            best_effort: i.best_effort,
+            requests: i.requests,
+        }
+    }
+
+    /// `SHA-256("freemkv-vid-fp-v1" ‖ VID)`: the only form of the VID that may be written
+    /// (KU §4.1). `None` when the set holds no VID.
+    pub fn vid_fingerprint(&self) -> Option<[u8; 32]> {
+        self.0.vid.map(|v| sha256(b"freemkv-vid-fp-v1", &v))
+    }
+
+    /// `SHA-256("freemkv-key-fp-v1" ‖ key)[..8]` of each proven base key: the legacy
+    /// mapfile identity (KU §4.4). Empty when no base key was proven.
+    pub fn proven_key_fingerprints(&self) -> Vec<[u8; 8]> {
+        self.0
+            .proven
+            .iter()
+            .filter_map(|&s| self.0.pool.get(s))
+            .map(|k| {
+                let d = sha256(b"freemkv-key-fp-v1", k);
+                let mut fp = [0u8; 8];
+                fp.copy_from_slice(&d[..8]);
+                fp
+            })
+            .collect()
+    }
+
+    /// `[start, end)` sector ranges left to the on-arrival proof (KU §2.4).
+    pub fn lazy(&self) -> &[(u32, u32)] {
+        &self.0.lazy
+    }
+
+    /// Every index-1 forensic anchor read faulted (KU §5.4): a decrypting single-pass rip
+    /// or disc→ISO copy must refuse (E7026); a raw capture asks once from its image later.
+    pub fn forensic_pending(&self) -> bool {
+        self.0.forensic == ForensicState::Pending
+    }
+
+    /// Key-source requests `resolve` made.
+    pub fn source_requests(&self) -> u32 {
+        self.0.requests
+    }
+
+    /// The rip's on-arrival proof record, shared by every clone of this set (J16).
+    pub fn proof_cache(&self) -> &ProofCache {
+        &self.0.proofs
+    }
+
+    // The decrypt keys the readers use: base keys (slot = index) then forensic index keys.
+    pub(crate) fn decrypt_keys(&self) -> DecryptKeys {
+        let i = &self.0;
+        let mut unit_keys: Vec<(u32, [u8; 16])> = i
+            .pool
+            .iter()
+            .enumerate()
+            .map(|(s, k)| (s as u32 + 1, *k))
+            .collect();
+        for (j, k) in i.fmts_keys.iter().enumerate() {
+            let tag = crate::mux::resolve::FMTS_POOL_TAG_BASE.saturating_add(j as u32);
+            unit_keys.push((tag, *k));
+        }
+        DecryptKeys::Aacs {
+            unit_keys,
+            format: i.content_format,
+        }
+    }
+
+    pub(crate) fn key_map(&self) -> Arc<AacsKeyMap> {
+        self.0.map.clone()
+    }
+
+    // Whether every sector of `extents` lies in a piece this set resolved.
+    pub(crate) fn covers_extents(&self, extents: &[crate::disc::Extent]) -> bool {
+        let mut spans: Vec<(u64, u64)> = self
+            .0
+            .spans
+            .iter()
+            .map(|&(s, n, _)| (s as u64, s as u64 + n as u64))
+            .collect();
+        spans.sort_unstable();
+        extents.iter().all(|e| {
+            let (mut at, end) = (
+                e.start_lba as u64,
+                e.start_lba as u64 + e.sector_count as u64,
+            );
+            for &(s, t) in &spans {
+                if s <= at && at < t {
+                    at = t;
+                }
+            }
+            at >= end
+        })
+    }
+
+    // The E7013 caller-bug refusal (KU §6): a set used on the wrong disc, outside its scope,
+    // or over a source that cannot seek.
+    fn caller_bug(what: &'static str) -> Error {
+        tracing::error!(target: "freemkv::keys", what, "key set used incorrectly (caller bug)");
+        debug_assert!(false, "key set used incorrectly: {what}");
+        Error::DecryptFailed
+    }
+
+    // The reader checks (KU §2.4, §5.4): refuse a non-random-access inner source and, unless
+    // `allow_pending`, content touching a forensic clip whose keys are Pending (E7026).
+    pub(crate) fn gate(
+        &self,
+        random_access: bool,
+        extents: Option<&[(u32, u32)]>,
+        allow_pending: bool,
+    ) -> Result<()> {
+        if !random_access {
+            // KU §2.4: side reads need random access; `Prefetched(Decrypting(..))`, never
+            // the other way round (SG27).
+            return Err(Self::caller_bug("decrypting reader over a prefetcher"));
+        }
+        if self.forensic_pending() && self.touches_clip(extents) && !allow_pending {
+            tracing::error!(
+                target: "freemkv::keys",
+                code = crate::error::E_FMTS_KEY_MISSING,
+                "forensic keys pending (every anchor unreadable): a decrypting single pass \
+                 cannot key the forensic segments; copy raw, then convert"
+            );
+            return Err(Error::FmtsKeyMissing);
+        }
+        Ok(())
+    }
+
+    fn touches_clip(&self, extents: Option<&[(u32, u32)]>) -> bool {
+        let clip = &self.0.forensic_clip;
+        match extents {
+            None => !clip.is_empty(),
+            Some(ex) => ex.iter().any(|&(s, e)| overlaps(clip, s, e)),
+        }
+    }
+
+    // The on-arrival proof for this set's readers, if any piece needs one (KU §2.4).
+    pub(crate) fn arrival(&self, stop: StopKind) -> Option<Arrival> {
+        (!self.0.arrival.is_empty() && !self.0.best_effort).then(|| Arrival::new(self, stop))
+    }
+
+    // The E7022 stop for a title read of this set's disc.
+    pub(crate) fn title_stop(&self) -> StopKind {
+        StopKind::Title {
+            disc_hash: crate::hex::strip_hex_prefix(&self.0.disc_hash).to_string(),
+        }
+    }
+
+    pub(crate) fn capacity(&self) -> u32 {
+        self.0.capacity
+    }
+
+    pub(crate) fn content_format(&self) -> ContentFormat {
+        self.0.content_format
+    }
+
+    // The decrypting view over `inner` for content touching `extents` (`None`: the whole
+    // disc), after [`gate`](Self::gate).
+    pub(crate) fn decrypting<S: SectorSource>(
+        &self,
+        inner: S,
+        extents: Option<&[(u32, u32)]>,
+        stop: StopKind,
+        allow_pending: bool,
+    ) -> Result<DecryptingSectorSource<S>> {
+        let i = &self.0;
+        self.gate(inner.random_access(), extents, allow_pending)?;
+        let mut dec =
+            DecryptingSectorSource::new(inner, self.decrypt_keys()).with_key_map(self.key_map());
+        if self.forensic_pending() && self.touches_clip(extents) {
+            // Pending segments stay ciphertext (never a stop, KU §2.4): outside content.
+            let spans: Vec<(u32, u32)> = i.spans.iter().map(|&(s, n, _)| (s, n)).collect();
+            let content: Vec<(u32, u32)> = crate::whole_disc::subtract_ranges(&spans, &i.segments)
+                .into_iter()
+                .map(|(s, e)| (s, e - s))
+                .collect();
+            dec = dec.with_content_ranges(Arc::from(crate::whole_disc::merge_ranges(content)));
+        }
+        if let Some(a) = self.arrival(stop) {
+            dec = dec.with_arrival(a);
+        }
+        Ok(dec)
+    }
+
+    // A set holding no key over `title` (KU §3.1: "`keys` must be `Some` for AACS"): every
+    // extent is left to the on-arrival proof, which has no key to try, so the first
+    // AACS-flagged unit stops with E7022 and clear units pass. For a mux given no AACS set.
+    pub(crate) fn keyless_for(title: &crate::disc::DiscTitle, format: ContentFormat) -> Self {
+        let mut i = Inner::empty();
+        i.aacs = true;
+        i.content_format = format;
+        i.scope = KeyScope::WholeDisc;
+        // Arrival spans must be disjoint: overlapping extents (a clip played twice, or one
+        // inside another) merge into one span. Adjacent clips keep their own unit grids.
+        let mut ranges: Vec<(u32, u32)> = title
+            .extents
+            .iter()
+            .filter(|e| e.sector_count > 0)
+            .map(|e| (e.start_lba, e.start_lba.saturating_add(e.sector_count)))
+            .collect();
+        ranges.sort_unstable();
+        let mut merged: Vec<(u32, u32)> = Vec::with_capacity(ranges.len());
+        for (s, e) in ranges {
+            match merged.last_mut() {
+                Some(last) if s < last.1 => last.1 = last.1.max(e),
+                _ => merged.push((s, e)),
+            }
+        }
+        for (s, e) in merged {
+            let span = (s, e - s, s as u64);
+            i.spans.push(span);
+            i.arrival.push(ArrivalPiece {
+                id: s,
+                spans: vec![span],
+                candidate: None,
+            });
+        }
+        ResolvedKeySet(Arc::new(i))
+    }
+
+    // `keyless_for` title `idx` of `disc`, carrying the disc's identity: its E7022 names the
+    // disc, and it passes `is_for` on that disc (a Session mux given no set, over no key).
+    pub(crate) fn keyless_for_disc(disc: &Disc, idx: usize) -> Option<Self> {
+        let title = disc.titles.get(idx)?;
+        let mut set = Self::keyless_for(title, disc.content_format);
+        let i = Arc::get_mut(&mut set.0).expect("a fresh set has one owner");
+        i.disc_hash = disc
+            .aacs
+            .as_ref()
+            .map(|a| a.disc_hash.clone())
+            .unwrap_or_default();
+        i.format = disc.format;
+        i.capacity = disc.capacity_sectors;
+        Some(set)
+    }
+
+    // A set keying `ranges` (`[start, end)`) with `key` as proven pieces: the mux tests'
+    // stand-in for a set `resolve` built over a disc they do not lay out.
+    #[cfg(test)]
+    pub(crate) fn keyed_for_test(disc: &Disc, key: [u8; 16], ranges: &[(u32, u32)]) -> Self {
+        let mut i = Inner::empty();
+        i.aacs = true;
+        i.disc_hash = disc
+            .aacs
+            .as_ref()
+            .map(|a| a.disc_hash.clone())
+            .unwrap_or_default();
+        i.capacity = disc.capacity_sectors;
+        i.format = disc.format;
+        i.content_format = disc.content_format;
+        i.scope = KeyScope::WholeDisc;
+        i.pool = vec![key];
+        i.proven = vec![0];
+        i.keyed = ranges.len();
+        i.spans = ranges.iter().map(|&(s, e)| (s, e - s, s as u64)).collect();
+        i.map = Arc::new(AacsKeyMap::from_ranges(
+            ranges.iter().map(|&(s, e)| (s, e, 0)).collect(),
+        ));
+        ResolvedKeySet(Arc::new(i))
+    }
+
+    /// The decrypting reader for title `idx` of `disc` over the raw, random-access `inner`:
+    /// keyed pieces through the set's map, the rest proven on arrival (a readable unit no
+    /// held key opens stops with E7022). E7013 if the set is not for `disc`, does not cover
+    /// the title, or `inner` cannot seek; E7026 if the title needs Pending forensic keys.
+    pub fn title_reader<S: SectorSource>(
+        &self,
+        disc: &Disc,
+        idx: usize,
+        inner: S,
+    ) -> Result<DecryptingSectorSource<S>> {
+        if !self.is_for(disc) {
+            return Err(Self::caller_bug("key set is not for this disc"));
+        }
+        let title = disc.titles.get(idx).ok_or(Error::DiscTitleRange {
+            index: idx,
+            count: disc.titles.len(),
+        })?;
+        if !self.0.aacs {
+            if !inner.random_access() {
+                return Err(Self::caller_bug("decrypting reader over a prefetcher"));
+            }
+            return Ok(DecryptingSectorSource::new(inner, disc.decrypt_keys()));
+        }
+        if !self.covers(&KeyScope::Titles(vec![idx])) {
+            return Err(Self::caller_bug("title outside the key set's scope"));
+        }
+        let extents: Vec<(u32, u32)> = title
+            .extents
+            .iter()
+            .map(|e| (e.start_lba, e.start_lba.saturating_add(e.sector_count)))
+            .collect();
+        let stop = StopKind::Title {
+            disc_hash: disc.aacs_disc_hash(),
+        };
+        self.decrypting(inner, Some(&extents), stop, false)
+    }
+
+    /// The whole-disc decrypting reader for a decrypted image or folder copy over the raw,
+    /// random-access `inner`, on each stream file's own unit grid. A readable unit no held
+    /// key opens stops with E7032. E7013 if the set is not for `disc`, does not cover the
+    /// whole disc, or `inner` cannot seek; E7026 if forensic keys are Pending.
+    pub fn whole_disc_reader<S: SectorSource>(
+        &self,
+        disc: &Disc,
+        inner: S,
+        halt: Option<&Halt>,
+    ) -> Result<WholeDiscReader<S>> {
+        if halt.is_some_and(|h| h.is_cancelled()) {
+            return Err(Error::Halted);
+        }
+        if !self.is_for(disc) {
+            return Err(Self::caller_bug("key set is not for this disc"));
+        }
+        let mut content = disc.encrypted_content_ranges();
+        if !self.0.aacs {
+            if !inner.random_access() {
+                return Err(Self::caller_bug("decrypting reader over a prefetcher"));
+            }
+            let mut dec = DecryptingSectorSource::new(inner, disc.decrypt_keys());
+            if !content.is_empty() {
+                dec = dec.with_content_ranges(Arc::from(content));
+            }
+            return Ok(crate::whole_disc::UnitAligned::new(dec, Vec::new()));
+        }
+        if !self.covers(&KeyScope::WholeDisc) {
+            return Err(Self::caller_bug("whole disc outside the key set's scope"));
+        }
+        let mut dec = self.decrypting(inner, None, StopKind::Image, false)?;
+        content.extend(self.0.spans.iter().map(|&(s, n, _)| (s, n)));
+        let content = crate::whole_disc::merge_ranges(content);
+        if !content.is_empty() {
+            dec = dec.with_content_ranges(Arc::from(content));
+        }
+        Ok(crate::whole_disc::UnitAligned::new(
+            dec,
+            self.0.spans.clone(),
+        ))
+    }
+}
+
+/// Whether `disc` can be decrypted with `keys` (KU §3.5). CSS is read from `disc.css` /
+/// `css_error` exactly as [`Disc::ensure_decryptable_keys`] does. With `keys == None` the
+/// legacy disc-banked AACS keys decide (the library fallback until KU-X2).
+pub fn decrypt_status(disc: &Disc, keys: Option<&ResolvedKeySet>) -> DecryptStatus {
+    if disc.css_error.is_some() {
+        return DecryptStatus::CssNotCracked(Error::CssNoDiscKey);
+    }
+    if disc.css.is_some() {
+        return DecryptStatus::Ready;
+    }
+    if disc.aacs.is_none() && !disc.encrypted {
+        return DecryptStatus::NotEncrypted;
+    }
+    match keys {
+        Some(set) if set.is_aacs() => {
+            if !set.is_for(disc) {
+                DecryptStatus::AacsKeysMissing(Error::DecryptFailed)
+            } else if set.forensic_pending() {
+                DecryptStatus::ForensicPending
+            } else {
+                DecryptStatus::Ready
+            }
+        }
+        _ => match disc.ensure_decryptable_keys(false, &disc.decrypt_keys()) {
+            Ok(()) => DecryptStatus::Ready,
+            Err(e) => DecryptStatus::AacsKeysMissing(e),
+        },
+    }
+}
+
+/// The pre-flight decrypt gate over a set (KU §3.5): `Ok` when `scope` can be decrypted
+/// from `keys`, else the typed refusal, before any output. `raw` always passes. CSS is read
+/// from `disc.css` / `css_error` as [`Disc::ensure_decryptable_keys`] does; `keys == None`
+/// falls back to the legacy disc-banked AACS keys (until KU-X2).
+pub fn check_decryptable(
+    disc: &Disc,
+    raw: bool,
+    keys: Option<&ResolvedKeySet>,
+    scope: &KeyScope,
+) -> Result<()> {
+    if raw || matches!(scope, KeyScope::None) {
+        return Ok(());
+    }
+    match keys {
+        Some(set) if set.is_aacs() && disc.css.is_none() && disc.css_error.is_none() => {
+            if !set.is_for(disc) {
+                return Err(ResolvedKeySet::caller_bug("key set is not for this disc"));
+            }
+            if !set.covers(scope) {
+                return Err(ResolvedKeySet::caller_bug("scope outside the key set"));
+            }
+            if set.forensic_pending() {
+                return Err(Error::FmtsKeyMissing);
+            }
+            Ok(())
+        }
+        Some(_) if disc.aacs.is_some() && disc.css.is_none() => {
+            disc.ensure_decryptable_keys(false, &DecryptKeys::None)
+        }
+        _ => disc.ensure_decryptable(false),
+    }
+}

src/keys/resolve.rs

@@ -0,0 +1,986 @@
+//! [`ResolvedKeySet::resolve`] (KU §2.3): pieces, probes, sources, verdicts.
+
+use super::{
+    ArrivalPiece, ForensicState, Inner, KeyResolution, KeyScope, ProofCache, ResolveKeysOptions,
+    ResolvedKeySet,
+};
+use crate::aacs::content::{ALIGNED_UNIT_LEN, aacs_unit_encrypted, decrypt_unit, is_clean};
+use crate::aacs::trace::{KeyNode, KeyOutcome, KeyStep, ResolutionTrace};
+use crate::decrypt::{AacsKeyMap, Phase};
+use crate::disc::{ContentFormat, Disc, DiscFormat, Extent};
+use crate::error::{Error, Result};
+use crate::halt::Halt;
+use crate::keysource::{DiscInputs, DiscInputsCtx, KeySource, MIN_SAMPLE_UNITS};
+use crate::sector::SectorSource;
+use crate::session::KeySourceFactory;
+use crate::whole_disc::{UNIT, UnitSpan, probe_units, subtract_ranges, unit_head};
+use std::collections::HashMap;
+use std::sync::Arc;
+use std::time::{Duration, Instant};
+
+/// First wait before re-asking a source that gave no answer (J13).
+const RETRY_FIRST: Duration = Duration::from_secs(1);
+/// The retry wait doubles up to this (J13).
+const RETRY_CAP: Duration = Duration::from_secs(8);
+/// A source is given up after this long with no answer (J13; Stop T31). A stall window, not
+/// a total: any answer ends the retry at once.
+const NO_ANSWER_WINDOW: Duration = Duration::from_secs(60);
+/// Encrypted units sent in one per-piece request, at most.
+const SAMPLE_CAP: usize = 32;
+
+/// Time for the J13 retry: injected so tests use a fake clock and never sleep.
+pub(crate) trait Clock {
+    fn now(&self) -> Duration;
+    /// Wait `d`, returning `Err(Halted)` as soon as `halt` is cancelled.
+    fn sleep(&self, d: Duration, halt: Option<&Halt>) -> Result<()>;
+}
+
+/// The wall clock; waits are halt-aware ([`Halt::wait`]).
+pub(crate) struct RealClock(Instant);
+
+impl RealClock {
+    pub(crate) fn new() -> Self {
+        RealClock(Instant::now())
+    }
+}
+
+impl Clock for RealClock {
+    fn now(&self) -> Duration {
+        self.0.elapsed()
+    }
+    fn sleep(&self, d: Duration, halt: Option<&Halt>) -> Result<()> {
+        match halt {
+            Some(h) => h.wait(d),
+            None => {
+                std::thread::sleep(d);
+                Ok(())
+            }
+        }
+    }
+}
+
+/// A piece's verdict (KU §2.3 step 9). `Ask` is transient: resolved before `resolve` ends.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+enum Verdict {
+    Keyed(usize),
+    Clear,
+    Lazy(Option<usize>),
+    Missing,
+    Ask,
+}
+
+// One Clip AV stream file (or an uncovered title extent) in scope, with its probes.
+struct Piece {
+    spans: Vec<UnitSpan>,
+    titles: Vec<usize>,
+    rank: u64,
+    units: u64,
+    enc: Vec<Vec<u8>>,
+    faults: usize,
+    verdict: Verdict,
+}
+
+impl Piece {
+    fn id(&self) -> u32 {
+        self.spans[0].0
+    }
+
+    fn ranges(&self) -> impl Iterator<Item = (u32, u32)> + '_ {
+        self.spans.iter().map(|&(s, n, _)| (s, s.saturating_add(n)))
+    }
+
+    // Unit heads on the piece's grid, as (first head, unit count) per span. A unit that
+    // would cross its span's end is not a unit of the piece (KS-7, Informative).
+    fn grid(&self) -> Vec<(u64, u64)> {
+        self.spans
+            .iter()
+            .map(|&(s, n, anchor)| {
+                let head = unit_head(s, anchor);
+                let end = s as u64 + n as u64;
+                (head, end.saturating_sub(head) / UNIT)
+            })
+            .collect()
+    }
+}
+
+// What one request to one source produced.
+enum Asked {
+    Keys(Vec<[u8; 16]>),
+    Empty,
+    /// The source failed; the error is the run's `first_failure` if it came first.
+    Failed,
+    Skipped,
+}
+
+// The state of one `resolve` run.
+struct Run<'a> {
+    halt: Option<&'a Halt>,
+    clock: &'a dyn Clock,
+    format: ContentFormat,
+    sources: Vec<Box<dyn KeySource>>,
+    dead: Vec<bool>,
+    inputs: DiscInputs,
+    pool: Vec<[u8; 16]>,
+    origin: Vec<&'static str>,
+    requests: u32,
+    trace: ResolutionTrace,
+    // A source matched the disc and held a Media Key but had no VID (KU J23).
+    km_needs_vid: bool,
+    // The run's first source failure: a Missing refusal is this failure when there was one,
+    // whichever piece's request failed (review B-1, KU-E1b); never E7022, so never E7034.
+    first_failure: Option<Error>,
+}
+
+impl Run<'_> {
+    fn check_halt(&self) -> Result<()> {
+        match self.halt {
+            Some(h) => h.check(),
+            None => Ok(()),
+        }
+    }
+
+    // Add keys to the pool (deduplicated): the pool only grows during `resolve`.
+    fn add_keys(&mut self, keys: &[[u8; 16]], who: &'static str) {
+        for k in keys {
+            if !self.pool.contains(k) {
+                self.pool.push(*k);
+                self.origin.push(who);
+            }
+        }
+    }
+
+    fn opens(&self, unit: &[u8], slot: usize) -> bool {
+        let mut u = unit.to_vec();
+        decrypt_unit(&mut u, &self.pool[slot]);
+        is_clean(&u, self.format)
+    }
+
+    // One request to source `i` with `samples`, retried only while it gives no answer (J13,
+    // J15): `last_failure_was_transport()`, backoff 1 s doubling to 8 s, given up after 60 s
+    // with no answer. Any answer ends it; a given-up or answered-with-failure source is dead.
+    fn ask(&mut self, i: usize, samples: &[Vec<u8>], forensic: bool) -> Result<Asked> {
+        if self.dead[i] {
+            return Ok(Asked::Skipped);
+        }
+        self.requests += 1;
+        let mut inputs = self.inputs.clone();
+        inputs.samples = samples.to_vec();
+        let ctx = DiscInputsCtx::new(&inputs);
+        let start = self.clock.now();
+        let mut wait = RETRY_FIRST;
+        let mut attempted = false;
+        loop {
+            if let Err(stop) = self.check_halt() {
+                if attempted {
+                    self.unanswered(i);
+                }
+                return Err(stop);
+            }
+            attempted = true;
+            let src = &self.sources[i];
+            let who = src.label().to_string();
+            let answer = if forensic {
+                src.get_fmts_indexes(&ctx).map(|k| (k, None))
+            } else {
+                src.resolve_unit_keys(&ctx).map(|r| {
+                    let info = (r.matched, r.matched_entry, r.store_entries, r.miss_path);
+                    (r.keys, Some(info))
+                })
+            };
+            match answer {
+                Ok((keys, info)) => {
+                    let (matched, entry, store, miss) =
+                        info.unwrap_or((false, None, None, Vec::new()));
+                    self.km_needs_vid |= miss.contains(&KeyNode::NoVid);
+                    let (path, outcome) = match (keys.is_empty(), matched) {
+                        (false, _) => (vec![KeyNode::FoundUnitKeys], KeyOutcome::Resolved),
+                        // The source's own reason (e.g. `NoVid`: `MissingVid`), else a bare no-key.
+                        (true, true) => {
+                            let outcome = if miss.contains(&KeyNode::NoVid) {
+                                KeyOutcome::MissingVid
+                            } else {
+                                KeyOutcome::NoKey
+                            };
+                            let why = if miss.is_empty() {
+                                vec![KeyNode::NoDerivableKey]
+                            } else {
+                                miss
+                            };
+                            ([vec![KeyNode::MatchedDisc], why].concat(), outcome)
+                        }
+                        (true, false) => (vec![KeyNode::NoEntry], KeyOutcome::NoKey),
+                    };
+                    self.trace.keys.push(KeyStep {
+                        who,
+                        path,
+                        outcome,
+                        matched_entry: entry,
+                        store_entries: store,
+                    });
+                    return Ok(if keys.is_empty() {
+                        Asked::Empty
+                    } else {
+                        Asked::Keys(keys.into_iter().map(|u| u.key).collect())
+                    });
+                }
+                Err(Error::Halted) => {
+                    // A source stopped mid-request: the request went out (minor 1).
+                    self.unanswered(i);
+                    return Err(Error::Halted);
+                }
+                Err(e) => {
+                    let waited = self.clock.now().saturating_sub(start);
+                    if src.last_failure_was_transport() && waited < NO_ANSWER_WINDOW {
+                        tracing::info!(
+                            target: "freemkv::keys",
+                            who,
+                            waited_secs = waited.as_secs(),
+                            "key source gave no answer; retrying"
+                        );
+                        let pause = wait.min(NO_ANSWER_WINDOW - waited);
+                        if let Err(stop) = self.clock.sleep(pause, self.halt) {
+                            self.unanswered(i);
+                            return Err(stop);
+                        }
+                        wait = (wait * 2).min(RETRY_CAP);
+                        continue;
+                    }
+                    self.dead[i] = true;
+                    self.trace.keys.push(KeyStep {
+                        who,
+                        path: Vec::new(),
+                        outcome: KeyOutcome::NoKey,
+                        matched_entry: None,
+                        store_entries: None,
+                    });
+                    self.first_failure.get_or_insert(e);
+                    return Ok(Asked::Failed);
+                }
+            }
+        }
+    }
+
+    // A request to source `i` went out and a Stop ended its retries with no answer: the
+    // trace keeps it, so a caller counting requests sees one was made.
+    fn unanswered(&mut self, i: usize) {
+        self.trace.keys.push(KeyStep {
+            who: self.sources[i].label().to_string(),
+            path: Vec::new(),
+            outcome: KeyOutcome::NoKey,
+            matched_entry: None,
+            store_entries: None,
+        });
+    }
+
+    // KU §2.3 step 9, rules 1 and 3–5, against the current pool. `Ask` means rule 2 applies:
+    // an `Enc` probe no held key opens.
+    fn judge(&self, p: &Piece) -> Verdict {
+        if p.units == 0 {
+            return Verdict::Clear;
+        }
+        let mut count = vec![0usize; self.pool.len()];
+        let mut first_opened = None;
+        let mut unopened = false;
+        for u in &p.enc {
+            match (0..self.pool.len()).find(|&s| self.opens(u, s)) {
+                Some(s) => {
+                    count[s] += 1;
+                    first_opened.get_or_insert(s);
+                }
+                None => unopened = true,
+            }
+        }
+        let best = (0..count.len()).max_by_key(|&s| (count[s], std::cmp::Reverse(s)));
+        match best {
+            // Rule 1: two probes, or the one probe of a one-unit piece.
+            Some(s) if count[s] >= 2 || (p.units == 1 && count[s] == 1) => Verdict::Keyed(s),
+            _ if unopened => Verdict::Ask,
+            // Rule 3 (J21): one probe opened, never keyed from it.
+            _ if p.enc.len() == 1 => Verdict::Lazy(first_opened),
+            // Rule 4: every probe read clear.
+            _ if p.enc.is_empty() && p.faults == 0 => Verdict::Clear,
+            _ => Verdict::Lazy(first_opened),
+        }
+    }
+}
+
+fn sorted_ranges(mut v: Vec<(u32, u32)>) -> Vec<(u32, u32)> {
+    v.sort_unstable();
+    let mut out: Vec<(u32, u32)> = Vec::with_capacity(v.len());
+    for (s, e) in v {
+        match out.last_mut() {
+            Some(last) if s <= last.1 => last.1 = last.1.max(e),
+            _ => out.push((s, e)),
+        }
+    }
+    out
+}
+
+fn overlaps(a: &[(u32, u32)], s: u32, e: u32) -> bool {
+    a.iter().any(|&(x, y)| x < e && s < y)
+}
+
+// The pieces of `scope`: each content file the scope touches (on its own unit grid, KS-1),
+// minus sectors an earlier file claimed (an SSIF re-lists m2ts sectors), then title extents
+// no file covers (anchored at the extent start).
+fn pieces(disc: &Disc, files: &[Vec<(u32, u32)>], sel: &[usize], whole: bool) -> Vec<Piece> {
+    let title_ranges: Vec<(usize, u32, u32)> = sel
+        .iter()
+        .flat_map(|&t| {
+            disc.titles[t]
+                .extents
+                .iter()
+                .filter(|e| e.sector_count > 0)
+                .map(move |e| (t, e.start_lba, e.start_lba.saturating_add(e.sector_count)))
+        })
+        .collect();
+    let touches = |ranges: &[(u32, u32)]| -> Vec<usize> {
+        let mut t: Vec<usize> = title_ranges
+            .iter()
+            .filter(|&&(_, s, e)| overlaps(ranges, s, e))
+            .map(|&(t, _, _)| t)
+            .collect();
+        t.sort_unstable();
+        t.dedup();
+        t
+    };
+    let mut claimed: Vec<(u32, u32)> = Vec::new();
+    let mut out = Vec::new();
+    let push = |spans: Vec<UnitSpan>, out: &mut Vec<Piece>| {
+        let ranges: Vec<(u32, u32)> = spans.iter().map(|&(s, n, _)| (s, s + n)).collect();
+        let titles = touches(&ranges);
+        let rank = titles
+            .iter()
+            .map(|&t| disc.titles[t].size_bytes)
+            .max()
+            .unwrap_or(0);
+        out.push(Piece {
+            spans,
+            titles,
+            rank,
+            units: 0,
+            enc: Vec::new(),
+            faults: 0,
+            verdict: Verdict::Ask,
+        });
+    };
+    for file in files {
+        let ends: Vec<(u32, u32)> = file.iter().map(|&(s, n)| (s, s + n)).collect();
+        if !whole && touches(&ends).is_empty() {
+            continue;
+        }
+        let mut spans = Vec::new();
+        let mut off = 0u64;
+        for &(lba, n) in file {
+            let anchor = (lba as u64).saturating_sub(off % UNIT);
+            for (s, e) in subtract_ranges(&[(lba, n)], &claimed) {
+                spans.push((s, e - s, anchor));
+            }
+            off += n as u64;
+        }
+        claimed = sorted_ranges(claimed.into_iter().chain(ends).collect());
+        if !spans.is_empty() {
+            push(spans, &mut out);
+        }
+    }
+    for &(_, s, e) in &title_ranges {
+        let left = subtract_ranges(&[(s, e - s)], &claimed);
+        if left.is_empty() {
+            continue;
+        }
+        let spans = left.iter().map(|&(a, b)| (a, b - a, s as u64)).collect();
+        claimed = sorted_ranges(claimed.into_iter().chain(left).collect());
+        push(spans, &mut out);
+    }
+    out.sort_by_key(|p| p.id());
+    out
+}
+
+fn in_segment(segments: &[(u32, u32)], lba: u32) -> bool {
+    let i = segments.partition_point(|s| s.0 <= lba);
+    i > 0 && lba < segments[i - 1].1
+}
+
+// KU §2.3 step 7: up to 32 units per piece on its grid, each `Enc`, `Clr` or `Fault`.
+// FMTS segment units are not base probes (KU §2.3 step 6).
+fn probe(
+    reader: &mut dyn SectorSource,
+    p: &mut Piece,
+    segments: &[(u32, u32)],
+    format: ContentFormat,
+    halt: Option<&Halt>,
+) -> Result<()> {
+    let grid = p.grid();
+    p.units = grid.iter().map(|g| g.1).sum();
+    let mut buf = vec![0u8; ALIGNED_UNIT_LEN];
+    for idx in probe_units(p.units) {
+        if halt.is_some_and(|h| h.is_cancelled()) {
+            return Err(Error::Halted);
+        }
+        let mut k = idx;
+        let Some(&(head, _)) = grid.iter().find(|g| {
+            let hit = k < g.1;
+            if !hit {
+                k -= g.1;
+            }
+            hit
+        }) else {
+            continue;
+        };
+        let Ok(lba) = u32::try_from(head + k * UNIT) else {
+            continue;
+        };
+        if in_segment(segments, lba) {
+            continue;
+        }
+        match reader.read_sectors(lba, UNIT as u16, &mut buf, false) {
+            Ok(n) if n == buf.len() => {
+                if aacs_unit_encrypted(&buf, format) {
+                    p.enc.push(buf.clone());
+                }
+            }
+            _ => p.faults += 1,
+        }
+    }
+    Ok(())
+}
+
+fn missing_error(scope: &KeyScope, disc: &Disc) -> Error {
+    match scope {
+        KeyScope::WholeDisc => Error::WholeDiscKeyMissing,
+        _ => Error::NoDiscKey {
+            disc_hash: disc.aacs_disc_hash(),
+        },
+    }
+}
+
+/// The body of [`ResolvedKeySet::resolve`] with an injected clock.
+pub(crate) fn resolve(
+    disc: &Disc,
+    reader: &mut dyn SectorSource,
+    scope: KeyScope,
+    sources: &KeySourceFactory,
+    opts: ResolveKeysOptions,
+    clock: &dyn Clock,
+) -> Result<KeyResolution> {
+    let halt = opts.halt;
+    if let Some(h) = halt {
+        h.check()?;
+    }
+    // Step 1: not AACS, or nothing decrypted: no source call.
+    let (Some(aacs), false) = (disc.aacs.as_ref(), scope == KeyScope::None) else {
+        return Ok(KeyResolution {
+            keys: ResolvedKeySet::none(),
+            trace: ResolutionTrace::new(),
+        });
+    };
+    let sel: Vec<usize> = match &scope {
+        KeyScope::Titles(v) => {
+            if let Some(&bad) = v.iter().find(|&&t| t >= disc.titles.len()) {
+                return Err(Error::DiscTitleRange {
+                    index: bad,
+                    count: disc.titles.len(),
+                });
+            }
+            v.clone()
+        }
+        _ => (0..disc.titles.len()).collect(),
+    };
+    // Step 3: KS-14 [BD] §3.9.3 "Num_of_CPS_Unit field (16 bits) indicates the number of CPS
+    // Units on the disc" — the declared count; `None` (unparseable) fails closed.
+    let n_decl = disc.declared_cps_units();
+    let seed = opts.seed.filter(|s| s.is_aacs() && s.is_for(disc));
+    // Step 4: the VID, in memory only.
+    let vid = Some(aacs.volume_id)
+        .filter(|v| *v != [0u8; 16])
+        .or(opts.vid)
+        .or(seed.and_then(|s| s.0.vid));
+    let mut inputs = disc.inputs().unwrap_or_else(|| DiscInputs {
+        disc_hash: aacs.disc_hash.clone(),
+        volume_id: [0u8; 16],
+        version: aacs.version,
+        mkb: Vec::new(),
+        unit_key_ro: Vec::new(),
+        samples: Vec::new(),
+        volume_label: None,
+    });
+    inputs.volume_id = vid.unwrap_or([0u8; 16]);
+    let built = sources();
+    let n_src = built.len();
+    let vid_consumer = built.iter().any(|s| s.uses_vid());
+    let mut run = Run {
+        halt,
+        clock,
+        format: disc.content_format,
+        sources: built,
+        dead: vec![false; n_src],
+        inputs,
+        pool: Vec::new(),
+        origin: Vec::new(),
+        requests: 0,
+        trace: ResolutionTrace::new(),
+        km_needs_vid: false,
+        first_failure: None,
+    };
+    if let Some(s) = seed {
+        run.add_keys(&s.0.pool, "seed");
+    }
+    let result = if disc.format == DiscFormat::HdDvd {
+        resolve_hddvd(disc, reader, &scope, &sel, n_decl, &mut run)
+    } else {
+        resolve_bd(disc, reader, &scope, &sel, n_decl, seed, &mut run)
+    };
+    // Every source is dropped here: nothing can ask after `resolve` (LK7).
+    let trace = std::mem::take(&mut run.trace);
+    drop(run.sources);
+    if let Some(out) = opts.trace {
+        *out.lock().unwrap_or_else(|e| e.into_inner()) = trace.clone();
+    }
+    // KU J23: the VID would help only through a Km path or a source that consumes it.
+    if let Some(flag) = opts.vid_would_help
+        && vid.is_none()
+        && (run.km_needs_vid || vid_consumer)
+    {
+        flag.store(true, std::sync::atomic::Ordering::SeqCst);
+    }
+    let mut inner = result?;
+    inner.disc_hash = aacs.disc_hash.clone();
+    inner.capacity = disc.capacity_sectors;
+    inner.format = disc.format;
+    inner.content_format = disc.content_format;
+    inner.vid = vid;
+    inner.n_decl = n_decl;
+    inner.scope = scope;
+    inner.requests = run.requests;
+    tracing::info!(
+        target: "freemkv::keys",
+        requests = inner.requests,
+        keyed = inner.keyed,
+        clear = inner.clear,
+        lazy = inner.arrival.len().saturating_sub(inner.clear),
+        forensic = ?inner.forensic,
+        declared = ?n_decl,
+        "keys: resolved up front"
+    );
+    Ok(KeyResolution {
+        keys: ResolvedKeySet(Arc::new(inner)),
+        trace,
+    })
+}
+
+// A fresh `Inner` for an AACS set, filled by the caller with the disc identity.
+fn aacs_inner(run: &Run) -> Inner {
+    let mut inner = Inner::empty();
+    inner.aacs = true;
+    inner.pool = run.pool.clone();
+    inner.proofs = ProofCache::default();
+    inner
+}
+
+// KU §2.6: HD DVD is best effort. Not probed (its encrypted flag is unverified, KS-27):
+// a single declared key is applied to every piece; several, or none parseable, refuse.
+fn resolve_hddvd(
+    disc: &Disc,
+    reader: &mut dyn SectorSource,
+    scope: &KeyScope,
+    sel: &[usize],
+    n_decl: Option<usize>,
+    run: &mut Run,
+) -> Result<Inner> {
+    if n_decl != Some(1) {
+        tracing::error!(target: "freemkv::keys", declared = ?n_decl, "multi-key HD DVD cannot be matched reliably");
+        return Err(Error::NoDiscKey {
+            disc_hash: disc.aacs_disc_hash(),
+        });
+    }
+    let files = crate::whole_disc::content_files(reader).unwrap_or_default();
+    let ps = pieces(disc, &files, sel, *scope == KeyScope::WholeDisc);
+    if run.pool.is_empty() {
+        let samples = disc.content_samples(reader, MIN_SAMPLE_UNITS);
+        for i in 0..run.sources.len() {
+            if let Asked::Keys(k) = run.ask(i, &samples, false)? {
+                let who = run.sources[i].label();
+                run.add_keys(&k[..1], who);
+                break;
+            }
+        }
+        if run.pool.is_empty() {
+            let failure = run.first_failure.take();
+            return Err(failure.unwrap_or_else(|| missing_error(scope, disc)));
+        }
+    }
+    let mut inner = aacs_inner(run);
+    inner.best_effort = true;
+    inner.origin = run.origin.first().copied();
+    inner.proven = vec![0];
+    inner.keyed = ps.len();
+    let mut ranges = Vec::new();
+    for p in &ps {
+        ranges.extend(p.ranges().map(|(s, e)| (s, e, 0usize)));
+        inner.spans.extend(p.spans.iter().copied());
+    }
+    inner.map = Arc::new(AacsKeyMap::from_ranges(ranges));
+    Ok(inner)
+}
+
+// KU §2.3 steps 5–14 for BD/UHD content.
+fn resolve_bd(
+    disc: &Disc,
+    reader: &mut dyn SectorSource,
+    scope: &KeyScope,
+    sel: &[usize],
+    n_decl: Option<usize>,
+    seed: Option<&ResolvedKeySet>,
+    run: &mut Run,
+) -> Result<Inner> {
+    let format = disc.content_format;
+    // Step 5: pieces. An unreadable filesystem fails a whole-disc copy (it would otherwise
+    // ship ciphertext); a title rip keys its title extents instead.
+    let fs = match crate::udf::read_filesystem(reader) {
+        Ok(fs) => Some(fs),
+        Err(e) if *scope == KeyScope::WholeDisc => return Err(e),
+        Err(e) => {
+            tracing::warn!(target: "freemkv::keys", error = %e, "no filesystem: keying title extents");
+            None
+        }
+    };
+    let files = match &fs {
+        Some(fs) => match crate::whole_disc::content_files_in(fs, reader) {
+            Ok(f) => f,
+            Err(e) if *scope == KeyScope::WholeDisc => return Err(e),
+            Err(_) => Vec::new(),
+        },
+        None => Vec::new(),
+    };
+    let mut ps = pieces(disc, &files, sel, *scope == KeyScope::WholeDisc);
+    // Step 6: FMTS, when the forensic clip is in scope.
+    let layout = match &fs {
+        Some(fs) => super::fmts::layout(fs, reader)?,
+        None => None,
+    };
+    let clip: Vec<(u32, u32)> = layout
+        .as_ref()
+        .map(|l| {
+            l.clip
+                .iter()
+                .map(|e| (e.start_lba, e.start_lba.saturating_add(e.sector_count)))
+                .collect()
+        })
+        .unwrap_or_default();
+    let layout = layout.filter(|_| {
+        ps.iter()
+            .any(|p| p.ranges().any(|(s, e)| overlaps(&clip, s, e)))
+    });
+    let segments: Vec<(u32, u32)> = layout
+        .as_ref()
+        .map(|l| sorted_ranges(l.ranges.iter().map(|&(s, e, _)| (s, e)).collect()))
+        .unwrap_or_default();
+    // Step 7: probes.
+    for p in &mut ps {
+        probe(reader, p, &segments, format, run.halt)?;
+    }
+    // Step 8: sample-independent sources, once, with the main title's samples.
+    let needs_keys = ps
+        .iter()
+        .any(|p| p.units > 0 && (!p.enc.is_empty() || p.faults > 0));
+    if needs_keys {
+        let main = disc.main_title().map(|t| {
+            t.extents
+                .iter()
+                .map(|e| (e.start_lba, e.start_lba.saturating_add(e.sector_count)))
+                .collect::<Vec<_>>()
+        });
+        let mut samples: Vec<Vec<u8>> = Vec::new();
+        for p in ps.iter().filter(|p| {
+            main.as_ref()
+                .is_none_or(|m| p.ranges().any(|(s, e)| overlaps(m, s, e)))
+        }) {
+            samples.extend(p.enc.iter().take(MIN_SAMPLE_UNITS).cloned());
+        }
+        samples.truncate(SAMPLE_CAP);
+        for i in 0..run.sources.len() {
+            if run.sources[i].answer_depends_on_samples() {
+                continue;
+            }
+            if let Asked::Keys(k) = run.ask(i, &samples, false)? {
+                let who = run.sources[i].label();
+                run.add_keys(&k, who);
+            }
+        }
+    }
+    // Step 9: verdicts from the pool, then rule 2 for the rest, largest title first.
+    for p in &mut ps {
+        p.verdict = run.judge(p);
+    }
+    let mut order: Vec<usize> = (0..ps.len())
+        .filter(|&i| ps[i].verdict == Verdict::Ask)
+        .collect();
+    order.sort_by_key(|&i| (std::cmp::Reverse(ps[i].rank), ps[i].id()));
+    // Step 11: sample-dependent requests ≤ n_decl (or the number of pieces).
+    let mut budget = n_decl.unwrap_or(ps.len());
+    for &i in &order {
+        run.check_halt()?;
+        let v = run.judge(&ps[i]);
+        if v != Verdict::Ask {
+            ps[i].verdict = v;
+            continue;
+        }
+        let samples = borrow_samples(run, &ps, i);
+        if samples.len() < MIN_SAMPLE_UNITS {
+            // Step 9.2: too little ciphertext to ask with: Lazy, proven on arrival.
+            ps[i].verdict = Verdict::Lazy(None);
+            continue;
+        }
+        let dependent: Vec<usize> = (0..run.sources.len())
+            .filter(|&s| run.sources[s].answer_depends_on_samples())
+            .collect();
+        if budget == 0 || dependent.is_empty() {
+            ps[i].verdict = Verdict::Missing;
+            continue;
+        }
+        budget -= 1;
+        for s in dependent {
+            if let Asked::Keys(k) = run.ask(s, &samples, false)? {
+                let who = run.sources[s].label();
+                run.add_keys(&k, who);
+                if run.judge(&ps[i]) != Verdict::Ask {
+                    break;
+                }
+            }
+        }
+        // Proven on this piece's own units: ≥ 2 → Keyed; an unopened unit → Missing. A piece
+        // with a single own `Enc` unit that the answer opens stays Lazy (J21).
+        ps[i].verdict = match run.judge(&ps[i]) {
+            Verdict::Ask => Verdict::Missing,
+            v => v,
+        };
+    }
+    apply_single_unit_rule(run, &mut ps, n_decl)?;
+    if let Some(i) = ps.iter().position(|p| p.verdict == Verdict::Missing) {
+        let lba = ps[i].id();
+        let failure = run.first_failure.take();
+        let err = failure.unwrap_or_else(|| missing_error(scope, disc));
+        tracing::error!(target: "freemkv::keys", lba, code = err.code(), "a stream file in scope has no key; refusing before any output");
+        return Err(err);
+    }
+    // KU §2.7: an empty pool on an encrypted scope is today's keyless case.
+    if run.pool.is_empty() && ps.iter().any(|p| matches!(p.verdict, Verdict::Lazy(_))) {
+        let failure = run.first_failure.take();
+        return Err(failure.unwrap_or_else(|| missing_error(scope, disc)));
+    }
+    let mut inner = aacs_inner(run);
+    // Step 6 (continued): forensic keys, reused from the seed or anchored once.
+    let forensic = match &layout {
+        None => None,
+        Some(l) => Some(resolve_forensic(reader, l, seed, run, format)?),
+    };
+    build(&mut inner, &ps, layout.as_ref(), forensic, run, clip);
+    Ok(inner)
+}
+
+// Step 9.2: this piece's unopened `Enc` units first, topped up from other unopened pieces
+// of the same playlist, never across playlists (SG23).
+fn borrow_samples(run: &Run, ps: &[Piece], i: usize) -> Vec<Vec<u8>> {
+    // KS-10 [BD] §3.9.2: "All AV stream files that are referred to by one Title are included
+    // in the same CPS Unit" — so only pieces sharing a title may lend units.
+    let unopened = |p: &Piece| -> Vec<Vec<u8>> {
+        p.enc
+            .iter()
+            .filter(|u| !(0..run.pool.len()).any(|s| run.opens(u, s)))
+            .cloned()
+            .collect()
+    };
+    let mut out = unopened(&ps[i]);
+    for (j, q) in ps.iter().enumerate() {
+        if out.len() >= SAMPLE_CAP {
+            break;
+        }
+        if j == i || q.verdict != Verdict::Ask || !q.titles.iter().any(|t| ps[i].titles.contains(t))
+        {
+            continue;
+        }
+        out.extend(unopened(q));
+    }
+    out.truncate(SAMPLE_CAP);
+    out
+}
+
+// Step 10: the `n_decl == 1` rule.
+fn apply_single_unit_rule(run: &Run, ps: &mut [Piece], n_decl: Option<usize>) -> Result<()> {
+    // KS-14 [BD] §3.9.3: "Num_of_CPS_Unit field (16 bits) indicates the number of CPS Units
+    // on the disc" — with one declared, a key proven on one piece keys them all.
+    if n_decl != Some(1) {
+        return Ok(());
+    }
+    if let Some(slot) = ps.iter().find_map(|p| match p.verdict {
+        Verdict::Keyed(s) => Some(s),
+        _ => None,
+    }) {
+        for p in ps.iter_mut() {
+            // J22 ("never guessed"; J21 parity): keyed only on step 9's evidence, the key
+            // opening every Enc probe, two or more (or one of a one-unit piece); else Lazy.
+            let opens_all = !p.enc.is_empty() && p.enc.iter().all(|u| run.opens(u, slot));
+            let enough = p.enc.len() >= 2 || p.units == 1;
+            match p.verdict {
+                Verdict::Lazy(_) if opens_all && enough => p.verdict = Verdict::Keyed(slot),
+                Verdict::Lazy(_) if opens_all || p.enc.is_empty() => {
+                    p.verdict = Verdict::Lazy(Some(slot));
+                }
+                _ => {}
+            }
+        }
+        return Ok(());
+    }
+    let any_enc = ps.iter().any(|p| !p.enc.is_empty());
+    let any_opened = ps
+        .iter()
+        .flat_map(|p| &p.enc)
+        .any(|u| (0..run.pool.len()).any(|s| run.opens(u, s)));
+    if any_enc && !run.pool.is_empty() && !any_opened {
+        tracing::error!(
+            target: "freemkv::keys",
+            code = crate::error::E_DECRYPT_FAILED,
+            "one CPS unit declared and the held key opens none of the ciphertext: wrong key"
+        );
+        return Err(Error::DecryptFailed);
+    }
+    Ok(())
+}
+
+// The forensic outcome: resolved keys and phases, or Pending.
+enum Forensic {
+    Resolved(Vec<[u8; 16]>, HashMap<u16, Phase>),
+    Pending,
+}
+
+fn resolve_forensic(
+    reader: &mut dyn SectorSource,
+    layout: &super::fmts::Layout,
+    seed: Option<&ResolvedKeySet>,
+    run: &mut Run,
+    format: ContentFormat,
+) -> Result<Forensic> {
+    if layout.unresolved {
+        return Err(Error::FmtsKeyMissing);
+    }
+    if let Some(s) = seed.filter(|s| s.0.forensic == ForensicState::Resolved) {
+        return Ok(Forensic::Resolved(
+            s.0.fmts_keys.clone(),
+            s.0.fmts_phases.clone(),
+        ));
+    }
+    let halt = run.halt;
+    // A failed request is kept as the run's first failure, not returned per batch.
+    let mut ask = |batch: &[Vec<u8>]| -> Result<Option<Vec<[u8; 16]>>> {
+        for i in 0..run.sources.len() {
+            if let Asked::Keys(k) = run.ask(i, batch, true)? {
+                return Ok(Some(k));
+            }
+        }
+        Ok(None)
+    };
+    let keys = match super::fmts::anchor(reader, layout, halt, &mut ask)? {
+        super::fmts::Anchor::Keys(k) => k,
+        super::fmts::Anchor::Pending => {
+            tracing::warn!(target: "freemkv::keys", "fmts: every index-1 anchor read faulted: forensic keys pending");
+            return Ok(Forensic::Pending);
+        }
+        super::fmts::Anchor::Missing(failure) => {
+            let failure = failure.or(run.first_failure.take());
+            return Err(failure.unwrap_or(Error::FmtsKeyMissing));
+        }
+    };
+    // Every segment must map to a held index key; a hole would fall to the base key.
+    if layout
+        .ranges
+        .iter()
+        .any(|&(_, _, idx)| idx == 0 || idx as usize > keys.len())
+    {
+        return Err(Error::FmtsKeyMissing);
+    }
+    match super::fmts::phases(reader, layout, &keys, format, halt)? {
+        Some(p) => Ok(Forensic::Resolved(keys, p)),
+        None => Err(Error::FmtsKeyMissing),
+    }
+}
+
+// Step 14: maps from Keyed pieces (and the forensic ranges); arrival pieces for the rest.
+fn build(
+    inner: &mut Inner,
+    ps: &[Piece],
+    layout: Option<&super::fmts::Layout>,
+    forensic: Option<Forensic>,
+    run: &Run,
+    clip: Vec<(u32, u32)>,
+) {
+    let base = run.pool.len();
+    let mut ranges: Vec<(u32, u32, usize, Phase)> = Vec::new();
+    let mut seg_ranges: Vec<(u32, u32, usize, Phase)> = Vec::new();
+    if let (Some(l), Some(f)) = (layout, &forensic) {
+        inner.forensic_clip = clip.clone();
+        inner.segments = sorted_ranges(l.ranges.iter().map(|&(s, e, _)| (s, e)).collect());
+        match f {
+            Forensic::Pending => inner.forensic = ForensicState::Pending,
+            Forensic::Resolved(keys, phases) => {
+                inner.forensic = ForensicState::Resolved;
+                inner.fmts_keys = keys.clone();
+                inner.fmts_phases = phases.clone();
+                for &(s, e, idx) in &l.ranges {
+                    let slot = base + (idx as usize).saturating_sub(1);
+                    if (idx as usize) <= keys.len() && idx > 0 {
+                        let phase = phases.get(&idx).copied().unwrap_or(Phase::Verify);
+                        seg_ranges.push((s, e, slot, phase));
+                    }
+                }
+            }
+        }
+    }
+    let mut proven = Vec::new();
+    for p in ps {
+        inner.spans.extend(p.spans.iter().copied());
+        match p.verdict {
+            Verdict::Keyed(slot) => {
+                inner.keyed += 1;
+                if !proven.contains(&slot) {
+                    proven.push(slot);
+                }
+                inner.origin = inner.origin.or(run.origin.get(slot).copied());
+                let extents: Vec<Extent> = p
+                    .ranges()
+                    .map(|(s, e)| Extent {
+                        start_lba: s,
+                        sector_count: e - s,
+                    })
+                    .collect();
+                // Base content around the forensic segments keeps the piece's own key (K-3).
+                ranges.extend(crate::mux::resolve::fill_base_key_gaps(
+                    &extents,
+                    &seg_ranges,
+                    slot,
+                ));
+            }
+            Verdict::Clear | Verdict::Lazy(_) => {
+                if p.verdict == Verdict::Clear {
+                    inner.clear += 1;
+                } else {
+                    inner.lazy.extend(p.ranges());
+                }
+                inner.arrival.push(ArrivalPiece {
+                    id: p.id(),
+                    spans: p.spans.clone(),
+                    candidate: match p.verdict {
+                        Verdict::Lazy(c) => c,
+                        _ => None,
+                    },
+                });
+            }
+            Verdict::Missing | Verdict::Ask => {}
+        }
+    }
+    inner.lazy = sorted_ranges(std::mem::take(&mut inner.lazy));
+    inner.spans.sort_unstable_by_key(|s| s.0);
+    inner.proven = proven;
+    ranges.extend(seg_ranges);
+    ranges.sort_unstable_by_key(|r| r.0);
+    inner.map = Arc::new(AacsKeyMap::from_ranges_phased(ranges));
+}

src/keysource.rs

@@ -236,7 +236,8 @@ pub struct UnitKeyResolution {
     /// When `matched` and `keys` is empty: the derivation nodes walked AFTER the
     /// implicit `MatchedDisc` node — e.g. `[NoVid]` (had a Media Key, no VID) or
     /// `[NoDerivableKey]` (no material at all). Empty ⇒ the caller supplies a
-    /// bare `NoDerivableKey`. Ignored unless `matched` and `keys` is empty.
+    /// bare `NoDerivableKey`. Rendered only when `matched` and `keys` is empty; a `NoVid`
+    /// beside partial keys still tells `resolve` the VID would help (KU J23).
     pub miss_path: Vec<crate::aacs::trace::KeyNode>,
     /// When `matched`: a booleans-and-lengths shape of the matched entry (no key
     /// material), for the application to log. `None` for a miss or a source kind
@@ -304,6 +305,28 @@ pub trait KeySource {
     fn label(&self) -> &'static str {
         "source"
     }
+
+    /// Whether this source's answer depends on the content samples it is sent. `true` (the
+    /// default): `ResolvedKeySet::resolve` asks it once per unopened piece, with that piece's
+    /// samples. `false` (a keydb keyed by disc hash): asked once per resolve (KU §2.3 step 8).
+    fn answer_depends_on_samples(&self) -> bool {
+        true
+    }
+
+    /// Whether this source's most recent failed request got NO answer at all (DNS, connect or
+    /// idle stall: transport class). Only then does `resolve` retry it, until 60 s pass with no
+    /// answer (KU §2.3 step 8, J13, J15). Default `false`: never retried, so a source that
+    /// answered (a 5xx, 401, 429) is never asked twice.
+    fn last_failure_was_transport(&self) -> bool {
+        false
+    }
+
+    /// Whether this source derives keys from the disc's Volume ID it is sent (an online key
+    /// service: Kvu = AES-G(Km, IDv), KS-16), so a Missing piece might open with the VID in
+    /// hand (KU J23, E7034). Default `false`.
+    fn uses_vid(&self) -> bool {
+        false
+    }
 }
 
 /// Drive `sources` until one resolves Unit Keys that decrypt `disc`. Returns

src/lib.rs

@@ -100,6 +100,8 @@ pub(crate) mod ifo;
 #[cfg(feature = "rip")]
 pub mod io;
 #[cfg(feature = "rip")]
+pub mod keys;
+#[cfg(feature = "rip")]
 pub mod keysource;
 #[cfg(feature = "rip")]
 pub mod labels;
@@ -246,7 +248,7 @@ pub use mux::resolve_mux_key_map;
 #[cfg(feature = "rip")]
 pub use mux::select::{PidFilter, StreamSelection};
 #[cfg(feature = "rip")]
-pub use mux::{MuxEvents, MuxInput, MuxOptions, MuxOutcome, mux_stream};
+pub use mux::{MuxEvents, MuxInput, MuxOptions, MuxOutcome, MuxSource, mux_stream, mux_with_keys};
 pub use scsi::{
     DiscPresence, DriveInfo, ScsiSense, ScsiTransport, SenseFamily, disc_presence, drive_has_disc,
     list_drives,

src/mux/disc.rs

@@ -358,6 +358,13 @@ impl DiscStream {
         self
     }
 
+    /// Install a [`ResolvedKeySet`](crate::keys::ResolvedKeySet)'s on-arrival proof on the
+    /// inline reader (KU §2.4): a piece the set left unproven is proven when first read.
+    pub(crate) fn with_arrival(mut self, arrival: crate::keys::Arrival) -> Self {
+        self.reader.set_arrival(arrival);
+        self
+    }
+
     fn is_halted(&self) -> bool {
         self.halt
             .as_ref()
@@ -476,6 +483,13 @@ impl DiscStream {
             let res = self
                 .reader
                 .read_sectors(lba, sectors, &mut self.read_buf[..bytes], false);
+            // The key set's loud stop (KU §2.4, §6): a readable unit no held key opens is not
+            // bad media — never shrunk, recovered or skipped.
+            if let Err(e) = res.as_ref()
+                && is_key_stop(e)
+            {
+                return Err(res.unwrap_err().into());
+            }
 
             if let Ok(&got) = res.as_ref() {
                 // read_sectors returns bytes written into buf. All in-tree
@@ -533,6 +547,11 @@ impl DiscStream {
                 let rec = self
                     .reader
                     .read_sectors(lba, sectors, &mut self.read_buf[..bytes], true);
+                if let Err(e) = rec.as_ref()
+                    && is_key_stop(e)
+                {
+                    return Err(rec.unwrap_err().into());
+                }
                 if let Ok(&got) = rec.as_ref() {
                     debug_assert!(got <= bytes, "recovery read over-reported byte count");
                     if let Some(ev) = self.adaptive.on_success(sectors) {
@@ -614,6 +633,14 @@ impl DiscStream {
     }
 }
 
+// The key set's on-arrival loud stop: E7022 (a title) or E7032 (an image or folder).
+fn is_key_stop(e: &crate::error::Error) -> bool {
+    matches!(
+        e,
+        crate::error::Error::NoDiscKey { .. } | crate::error::Error::WholeDiscKeyMissing
+    )
+}
+
 // Per-stage profiling state, populated only when FREEMKV_PROFILE is set.
 // Logs a percentage breakdown via tracing (target "mux") every
 // PROFILE_INTERVAL. Zero overhead otherwise.

src/mux/driver.rs

@@ -482,6 +482,272 @@ pub fn mux_stream(
     )
 }
 
+/// Where [`mux_with_keys`] reads its PES frames from (KU §3.1). The same sources as
+/// [`MuxInput`] without key fields: keys come only from the rip's
+/// [`ResolvedKeySet`](crate::keys::ResolvedKeySet).
+pub enum MuxSource<'a> {
+    /// Live single-pass mux off an opened [`DiscSession`] (its reader staged).
+    Session {
+        session: &'a mut DiscSession,
+        /// Index into `session.disc().titles`.
+        title_index: usize,
+    },
+    /// An image file and a title the caller ALREADY scanned (the drive scan, J14): the
+    /// image is never rescanned, so a sweep's unreadable UDF/MPLS/AACS sectors do not
+    /// matter. The set must cover the title and be for a sector-exact image of its disc.
+    Iso {
+        path: &'a Path,
+        title: DiscTitle,
+        format: crate::disc::ContentFormat,
+    },
+    /// Live single-pass mux off a raw disc reader (the inline `DiscStream`).
+    Live {
+        reader: Box<dyn SectorSource>,
+        title: DiscTitle,
+        format: crate::disc::ContentFormat,
+    },
+    /// Any URL-addressed source; the only variant that scans. `keys` goes into
+    /// [`InputOptions::keys`].
+    Url { url: &'a str, opts: InputOptions },
+}
+
+/// [`mux_stream`] with the rip's up-front key set (KU §3.1): every AACS read goes through
+/// the set's readers (its map, and the on-arrival proof for pieces it left unproven), with
+/// no key lookup. `keys == None`, a non-AACS set, or `opts.raw`: no AACS decryption (CSS
+/// and clear discs as before; a `Session` keeps the legacy disc-banked keys until KU-X2).
+/// E7013 when the set is not for the disc or does not cover the title; E7026 when the title
+/// needs forensic keys that are Pending. `events` stays last, as in `mux_stream` (J17).
+pub fn mux_with_keys(
+    source: MuxSource,
+    keys: Option<&crate::keys::ResolvedKeySet>,
+    dest_url: &str,
+    opts: &MuxOptions,
+    halt: &Halt,
+    events: std::sync::Arc<dyn MuxEvents>,
+) -> std::io::Result<MuxOutcome> {
+    let set = keys.filter(|s| s.is_aacs() && !opts.raw);
+    // KU §3.1: "`keys` must be `Some` for AACS". A BD-TS Iso/Live mux with no AACS set reads
+    // through a keyless set: the first AACS-flagged unit is E7022, never muxed as content.
+    // MPEG-PS keeps the old path (a DVD cracks its own CSS key; `MuxSource` has no disc).
+    let keyless = match (&source, set) {
+        (MuxSource::Iso { title, format, .. } | MuxSource::Live { title, format, .. }, None)
+            if !opts.raw && *format == crate::disc::ContentFormat::BdTs =>
+        {
+            Some(crate::keys::ResolvedKeySet::keyless_for(title, *format))
+        }
+        // A Session over an AACS disc with no banked key: BD-TS as above; HD DVD is never
+        // probed (KU §2.6), so it refuses up front. Banked keys keep the legacy path (J20).
+        (
+            MuxSource::Session {
+                session,
+                title_index,
+            },
+            None,
+        ) if !opts.raw => match session.disc() {
+            Some(d) if d.aacs.is_some() && !d.decrypt_keys().is_encrypted() => {
+                if d.content_format != crate::disc::ContentFormat::BdTs {
+                    return Err(Error::NoDiscKey {
+                        disc_hash: d.aacs_disc_hash(),
+                    }
+                    .into());
+                }
+                crate::keys::ResolvedKeySet::keyless_for_disc(d, *title_index)
+            }
+            _ => None,
+        },
+        _ => None,
+    };
+    let set = set.or(keyless.as_ref());
+    let input = match (source, set) {
+        (MuxSource::Url { url, opts: mut o }, _) => {
+            if let Some(k) = keys {
+                o.keys = Some(k.clone());
+            }
+            MuxInput::Url { url, opts: o }
+        }
+        (
+            MuxSource::Session {
+                session,
+                title_index,
+            },
+            None,
+        ) => MuxInput::Session {
+            session,
+            title_index,
+        },
+        (
+            MuxSource::Iso {
+                path,
+                title,
+                format,
+            },
+            None,
+        ) => MuxInput::Iso {
+            path,
+            title,
+            format,
+            keys: DecryptKeys::None,
+            key_fetch: None,
+        },
+        (
+            MuxSource::Live {
+                reader,
+                title,
+                format,
+            },
+            None,
+        ) => MuxInput::Live {
+            reader,
+            title,
+            format,
+            keys: DecryptKeys::None,
+            key_map: None,
+        },
+        (src, Some(set)) => return mux_keyed(src, set, dest_url, opts, halt, events),
+    };
+    mux_stream(input, dest_url, opts, halt, events)
+}
+
+// The keyed arms of `mux_with_keys`: the set's readers, never a resolve.
+fn mux_keyed(
+    source: MuxSource,
+    set: &crate::keys::ResolvedKeySet,
+    dest_url: &str,
+    opts: &MuxOptions,
+    halt: &Halt,
+    events: std::sync::Arc<dyn MuxEvents>,
+) -> std::io::Result<MuxOutcome> {
+    let (stream, playlist, source): (Box<dyn Stream>, Option<String>, SourceInfo) = match source {
+        MuxSource::Iso { path, title, .. } => {
+            let mut title = title;
+            opts.selection
+                .apply(&mut title)
+                .map_err(std::io::Error::from)?;
+            let source = SourceInfo {
+                medium: Medium::Iso,
+                path: format!("iso://{}", path.display()),
+                playlist: title.playlist.clone(),
+                ..SourceInfo::default()
+            };
+            let reader = FileSectorSource::open(path)?;
+            let stream = super::resolve::build_iso_pipeline_keyed(
+                reader,
+                title,
+                set,
+                opts.batch_sectors,
+                Some(halt.clone()),
+                Some(reader_event_fn(events.clone())),
+            )?;
+            (Box::new(stream), None, source)
+        }
+        MuxSource::Session {
+            session,
+            title_index,
+        } => {
+            let (title, format, playlist, source) = {
+                let disc = session.disc().ok_or_else(|| Error::DeviceNotReady {
+                    path: session.device_path().to_string(),
+                })?;
+                let scope = crate::keys::KeyScope::Titles(vec![title_index]);
+                if !set.is_for(disc) || !set.covers(&scope) {
+                    tracing::error!(target: "freemkv::keys", "key set is not for this session's title");
+                    debug_assert!(false, "key set is not for this session's title");
+                    return Err(Error::DecryptFailed.into());
+                }
+                let title = disc
+                    .titles
+                    .get(title_index)
+                    .cloned()
+                    .ok_or(Error::MuxTrackRange {
+                        track: title_index,
+                        tracks: disc.titles.len(),
+                    })?;
+                let source = SourceInfo {
+                    medium: Medium::Disc,
+                    path: format!("disc://{}", session.device_path()),
+                    title: title_index,
+                    playlist: title.playlist.clone(),
+                    volume_id: disc.volume_id.clone(),
+                };
+                let playlist = disc
+                    .meta_title
+                    .clone()
+                    .unwrap_or_else(|| disc.volume_id.clone());
+                (title, disc.content_format, playlist, source)
+            };
+            let reader = session.take_reader().ok_or_else(|| Error::DeviceNotReady {
+                path: session.device_path().to_string(),
+            })?;
+            let stream = live_keyed(reader, title, format, set, opts, halt, &events)?;
+            (stream, Some(playlist), source)
+        }
+        MuxSource::Live {
+            reader,
+            title,
+            format,
+        } => {
+            let source = SourceInfo {
+                medium: Medium::Disc,
+                playlist: title.playlist.clone(),
+                ..SourceInfo::default()
+            };
+            let stream = live_keyed(reader, title, format, set, opts, halt, &events)?;
+            (stream, None, source)
+        }
+        MuxSource::Url { .. } => unreachable!("mux_with_keys routes Url to mux_stream"),
+    };
+    let mut source = source;
+    if let Some(name) = playlist.as_deref() {
+        source.playlist = name.to_string();
+    }
+    drive_mux(
+        stream,
+        dest_url,
+        halt,
+        events.as_ref(),
+        playlist.as_deref(),
+        effective_send_deadline(opts.send_deadline),
+        Some(&source),
+    )
+}
+
+// The inline live-drive `DiscStream` over the set: its keys, map and on-arrival proof.
+fn live_keyed(
+    reader: Box<dyn SectorSource>,
+    mut title: DiscTitle,
+    format: crate::disc::ContentFormat,
+    set: &crate::keys::ResolvedKeySet,
+    opts: &MuxOptions,
+    halt: &Halt,
+    events: &std::sync::Arc<dyn MuxEvents>,
+) -> std::io::Result<Box<dyn Stream>> {
+    opts.selection
+        .apply(&mut title)
+        .map_err(std::io::Error::from)?;
+    let ranges: Vec<(u32, u32)> = title
+        .extents
+        .iter()
+        .map(|e| (e.start_lba, e.start_lba.saturating_add(e.sector_count)))
+        .collect();
+    set.gate(reader.random_access(), Some(&ranges), false)?;
+    let mut stream = crate::mux::DiscStream::new(
+        reader,
+        title,
+        set.decrypt_keys(),
+        opts.batch_sectors,
+        format,
+        false,
+        Some(halt.clone()),
+    )?
+    .with_key_map(set.key_map());
+    if let Some(a) = set.arrival(set.title_stop()) {
+        stream = stream.with_arrival(a);
+    }
+    stream.skip_errors = opts.skip_errors;
+    stream.on_event(reader_event_fn(events.clone()));
+    Ok(Box::new(stream))
+}
+
 // Decrypt keys for the live `Session` mux of `disc`. A DVD is handed `DecryptKeys::None` so
 // `DiscStream::new` cracks the CORRECT per-title CSS key rather than the whole-disc
 // (largest-title) VTS key.
@@ -3159,4 +3425,244 @@ mod tests {
         assert!(!calls.is_empty() && calls.len() <= 3, "{calls:?}");
         assert_eq!(calls.last(), Some(&(20_000, 20_000)));
     }
+
+    // ── mux_with_keys (KU §3.1): keys come only from the rip's set ─────────────
+
+    fn keyed_opts() -> MuxOptions {
+        MuxOptions {
+            skip_errors: false,
+            batch_sectors: 3,
+            raw: false,
+            send_deadline: Some(Duration::from_secs(60)),
+            selection: Default::default(),
+        }
+    }
+
+    // One encrypted audio unit at LBA 0..3 under `key`, its title, and a set keying it.
+    fn keyed_live(key: [u8; 16]) -> (Box<AacsUnitReader>, DiscTitle, crate::keys::ResolvedKeySet) {
+        let reader = Box::new(AacsUnitReader {
+            unit: encrypted_audio_unit(&key),
+            capacity: 16,
+        });
+        let mut title = aac_audio_title(0x1100);
+        title.extents = vec![crate::disc::Extent {
+            start_lba: 0,
+            sector_count: 3,
+        }];
+        let mut disc = aacs_session_disc(title.clone(), [0u8; 16]);
+        disc.capacity_sectors = 16;
+        let set = crate::keys::ResolvedKeySet::keyed_for_test(&disc, key, &[(0, 3)]);
+        (reader, title, set)
+    }
+
+    // The plaintext audio ES inside `encrypted_audio_unit`, and a muxed MKV holding it.
+    const PLAIN_ES: [u8; 6] = [0xDE, 0xAD, 0xBE, 0xEF, 0x11, 0x22];
+    fn holds_plain_es(path: &std::path::Path) -> bool {
+        std::fs::read(path)
+            .unwrap()
+            .windows(PLAIN_ES.len())
+            .any(|w| w == PLAIN_ES)
+    }
+
+    /// `MuxSource::Live` decrypts through the set's reader: no key banked on a disc, no
+    /// resolution in the driver.
+    #[test]
+    fn mux_with_keys_live_decrypts_through_the_set() {
+        let (reader, title, set) = keyed_live([0x5A; 16]);
+        let dir = tempfile::tempdir().unwrap();
+        let out_path = dir.path().join("live.mkv");
+        let out = mux_with_keys(
+            MuxSource::Live {
+                reader,
+                title,
+                format: crate::disc::ContentFormat::BdTs,
+            },
+            Some(&set),
+            &format!("mkv://{}", out_path.display()),
+            &keyed_opts(),
+            &Halt::new(),
+            Arc::new(NoopEvents),
+        )
+        .expect("the set's key opens the unit");
+        assert!(out.completed);
+        assert!(
+            holds_plain_es(&out_path),
+            "the muxed audio is the plaintext"
+        );
+    }
+
+    /// `MuxSource::Session` uses the set, not the disc's banked keys (a wrong key here).
+    #[test]
+    fn mux_with_keys_session_uses_the_set_not_banked_keys() {
+        let key = [0x5A; 16];
+        let (reader, title, _) = keyed_live(key);
+        let mut disc = aacs_session_disc(title, [0x99; 16]);
+        disc.capacity_sectors = 16;
+        let set = crate::keys::ResolvedKeySet::keyed_for_test(&disc, key, &[(0, 3)]);
+        let mut session = DiscSession::from_parts_for_test(Some(disc), Some(reader), None);
+        let out = mux_with_keys(
+            MuxSource::Session {
+                session: &mut session,
+                title_index: 0,
+            },
+            Some(&set),
+            "null://",
+            &keyed_opts(),
+            &Halt::new(),
+            Arc::new(NoopEvents),
+        )
+        .expect("the set's key opens the unit");
+        assert!(out.completed && out.bytes_written > 0);
+    }
+
+    /// J14: `MuxSource::Iso` muxes the caller's already-scanned title out of an image with
+    /// no filesystem at all; opening the same image by URL (which scans) fails.
+    #[test]
+    fn mux_with_keys_iso_muxes_a_scanned_title_without_rescanning() {
+        let _serial = crate::sector::prefetched::holder_test_lock();
+        let key = [0x5A; 16];
+        let (_, title, set) = keyed_live(key);
+        let dir = tempfile::tempdir().unwrap();
+        let path = dir.path().join("staged.iso");
+        let mut image = encrypted_audio_unit(&key);
+        image.resize(16 * 2048, 0);
+        std::fs::write(&path, &image).unwrap();
+        let out_path = dir.path().join("iso.mkv");
+        let out = mux_with_keys(
+            MuxSource::Iso {
+                path: &path,
+                title,
+                format: crate::disc::ContentFormat::BdTs,
+            },
+            Some(&set),
+            &format!("mkv://{}", out_path.display()),
+            &keyed_opts(),
+            &Halt::new(),
+            Arc::new(NoopEvents),
+        )
+        .expect("no rescan: the scanned title is muxed as given");
+        assert!(out.completed);
+        assert!(
+            holds_plain_es(&out_path),
+            "the muxed audio is the plaintext"
+        );
+        let url = format!("iso://{}", path.display());
+        let rescanned = mux_with_keys(
+            MuxSource::Url {
+                url: &url,
+                opts: InputOptions::default(),
+            },
+            Some(&set),
+            "null://",
+            &keyed_opts(),
+            &Halt::new(),
+            Arc::new(NoopEvents),
+        );
+        assert!(
+            rescanned.is_err(),
+            "a URL source scans the image, which has no filesystem"
+        );
+    }
+
+    /// KU §3.1 (review item 3): "`keys` must be `Some` for AACS". With no AACS set (none, or a
+    /// non-AACS set) and not raw, `Iso` and `Live` refuse E7022 at the first AACS-flagged
+    /// unit rather than muxing ciphertext; raw passes it through, and clear content muxes.
+    #[test]
+    fn mux_with_keys_without_an_aacs_set_refuses_aacs_content() {
+        let _serial = crate::sector::prefetched::holder_test_lock();
+        let key = [0x5A; 16];
+        let none = crate::keys::ResolvedKeySet::none();
+        let live = |keys: Option<&crate::keys::ResolvedKeySet>, raw: bool, unit: Vec<u8>| {
+            let (_, title, _) = keyed_live(key);
+            let reader = Box::new(AacsUnitReader { unit, capacity: 16 });
+            let opts = MuxOptions {
+                raw,
+                ..keyed_opts()
+            };
+            mux_with_keys(
+                MuxSource::Live {
+                    reader,
+                    title,
+                    format: crate::disc::ContentFormat::BdTs,
+                },
+                keys,
+                "null://",
+                &opts,
+                &Halt::new(),
+                Arc::new(NoopEvents),
+            )
+        };
+        let code = |r: std::io::Result<MuxOutcome>| r.err().and_then(|e| crate::error_code(&e));
+        let e7022 = Some(crate::error::E_NO_DISC_KEY);
+        assert_eq!(code(live(None, false, encrypted_audio_unit(&key))), e7022);
+        assert_eq!(
+            code(live(Some(&none), false, encrypted_audio_unit(&key))),
+            e7022
+        );
+        assert!(
+            live(None, true, encrypted_audio_unit(&key)).is_ok(),
+            "raw passes"
+        );
+        let mut clear = encrypted_audio_unit(&key);
+        crate::test_util::decrypt_unit(&mut clear, &key);
+        clear.chunks_mut(192).for_each(|p| p[0] &= 0x3F);
+        let out = live(None, false, clear).expect("clear content needs no key");
+        assert!(out.completed && out.bytes_written > 0);
+
+        let (_, title, _) = keyed_live(key);
+        let dir = tempfile::tempdir().unwrap();
+        let path = dir.path().join("staged.iso");
+        let mut image = encrypted_audio_unit(&key);
+        image.resize(16 * 2048, 0);
+        std::fs::write(&path, &image).unwrap();
+        let iso = mux_with_keys(
+            MuxSource::Iso {
+                path: &path,
+                title,
+                format: crate::disc::ContentFormat::BdTs,
+            },
+            None,
+            "null://",
+            &keyed_opts(),
+            &Halt::new(),
+            Arc::new(NoopEvents),
+        );
+        assert_eq!(code(iso), e7022);
+
+        // Session over an AACS disc with no banked key (review B1): BD-TS stops at the first
+        // AACS unit; HD DVD refuses up front (it is never probed, KU §2.6).
+        for format in [crate::DiscFormat::Uhd, crate::DiscFormat::HdDvd] {
+            let (_, title, _) = keyed_live(key);
+            let mut disc = aacs_session_disc(title, key);
+            disc.aacs.as_mut().unwrap().unit_keys.clear();
+            disc.capacity_sectors = 16;
+            disc.format = format;
+            if format == crate::DiscFormat::HdDvd {
+                disc.content_format = crate::ContentFormat::MpegPs;
+            }
+            let reader = Box::new(AacsUnitReader {
+                unit: encrypted_audio_unit(&key),
+                capacity: 16,
+            });
+            let mut session = DiscSession::from_parts_for_test(Some(disc), Some(reader), None);
+            let r = mux_with_keys(
+                MuxSource::Session {
+                    session: &mut session,
+                    title_index: 0,
+                },
+                None,
+                "null://",
+                &keyed_opts(),
+                &Halt::new(),
+                Arc::new(NoopEvents),
+            );
+            let err = r.expect_err("no key, no mux");
+            assert_eq!(crate::error_code(&err), e7022, "{format:?}");
+            let text = err.to_string().to_ascii_lowercase();
+            assert!(
+                text.contains("abc"),
+                "{format:?}: E7022 names the disc: {text}"
+            );
+        }
+    }
 }

src/mux/mod.rs

@@ -82,7 +82,9 @@ pub(crate) mod videomap;
 // The provenance types ARE public: `output()` takes a `SourceInfo` so an `fvi://`
 // destination records the INPUT it was built from (§6.2), not the file written.
 pub use disc::DiscStream;
-pub use driver::{MuxEvents, MuxInput, MuxOptions, MuxOutcome, mux_stream};
+pub use driver::{
+    MuxEvents, MuxInput, MuxOptions, MuxOutcome, MuxSource, mux_stream, mux_with_keys,
+};
 pub use m2ts::M2tsStream;
 pub use mkvstream::{
     MkvProbe, MkvProbeTrack, MkvStream, MkvTrackKind, parse_freemkv_version, probe_mkv,

src/mux/resolve.rs

@@ -337,6 +337,10 @@ pub struct InputOptions {
     /// after the title-index bounds check. Default keeps every stream (video is
     /// always kept). See [`crate::StreamSelection`].
     pub selection: crate::StreamSelection,
+    /// The rip's up-front key set (KU §3.1). For an AACS image it replaces `unit_keys` and
+    /// `key_fetch`: the title is read through the set's reader, with no lookup here.
+    /// `None` keeps the legacy disc-banked keys (until KU-X2).
+    pub keys: Option<crate::keys::ResolvedKeySet>,
 }
 
 // `KeyFetchFactory` holds a trait object that is not `Debug`; hand-roll the
@@ -350,6 +354,7 @@ impl std::fmt::Debug for InputOptions {
             .field("raw", &self.raw)
             .field("key_fetch", &self.key_fetch.is_some())
             .field("selection", &self.selection)
+            .field("keys", &self.keys.is_some())
             .finish()
     }
 }
@@ -473,6 +478,9 @@ where
         crate::session::apply_folder_encryption_verdict(&mut reader, &mut disc)
             .map_err(|e| -> io::Error { e.into() })?;
     }
+    if let Some(set) = opts.keys.as_ref().filter(|s| s.is_aacs() && !opts.raw) {
+        return keyed_image_input(reader, disc, opts, set, reopen);
+    }
     // Apply the caller-resolved keys (lookup-free); decrypt_keys() then
     // yields them for the stream below. Propagate a failed application
     // rather than silently muxing an undecryptable stream.
@@ -609,6 +617,50 @@ where
     Ok(stream)
 }
 
+// `image_input` over the rip's key set (KU §3.1, §3.5): the gate is `check_decryptable`,
+// and the TrueHD probe and the mux read through the set's reader. No lookup, no banking.
+fn keyed_image_input<S, F>(
+    reader: S,
+    mut disc: crate::disc::Disc,
+    opts: &InputOptions,
+    set: &crate::keys::ResolvedKeySet,
+    reopen: F,
+) -> io::Result<PipelinedPesStream>
+where
+    S: SectorSource + Send + 'static,
+    F: FnOnce() -> io::Result<S>,
+{
+    if disc.titles.is_empty() {
+        return Err(crate::error::Error::NoStreams.into());
+    }
+    let idx = opts.title_index.unwrap_or(0);
+    if idx >= disc.titles.len() {
+        return Err(crate::error::Error::DiscTitleRange {
+            index: idx,
+            count: disc.titles.len(),
+        }
+        .into());
+    }
+    let scope = crate::keys::KeyScope::Titles(vec![idx]);
+    crate::keys::check_decryptable(&disc, false, Some(set), &scope)
+        .map_err(|e| -> io::Error { e.into() })?;
+    opts.selection
+        .apply(&mut disc.titles[idx])
+        .map_err(|e| -> io::Error { e.into() })?;
+    match reopen() {
+        Ok(probe) => match set.title_reader(&disc, idx, probe) {
+            Ok(mut dec) => crate::disc::correct_truehd_channels(&mut dec, &mut disc.titles[idx]),
+            Err(e) => tracing::debug!(target: "mux", error = %e, "TrueHD probe reader refused"),
+        },
+        Err(e) => {
+            tracing::debug!(target: "mux", error = %e, "TrueHD channel-correction probe re-open failed")
+        }
+    }
+    const ISO_MUX_BATCH_SECTORS: u16 = 8192;
+    let title = disc.titles[idx].clone();
+    build_iso_pipeline_keyed(reader, title, set, ISO_MUX_BATCH_SECTORS, None, None)
+}
+
 /// Open a PES output stream (consumes PES frames).
 ///
 /// `source` is the provenance of the material being written — the INPUT the caller is muxing
@@ -823,7 +875,7 @@ pub(crate) fn build_demux_state(title: &DiscTitle, format: ContentFormat) -> Dem
 
 // Tag for an FMTS forensic index key banked into the key pool (base + slot); separates disc
 // BASE CPS unit keys (< this) from forensic ones (>= this).
-const FMTS_POOL_TAG_BASE: u32 = 1 << 24;
+pub(crate) const FMTS_POOL_TAG_BASE: u32 = 1 << 24;
 
 // Pool slot of the disc's ONE base CPS Unit Key, or None for a genuine multi-CPS disc; excludes
 // forensic index keys.
@@ -1322,7 +1374,7 @@ fn probe_fmts_index_keys(
 
 // FMTS forensic feature clip's own extents — the byte space every `IndividualSegment.tbl` SPN
 // is relative to — or None if not exactly one such clip.
-fn forensic_clip_extents(
+pub(crate) fn forensic_clip_extents(
     udf: &crate::udf::UdfFs,
     reader: &mut dyn SectorSource,
 ) -> io::Result<Option<Vec<crate::disc::Extent>>> {
@@ -1373,7 +1425,7 @@ fn extents_overlap(a: &[crate::disc::Extent], b: &[crate::disc::Extent]) -> bool
 // Keep only forensic segments addressable within the FORENSIC CLIP's extents; stale/foreign
 // records past the clip's end are dropped. Extracted from resolve_fmts_key_map for direct
 // testing.
-fn filter_addressable_segments(
+pub(crate) fn filter_addressable_segments(
     segments: Vec<crate::aacs::segment::Segment>,
     extents: &[crate::disc::Extent],
 ) -> Vec<crate::aacs::segment::Segment> {
@@ -1401,7 +1453,7 @@ fn resolve_tie_phase(even_clean: usize, odd_clean: usize) -> io::Result<crate::d
 // Outcome of probing ONE forensic index's decrypt phase; the load-bearing split is WrongKey vs
 // ReadFault — only the former is a real FmtsKeyMissing.
 #[derive(Debug, PartialEq, Eq)]
-enum IndexProbe {
+pub(crate) enum IndexProbe {
     /// A parity decrypted clean under this index's key (or a padding tie) → its phase.
     Phase(crate::decrypt::Phase),
     /// At least one unit was READ and decrypt-attempted, yet NEITHER parity came up
@@ -1415,7 +1467,7 @@ enum IndexProbe {
 
 // Probe one forensic index's decrypt phase (EVEN vs ODD aligned units) under `key`, tolerating
 // read faults without masking a genuine wrong key.
-fn probe_index_phase(
+pub(crate) fn probe_index_phase(
     segments: &[crate::aacs::segment::Segment],
     tag: u16,
     batch_units: usize,
@@ -1466,7 +1518,7 @@ fn probe_index_phase(
 
 // Back-fill LBA gaps NOT covered by forensic segment ranges with the base Unit Key, so the map
 // is a COMPLETE positive list over the title's content extents.
-fn fill_base_key_gaps(
+pub(crate) fn fill_base_key_gaps(
     extents: &[crate::disc::Extent],
     forensic_ranges: &[(u32, u32, usize, crate::decrypt::Phase)],
     base_idx: usize,
@@ -1785,15 +1837,93 @@ pub fn build_iso_pipeline<S: SectorSource + Send + 'static>(
         Some(map) => map.read_plan(&extents, unit_align as u32),
         None => extents,
     };
-    // The plan and clip feed spans must describe the SAME bytes: if the plan
-    // drops alternate-phase units, frames drift out of place near a join —
-    // undetectable by the tile check. Fall back to timestamps if mismatched.
-    let feed_matches_spans = extents == full_extents;
     let mut decrypting =
         crate::sector::DecryptingSectorSource::new(Box::new(reader) as Box<dyn SectorSource>, keys);
     if let Some(map) = key_map {
         decrypting = decrypting.with_key_map(map);
     }
+    let plan = IsoPlan {
+        extents,
+        full_extents,
+        batch_sectors,
+        unit_align,
+        format,
+    };
+    iso_pipeline_tail(decrypting, plan, title, halt, event_fn)
+}
+
+// The read plan of an ISO title mux: the extents read (FMTS alternate phase dropped), the
+// title's full extents, and the read geometry.
+struct IsoPlan {
+    extents: Vec<crate::disc::Extent>,
+    full_extents: Vec<crate::disc::Extent>,
+    batch_sectors: u16,
+    unit_align: u16,
+    format: ContentFormat,
+}
+
+/// The ISO title mux over a [`ResolvedKeySet`](crate::keys::ResolvedKeySet) (KU §3.1): the
+/// set's decrypting reader (its map and on-arrival proof) under the prefetcher, with no
+/// resolution here. `title` is an already-scanned title; E7013 if the set does not cover
+/// its extents or the image is not a sector-exact copy of the set's disc.
+pub(crate) fn build_iso_pipeline_keyed<S: SectorSource + Send + 'static>(
+    reader: S,
+    title: DiscTitle,
+    set: &crate::keys::ResolvedKeySet,
+    batch_sectors: u16,
+    halt: Option<crate::halt::Halt>,
+    event_fn: Option<crate::sector::prefetched::EventFn>,
+) -> io::Result<PipelinedPesStream> {
+    let cap = reader.capacity_sectors();
+    if (cap != 0 && set.capacity() != 0 && cap != set.capacity())
+        || !set.covers_extents(&title.extents)
+    {
+        tracing::error!(target: "freemkv::keys", "key set does not cover this image title");
+        debug_assert!(false, "key set does not cover this image title");
+        return Err(crate::error::Error::DecryptFailed.into());
+    }
+    let full_extents = title.extents.clone();
+    let ranges: Vec<(u32, u32)> = full_extents
+        .iter()
+        .map(|e| (e.start_lba, e.start_lba.saturating_add(e.sector_count)))
+        .collect();
+    let decrypting = set
+        .decrypting(
+            Box::new(reader) as Box<dyn SectorSource>,
+            Some(&ranges),
+            set.title_stop(),
+            false,
+        )
+        .map_err(io::Error::from)?;
+    let plan = IsoPlan {
+        extents: set.key_map().read_plan(&full_extents, 3),
+        full_extents,
+        batch_sectors,
+        unit_align: 3,
+        format: set.content_format(),
+    };
+    iso_pipeline_tail(decrypting, plan, title, halt, event_fn)
+}
+
+// The shared tail of the ISO title mux: decrypting reader → prefetcher → demux → parse.
+fn iso_pipeline_tail(
+    mut decrypting: crate::sector::DecryptingSectorSource<Box<dyn SectorSource>>,
+    plan: IsoPlan,
+    title: DiscTitle,
+    halt: Option<crate::halt::Halt>,
+    event_fn: Option<crate::sector::prefetched::EventFn>,
+) -> io::Result<PipelinedPesStream> {
+    let IsoPlan {
+        extents,
+        full_extents,
+        batch_sectors,
+        unit_align,
+        format,
+    } = plan;
+    // The plan and clip feed spans must describe the SAME bytes: if the plan
+    // drops alternate-phase units, frames drift out of place near a join —
+    // undetectable by the tile check. Fall back to timestamps if mismatched.
+    let feed_matches_spans = extents == full_extents;
     // The mux does not tally decrypt-quality misses (`lost_bytes()` is read-loss
     // only); a missing key is an up-front resolve failure. Wrong-substream fix
     // below: re-route declared AC-3 audio onto correct `0x8x` sub-streams.

src/sector/bus_removal.rs

@@ -506,6 +506,10 @@ impl<S: SectorSource> SectorSource for BusRemovalSectorSource<S> {
             _ => self.inner.unmapped_stream_files(),
         }
     }
+
+    fn random_access(&self) -> bool {
+        self.inner.random_access()
+    }
 }
 
 // Shared across every module's "wrappers forward `unmapped_stream_files`" test.

src/sector/decrypting.rs

@@ -95,6 +95,9 @@ pub struct DecryptingSectorSource<S: SectorSource> {
     /// (CSS self-descrambles in `decrypt_sectors`); an AACS source without a map
     /// is a bug and fails loud on the first unit (AACS decrypts only via the map).
     key_map: Option<Arc<crate::decrypt::AacsKeyMap>>,
+    /// The key set's on-arrival proof for pieces `resolve` could not prove up front
+    /// (KU §2.4). `None` for every reader not built by a `ResolvedKeySet`.
+    arrival: Option<Box<crate::keys::Arrival>>,
 }
 
 /// Does the sector span `[lba, lba+count)` intersect any encrypted-content range?
@@ -117,9 +120,22 @@ impl<S: SectorSource> DecryptingSectorSource<S> {
             unit_base: None,
             content_ranges: None,
             key_map: None,
+            arrival: None,
         }
     }
 
+    /// Install the key set's on-arrival proof (KU §2.4): an encrypted unit of a piece
+    /// `resolve` left unproven is decrypted with a held key proven on it here.
+    pub(crate) fn with_arrival(mut self, arrival: crate::keys::Arrival) -> Self {
+        self.arrival = Some(Box::new(arrival));
+        self
+    }
+
+    /// `&mut` counterpart of [`with_arrival`](Self::with_arrival).
+    pub(crate) fn set_arrival(&mut self, arrival: crate::keys::Arrival) {
+        self.arrival = Some(Box::new(arrival));
+    }
+
     /// Install a proactive [`AacsKeyMap`](crate::decrypt::AacsKeyMap): the caller
     /// resolved one key per CPS unit / segment up front, so every aligned unit is
     /// decrypted with its MAPPED key and trusted — no per-unit `is_clean` check.
@@ -206,6 +222,9 @@ impl<S: SectorSource> SectorSource for DecryptingSectorSource<S> {
     fn unmapped_stream_files(&self) -> &[crate::sector::bus_removal::UnmappedStreamFile] {
         self.inner.unmapped_stream_files()
     }
+    fn random_access(&self) -> bool {
+        self.inner.random_access()
+    }
 
     fn read_sectors(
         &mut self,
@@ -249,6 +268,13 @@ impl<S: SectorSource> SectorSource for DecryptingSectorSource<S> {
             .inner
             .read_sectors_fua(lba, count, buf, recovery, fua)?;
 
+        // KU §2.4: units of a piece left unproven are proven now, from the held keys only.
+        if let (Some(arrival), DecryptKeys::Aacs { unit_keys, .. }) =
+            (self.arrival.as_deref(), &self.keys)
+        {
+            arrival.process(&mut self.inner, lba, &mut buf[..n], unit_keys)?;
+        }
+
         // Proactive map path (storm-free mux): keys were resolved per unit up front,
         // so decrypt with the mapped key and trust it (no per-unit `is_clean`). A
         // resolver gap fails loud; bad TS and units outside content extents pass through.

src/sector/mod.rs

@@ -90,6 +90,14 @@ pub trait SectorSource: Send {
     fn unmapped_stream_files(&self) -> &[bus_removal::UnmappedStreamFile] {
         &[]
     }
+
+    /// Whether `read_sectors` honours the requested `lba`/`count` (random access). `false`
+    /// for a prefetcher, whose "lba/count are advisory". The key set's decrypting readers
+    /// side-read neighbouring units, so they refuse a non-random-access inner source (KU
+    /// §2.4); wrappers must forward it.
+    fn random_access(&self) -> bool {
+        true
+    }
 }
 
 // Forwarding impls so `Box<dyn SectorSource>` and `&mut dyn SectorSource`
@@ -132,6 +140,10 @@ impl SectorSource for Box<dyn SectorSource> {
     fn unmapped_stream_files(&self) -> &[bus_removal::UnmappedStreamFile] {
         (**self).unmapped_stream_files()
     }
+
+    fn random_access(&self) -> bool {
+        (**self).random_access()
+    }
 }
 
 impl SectorSource for &mut (dyn SectorSource + '_) {
@@ -171,6 +183,10 @@ impl SectorSource for &mut (dyn SectorSource + '_) {
     fn unmapped_stream_files(&self) -> &[bus_removal::UnmappedStreamFile] {
         (**self).unmapped_stream_files()
     }
+
+    fn random_access(&self) -> bool {
+        (**self).random_access()
+    }
 }
 
 pub use crate::io::file_sector_source::FileSectorSource;

src/sector/prefetched.rs

@@ -342,6 +342,11 @@ impl SectorSource for PrefetchedSectorSource {
         &self.unmapped
     }
 
+    // The producer decides the next batch; "lba/count are advisory" (read_sectors below).
+    fn random_access(&self) -> bool {
+        false
+    }
+
     fn read_sectors(
         &mut self,
         _lba: u32,

src/session.rs

@@ -266,6 +266,33 @@ impl DiscSession {
         Ok(resolved.trace)
     }
 
+    /// Resolve the rip's key set for `scope` up front (KU §3.1): one
+    /// [`ResolvedKeySet::resolve`](crate::keys::ResolvedKeySet::resolve) through the
+    /// session's staged reader, else its drive. The session keeps nothing: the set is the
+    /// caller's, and no source is retained. Requires [`Self::scan`] to have run.
+    pub fn resolve_key_set(
+        &mut self,
+        scope: crate::keys::KeyScope,
+        sources: &KeySourceFactory,
+        opts: crate::keys::ResolveKeysOptions,
+    ) -> Result<crate::keys::KeyResolution> {
+        let Some(disc) = self.disc.as_ref() else {
+            return Err(Error::DeviceNotReady {
+                path: self.device.clone(),
+            });
+        };
+        let reader: &mut dyn SectorSource = match (self.reader.as_mut(), self.drive.as_mut()) {
+            (Some(r), _) => r.as_mut(),
+            (None, Some(d)) => d,
+            (None, None) => {
+                return Err(Error::DeviceNotReady {
+                    path: self.device.clone(),
+                });
+            }
+        };
+        crate::keys::ResolvedKeySet::resolve(disc, reader, scope, sources, opts)
+    }
+
     /// The read-time AACS fetch closure retained by [`Self::resolve_keys`], for a
     /// later mux (step 4) to install into the decrypt decorator. `None` before
     /// keys are resolved, or for a non-AACS disc.

src/test_util.rs

@@ -366,6 +366,12 @@ impl<S: SectorSource> SectorSource for CountingSource<S> {
     fn set_unit_base(&mut self, lba: u32) {
         self.inner.set_unit_base(lba)
     }
+    fn unmapped_stream_files(&self) -> &[crate::sector::bus_removal::UnmappedStreamFile] {
+        self.inner.unmapped_stream_files()
+    }
+    fn random_access(&self) -> bool {
+        self.inner.random_access()
+    }
 }
 
 /// Decrypt one 6144-byte aligned unit in place with a CPS unit key (KS-3, KS-4). The

src/udf.rs

@@ -1821,6 +1821,9 @@ impl<S: SectorSource + ?Sized> SectorSource for BufferedSectorReader<'_, S> {
     fn unmapped_stream_files(&self) -> &[crate::sector::bus_removal::UnmappedStreamFile] {
         self.inner.unmapped_stream_files()
     }
+    fn random_access(&self) -> bool {
+        self.inner.random_access()
+    }
 }
 
 /// Read a single 2048-byte sector from the drive.

src/whole_disc.rs

@@ -13,7 +13,7 @@ use crate::error::{Error, Result};
 use crate::sector::{DecryptingSectorSource, KeyFetch, SectorSource};
 
 /// Sectors in one AACS aligned unit (6144 bytes).
-const UNIT: u64 = (crate::aacs::content::ALIGNED_UNIT_LEN / 2048) as u64;
+pub(crate) const UNIT: u64 = (crate::aacs::content::ALIGNED_UNIT_LEN / 2048) as u64;
 
 /// Units probed per unplayed content file: its first unit, then evenly across it.
 const PROBES: u64 = 32;
@@ -92,8 +92,16 @@ pub fn whole_disc_reader<S: SectorSource>(
 // Every AACS content file's extents, one entry per file (contiguous extents joined):
 // `/BDMV/STREAM` (m2ts before SSIF, which re-lists them) or else HD DVD `/HVDVD_TS/*.EVO`.
 // An unreadable UDF or unmappable file fails loud: it would otherwise ship as ciphertext.
-fn content_files(reader: &mut dyn SectorSource) -> Result<Vec<Vec<(u32, u32)>>> {
+pub(crate) fn content_files(reader: &mut dyn SectorSource) -> Result<Vec<Vec<(u32, u32)>>> {
     let fs = crate::udf::read_filesystem(reader)?;
+    content_files_in(&fs, reader)
+}
+
+// `content_files` over an already-read filesystem.
+pub(crate) fn content_files_in(
+    fs: &crate::udf::UdfFs,
+    reader: &mut dyn SectorSource,
+) -> Result<Vec<Vec<(u32, u32)>>> {
     let (top, evo_only) = if fs.find_dir("/BDMV/STREAM").is_some() {
         ("/BDMV/STREAM", false)
     } else if fs.find_dir("/HVDVD_TS").is_some() {
@@ -137,12 +145,12 @@ fn content_files(reader: &mut dyn SectorSource) -> Result<Vec<Vec<(u32, u32)>>>
 
 /// A content extent `(start, count)` and the LBA its unit grid is anchored at: the
 /// owning file's first sector, carried across extents by file offset.
-type UnitSpan = (u32, u32, u64);
+pub(crate) type UnitSpan = (u32, u32, u64);
 
 // The unit grid of every content extent: per file by file offset (never a merged run:
 // adjacent files each start their own grid), then title extents no file covers (anchored
 // at the extent start). Sorted, disjoint; on overlap the earlier-listed source wins.
-fn unit_spans(files: &[Vec<(u32, u32)>], title_extents: &[(u32, u32)]) -> Vec<UnitSpan> {
+pub(crate) fn unit_spans(files: &[Vec<(u32, u32)>], title_extents: &[(u32, u32)]) -> Vec<UnitSpan> {
     let mut raw: Vec<(UnitSpan, usize)> = Vec::new();
     for (i, file) in files.iter().enumerate() {
         let mut off = 0u64;
@@ -171,7 +179,7 @@ fn unit_spans(files: &[Vec<(u32, u32)>], title_extents: &[(u32, u32)]) -> Vec<Un
 }
 
 // The span holding `lba`, if any.
-fn span_at(spans: &[UnitSpan], lba: u64) -> Option<UnitSpan> {
+pub(crate) fn span_at(spans: &[UnitSpan], lba: u64) -> Option<UnitSpan> {
     let i = spans
         .partition_point(|&(s, _, _)| s as u64 <= lba)
         .checked_sub(1)?;
@@ -180,7 +188,7 @@ fn span_at(spans: &[UnitSpan], lba: u64) -> Option<UnitSpan> {
 }
 
 // The first unit head at or after `lba` on the grid anchored at `anchor`.
-fn unit_head(lba: u32, anchor: u64) -> u64 {
+pub(crate) fn unit_head(lba: u32, anchor: u64) -> u64 {
     lba as u64 + (UNIT - (lba as u64).saturating_sub(anchor) % UNIT) % UNIT
 }
 
@@ -393,7 +401,7 @@ impl Probe<'_> {
 
 // Which of a piece's `units` to probe, in order: all of them when few, else the
 // first unit, then `PROBES - 1` more spread evenly to the end.
-fn probe_units(units: u64) -> Vec<u64> {
+pub(crate) fn probe_units(units: u64) -> Vec<u64> {
     if units <= PROBES {
         return (0..units).collect();
     }
@@ -440,7 +448,7 @@ fn merge_key_ranges(mut ranges: Vec<(u32, u32, usize, Phase)>) -> Vec<(u32, u32,
 
 // `content` `(start, count)` ranges minus the sorted, disjoint `[start, end)` `keyed`
 // ranges, as `[start, end)` pieces.
-fn subtract_ranges(content: &[(u32, u32)], keyed: &[(u32, u32)]) -> Vec<(u32, u32)> {
+pub(crate) fn subtract_ranges(content: &[(u32, u32)], keyed: &[(u32, u32)]) -> Vec<(u32, u32)> {
     let mut out = Vec::new();
     for &(start, count) in content {
         let end = start.saturating_add(count);
@@ -463,7 +471,7 @@ fn subtract_ranges(content: &[(u32, u32)], keyed: &[(u32, u32)]) -> Vec<(u32, u3
 
 // Sort + coalesce overlapping/adjacent `(start, count)` ranges (the content gate's
 // binary search needs them sorted and disjoint); empty ranges are dropped.
-fn merge_ranges(mut ranges: Vec<(u32, u32)>) -> Vec<(u32, u32)> {
+pub(crate) fn merge_ranges(mut ranges: Vec<(u32, u32)>) -> Vec<(u32, u32)> {
     ranges.retain(|&(_, count)| count > 0);
     ranges.sort_unstable();
     let mut out: Vec<(u32, u32)> = Vec::with_capacity(ranges.len());
@@ -508,7 +516,7 @@ pub struct UnitAligned<S> {
 }
 
 impl<S: SectorSource> UnitAligned<S> {
-    fn new(inner: S, spans: Vec<UnitSpan>) -> Self {
+    pub(crate) fn new(inner: S, spans: Vec<UnitSpan>) -> Self {
         Self {
             inner,
             spans,
@@ -565,6 +573,9 @@ impl<S: SectorSource> SectorSource for UnitAligned<S> {
     fn unmapped_stream_files(&self) -> &[crate::sector::bus_removal::UnmappedStreamFile] {
         self.inner.unmapped_stream_files()
     }
+    fn random_access(&self) -> bool {
+        self.inner.random_access()
+    }
 
     fn read_sectors(
         &mut self,

tests/spec_sites.txt

@@ -26,3 +26,13 @@ src/drive/allow.rs	is_eject	SS-6
 src/drive/mod.rs	eject	SS-6
 src/drive/mod.rs	start_unit	SS-6
 src/drive/mod.rs	note_ledger	SS-7
+src/keys/resolve.rs	resolve	KS-14
+src/keys/resolve.rs	borrow_samples	KS-10
+src/keys/resolve.rs	apply_single_unit_rule	KS-14
+src/keys/arrival.rs	decrypt_proven	KS-5
+src/keys/arrival.rs	side_partners	KS-1
+src/keys/arrival.rs	side_partners	KS-2
+src/keys/arrival.rs	side_partners	KS-7
+src/keys/fmts.rs	anchor	KS-26
+src/keys/fmts.rs	phases	KS-25
+src/keys/fmts.rs	phases	KS-26