par2_rs/

verify.rs

use std::collections::HashMap;
use std::io::{self, Read, Seek};
use std::path::PathBuf;

use crate::checksum;
use crate::md5_simd;
use crate::par2_set::Par2FileSet;
use crate::types::{
    CancellationToken, FileId, MAX_SLICES_PER_FILE, ProgressCallback, ProgressPhase, ProgressStage,
    ProgressUpdate, SliceChecksum,
};

const VERIFY_SLICE_CHUNK_BYTES: usize = 64 * 1024;
const VERIFY_SIMD_BATCH_MEMORY_BYTES: usize = 4 * 1024 * 1024;
/// Upper bound on the multi-buffer MD5 kernel width, used only to size stack
/// arrays. The width actually driven is [`md5_simd::max_lanes`], which selects
/// by ISA (8 on AVX2, 4 on NEON/SSE2/simd128, 1 scalar) and never exceeds this.
const VERIFY_SIMD_MAX_LANES: usize = 8;
const QUICK_CHECK_16K_BYTES: usize = 16 * 1024;
const VERIFY_FULL_HASH_CHUNK_BYTES: usize = 1024 * 1024;
/// Read span per task for slice-parallel staged-file verification.
const VERIFY_SPAN_TARGET_BYTES: usize = 4 * 1024 * 1024;

/// Seekable reader used by repair streams that revisit many ranges in one
/// source file. Implementations may return one to avoid reopening per slice.
pub trait FileRangeReader: Read + Seek {}

impl<T: Read + Seek> FileRangeReader for T {}

/// Abstraction for file I/O during verification and repair.
pub trait FileAccess {
    /// Read a range of bytes from a file identified by its FileId.
    fn read_file_range(&self, file_id: &FileId, offset: u64, len: u64) -> io::Result<Vec<u8>>;

    /// Read bytes into a caller-provided buffer, returning the number of bytes read.
    fn read_file_range_into(
        &self,
        file_id: &FileId,
        offset: u64,
        dst: &mut [u8],
    ) -> io::Result<usize> {
        let data = self.read_file_range(file_id, offset, dst.len() as u64)?;
        let read_len = data.len().min(dst.len());
        dst[..read_len].copy_from_slice(&data[..read_len]);
        Ok(read_len)
    }

    /// Open a forward-only reader for hot paths that consume a whole file from offset 0.
    fn open_sequential_reader(&self, _file_id: &FileId) -> io::Result<Option<Box<dyn Read>>> {
        Ok(None)
    }

    /// Open a reusable seekable reader for range-oriented repair input.
    fn open_range_reader(&self, _file_id: &FileId) -> io::Result<Option<Box<dyn FileRangeReader>>> {
        Ok(None)
    }

    /// Check if a file exists on disk.
    fn file_exists(&self, file_id: &FileId) -> bool;

    /// Get the actual length of a file on disk.
    fn file_length(&self, file_id: &FileId) -> Option<u64>;

    /// Read the entire file content.
    fn read_file(&self, file_id: &FileId) -> io::Result<Vec<u8>>;

    /// Write data to a file at the given offset.
    fn write_file_range(&mut self, file_id: &FileId, offset: u64, data: &[u8]) -> io::Result<()>;
}

/// Simple in-memory file access for testing.
pub struct MemoryFileAccess {
    files: HashMap<FileId, Vec<u8>>,
}

impl MemoryFileAccess {
    pub fn new() -> Self {
        Self {
            files: HashMap::new(),
        }
    }

    pub fn add_file(&mut self, file_id: FileId, data: Vec<u8>) {
        self.files.insert(file_id, data);
    }
}

impl Default for MemoryFileAccess {
    fn default() -> Self {
        Self::new()
    }
}

impl FileAccess for MemoryFileAccess {
    fn read_file_range(&self, file_id: &FileId, offset: u64, len: u64) -> io::Result<Vec<u8>> {
        let data = self
            .files
            .get(file_id)
            .ok_or_else(|| io::Error::new(io::ErrorKind::NotFound, "file not found"))?;
        let offset = offset as usize;
        let end = (offset + len as usize).min(data.len());
        if offset >= data.len() {
            return Ok(Vec::new());
        }
        Ok(data[offset..end].to_vec())
    }

    fn read_file_range_into(
        &self,
        file_id: &FileId,
        offset: u64,
        dst: &mut [u8],
    ) -> io::Result<usize> {
        let data = self
            .files
            .get(file_id)
            .ok_or_else(|| io::Error::new(io::ErrorKind::NotFound, "file not found"))?;
        let offset = offset as usize;
        if offset >= data.len() {
            return Ok(0);
        }
        let end = (offset + dst.len()).min(data.len());
        let read_len = end - offset;
        dst[..read_len].copy_from_slice(&data[offset..end]);
        Ok(read_len)
    }

    fn file_exists(&self, file_id: &FileId) -> bool {
        self.files.contains_key(file_id)
    }

    fn file_length(&self, file_id: &FileId) -> Option<u64> {
        self.files.get(file_id).map(|d| d.len() as u64)
    }

    fn read_file(&self, file_id: &FileId) -> io::Result<Vec<u8>> {
        self.files
            .get(file_id)
            .cloned()
            .ok_or_else(|| io::Error::new(io::ErrorKind::NotFound, "file not found"))
    }

    fn write_file_range(&mut self, file_id: &FileId, offset: u64, data: &[u8]) -> io::Result<()> {
        let file_data = self
            .files
            .get_mut(file_id)
            .ok_or_else(|| io::Error::new(io::ErrorKind::NotFound, "file not found"))?;
        let offset = offset as usize;
        let end = offset + data.len();
        // Extend the file if needed.
        if end > file_data.len() {
            file_data.resize(end, 0);
        }
        file_data[offset..end].copy_from_slice(data);
        Ok(())
    }
}

/// Status of a single file after verification.
#[derive(Debug, Clone)]
pub enum FileStatus {
    /// File is complete: full MD5 matches.
    Complete,
    /// File is damaged: N slices are bad.
    Damaged(u32),
    /// File was not found on disk.
    Missing,
    /// File was found at a different path (identified via 16k hash).
    Renamed(PathBuf),
}

/// Verification result for a single file.
#[derive(Debug, Clone)]
pub struct FileVerification {
    pub file_id: FileId,
    pub filename: String,
    pub status: FileStatus,
    /// Per-slice validity: true = valid, false = damaged/missing.
    pub valid_slices: Vec<bool>,
    /// Number of damaged/missing slices.
    pub missing_slice_count: u32,
}

/// Repairability assessment.
#[derive(Debug, Clone)]
pub enum Repairability {
    /// No repair needed; all files are intact.
    NotNeeded,
    /// Repair is possible with available recovery data.
    Repairable {
        blocks_needed: u32,
        blocks_available: u32,
    },
    /// Insufficient recovery data to repair.
    Insufficient {
        blocks_needed: u32,
        blocks_available: u32,
        deficit: u32,
    },
    /// Verification could not continue within configured resource limits.
    ResourceLimited { reason: String },
}

/// Overall verification result.
#[derive(Debug, Clone)]
pub struct VerificationResult {
    pub files: Vec<FileVerification>,
    /// Recovery blocks the set describes. Verification never reads a recovery
    /// payload, so this is [`Par2FileSet::recovery_block_count`]'s upper bound:
    /// a block with a damaged payload is counted until repair validates it.
    pub recovery_blocks_available: u32,
    pub total_missing_blocks: u32,
    pub repairable: Repairability,
}

impl VerificationResult {
    pub fn needs_repair(&self) -> bool {
        self.total_missing_blocks > 0
            || self
                .files
                .iter()
                .any(|file| !matches!(file.status, FileStatus::Complete))
    }

    pub fn refresh_repairability(&mut self) {
        if let Repairability::ResourceLimited { .. } = self.repairable
            && (self.total_missing_blocks > 0
                || self
                    .files
                    .iter()
                    .any(|file| !matches!(file.status, FileStatus::Complete)))
        {
            return;
        }

        self.repairable = repairability_for_result(
            &self.files,
            self.total_missing_blocks,
            self.recovery_blocks_available,
        );
    }
}

fn repairability_for_result(
    files: &[FileVerification],
    total_missing_blocks: u32,
    recovery_blocks_available: u32,
) -> Repairability {
    if total_missing_blocks == 0
        && files
            .iter()
            .all(|file| matches!(file.status, FileStatus::Complete))
    {
        Repairability::NotNeeded
    } else {
        repairability_for_counts(total_missing_blocks, recovery_blocks_available)
    }
}

fn repairability_for_result_with_resource_limit(
    files: &[FileVerification],
    total_missing_blocks: u32,
    recovery_blocks_available: u32,
    resource_limit_reason: Option<String>,
) -> Repairability {
    match resource_limit_reason {
        Some(reason) => Repairability::ResourceLimited { reason },
        None => repairability_for_result(files, total_missing_blocks, recovery_blocks_available),
    }
}

fn repairability_for_counts(
    total_missing_blocks: u32,
    recovery_blocks_available: u32,
) -> Repairability {
    if total_missing_blocks == 0 {
        Repairability::NotNeeded
    } else if total_missing_blocks <= recovery_blocks_available {
        Repairability::Repairable {
            blocks_needed: total_missing_blocks,
            blocks_available: recovery_blocks_available,
        }
    } else {
        Repairability::Insufficient {
            blocks_needed: total_missing_blocks,
            blocks_available: recovery_blocks_available,
            deficit: total_missing_blocks - recovery_blocks_available,
        }
    }
}

fn bounded_slice_count(par2: &Par2FileSet, length: u64) -> Option<usize> {
    let count = usize::try_from(par2.slice_count_for_file(length)).ok()?;
    (count <= MAX_SLICES_PER_FILE).then_some(count)
}

fn resource_limited_verification(file_id: FileId, filename: String) -> FileVerification {
    FileVerification {
        file_id,
        filename,
        status: FileStatus::Damaged(0),
        valid_slices: Vec::new(),
        missing_slice_count: 0,
    }
}

/// Perform a 16KB quick-check on a file.
///
/// Reads the first 16384 bytes (or less if file is shorter) and computes MD5,
/// comparing against the stored `hash_16k` from the File Description packet.
pub fn quick_check_16k(
    par2: &Par2FileSet,
    file_id: &FileId,
    access: &dyn FileAccess,
) -> Option<bool> {
    let mut scratch = [0u8; QUICK_CHECK_16K_BYTES];
    quick_check_16k_with_scratch(par2, file_id, access, &mut scratch)
}

fn quick_check_16k_with_scratch(
    par2: &Par2FileSet,
    file_id: &FileId,
    access: &dyn FileAccess,
    scratch: &mut [u8; QUICK_CHECK_16K_BYTES],
) -> Option<bool> {
    let desc = par2.file_description(file_id)?;

    if !access.file_exists(file_id) {
        return Some(false);
    }

    let read_len = if let Some(mut reader) = access.open_sequential_reader(file_id).ok()? {
        let scratch_len = scratch.len();
        read_from_sequential_reader(&mut *reader, scratch, scratch_len).ok()?
    } else {
        access.read_file_range_into(file_id, 0, scratch).ok()?
    };
    let hash = checksum::md5(&scratch[..read_len]);
    Some(hash == desc.hash_16k)
}

/// Perform full-file MD5 verification.
pub fn verify_full_hash(
    par2: &Par2FileSet,
    file_id: &FileId,
    access: &dyn FileAccess,
) -> Option<bool> {
    let desc = par2.file_description(file_id)?;
    let actual_len = access.file_length(file_id)?;
    if actual_len != desc.length {
        return Some(false);
    }
    verify_full_hash_streaming(desc.hash_full, actual_len, file_id, access)
}

/// What the strict single-file streaming pass established.
enum StrictStreamOutcome {
    /// A slice's CRC32 disagreed with its IFSC entry, so the file's bytes
    /// provably differ from the protected content. The stream stopped there:
    /// the whole-file hashes were never finished, and could not have changed
    /// the verdict (see [`verify_selected_file_ids_resolved`]).
    SliceCrcMismatch,
    /// The whole file streamed to the end. Verdicts for the 16 KiB prefix and
    /// the whole-file MD5, exactly as the unfused pass reported them.
    Hashes { quick_ok: bool, full_ok: bool },
}

/// Slice-aligned chunk size for the fused strict stream: the largest whole
/// number of slices that fits in [`VERIFY_FULL_HASH_CHUNK_BYTES`], so every
/// chunk ends on a slice boundary and the per-slice CRC32 needs no carry
/// state between chunks. `None` when a single slice does not fit, in which
/// case the stream runs unfused at the original chunk size. The buffer is
/// therefore never larger than the unfused pass's, so the resident-memory
/// accounting of a verify is unchanged.
fn fused_chunk_bytes(slice_size: u64) -> Option<usize> {
    let slice = usize::try_from(slice_size).ok()?;
    if slice == 0 || slice > VERIFY_FULL_HASH_CHUNK_BYTES {
        return None;
    }
    Some((VERIFY_FULL_HASH_CHUNK_BYTES / slice) * slice)
}

/// One streaming pass over a file that feeds the 16 KiB and whole-file MD5
/// chains and, when `slice_checksums` is supplied, CRC32s each slice out of the
/// same buffer.
///
/// The CRC32 rides along on bytes already in cache for the MD5 chain, so it
/// costs no extra read, no extra buffer, and no extra thread — a clean file
/// pays only the CRC32 arithmetic. Its value is the early exit: the first slice
/// whose CRC32 disagrees with its IFSC entry proves the file's content differs
/// from the protected content, and from there the serial MD5 chain can only
/// confirm what is already known, so the pass abandons it.
///
/// `slice_checksums` must be `Some` only when the entry count matches the
/// file's slice count and the on-disk length matches the description — the
/// same preconditions under which [`verify_slices`] indexes those entries.
fn stream_strict_hashes(
    par2: &Par2FileSet,
    file_id: &FileId,
    access: &dyn FileAccess,
    slice_checksums: Option<&[SliceChecksum]>,
) -> Option<StrictStreamOutcome> {
    let desc = par2.file_description(file_id)?;
    let actual_len = access.file_length(file_id)?;
    let slice_size = par2.slice_size;
    // `fused_chunk_bytes` is `None` for a zero or oversized slice size, so a
    // fused pass always has a positive, slice-aligned chunk.
    let fused = slice_checksums.zip(fused_chunk_bytes(slice_size));
    let chunk_bytes = fused.map_or(VERIFY_FULL_HASH_CHUNK_BYTES, |(_, chunk)| chunk);

    let mut quick_state = checksum::FileHashState::new();
    let mut full_state = None;
    let mut buf = vec![0u8; chunk_bytes];
    let mut total_read = 0u64;
    let mut reader = access.open_sequential_reader(file_id).ok()?;

    while total_read < actual_len {
        let want = ((actual_len - total_read) as usize).min(chunk_bytes);
        let read_len = match reader.as_mut() {
            // Chunks are filled exactly, not by a single `read`, so the chunk
            // boundary stays slice-aligned for the CRC32 ride-along.
            Some(reader) => {
                read_from_sequential_reader(&mut **reader, &mut buf[..want], want).ok()?
            }
            None => access
                .read_file_range_into(file_id, total_read, &mut buf[..want])
                .ok()?,
        };
        if read_len == 0 {
            break;
        }
        let data = &buf[..read_len];
        update_quick_and_full_hash_states(&mut quick_state, &mut full_state, data);
        if let Some((checksums, _)) = fused
            && chunk_proves_slice_damage(data, total_read, slice_size, actual_len, checksums)
        {
            return Some(StrictStreamOutcome::SliceCrcMismatch);
        }
        total_read += read_len as u64;
        if read_len < want {
            break;
        }
    }

    // A file that grew past its reported length is not the described file. The
    // unfused pass caught that by reading to EOF; the fused pass stops at
    // `actual_len`, so it probes for a trailing byte instead.
    let grew = match reader.as_mut() {
        Some(reader) => {
            let mut probe = [0u8; 1];
            read_from_sequential_reader(&mut **reader, &mut probe, 1).ok()? != 0
        }
        None => false,
    };
    if total_read != actual_len || grew {
        return Some(StrictStreamOutcome::Hashes {
            quick_ok: false,
            full_ok: false,
        });
    }

    let quick_hash = quick_state.finalize();
    let full_hash = full_state.map_or(quick_hash, checksum::FileHashState::finalize);
    Some(StrictStreamOutcome::Hashes {
        quick_ok: quick_hash == desc.hash_16k,
        full_ok: actual_len == desc.length && full_hash == desc.hash_full,
    })
}

/// True when some slice wholly contained in this slice-aligned chunk has a
/// CRC32 that disagrees with its IFSC entry.
///
/// Only a window that is either a full slice or the file's final (zero-padded)
/// slice is judged; a short window anywhere else is the product of a short
/// read, and padding it would manufacture a mismatch. Padding matches
/// [`check_slice_span`] and [`verify_slices`] exactly, so a slice this rejects
/// is one they reject too.
fn chunk_proves_slice_damage(
    data: &[u8],
    chunk_offset: u64,
    slice_size: u64,
    file_len: u64,
    checksums: &[SliceChecksum],
) -> bool {
    debug_assert!(chunk_offset.is_multiple_of(slice_size));
    let slice = slice_size as usize;
    let first_slice = (chunk_offset / slice_size) as usize;
    for (index, window) in data.chunks(slice).enumerate() {
        let Some(expected) = checksums.get(first_slice + index) else {
            return false;
        };
        let window_end = chunk_offset + (index as u64) * slice_size + window.len() as u64;
        if window.len() != slice && window_end != file_len {
            return false;
        }
        if checksum::crc32_padded(window, slice_size) != expected.crc32 {
            return true;
        }
    }
    false
}

fn verify_full_hash_streaming(
    expected_hash: [u8; 16],
    actual_len: u64,
    file_id: &FileId,
    access: &dyn FileAccess,
) -> Option<bool> {
    let mut state = checksum::FileHashState::new();
    let mut buf = vec![0u8; VERIFY_FULL_HASH_CHUNK_BYTES];

    if let Some(mut reader) = access.open_sequential_reader(file_id).ok()? {
        let mut total_read = 0u64;
        loop {
            let read_len = reader.read(&mut buf).ok()?;
            if read_len == 0 {
                break;
            }
            state.update(&buf[..read_len]);
            total_read += read_len as u64;
        }
        if total_read != actual_len {
            return Some(false);
        }
    } else {
        let mut offset = 0u64;
        while offset < actual_len {
            let chunk_len = ((actual_len - offset) as usize).min(buf.len());
            let read_len = access
                .read_file_range_into(file_id, offset, &mut buf[..chunk_len])
                .ok()?;
            if read_len == 0 {
                return Some(false);
            }
            state.update(&buf[..read_len]);
            offset += read_len as u64;
        }
    }

    Some(state.finalize() == expected_hash)
}

fn update_quick_and_full_hash_states(
    quick_state: &mut checksum::FileHashState,
    full_state: &mut Option<checksum::FileHashState>,
    data: &[u8],
) {
    if data.is_empty() {
        return;
    }
    if let Some(full_state) = full_state.as_mut() {
        full_state.update(data);
        return;
    }

    let quick_remaining = QUICK_CHECK_16K_BYTES.saturating_sub(quick_state.bytes_fed() as usize);
    if data.len() <= quick_remaining {
        quick_state.update(data);
        return;
    }

    quick_state.update(&data[..quick_remaining]);
    let mut cloned = quick_state.clone();
    cloned.update(&data[quick_remaining..]);
    *full_state = Some(cloned);
}

/// Perform slice-level verification of a single file.
///
/// Returns a vector of booleans, one per slice: true = valid, false = damaged.
pub fn verify_slices(
    par2: &Par2FileSet,
    file_id: &FileId,
    access: &dyn FileAccess,
) -> Option<Vec<bool>> {
    let desc = par2.file_description(file_id)?;
    let checksums = par2.file_checksums(file_id)?;
    let slice_size = par2.slice_size;
    let expected_slices = bounded_slice_count(par2, desc.length)?;
    if checksums.len() != expected_slices {
        return Some(vec![false; expected_slices]);
    }

    if !access.file_exists(file_id) {
        return Some(vec![false; expected_slices]);
    }

    if let Some(results) =
        verify_slices_batched_md5(file_id, desc.length, checksums, slice_size, access).ok()?
    {
        return Some(results);
    }

    let mut results = vec![false; expected_slices];
    let mut buf = vec![0u8; VERIFY_SLICE_CHUNK_BYTES];

    for (i, result) in results.iter_mut().enumerate() {
        let offset = i as u64 * slice_size;
        let expected_data_len = desc.length.saturating_sub(offset).min(slice_size);
        let actual = checksum_file_slice_padded(
            file_id,
            offset,
            expected_data_len,
            slice_size,
            access,
            &mut buf,
        )
        .ok()?;
        *result = actual.crc32 == checksums[i].crc32 && actual.md5 == checksums[i].md5;
    }

    Some(results)
}

fn verify_slices_batched_md5(
    file_id: &FileId,
    file_len: u64,
    checksums: &[SliceChecksum],
    slice_size: u64,
    access: &dyn FileAccess,
) -> io::Result<Option<Vec<bool>>> {
    if checksums.is_empty() {
        return Ok(Some(Vec::new()));
    }
    let Ok(slice_size_usize) = usize::try_from(slice_size) else {
        return Ok(None);
    };
    if slice_size_usize == 0 {
        return Ok(None);
    }

    // Lane count is the narrower of what the kernel offers and what the batch
    // memory budget affords, so a wide host is used fully without the staging
    // buffers growing past the budget.
    let max_lanes = (VERIFY_SIMD_BATCH_MEMORY_BYTES / slice_size_usize).min(md5_simd::max_lanes());
    debug_assert!(max_lanes <= VERIFY_SIMD_MAX_LANES);
    if max_lanes < 2 {
        return Ok(None);
    }

    let mut results = vec![false; checksums.len()];
    let mut buffers = (0..max_lanes)
        .map(|_| vec![0u8; slice_size_usize])
        .collect::<Vec<_>>();

    if let Some(mut reader) = access.open_sequential_reader(file_id)? {
        return verify_slices_batched_md5_from_reader(
            &mut *reader,
            file_len,
            checksums,
            slice_size,
            &mut buffers,
        )
        .map(Some);
    }

    let mut index = 0usize;

    while index < checksums.len() {
        let lanes = max_lanes.min(checksums.len() - index);
        let mut read_lens = Vec::with_capacity(lanes);
        let mut crc32s = Vec::with_capacity(lanes);

        for (lane, buffer) in buffers.iter_mut().take(lanes).enumerate() {
            let slice_index = index + lane;
            let offset = slice_index as u64 * slice_size;
            let expected_data_len = file_len.saturating_sub(offset).min(slice_size);
            let read_len =
                read_file_slice_into(file_id, offset, expected_data_len, access, buffer)?;
            read_lens.push(read_len);
            crc32s.push(checksum::crc32_padded(&buffer[..read_len], slice_size));
        }

        let inputs = buffers
            .iter()
            .take(lanes)
            .zip(read_lens.iter())
            .map(|(buffer, read_len)| &buffer[..*read_len])
            .collect::<Vec<_>>();
        let md5s = md5_simd::md5_multi(&inputs, Some(slice_size));

        for lane in 0..lanes {
            let expected = checksums[index + lane];
            results[index + lane] = crc32s[lane] == expected.crc32 && md5s[lane] == expected.md5;
        }

        index += lanes;
    }

    Ok(Some(results))
}

fn verify_slices_batched_md5_from_reader(
    reader: &mut dyn Read,
    file_len: u64,
    checksums: &[SliceChecksum],
    slice_size: u64,
    buffers: &mut [Vec<u8>],
) -> io::Result<Vec<bool>> {
    let mut results = vec![false; checksums.len()];
    let max_lanes = buffers.len();
    let mut index = 0usize;

    while index < checksums.len() {
        let lanes = max_lanes.min(checksums.len() - index);
        let mut read_lens = Vec::with_capacity(lanes);
        let mut crc32s = Vec::with_capacity(lanes);

        for (lane, buffer) in buffers.iter_mut().take(lanes).enumerate() {
            let slice_index = index + lane;
            let offset = slice_index as u64 * slice_size;
            let expected_data_len = file_len.saturating_sub(offset).min(slice_size) as usize;
            let read_len = read_from_sequential_reader(reader, buffer, expected_data_len)?;
            read_lens.push(read_len);
            crc32s.push(checksum::crc32_padded(&buffer[..read_len], slice_size));
        }

        let inputs = buffers
            .iter()
            .take(lanes)
            .zip(read_lens.iter())
            .map(|(buffer, read_len)| &buffer[..*read_len])
            .collect::<Vec<_>>();
        let md5s = md5_simd::md5_multi(&inputs, Some(slice_size));

        for lane in 0..lanes {
            let expected = checksums[index + lane];
            results[index + lane] = crc32s[lane] == expected.crc32 && md5s[lane] == expected.md5;
        }

        index += lanes;
    }

    Ok(results)
}

fn read_from_sequential_reader(
    reader: &mut dyn Read,
    dst: &mut [u8],
    expected_len: usize,
) -> io::Result<usize> {
    let expected_len = expected_len.min(dst.len());
    let mut read_len = 0usize;
    while read_len < expected_len {
        let n = reader.read(&mut dst[read_len..expected_len])?;
        if n == 0 {
            break;
        }
        read_len += n;
    }
    Ok(read_len)
}

fn read_file_slice_into(
    file_id: &FileId,
    offset: u64,
    expected_data_len: u64,
    access: &dyn FileAccess,
    dst: &mut [u8],
) -> io::Result<usize> {
    let mut consumed = 0u64;

    while consumed < expected_data_len {
        let start = consumed as usize;
        let remaining_capacity = dst.len().saturating_sub(start);
        if remaining_capacity == 0 {
            break;
        }
        let take = (expected_data_len - consumed).min(remaining_capacity as u64) as usize;
        let read_len = access.read_file_range_into(
            file_id,
            offset + consumed,
            &mut dst[start..start + take],
        )?;
        if read_len == 0 {
            break;
        }
        consumed += read_len as u64;
        if read_len < take {
            break;
        }
    }

    Ok(consumed as usize)
}

fn checksum_file_slice_padded(
    file_id: &FileId,
    offset: u64,
    expected_data_len: u64,
    slice_size: u64,
    access: &dyn FileAccess,
    buf: &mut [u8],
) -> io::Result<SliceChecksum> {
    let mut state = checksum::SliceChecksumState::new();
    let mut consumed = 0u64;

    while consumed < expected_data_len {
        let take = (expected_data_len - consumed).min(buf.len() as u64) as usize;
        let read_len = access.read_file_range_into(file_id, offset + consumed, &mut buf[..take])?;
        if read_len == 0 {
            break;
        }
        state.update(&buf[..read_len]);
        consumed += read_len as u64;
        if read_len < take {
            break;
        }
    }

    let (crc32, md5) = state.finalize(Some(slice_size));
    Ok(SliceChecksum { crc32, md5 })
}

/// Verify file slices using pre-computed CRC32 values instead of reading from disk.
///
/// This is a CRC-only check (no MD5 verification). Useful when the download layer
/// has already computed CRC32s for each slice during decode, avoiding disk re-reads.
///
/// Returns a vector of booleans per slice (true = CRC matches), or `None` if the
/// file is not in the PAR2 set.
pub fn verify_slices_from_crcs(
    par2: &Par2FileSet,
    file_id: &FileId,
    slice_crcs: &[u32],
) -> Option<Vec<bool>> {
    let checksums = par2.file_checksums(file_id)?;

    let results: Vec<bool> = checksums
        .iter()
        .enumerate()
        .map(|(i, expected)| {
            slice_crcs
                .get(i)
                .map(|&crc| crc == expected.crc32)
                .unwrap_or(false)
        })
        .collect();

    Some(results)
}

/// Options controlling verification behavior.
#[derive(Default)]
// Verification learns new dials over time; a new one should not cost every
// consumer a major version. Build with `..Default::default()`.
#[non_exhaustive]
pub struct VerifyOptions {
    /// If set, verification will check this token and stop early if cancelled.
    pub cancel: Option<CancellationToken>,
    /// If set, called with progress updates after each file is verified.
    pub progress: Option<ProgressCallback>,
    /// Per-file slices the caller has already proven, keyed by [`FileId`].
    ///
    /// Each vector is one entry per slice of that file, in the set's slice
    /// order and of exactly the length the set's layout gives that file —
    /// the same shape as [`FileVerification::valid_slices`], so a caller can
    /// feed back what an earlier pass produced. `true` means "this slice is
    /// proven intact and need not be read"; `false` means "say nothing about
    /// it", which is what an entirely absent entry means for every slice.
    ///
    /// Verification then **reads only the unproven slices**, seeking over the
    /// byte ranges the proven ones cover, and reports a result shaped exactly
    /// as a full read would have: per-slice accounting is exact, and the file
    /// is `Complete` only when every one of its slices is accounted for by
    /// evidence or by a read that matched.
    ///
    /// # Trust contract
    ///
    /// This is caller-attested and is not re-derived. A wrong attestation
    /// produces a false `Complete` for the slices it covers — verification
    /// reads nothing that could contradict it. That is the same trust class
    /// as the crate's other evidence paths: what a caller vouches for, it
    /// vouches for. Attest only slices whose bytes you checksummed against
    /// this set's own per-slice checksums, on a file that is no longer being
    /// written.
    ///
    /// # When it is ignored
    ///
    /// Evidence applies only where a per-slice verdict can mean anything: the
    /// file must exist, its on-disk length must equal the description's, it
    /// must carry a complete IFSC checksum set, and it must be non-empty. A
    /// file failing any of those takes the pipeline it always took, evidence
    /// or no evidence — in particular, evidence about a **missing** file does
    /// not resurrect it, and evidence about a file whose length no longer
    /// matches is discarded rather than believed, because the offsets those
    /// verdicts were about are not the offsets in the file now on disk. A
    /// vector of the wrong length is ignored in full for the same reason.
    ///
    /// An entry that proves nothing (all `false`) is treated as absent, so an
    /// empty or all-`false` map leaves verification byte-for-byte what it was.
    pub proven_slices: HashMap<FileId, Vec<bool>>,
    /// Opt-in fast-verify mode (default `false`). When enabled, an intact
    /// candidate file whose on-disk length already matches the description is
    /// proven complete from its per-slice IFSC checksums (CRC32 + MD5, tail
    /// slice zero-padded) scanned span-parallel at read speed, skipping the
    /// inherently serial full-file MD5. Off by default, in which case
    /// verification is byte-identical to the strict full-MD5 pipeline. The
    /// `WEAVER_PAR2_FAST_VERIFY` environment variable overrides this per verify
    /// call, taking precedence over whatever is set here.
    pub fast_verify: bool,
}

impl VerifyOptions {
    /// The proven-slice evidence that applies to `file_id` under a layout of
    /// `slice_count` slices, or `None` when this call has nothing usable to
    /// say about that file.
    ///
    /// A vector of the wrong length describes a different layout than the one
    /// being verified, and there is no honest way to line the two up, so it is
    /// dropped whole rather than truncated or padded. A vector that proves
    /// nothing is dropped too, so that the no-evidence path is reached by
    /// exactly the same branch whether the caller passed no map, an empty map,
    /// or a map of all-`false` vectors.
    fn proven_slices_for(&self, file_id: &FileId, slice_count: usize) -> Option<&[bool]> {
        if slice_count == 0 {
            return None;
        }
        let proven = self.proven_slices.get(file_id)?;
        if proven.len() != slice_count {
            return None;
        }
        proven
            .iter()
            .any(|proven| *proven)
            .then_some(proven.as_slice())
    }
}

/// [`VerifyOptions::proven_slices_for`] for a caller that has not already
/// computed the file's slice count. Returns `None` for a file the set does not
/// describe or whose slice count exceeds the verifier's limits, which are the
/// same files evidence could not be applied to anyway.
fn proven_slices_for_file<'a>(
    par2: &Par2FileSet,
    options: &'a VerifyOptions,
    file_id: &FileId,
) -> Option<&'a [bool]> {
    let desc = par2.file_description(file_id)?;
    let slice_count = bounded_slice_count(par2, desc.length)?;
    options.proven_slices_for(file_id, slice_count)
}

/// Resolve whether the fast-verify path runs for this call. Precedence: the
/// `WEAVER_PAR2_FAST_VERIFY` environment variable wins — `"1"` forces the fast
/// path on, `"0"` forces it off — and any other or absent value falls back to
/// the caller's `flag`. Read once per verify call so a single process can flip
/// the behavior via the environment without recompiling.
fn fast_verify_enabled(flag: bool) -> bool {
    resolve_fast_verify(
        std::env::var("WEAVER_PAR2_FAST_VERIFY").ok().as_deref(),
        flag,
    )
}

/// Pure precedence rule behind [`fast_verify_enabled`], split out so the
/// override semantics can be unit-tested without mutating process environment
/// state shared by concurrently running tests.
fn resolve_fast_verify(env_value: Option<&str>, flag: bool) -> bool {
    match env_value {
        Some("1") => true,
        Some("0") => false,
        _ => flag,
    }
}

/// Verify all files in a PAR2 set.
pub fn verify_all(par2: &Par2FileSet, access: &dyn FileAccess) -> VerificationResult {
    verify_all_with_options(par2, access, &VerifyOptions::default())
}

/// Verify only the selected PAR2 file IDs.
pub fn verify_selected_file_ids(
    par2: &Par2FileSet,
    access: &dyn FileAccess,
    file_ids: &[FileId],
) -> VerificationResult {
    verify_selected_file_ids_with_options(par2, access, file_ids, &VerifyOptions::default())
}

/// Verify all files in a PAR2 set with cancellation and progress support.
pub fn verify_all_with_options(
    par2: &Par2FileSet,
    access: &dyn FileAccess,
    options: &VerifyOptions,
) -> VerificationResult {
    verify_selected_file_ids_with_options(par2, access, &par2.recovery_file_ids, options)
}

/// Verify selected PAR2 file IDs with one rayon task per file. Each file
/// runs the serial single-file pipeline unchanged (no cancellation or
/// progress on this path), so results are identical to
/// [`verify_selected_file_ids`] with the set-level repairability assessed
/// once over the combined outcome.
pub fn verify_selected_file_ids_parallel(
    par2: &Par2FileSet,
    access: &(dyn FileAccess + Sync),
    file_ids: &[FileId],
) -> VerificationResult {
    verify_selected_file_ids_parallel_with_options(
        par2,
        access,
        file_ids,
        &VerifyOptions::default(),
    )
}

/// Like [`verify_selected_file_ids_parallel`] but honoring [`VerifyOptions`].
/// The options consulted on this path are `fast_verify` and `proven_slices`
/// (cancellation and progress are not surfaced here, as documented on the base
/// function). Per-slice evidence is applied first and reads only the slices it
/// does not already prove; it subsumes the fast arm for the files it covers,
/// because both prove completeness from the same IFSC entries. When
/// fast verify resolves on (via the flag or `WEAVER_PAR2_FAST_VERIFY`), each
/// eligible file is proven complete from its IFSC slice checksums instead of a
/// full-file MD5 — span-parallel for a lone file and file-parallel otherwise,
/// the same single-axis rule as the post-repair readback. Ineligible files
/// fall back to the serial pipeline. With fast verify off this is
/// byte-identical to [`verify_selected_file_ids_parallel`].
pub fn verify_selected_file_ids_parallel_with_options(
    par2: &Par2FileSet,
    access: &(dyn FileAccess + Sync),
    file_ids: &[FileId],
    options: &VerifyOptions,
) -> VerificationResult {
    use rayon::prelude::*;
    let fast_verify = fast_verify_enabled(options.fast_verify);
    // Parallelism runs on exactly one axis: span-parallel for a lone file,
    // else file-parallel with each file's spans serial (mirrors
    // `verify_repaired_file_ids_parallel`).
    let span_parallel = file_ids.len() == 1;
    let partials: Vec<VerificationResult> = file_ids
        .par_iter()
        .map(|file_id| {
            // Per-slice evidence first: it subsumes the fast arm (both prove
            // from the same IFSC entries) and reads strictly less.
            if let Some(proven) = proven_slices_for_file(par2, options, file_id)
                && let Some(file) = verify_file_sliced_with_evidence(
                    par2,
                    access,
                    file_id,
                    span_parallel,
                    Some(proven),
                )
            {
                return single_file_result(par2, file);
            }
            if fast_verify
                && let Some(file) = verify_file_sliced(par2, access, file_id, span_parallel)
            {
                return single_file_result(par2, file);
            }
            // A lone file has no other rayon axis in flight, so the strict
            // pipeline's damaged branch may fan out across this file's spans.
            // Otherwise identical to the `verify_selected_file_ids` call this
            // replaced, `fast_verify` included — an eligible file was already
            // handled above, and an ineligible one reaches the same branch
            // either way.
            verify_selected_file_ids_resolved(
                par2,
                access,
                span_parallel.then_some(access),
                std::slice::from_ref(file_id),
                &VerifyOptions::default(),
                fast_verify,
            )
        })
        .collect();
    combine_partial_results(par2, partials)
}

/// Post-repair verification for staged files: every byte is read back and
/// checked against its IFSC slice checksum (MD5 + CRC32, tail padded),
/// with slice spans verified concurrently. Coverage matches the serial
/// full-hash pipeline — these are the same per-slice checksums that gate
/// block reuse during scanning — but the whole-file MD5 chain (inherently
/// serial) is replaced by slice-parallel hashing, so a repaired multi-GB
/// file verifies at read speed. `Complete` from slice proof is sound here
/// because a staged file's identity is definitional (the repair just built
/// it from the set's own layout); the scanner's stricter full-MD5 rule
/// exists to establish identity of found files, not to re-check content.
/// Files without IFSC data or with zero length take the serial single-file
/// pipeline, full-file MD5 included.
pub fn verify_repaired_file_ids_parallel(
    par2: &Par2FileSet,
    access: &(dyn FileAccess + Sync),
    file_ids: &[FileId],
) -> VerificationResult {
    use rayon::prelude::*;
    // Parallelism runs on exactly one axis: across files when there are
    // several, across a lone file's slice spans otherwise. Nesting the span
    // fan-out inside the file fan-out lets a blocked worker steal other
    // files' verification frames onto one stack — the same steal-on-block
    // recursion that overflowed worker stacks in the candidate scanner.
    let span_parallel = file_ids.len() == 1;
    let partials: Vec<VerificationResult> = file_ids
        .par_iter()
        .map(|file_id| {
            verify_repaired_file_sliced(par2, access, file_id, span_parallel).unwrap_or_else(|| {
                verify_selected_file_ids(par2, access, std::slice::from_ref(file_id))
            })
        })
        .collect();
    combine_partial_results(par2, partials)
}

/// Precomputed layout for slice-proof verification of one eligible file: its
/// filename, on-disk length, slice size, slice count, and the work to do.
/// Produced by [`sliced_verify_plan`] only for files that pass the shared
/// fast-path / post-repair preconditions.
struct SlicedVerifyPlan {
    filename: String,
    length: u64,
    slice_size: u64,
    slice_count: usize,
    work: SlicedVerifyWork,
    /// Caller-attested per-slice verdicts, or `None` when this plan reads the
    /// whole file. When present it is exactly `slice_count` long, and every
    /// `true` entry names a slice the work does not cover.
    proven: Option<Vec<bool>>,
}

/// What one file's slice-proof pass has to read.
///
/// The two shapes are not interchangeable, and the difference is the whole
/// reason evidence pays off instead of costing.
///
/// Without evidence every slice is read, so the work is contiguous spans: one
/// span read fills one buffer that many slices are then hashed out of, in SIMD
/// lanes. That is the shape the fast-verify and post-repair readbacks have
/// always used, and it is untouched here.
///
/// With evidence the slices still to read can be scattered anywhere in the
/// file, and spanning them the same way would hand `md5_multi` one slice per
/// call for a mask that alternates — one lane of an eight-lane engine. That is
/// slower than reading the whole file, which would make the option a
/// pessimisation on exactly the masks a real host produces. So evidence work is
/// a list of slice indices instead, and the lanes are filled from wherever
/// those slices are: batching by lane rather than by adjacency. Reading one
/// slice per lane at an arbitrary offset is the same I/O discipline
/// [`verify_slices_batched_md5`] already uses for a whole file.
enum SlicedVerifyWork {
    /// `(start, count)` runs that together cover every slice, in order.
    Spans(Vec<(usize, usize)>),
    /// The slice indices to read, ascending. May be non-contiguous, and is
    /// empty for a fully proven file — the zero-read case.
    Slices(Vec<usize>),
}

/// Build the [`SlicedVerifyPlan`] for `file_id`, or `None` when the file is
/// ineligible for slice-proof verification: no file description, no or
/// incomplete IFSC checksums (the count must equal the expected slice count),
/// zero length or slice size, or an on-disk length that does not match the
/// description. These are exactly the preconditions the opt-in fast-verify
/// path and the post-repair readback share; an ineligible file falls back to
/// the serial full-hash pipeline.
///
/// `proven` folds in caller-attested per-slice evidence: the work then names
/// only the unproven slices. The eligibility gate above is deliberately
/// unchanged by it — evidence buys a shorter read of a file that was already
/// verifiable slice by slice, and buys nothing anywhere else. In particular the
/// on-disk length must still equal the description's, which is what makes a
/// per-slice verdict addressable at all: the offsets the caller's verdicts were
/// about are the offsets in this file only while its length is the described
/// one. `proven` must be `slice_count` long; callers reach this through
/// [`VerifyOptions::proven_slices_for`], which drops any other shape, and it is
/// re-checked here so no future caller can route around that.
fn sliced_verify_plan(
    par2: &Par2FileSet,
    access: &dyn FileAccess,
    file_id: &FileId,
    proven: Option<&[bool]>,
) -> Option<SlicedVerifyPlan> {
    let desc = par2.file_description(file_id)?;
    let checksums = par2.file_checksums(file_id)?;
    let slice_size = par2.slice_size;
    if desc.length == 0 || slice_size == 0 {
        return None;
    }
    let expected_slices = bounded_slice_count(par2, desc.length)?;
    if expected_slices == 0 || checksums.len() != expected_slices {
        return None;
    }
    if access.file_length(file_id) != Some(desc.length) {
        return None;
    }
    let proven = proven.filter(|proven| proven.len() == expected_slices);

    let work = match proven {
        None => {
            let span_slices = ((VERIFY_SPAN_TARGET_BYTES as u64 / slice_size).max(1) as usize)
                .min(expected_slices);
            SlicedVerifyWork::Spans(
                (0..expected_slices)
                    .step_by(span_slices)
                    .map(|start| (start, span_slices.min(expected_slices - start)))
                    .collect(),
            )
        }
        Some(proven) => SlicedVerifyWork::Slices(
            proven
                .iter()
                .enumerate()
                .filter_map(|(index, proven)| (!*proven).then_some(index))
                .collect(),
        ),
    };
    Some(SlicedVerifyPlan {
        filename: desc.filename.clone(),
        length: desc.length,
        slice_size,
        slice_count: expected_slices,
        work,
        proven: proven.map(<[bool]>::to_vec),
    })
}

/// How many slices one [`md5_simd::md5_multi`] call takes at this slice size:
/// the narrower of what the kernel offers and what the batch memory budget
/// affords, and never zero. Identical to the rule
/// [`verify_slices_batched_md5`] uses, so a scattered evidence read and a
/// whole-file read cost the same per slice and hold the same working set.
fn slice_batch_lanes(slice_size: u64) -> usize {
    let Ok(slice_size) = usize::try_from(slice_size) else {
        return 1;
    };
    if slice_size == 0 {
        return 1;
    }
    (VERIFY_SIMD_BATCH_MEMORY_BYTES / slice_size)
        .min(md5_simd::max_lanes())
        .clamp(1, VERIFY_SIMD_MAX_LANES)
}

/// How many slices one parallel task takes, so a scattered read fans out in
/// units comparable to a span rather than one lane batch at a time. Always a
/// whole number of lane batches, so no task ends mid-batch.
fn slice_task_slices(slice_size: u64, lanes: usize) -> usize {
    let per_task = (VERIFY_SPAN_TARGET_BYTES as u64 / slice_size.max(1)).max(1) as usize;
    per_task.div_ceil(lanes).max(1) * lanes
}

/// Weave the read-backs into one per-slice validity vector.
///
/// The starting point is the caller's evidence, so a proven slice keeps its
/// attested verdict, and every slice the work covered is overwritten by what
/// the read found. Without evidence the spans tile the file, so this is a
/// scatter-shaped `concat`: the result is identical to the full-read vector,
/// which is the property `verify_selected_file_ids` with no evidence relies on.
fn assemble_valid_slices(plan: &SlicedVerifyPlan, results: Vec<Vec<bool>>) -> Vec<bool> {
    let mut valid = match &plan.proven {
        Some(proven) => proven.clone(),
        None => vec![false; plan.slice_count],
    };
    match &plan.work {
        SlicedVerifyWork::Spans(spans) => {
            for (&(start, count), results) in spans.iter().zip(results) {
                debug_assert_eq!(results.len(), count, "a span reports one verdict per slice");
                let end = (start + count).min(valid.len());
                valid[start..end].copy_from_slice(&results[..end - start]);
            }
        }
        SlicedVerifyWork::Slices(indices) => {
            for (index, result) in indices.iter().zip(results.into_iter().flatten()) {
                valid[*index] = result;
            }
        }
    }
    valid
}

/// Verify one lane batch of slices, wherever in the file they are.
///
/// Each slice is read into its own lane buffer at its own offset and the
/// batch is hashed in one [`md5_simd::md5_multi`] call, so a scattered set of
/// slices costs the same per slice as an adjacent one. Padding semantics match
/// [`check_slice_span`] exactly — `md5_multi(.., Some(slice_size))` and
/// [`checksum::crc32_padded`] zero-pad a short tail slice — so a slice this
/// judges is judged the same way the whole-file paths judge it.
///
/// A read error or a short read fails only the slice it happened to, not the
/// batch: a lane batch is an arbitrary grouping, not a contiguous run, so
/// failing its neighbours would report damage that was never observed.
fn check_slice_batch(
    access: &dyn FileAccess,
    file_id: &FileId,
    checksums: &[SliceChecksum],
    plan: &SlicedVerifyPlan,
    indices: &[usize],
    buffers: &mut Vec<Vec<u8>>,
) -> Vec<bool> {
    let slice_size = plan.slice_size;
    let Ok(slice_len) = usize::try_from(slice_size) else {
        return vec![false; indices.len()];
    };
    while buffers.len() < indices.len() {
        buffers.push(vec![0u8; slice_len]);
    }

    let mut read_lens = Vec::with_capacity(indices.len());
    let mut read_complete = Vec::with_capacity(indices.len());
    for (lane, &index) in indices.iter().enumerate() {
        let offset = index as u64 * slice_size;
        let want = plan.length.saturating_sub(offset).min(slice_size);
        let buffer = &mut buffers[lane];
        if buffer.len() < slice_len {
            buffer.resize(slice_len, 0);
        }
        match read_file_slice_into(file_id, offset, want, access, &mut buffer[..slice_len]) {
            Ok(read) => {
                read_lens.push(read);
                read_complete.push(read as u64 == want);
            }
            Err(_) => {
                read_lens.push(0);
                read_complete.push(false);
            }
        }
    }

    let inputs: Vec<&[u8]> = buffers
        .iter()
        .zip(read_lens.iter())
        .map(|(buffer, read)| &buffer[..*read])
        .collect();
    let md5s = md5_simd::md5_multi(&inputs, Some(slice_size));
    indices
        .iter()
        .enumerate()
        .map(|(lane, index)| {
            let expected = &checksums[*index];
            read_complete[lane]
                && checksum::crc32_padded(inputs[lane], slice_size) == expected.crc32
                && md5s[lane] == expected.md5
        })
        .collect()
}

/// Verify a run of slice indices serially, lane batch by lane batch.
fn check_slice_indices(
    access: &dyn FileAccess,
    file_id: &FileId,
    checksums: &[SliceChecksum],
    plan: &SlicedVerifyPlan,
    indices: &[usize],
    buffers: &mut Vec<Vec<u8>>,
) -> Vec<bool> {
    let lanes = slice_batch_lanes(plan.slice_size);
    indices
        .chunks(lanes)
        .flat_map(|batch| check_slice_batch(access, file_id, checksums, plan, batch, buffers))
        .collect()
}

/// Verify one `(start, count)` slice span against its IFSC checksums (CRC32 +
/// MD5, tail slice zero-padded to `slice_size`). Returns one bool per slice in
/// the span; a read error or short read fails the whole span. Shared verbatim
/// by the serial and span-parallel drivers.
///
/// The span is read into the caller's `scratch` buffer (grown once, reused
/// across spans) through the same short-read-looping [`read_file_slice_into`]
/// the strict pipeline uses — a bare `read_file_range_into` is one `read(2)`
/// and may come up short on a multi-megabyte span, which would report the
/// whole span damaged. Its slices are then hashed in SIMD lanes
/// ([`md5_simd::md5_multi`], [`md5_simd::max_lanes`] at a time) rather
/// than one scalar MD5 per slice, matching the strict path's
/// [`verify_slices_batched_md5`]. Padding semantics are unchanged:
/// `md5_multi(.., Some(slice_size))` and [`checksum::crc32_padded`] zero-pad a
/// short tail slice exactly as `SliceChecksumState::finalize(Some(slice_size))`
/// did.
fn check_slice_span(
    access: &dyn FileAccess,
    file_id: &FileId,
    checksums: &[SliceChecksum],
    plan: &SlicedVerifyPlan,
    start: usize,
    count: usize,
    scratch: &mut Vec<u8>,
) -> Vec<bool> {
    let slice_size = plan.slice_size;
    let offset = start as u64 * slice_size;
    let want = (plan.length - offset).min(count as u64 * slice_size);
    let Ok(want_len) = usize::try_from(want) else {
        return vec![false; count];
    };
    if scratch.len() < want_len {
        scratch.resize(want_len, 0);
    }
    let Ok(read) = read_file_slice_into(file_id, offset, want, access, &mut scratch[..want_len])
    else {
        return vec![false; count];
    };
    if read != want_len {
        return vec![false; count];
    }
    let data = &scratch[..want_len];

    let mut valid = Vec::with_capacity(count);
    let mut index = 0usize;
    let kernel_lanes = md5_simd::max_lanes().min(VERIFY_SIMD_MAX_LANES);
    while index < count {
        let lanes = kernel_lanes.min(count - index);
        let mut inputs: [&[u8]; VERIFY_SIMD_MAX_LANES] = [&[]; VERIFY_SIMD_MAX_LANES];
        for (lane, input) in inputs.iter_mut().take(lanes).enumerate() {
            let slice_index = (index + lane) as u64;
            let lo = (slice_index * slice_size).min(want) as usize;
            let hi = ((slice_index + 1) * slice_size).min(want) as usize;
            *input = &data[lo..hi];
        }
        let inputs = &inputs[..lanes];
        let md5s = md5_simd::md5_multi(inputs, Some(slice_size));
        for (lane, input) in inputs.iter().enumerate() {
            let expected = &checksums[start + index + lane];
            let crc32 = checksum::crc32_padded(input, slice_size);
            valid.push(crc32 == expected.crc32 && md5s[lane] == expected.md5);
        }
        index += lanes;
    }
    valid
}

/// Assemble the per-file outcome from a completed slice-validity vector:
/// `Complete` when every slice matched, else `Damaged(n)` carrying the same
/// per-slice validity vector the serial pipeline produces (repair planning
/// consumes `valid_slices` identically, so the damaged shape must match).
fn finish_sliced_verification(
    file_id: &FileId,
    plan: &SlicedVerifyPlan,
    valid_slices: Vec<bool>,
) -> FileVerification {
    let damaged = valid_slices.iter().filter(|valid| !**valid).count() as u32;
    let status = if damaged == 0 {
        FileStatus::Complete
    } else {
        FileStatus::Damaged(damaged)
    };
    FileVerification {
        file_id: *file_id,
        filename: plan.filename.clone(),
        status,
        valid_slices,
        missing_slice_count: damaged,
    }
}

/// Per-slice validity for the strict pipeline's damaged branch.
///
/// Uses the span-parallel scanner when the caller established there is no
/// other rayon axis in flight and the file is eligible for it, else the serial
/// [`verify_slices`]. Both compare the same per-slice CRC32 and zero-padded
/// MD5 against the same IFSC entries, so the vectors agree; only the span
/// iteration differs. An ineligible file (missing or short IFSC, length
/// mismatch) falls back, which is also where a length mismatch lands.
fn strict_slice_validity(
    par2: &Par2FileSet,
    access: &dyn FileAccess,
    span_access: Option<&(dyn FileAccess + Sync)>,
    file_id: &FileId,
    slice_count: usize,
) -> Vec<bool> {
    if let Some(span_access) = span_access
        && let Some(file) = verify_file_sliced(par2, span_access, file_id, true)
        && file.valid_slices.len() == slice_count
    {
        return file.valid_slices;
    }
    verify_slices(par2, file_id, access).unwrap_or_else(|| vec![false; slice_count])
}

/// Wrap a single [`FileVerification`] into a set-level [`VerificationResult`],
/// assessing repairability over just that file. Used by the per-file parallel
/// drivers that fold their partials with [`combine_partial_results`].
fn single_file_result(par2: &Par2FileSet, file: FileVerification) -> VerificationResult {
    let damaged = file.missing_slice_count;
    let files = vec![file];
    let recovery_blocks_available = par2.recovery_block_count();
    let repairable = repairability_for_result_with_resource_limit(
        &files,
        damaged,
        recovery_blocks_available,
        None,
    );
    VerificationResult {
        files,
        recovery_blocks_available,
        total_missing_blocks: damaged,
        repairable,
    }
}

/// Slice-span verification of one file against its IFSC checksums, spans run
/// concurrently when `span_parallel` is set. Returns `None` for ineligible
/// files (see [`sliced_verify_plan`]); the caller falls back to the serial
/// full-hash pipeline.
///
/// Why slice proof + length is sound (and why strict full-MD5 stays the
/// default): the per-slice IFSC checksums are the very CRC32 + MD5 pairs that
/// gate block reuse during scanning, and together they cover every byte of the
/// file (the final slice is zero-padded to `slice_size`). With the on-disk
/// length equal to the description length, a file whose every slice matches
/// CRC32 **and** MD5 can only diverge from the whole-file MD5 if some slice
/// suffered a simultaneous MD5 **and** CRC32 collision. The threat model here
/// is random media damage / bit rot, against which per-slice MD5+CRC32 is as
/// decisive as a whole-file MD5 — so fast verify trades the inherently serial
/// full-file MD5 (~0.8 GB/s wall on fast disks) for the same slice checks run
/// at read speed. It stays **opt-in**: identity establishment and the
/// misplaced/renamed-file scanner keep the stricter whole-file rule, which
/// exists to *establish* a found file's identity, not to re-check the content
/// of a file already matched by length.
fn verify_file_sliced(
    par2: &Par2FileSet,
    access: &(dyn FileAccess + Sync),
    file_id: &FileId,
    span_parallel: bool,
) -> Option<FileVerification> {
    verify_file_sliced_with_evidence(par2, access, file_id, span_parallel, None)
}

/// [`verify_file_sliced`] over only the slices the caller has not proven.
///
/// The slices that *are* read are checked against the identical IFSC entries
/// with the identical padding, so a slice's verdict does not depend on whether
/// its neighbours were skipped; only which slices get read changes.
/// `Complete` still means every slice is accounted for — the difference is
/// that some are accounted for by the caller's attestation rather than by this
/// read. See [`VerifyOptions::proven_slices`] for what that costs if the
/// attestation is wrong.
fn verify_file_sliced_with_evidence(
    par2: &Par2FileSet,
    access: &(dyn FileAccess + Sync),
    file_id: &FileId,
    span_parallel: bool,
    proven: Option<&[bool]>,
) -> Option<FileVerification> {
    use rayon::prelude::*;
    let plan = sliced_verify_plan(par2, access, file_id, proven)?;
    let checksums = par2.file_checksums(file_id)?;
    let results: Vec<Vec<bool>> = match (&plan.work, span_parallel) {
        // `map_init` hands each rayon job its own read buffer, reused across
        // the units that job takes; `collect` preserves order.
        (SlicedVerifyWork::Spans(spans), true) => spans
            .par_iter()
            .map_init(Vec::new, |scratch, &(start, count)| {
                check_slice_span(access, file_id, checksums, &plan, start, count, scratch)
            })
            .collect(),
        (SlicedVerifyWork::Spans(spans), false) => {
            let mut scratch = Vec::new();
            spans
                .iter()
                .map(|&(start, count)| {
                    check_slice_span(
                        access,
                        file_id,
                        checksums,
                        &plan,
                        start,
                        count,
                        &mut scratch,
                    )
                })
                .collect()
        }
        (SlicedVerifyWork::Slices(indices), true) => {
            let lanes = slice_batch_lanes(plan.slice_size);
            indices
                .par_chunks(slice_task_slices(plan.slice_size, lanes))
                .map_init(Vec::new, |buffers, chunk| {
                    check_slice_indices(access, file_id, checksums, &plan, chunk, buffers)
                })
                .collect()
        }
        (SlicedVerifyWork::Slices(indices), false) => {
            let mut buffers = Vec::new();
            vec![check_slice_indices(
                access,
                file_id,
                checksums,
                &plan,
                indices,
                &mut buffers,
            )]
        }
    };
    let valid_slices = assemble_valid_slices(&plan, results);
    Some(finish_sliced_verification(file_id, &plan, valid_slices))
}

/// Serial (span-sequential) sibling of [`verify_file_sliced`] for callers
/// holding a non-`Sync` [`FileAccess`] — namely the serial per-file verify
/// loop, which may itself run inside a file-parallel `par_iter` and so must
/// not open a second rayon axis. Shares the same eligibility gate and per-span
/// checks; only the span iteration differs.
fn verify_file_sliced_serial(
    par2: &Par2FileSet,
    access: &dyn FileAccess,
    file_id: &FileId,
) -> Option<FileVerification> {
    verify_file_sliced_serial_with_evidence(par2, access, file_id, None)
}

/// [`verify_file_sliced_serial`] over only the slices the caller has not
/// proven — the span-serial sibling of
/// [`verify_file_sliced_with_evidence`], used by the per-file verify loop
/// which may itself be running inside a file-parallel `par_iter`.
fn verify_file_sliced_serial_with_evidence(
    par2: &Par2FileSet,
    access: &dyn FileAccess,
    file_id: &FileId,
    proven: Option<&[bool]>,
) -> Option<FileVerification> {
    let plan = sliced_verify_plan(par2, access, file_id, proven)?;
    let checksums = par2.file_checksums(file_id)?;
    let results: Vec<Vec<bool>> = match &plan.work {
        SlicedVerifyWork::Spans(spans) => {
            let mut scratch = Vec::new();
            spans
                .iter()
                .map(|&(start, count)| {
                    check_slice_span(
                        access,
                        file_id,
                        checksums,
                        &plan,
                        start,
                        count,
                        &mut scratch,
                    )
                })
                .collect()
        }
        SlicedVerifyWork::Slices(indices) => {
            let mut buffers = Vec::new();
            vec![check_slice_indices(
                access,
                file_id,
                checksums,
                &plan,
                indices,
                &mut buffers,
            )]
        }
    };
    let valid_slices = assemble_valid_slices(&plan, results);
    Some(finish_sliced_verification(file_id, &plan, valid_slices))
}

/// Slice-level readback verification of one staged file. Thin wrapper over
/// [`verify_file_sliced`] (see there for the completeness argument) that lifts
/// the per-file outcome into a set-level result. `Complete` from slice proof
/// is additionally sound for a staged file because the repair just built it
/// from the set's own layout, so its identity is definitional. Returns `None`
/// for ineligible files (no description, no or incomplete IFSC, zero length,
/// or a length mismatch) — the caller falls back to the serial pipeline.
fn verify_repaired_file_sliced(
    par2: &Par2FileSet,
    access: &(dyn FileAccess + Sync),
    file_id: &FileId,
    span_parallel: bool,
) -> Option<VerificationResult> {
    verify_file_sliced(par2, access, file_id, span_parallel)
        .map(|file| single_file_result(par2, file))
}

/// Fold per-file verification results into one set-level result, summing
/// totals and reassessing repairability once over the combined files.
fn combine_partial_results(
    par2: &Par2FileSet,
    partials: Vec<VerificationResult>,
) -> VerificationResult {
    let mut files = Vec::with_capacity(partials.len());
    let mut total_missing_blocks = 0u32;
    let mut resource_limit_reason = None;
    for partial in partials {
        total_missing_blocks = total_missing_blocks.saturating_add(partial.total_missing_blocks);
        if let Repairability::ResourceLimited { reason } = &partial.repairable {
            resource_limit_reason.get_or_insert_with(|| reason.clone());
        }
        files.extend(partial.files);
    }

    let recovery_blocks_available = par2.recovery_block_count();
    let repairable = repairability_for_result_with_resource_limit(
        &files,
        total_missing_blocks,
        recovery_blocks_available,
        resource_limit_reason,
    );
    VerificationResult {
        files,
        recovery_blocks_available,
        total_missing_blocks,
        repairable,
    }
}

/// Overlay `updated` per-file verifications onto `base` (matched by file
/// id), recomputing the totals and set-level repairability. Files absent
/// from `updated` keep their `base` entry — the caller asserts their state
/// on disk has not changed since `base` was computed.
pub fn merge_verification_results(
    par2: &Par2FileSet,
    base: &VerificationResult,
    updated: VerificationResult,
) -> VerificationResult {
    let mut updated_by_id: HashMap<FileId, FileVerification> = updated
        .files
        .into_iter()
        .map(|file| (file.file_id, file))
        .collect();
    let files: Vec<FileVerification> = base
        .files
        .iter()
        .map(|file| {
            updated_by_id
                .remove(&file.file_id)
                .unwrap_or_else(|| file.clone())
        })
        .collect();

    let mut total_missing_blocks = 0u32;
    for file in &files {
        total_missing_blocks = total_missing_blocks.saturating_add(file.missing_slice_count);
    }
    let resource_limit_reason = match &updated.repairable {
        Repairability::ResourceLimited { reason } => Some(reason.clone()),
        _ => match &base.repairable {
            Repairability::ResourceLimited { reason } => Some(reason.clone()),
            _ => None,
        },
    };

    let recovery_blocks_available = par2.recovery_block_count();
    let repairable = repairability_for_result_with_resource_limit(
        &files,
        total_missing_blocks,
        recovery_blocks_available,
        resource_limit_reason,
    );
    VerificationResult {
        files,
        recovery_blocks_available,
        total_missing_blocks,
        repairable,
    }
}

/// Verify selected PAR2 file IDs with cancellation and progress support.
pub fn verify_selected_file_ids_with_options(
    par2: &Par2FileSet,
    access: &dyn FileAccess,
    file_ids: &[FileId],
    options: &VerifyOptions,
) -> VerificationResult {
    // Resolved once per call: env override wins, otherwise the option flag.
    let fast_verify = fast_verify_enabled(options.fast_verify);
    verify_selected_file_ids_resolved(par2, access, None, file_ids, options, fast_verify)
}

/// [`verify_selected_file_ids_with_options`] with `fast_verify` already
/// resolved. Split out (like [`resolve_fast_verify`]) so tests that assert
/// a specific pipeline's I/O discipline can pin the mode instead of
/// inheriting an ambient `WEAVER_PAR2_FAST_VERIFY` from the test
/// environment.
///
/// `span_access` is the same object as `access`, supplied only by a caller
/// that has established there is no other rayon axis in flight (a lone file).
/// It lets the damaged branch's slice scan fan out across spans instead of
/// walking them serially; the per-slice results are identical either way.
fn verify_selected_file_ids_resolved(
    par2: &Par2FileSet,
    access: &dyn FileAccess,
    span_access: Option<&(dyn FileAccess + Sync)>,
    file_ids: &[FileId],
    options: &VerifyOptions,
    fast_verify: bool,
) -> VerificationResult {
    let mut files = Vec::new();
    let mut total_missing_blocks = 0u32;
    let mut resource_limit_reason = None;
    let total_files = file_ids.len() as u32;
    let mut bytes_processed = 0u64;

    for (file_index, file_id) in file_ids.iter().enumerate() {
        // Check cancellation before each file.
        if let Some(ref cancel) = options.cancel
            && cancel.is_cancelled()
        {
            break;
        }

        let desc = match par2.file_description(file_id) {
            Some(d) => d,
            None => continue,
        };
        let Some(slice_count) = bounded_slice_count(par2, desc.length) else {
            resource_limit_reason.get_or_insert_with(|| {
                format!("file {} exceeds verifier slice limits", desc.filename)
            });
            files.push(resource_limited_verification(
                *file_id,
                desc.filename.clone(),
            ));
            continue;
        };
        let slice_count_u32 = slice_count as u32;

        if !access.file_exists(file_id) {
            total_missing_blocks = total_missing_blocks.saturating_add(slice_count_u32);
            files.push(FileVerification {
                file_id: *file_id,
                filename: desc.filename.clone(),
                status: FileStatus::Missing,
                valid_slices: vec![false; slice_count],
                missing_slice_count: slice_count_u32,
            });
            continue;
        }

        let Some(actual_len) = access.file_length(file_id) else {
            total_missing_blocks = total_missing_blocks.saturating_add(slice_count_u32);
            files.push(FileVerification {
                file_id: *file_id,
                filename: desc.filename.clone(),
                status: FileStatus::Damaged(slice_count_u32),
                valid_slices: vec![false; slice_count],
                missing_slice_count: slice_count_u32,
            });
            continue;
        };

        // Empty file: PAR2 spec says 0-length files have 0 slices.
        // Just verify the file exists and has the expected length.
        if desc.length == 0 {
            let status = if actual_len == 0 {
                FileStatus::Complete
            } else {
                FileStatus::Damaged(0)
            };
            files.push(FileVerification {
                file_id: *file_id,
                filename: desc.filename.clone(),
                status,
                valid_slices: vec![],
                missing_slice_count: 0,
            });
            continue;
        }

        // Check for files that should have content but are empty on disk.
        if actual_len == 0 && desc.length > 0 {
            total_missing_blocks = total_missing_blocks.saturating_add(slice_count_u32);
            files.push(FileVerification {
                file_id: *file_id,
                filename: desc.filename.clone(),
                status: FileStatus::Damaged(slice_count_u32),
                valid_slices: vec![false; slice_count],
                missing_slice_count: slice_count_u32,
            });
            continue;
        }

        // If IFSC data is missing for this file, we can't do slice-level verification.
        // Fall back to full-file hash check only. This can happen with truncated PAR2 files.
        if par2.file_checksums(file_id).is_none() {
            let full_ok = verify_full_hash(par2, file_id, access).unwrap_or(false);
            if full_ok {
                files.push(FileVerification {
                    file_id: *file_id,
                    filename: desc.filename.clone(),
                    status: FileStatus::Complete,
                    valid_slices: vec![true; slice_count],
                    missing_slice_count: 0,
                });
            } else {
                // Can't determine which slices are bad without IFSC data.
                // Mark all slices as damaged — repair will treat the whole file as needing recovery.
                total_missing_blocks = total_missing_blocks.saturating_add(slice_count_u32);
                files.push(FileVerification {
                    file_id: *file_id,
                    filename: desc.filename.clone(),
                    status: FileStatus::Damaged(slice_count_u32),
                    valid_slices: vec![false; slice_count],
                    missing_slice_count: slice_count_u32,
                });
            }
            bytes_processed += desc.length;
            if let Some(ref progress) = options.progress {
                progress(ProgressUpdate {
                    stage: ProgressStage::Verifying,
                    current: file_index as u32 + 1,
                    total: total_files,
                    bytes_processed,
                    total_bytes: None,
                    phase: ProgressPhase::Whole,
                });
            }
            continue;
        }

        if actual_len != desc.length {
            let valid = strict_slice_validity(par2, access, span_access, file_id, slice_count);
            let damaged = valid.iter().filter(|&&v| !v).count() as u32;
            total_missing_blocks = total_missing_blocks.saturating_add(damaged);
            files.push(FileVerification {
                file_id: *file_id,
                filename: desc.filename.clone(),
                status: FileStatus::Damaged(damaged),
                valid_slices: valid,
                missing_slice_count: damaged,
            });
            continue;
        }

        // Per-slice evidence (opt-in): the caller has already proven some of
        // this file's slices, so read only the ones it has not. The spans that
        // are read take the same `check_slice_span` against the same IFSC
        // entries the fast and strict arms use, which is why composing this
        // with either is not a third verdict rule but the same one over fewer
        // slices. The whole-file MD5 arm cannot run once any slice is skipped —
        // there is no way to hash bytes that were never read — so completeness
        // is decided from slice proof, exactly the shape fast verify already
        // reports (see `verify_file_sliced` for why slice proof plus a matching
        // length is sound). An entry that proves nothing, or is the wrong shape
        // for this file's layout, never reaches here: `proven_slices_for` drops
        // it, and the file takes the pipeline it always took.
        if let Some(proven) = options.proven_slices_for(file_id, slice_count)
            && let Some(evidenced) =
                verify_file_sliced_serial_with_evidence(par2, access, file_id, Some(proven))
        {
            total_missing_blocks =
                total_missing_blocks.saturating_add(evidenced.missing_slice_count);
            files.push(evidenced);
            // Progress counts the file's coverage, not this pass's reads: the
            // evidence accounts for the bytes it did not read, and a progress
            // stream that stalled on a skipped range would misreport the work
            // remaining.
            bytes_processed += desc.length;
            if let Some(ref progress) = options.progress {
                progress(ProgressUpdate {
                    stage: ProgressStage::Verifying,
                    current: file_index as u32 + 1,
                    total: total_files,
                    bytes_processed,
                    total_bytes: None,
                    phase: ProgressPhase::Whole,
                });
            }
            continue;
        }

        // Fast-verify (opt-in): the on-disk length already matches the
        // description here, so prove completeness from the per-slice IFSC
        // checksums scanned at read speed instead of the inherently serial
        // full-file MD5. All slices valid -> Complete; any slice invalid ->
        // Damaged with the same per-slice vector the strict path yields, so no
        // full-MD5 fallback adds information (see `verify_file_sliced` for why
        // slice proof + length is sound and why strict stays the default). The
        // 16k quick check is intentionally skipped: the slice scan already
        // covers the file's first bytes. Ineligible files (e.g. an IFSC count
        // mismatch) return `None` and fall through to the strict pipeline
        // below. This loop can run inside a file-parallel `par_iter`, so it
        // uses the span-serial driver to avoid nesting a second rayon axis.
        if fast_verify && let Some(fast) = verify_file_sliced_serial(par2, access, file_id) {
            total_missing_blocks = total_missing_blocks.saturating_add(fast.missing_slice_count);
            files.push(fast);
            bytes_processed += desc.length;
            if let Some(ref progress) = options.progress {
                progress(ProgressUpdate {
                    stage: ProgressStage::Verifying,
                    current: file_index as u32 + 1,
                    total: total_files,
                    bytes_processed,
                    total_bytes: None,
                    phase: ProgressPhase::Whole,
                });
            }
            continue;
        }

        // One streaming pass computes the 16 KiB and whole-file MD5 chains and
        // CRC32s each slice out of the same buffer. Passing the IFSC entries is
        // gated on the count matching this file's slice count, which is the
        // precondition under which `verify_slices` indexes them; `actual_len ==
        // desc.length` is already established above.
        let checksums = par2
            .file_checksums(file_id)
            .filter(|checksums| checksums.len() == slice_count);
        let outcome = stream_strict_hashes(par2, file_id, access, checksums).unwrap_or(
            StrictStreamOutcome::Hashes {
                quick_ok: false,
                full_ok: false,
            },
        );

        // Only an intact 16 KiB prefix *and* an intact whole-file MD5 mark a
        // file complete; every other outcome is the damaged branch, which
        // reports the per-slice validity vector. That collapse is why the
        // stream may abandon the MD5 chain once a slice CRC32 has proven the
        // content differs: the abandoned chain's only reachable verdict was
        // `full_ok == false`, which lands in this same branch. Believing
        // otherwise would require an MD5 second-preimage — the assumption the
        // `full_ok == true` arm already rests on.
        let complete = matches!(
            outcome,
            StrictStreamOutcome::Hashes {
                quick_ok: true,
                full_ok: true
            }
        );
        if complete {
            files.push(FileVerification {
                file_id: *file_id,
                filename: desc.filename.clone(),
                status: FileStatus::Complete,
                valid_slices: vec![true; slice_count],
                missing_slice_count: 0,
            });
        } else {
            let valid = strict_slice_validity(par2, access, span_access, file_id, slice_count);
            let damaged = valid.iter().filter(|&&v| !v).count() as u32;
            total_missing_blocks = total_missing_blocks.saturating_add(damaged);
            // Slice checks can identify usable blocks, but only a full
            // length+MD5 match is allowed to mark a file complete, so a
            // zero-damage slice scan still reports `Damaged(0)` here.
            files.push(FileVerification {
                file_id: *file_id,
                filename: desc.filename.clone(),
                status: FileStatus::Damaged(damaged),
                valid_slices: valid,
                missing_slice_count: damaged,
            });
        }

        bytes_processed += desc.length;
        if let Some(ref progress) = options.progress {
            progress(ProgressUpdate {
                stage: ProgressStage::Verifying,
                current: file_index as u32 + 1,
                total: total_files,
                bytes_processed,
                total_bytes: None,
                phase: ProgressPhase::Whole,
            });
        }
    }

    let recovery_blocks_available = par2.recovery_block_count();
    let repairable = repairability_for_result_with_resource_limit(
        &files,
        total_missing_blocks,
        recovery_blocks_available,
        resource_limit_reason,
    );

    VerificationResult {
        files,
        recovery_blocks_available,
        total_missing_blocks,
        repairable,
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::checksum::{self, SliceChecksumState};
    use crate::packet::header;
    use crate::par2_set::Par2FileSet;
    use crate::types::SliceChecksum;
    use md5::{Digest, Md5};
    use std::collections::HashMap;
    use std::io::{self, Cursor};
    use std::sync::{
        Arc,
        atomic::{AtomicUsize, Ordering},
    };

    /// Helper to build a complete valid packet (header + body).
    fn make_full_packet(packet_type: &[u8; 16], body: &[u8], recovery_set_id: [u8; 16]) -> Vec<u8> {
        let length = (header::HEADER_SIZE + body.len()) as u64;
        let mut hash_input = Vec::new();
        hash_input.extend_from_slice(&recovery_set_id);
        hash_input.extend_from_slice(packet_type);
        hash_input.extend_from_slice(body);
        let packet_hash: [u8; 16] = Md5::digest(&hash_input).into();

        let mut data = Vec::new();
        data.extend_from_slice(header::MAGIC);
        data.extend_from_slice(&length.to_le_bytes());
        data.extend_from_slice(&packet_hash);
        data.extend_from_slice(&recovery_set_id);
        data.extend_from_slice(packet_type);
        data.extend_from_slice(body);
        data
    }

    /// Build a Par2FileSet and MemoryFileAccess for testing.
    /// Creates a single file with known content and matching checksums.
    fn setup_test_set(
        file_data: &[u8],
        slice_size: u64,
    ) -> (Par2FileSet, MemoryFileAccess, FileId) {
        setup_test_set_with_full_hash(file_data, slice_size, None)
    }

    /// [`setup_test_set`], optionally describing a different whole-file MD5
    /// than the data actually has. That combination cannot arise from real
    /// damage (it would need an MD5 collision), but it is the only way to
    /// exercise the strict pipeline's "whole-file hash disagrees while every
    /// slice checks out" verdict.
    fn setup_test_set_with_full_hash(
        file_data: &[u8],
        slice_size: u64,
        full_hash_override: Option<[u8; 16]>,
    ) -> (Par2FileSet, MemoryFileAccess, FileId) {
        let file_length = file_data.len() as u64;
        let hash_full = full_hash_override.unwrap_or_else(|| checksum::md5(file_data));
        let hash_16k_data = &file_data[..file_data.len().min(16384)];
        let hash_16k = checksum::md5(hash_16k_data);

        // Compute file_id: MD5(hash_16k || length || filename)
        let filename = b"testfile.dat";
        let mut id_input = Vec::new();
        id_input.extend_from_slice(&hash_16k);
        id_input.extend_from_slice(&file_length.to_le_bytes());
        id_input.extend_from_slice(filename);
        let file_id_bytes: [u8; 16] = Md5::digest(&id_input).into();
        let file_id = FileId::from_bytes(file_id_bytes);

        // Build slice checksums
        let num_slices = if file_length == 0 {
            0
        } else {
            file_length.div_ceil(slice_size) as usize
        };

        let mut checksums = Vec::new();
        for i in 0..num_slices {
            let offset = i as u64 * slice_size;
            let end = ((offset + slice_size) as usize).min(file_data.len());
            let slice_data = &file_data[offset as usize..end];

            let mut state = SliceChecksumState::new();
            state.update(slice_data);
            let pad_to = if (slice_data.len() as u64) < slice_size {
                Some(slice_size)
            } else {
                None
            };
            let (crc, md5) = state.finalize(pad_to);
            checksums.push(SliceChecksum { crc32: crc, md5 });
        }

        // Build main body and RSID
        let mut main_body = Vec::new();
        main_body.extend_from_slice(&slice_size.to_le_bytes());
        main_body.extend_from_slice(&1u32.to_le_bytes());
        main_body.extend_from_slice(&file_id_bytes);
        let rsid: [u8; 16] = Md5::digest(&main_body).into();

        // File description body
        let mut fd_body = Vec::new();
        fd_body.extend_from_slice(&file_id_bytes);
        fd_body.extend_from_slice(&hash_full);
        fd_body.extend_from_slice(&hash_16k);
        fd_body.extend_from_slice(&file_length.to_le_bytes());
        fd_body.extend_from_slice(filename);
        // Pad to multiple of 4
        while fd_body.len() % 4 != 0 {
            fd_body.push(0);
        }

        // IFSC body
        let mut ifsc_body = Vec::new();
        ifsc_body.extend_from_slice(&file_id_bytes);
        for cs in &checksums {
            ifsc_body.extend_from_slice(&cs.md5);
            ifsc_body.extend_from_slice(&cs.crc32.to_le_bytes());
        }

        // Build stream
        let mut stream = Vec::new();
        stream.extend_from_slice(&make_full_packet(header::TYPE_MAIN, &main_body, rsid));
        stream.extend_from_slice(&make_full_packet(header::TYPE_FILE_DESC, &fd_body, rsid));
        stream.extend_from_slice(&make_full_packet(header::TYPE_IFSC, &ifsc_body, rsid));

        let set = Par2FileSet::from_files(&[&stream]).unwrap();

        let mut access = MemoryFileAccess::new();
        access.add_file(file_id, file_data.to_vec());

        (set, access, file_id)
    }

    fn setup_test_set_multi(
        files: &[(&[u8], &str)],
        slice_size: u64,
    ) -> (Par2FileSet, MemoryFileAccess, Vec<FileId>) {
        let mut file_ids = Vec::new();
        let mut fd_bodies = Vec::new();
        let mut ifsc_bodies = Vec::new();
        let mut access = MemoryFileAccess::new();

        for &(file_data, filename) in files {
            let file_length = file_data.len() as u64;
            let hash_full = checksum::md5(file_data);
            let hash_16k = checksum::md5(&file_data[..file_data.len().min(16384)]);

            let mut id_input = Vec::new();
            id_input.extend_from_slice(&hash_16k);
            id_input.extend_from_slice(&file_length.to_le_bytes());
            id_input.extend_from_slice(filename.as_bytes());
            let file_id_bytes: [u8; 16] = Md5::digest(&id_input).into();
            let file_id = FileId::from_bytes(file_id_bytes);
            file_ids.push(file_id);
            access.add_file(file_id, file_data.to_vec());

            let num_slices = if file_length == 0 {
                0
            } else {
                file_length.div_ceil(slice_size) as usize
            };

            let mut checksums = Vec::new();
            for i in 0..num_slices {
                let offset = i as u64 * slice_size;
                let end = ((offset + slice_size) as usize).min(file_data.len());
                let slice_data = &file_data[offset as usize..end];

                let mut state = SliceChecksumState::new();
                state.update(slice_data);
                let pad_to = if (slice_data.len() as u64) < slice_size {
                    Some(slice_size)
                } else {
                    None
                };
                let (crc, md5) = state.finalize(pad_to);
                checksums.push(SliceChecksum { crc32: crc, md5 });
            }

            let mut fd_body = Vec::new();
            fd_body.extend_from_slice(&file_id_bytes);
            fd_body.extend_from_slice(&hash_full);
            fd_body.extend_from_slice(&hash_16k);
            fd_body.extend_from_slice(&file_length.to_le_bytes());
            fd_body.extend_from_slice(filename.as_bytes());
            while fd_body.len() % 4 != 0 {
                fd_body.push(0);
            }
            fd_bodies.push(fd_body);

            let mut ifsc_body = Vec::new();
            ifsc_body.extend_from_slice(&file_id_bytes);
            for cs in &checksums {
                ifsc_body.extend_from_slice(&cs.md5);
                ifsc_body.extend_from_slice(&cs.crc32.to_le_bytes());
            }
            ifsc_bodies.push(ifsc_body);
        }

        let mut main_body = Vec::new();
        main_body.extend_from_slice(&slice_size.to_le_bytes());
        main_body.extend_from_slice(&(file_ids.len() as u32).to_le_bytes());
        for file_id in &file_ids {
            main_body.extend_from_slice(file_id.as_bytes());
        }
        let rsid: [u8; 16] = Md5::digest(&main_body).into();

        let mut stream = Vec::new();
        stream.extend_from_slice(&make_full_packet(header::TYPE_MAIN, &main_body, rsid));
        for fd_body in &fd_bodies {
            stream.extend_from_slice(&make_full_packet(header::TYPE_FILE_DESC, fd_body, rsid));
        }
        for ifsc_body in &ifsc_bodies {
            stream.extend_from_slice(&make_full_packet(header::TYPE_IFSC, ifsc_body, rsid));
        }

        let set = Par2FileSet::from_files(&[&stream]).unwrap();
        (set, access, file_ids)
    }

    fn setup_oversized_file_set() -> Par2FileSet {
        let slice_size = 4u64;
        let file_length = (MAX_SLICES_PER_FILE as u64 + 1) * slice_size;
        let filename = b"huge.dat";
        let hash_full = [0u8; 16];
        let hash_16k = [0u8; 16];

        let mut id_input = Vec::new();
        id_input.extend_from_slice(&hash_16k);
        id_input.extend_from_slice(&file_length.to_le_bytes());
        id_input.extend_from_slice(filename);
        let file_id_bytes: [u8; 16] = Md5::digest(&id_input).into();

        let mut main_body = Vec::new();
        main_body.extend_from_slice(&slice_size.to_le_bytes());
        main_body.extend_from_slice(&1u32.to_le_bytes());
        main_body.extend_from_slice(&file_id_bytes);
        let rsid: [u8; 16] = Md5::digest(&main_body).into();

        let mut fd_body = Vec::new();
        fd_body.extend_from_slice(&file_id_bytes);
        fd_body.extend_from_slice(&hash_full);
        fd_body.extend_from_slice(&hash_16k);
        fd_body.extend_from_slice(&file_length.to_le_bytes());
        fd_body.extend_from_slice(filename);
        while fd_body.len() % 4 != 0 {
            fd_body.push(0);
        }

        let mut stream = Vec::new();
        stream.extend_from_slice(&make_full_packet(header::TYPE_MAIN, &main_body, rsid));
        stream.extend_from_slice(&make_full_packet(header::TYPE_FILE_DESC, &fd_body, rsid));

        Par2FileSet::from_files(&[&stream]).unwrap()
    }

    struct ReadIntoOnlyAccess {
        files: HashMap<FileId, Vec<u8>>,
    }

    impl FileAccess for ReadIntoOnlyAccess {
        fn read_file_range(
            &self,
            _file_id: &FileId,
            _offset: u64,
            _len: u64,
        ) -> io::Result<Vec<u8>> {
            panic!("read_file_range should not be used by quick_check_16k")
        }

        fn read_file_range_into(
            &self,
            file_id: &FileId,
            offset: u64,
            dst: &mut [u8],
        ) -> io::Result<usize> {
            let data = self
                .files
                .get(file_id)
                .ok_or_else(|| io::Error::new(io::ErrorKind::NotFound, "file not found"))?;
            let offset = offset as usize;
            if offset >= data.len() {
                return Ok(0);
            }
            let end = (offset + dst.len()).min(data.len());
            let read_len = end - offset;
            dst[..read_len].copy_from_slice(&data[offset..end]);
            Ok(read_len)
        }

        fn file_exists(&self, file_id: &FileId) -> bool {
            self.files.contains_key(file_id)
        }

        fn file_length(&self, file_id: &FileId) -> Option<u64> {
            self.files.get(file_id).map(|data| data.len() as u64)
        }

        fn read_file(&self, file_id: &FileId) -> io::Result<Vec<u8>> {
            self.files
                .get(file_id)
                .cloned()
                .ok_or_else(|| io::Error::new(io::ErrorKind::NotFound, "file not found"))
        }

        fn write_file_range(
            &mut self,
            _file_id: &FileId,
            _offset: u64,
            _data: &[u8],
        ) -> io::Result<()> {
            Err(io::Error::new(
                io::ErrorKind::Unsupported,
                "test access is read-only",
            ))
        }
    }

    struct SequentialOnlyAccess {
        files: HashMap<FileId, Vec<u8>>,
    }

    impl FileAccess for SequentialOnlyAccess {
        fn read_file_range(
            &self,
            _file_id: &FileId,
            _offset: u64,
            _len: u64,
        ) -> io::Result<Vec<u8>> {
            panic!("read_file_range should not be used when a sequential reader is available")
        }

        fn read_file_range_into(
            &self,
            _file_id: &FileId,
            _offset: u64,
            _dst: &mut [u8],
        ) -> io::Result<usize> {
            panic!("read_file_range_into should not be used when a sequential reader is available")
        }

        fn open_sequential_reader(&self, file_id: &FileId) -> io::Result<Option<Box<dyn Read>>> {
            let data = self
                .files
                .get(file_id)
                .ok_or_else(|| io::Error::new(io::ErrorKind::NotFound, "file not found"))?;
            Ok(Some(Box::new(Cursor::new(data.clone()))))
        }

        fn file_exists(&self, file_id: &FileId) -> bool {
            self.files.contains_key(file_id)
        }

        fn file_length(&self, file_id: &FileId) -> Option<u64> {
            self.files.get(file_id).map(|data| data.len() as u64)
        }

        fn read_file(&self, file_id: &FileId) -> io::Result<Vec<u8>> {
            self.files
                .get(file_id)
                .cloned()
                .ok_or_else(|| io::Error::new(io::ErrorKind::NotFound, "file not found"))
        }

        fn write_file_range(
            &mut self,
            _file_id: &FileId,
            _offset: u64,
            _data: &[u8],
        ) -> io::Result<()> {
            Err(io::Error::new(
                io::ErrorKind::Unsupported,
                "test access is read-only",
            ))
        }
    }

    struct CountingSequentialAccess {
        files: HashMap<FileId, Vec<u8>>,
        open_calls: Arc<AtomicUsize>,
    }

    impl FileAccess for CountingSequentialAccess {
        fn read_file_range(
            &self,
            _file_id: &FileId,
            _offset: u64,
            _len: u64,
        ) -> io::Result<Vec<u8>> {
            panic!("read_file_range should not be used when a sequential reader is available")
        }

        fn read_file_range_into(
            &self,
            _file_id: &FileId,
            _offset: u64,
            _dst: &mut [u8],
        ) -> io::Result<usize> {
            panic!("read_file_range_into should not be used when a sequential reader is available")
        }

        fn open_sequential_reader(&self, file_id: &FileId) -> io::Result<Option<Box<dyn Read>>> {
            self.open_calls.fetch_add(1, Ordering::Relaxed);
            let data = self
                .files
                .get(file_id)
                .ok_or_else(|| io::Error::new(io::ErrorKind::NotFound, "file not found"))?;
            Ok(Some(Box::new(Cursor::new(data.clone()))))
        }

        fn file_exists(&self, file_id: &FileId) -> bool {
            self.files.contains_key(file_id)
        }

        fn file_length(&self, file_id: &FileId) -> Option<u64> {
            self.files.get(file_id).map(|data| data.len() as u64)
        }

        fn read_file(&self, _file_id: &FileId) -> io::Result<Vec<u8>> {
            Err(io::Error::new(
                io::ErrorKind::Unsupported,
                "force streaming path in this test",
            ))
        }

        fn write_file_range(
            &mut self,
            _file_id: &FileId,
            _offset: u64,
            _data: &[u8],
        ) -> io::Result<()> {
            Err(io::Error::new(
                io::ErrorKind::Unsupported,
                "test access is read-only",
            ))
        }
    }

    #[test]
    fn verify_slices_from_crcs_intact() {
        let file_data = vec![0xABu8; 2048];
        let (set, _access, file_id) = setup_test_set(&file_data, 1024);

        // Compute per-slice CRC32s manually.
        let crc0 = checksum::crc32(&file_data[0..1024]);
        let crc1 = checksum::crc32(&file_data[1024..2048]);

        let result = verify_slices_from_crcs(&set, &file_id, &[crc0, crc1]).unwrap();
        assert_eq!(result, vec![true, true]);
    }

    #[test]
    fn verify_slices_from_crcs_damaged() {
        let file_data = vec![0xABu8; 2048];
        let (set, _access, file_id) = setup_test_set(&file_data, 1024);

        let crc0 = checksum::crc32(&file_data[0..1024]);
        let wrong_crc = 0xDEADBEEF;

        let result = verify_slices_from_crcs(&set, &file_id, &[crc0, wrong_crc]).unwrap();
        assert_eq!(result, vec![true, false]);
    }

    #[test]
    fn verify_slices_from_crcs_with_padding() {
        // 1500 bytes, 1024 slice size -> last slice is 476 bytes, zero-padded to 1024
        let file_data = vec![0xCDu8; 1500];
        let (set, _access, file_id) = setup_test_set(&file_data, 1024);

        let crc0 = checksum::crc32(&file_data[0..1024]);
        // PAR2 checksums include zero-padding on the last slice
        let mut padded_last = file_data[1024..1500].to_vec();
        padded_last.resize(1024, 0);
        let crc1 = checksum::crc32(&padded_last);

        let result = verify_slices_from_crcs(&set, &file_id, &[crc0, crc1]).unwrap();
        assert_eq!(result, vec![true, true]);
    }

    #[test]
    fn verify_intact_file() {
        let file_data = vec![0xABu8; 2048];
        let (set, access, file_id) = setup_test_set(&file_data, 1024);

        // Quick check
        assert_eq!(quick_check_16k(&set, &file_id, &access), Some(true));

        // Full hash
        assert_eq!(verify_full_hash(&set, &file_id, &access), Some(true));

        // Slice verification
        let slices = verify_slices(&set, &file_id, &access).unwrap();
        assert_eq!(slices.len(), 2);
        assert!(slices.iter().all(|&v| v));

        // Full verification
        let result = verify_all(&set, &access);
        assert_eq!(result.files.len(), 1);
        assert!(matches!(result.files[0].status, FileStatus::Complete));
        assert_eq!(result.total_missing_blocks, 0);
        assert!(matches!(result.repairable, Repairability::NotNeeded));
    }

    #[test]
    fn verify_damaged_file() {
        let mut file_data = vec![0xABu8; 2048];
        let (set, mut access, file_id) = setup_test_set(&file_data, 1024);

        // Corrupt the second slice
        file_data[1024] ^= 0xFF;
        file_data[1025] ^= 0xFF;
        access.add_file(file_id, file_data);

        // Slice verification should show second slice damaged
        let slices = verify_slices(&set, &file_id, &access).unwrap();
        assert!(slices[0]); // first slice intact
        assert!(!slices[1]); // second slice damaged

        // Full verification
        let result = verify_all(&set, &access);
        assert_eq!(result.files.len(), 1);
        assert!(matches!(result.files[0].status, FileStatus::Damaged(1)));
        assert_eq!(result.total_missing_blocks, 1);
    }

    #[test]
    fn verify_missing_file() {
        let file_data = vec![0xABu8; 2048];
        let (set, _, _file_id) = setup_test_set(&file_data, 1024);

        // Don't add file to access
        let access = MemoryFileAccess::new();

        let result = verify_all(&set, &access);
        assert_eq!(result.files.len(), 1);
        assert!(matches!(result.files[0].status, FileStatus::Missing));
        assert_eq!(result.total_missing_blocks, 2);
    }

    #[test]
    fn verify_file_with_partial_last_slice() {
        // File is 1500 bytes with 1024-byte slices -> 2 slices, last is 476 bytes
        let file_data = vec![0xCDu8; 1500];
        let (set, access, file_id) = setup_test_set(&file_data, 1024);

        let slices = verify_slices(&set, &file_id, &access).unwrap();
        assert_eq!(slices.len(), 2);
        assert!(slices[0]);
        assert!(slices[1]); // Last partial slice should still verify with padding
    }

    #[test]
    fn quick_check_missing_file() {
        let file_data = vec![0xABu8; 100];
        let (set, _, file_id) = setup_test_set(&file_data, 1024);
        let access = MemoryFileAccess::new();

        assert_eq!(quick_check_16k(&set, &file_id, &access), Some(false));
    }

    #[test]
    fn quick_check_short_file_matches_and_detects_corruption() {
        let file_data = b"short-par2-file".to_vec();
        let (set, mut access, file_id) = setup_test_set(&file_data, 1024);

        assert_eq!(quick_check_16k(&set, &file_id, &access), Some(true));

        let mut corrupted = file_data.clone();
        corrupted[3] ^= 0xFF;
        access.add_file(file_id, corrupted);

        assert_eq!(quick_check_16k(&set, &file_id, &access), Some(false));
    }

    #[test]
    fn quick_check_exact_16k_boundary_matches_and_detects_corruption() {
        let file_data = (0..QUICK_CHECK_16K_BYTES)
            .map(|i| (i % 251) as u8)
            .collect::<Vec<_>>();
        let (set, mut access, file_id) = setup_test_set(&file_data, 4096);

        assert_eq!(quick_check_16k(&set, &file_id, &access), Some(true));

        let mut corrupted = file_data.clone();
        corrupted[QUICK_CHECK_16K_BYTES - 1] ^= 0x55;
        access.add_file(file_id, corrupted);

        assert_eq!(quick_check_16k(&set, &file_id, &access), Some(false));
    }

    #[test]
    fn quick_check_uses_read_into_path() {
        let file_data = vec![0x5Au8; 4096];
        let (set, access, file_id) = setup_test_set(&file_data, 1024);
        let access = ReadIntoOnlyAccess {
            files: access.files,
        };

        assert_eq!(quick_check_16k(&set, &file_id, &access), Some(true));
    }

    #[test]
    fn quick_check_uses_sequential_reader_when_available() {
        let file_data = vec![0x6Bu8; 4096];
        let (set, access, file_id) = setup_test_set(&file_data, 1024);
        let access = SequentialOnlyAccess {
            files: access.files,
        };

        assert_eq!(quick_check_16k(&set, &file_id, &access), Some(true));
    }

    #[test]
    fn verify_full_hash_uses_sequential_reader_when_available() {
        let file_data = (0..(QUICK_CHECK_16K_BYTES + 4096))
            .map(|i| (i % 251) as u8)
            .collect::<Vec<_>>();
        let (set, access, file_id) = setup_test_set(&file_data, 1024);
        let access = SequentialOnlyAccess {
            files: access.files,
        };

        assert_eq!(verify_full_hash(&set, &file_id, &access), Some(true));
    }

    #[test]
    fn verify_selected_file_ids_uses_single_sequential_pass_for_healthy_file() {
        let file_data = (0..(QUICK_CHECK_16K_BYTES + 4096))
            .map(|i| (i % 241) as u8)
            .collect::<Vec<_>>();
        let (set, access, file_id) = setup_test_set(&file_data, 1024);
        let open_calls = Arc::new(AtomicUsize::new(0));
        let access = CountingSequentialAccess {
            files: access.files,
            open_calls: open_calls.clone(),
        };

        // Pinned strict: the single-sequential-pass discipline is a property
        // of the strict pipeline, and must not depend on an ambient
        // WEAVER_PAR2_FAST_VERIFY in the test environment (fast verify reads
        // spans via `read_file_range_into`, which this mock forbids).
        let verification = verify_selected_file_ids_resolved(
            &set,
            &access,
            None,
            &[file_id],
            &VerifyOptions::default(),
            false,
        );

        assert_eq!(verification.total_missing_blocks, 0);
        assert!(matches!(verification.files[0].status, FileStatus::Complete));
        assert_eq!(open_calls.load(Ordering::Relaxed), 1);
    }

    #[test]
    fn verify_slices_uses_batched_sequential_reader_when_available() {
        let file_data = (0..8192)
            .map(|i| (i as u8).wrapping_mul(13).wrapping_add(7))
            .collect::<Vec<_>>();
        let (set, access, file_id) = setup_test_set(&file_data, 1024);
        let access = SequentialOnlyAccess {
            files: access.files,
        };

        let slices = verify_slices(&set, &file_id, &access).unwrap();
        assert_eq!(slices, vec![true; 8]);
    }

    #[test]
    fn verifier_falls_back_to_read_into_when_no_sequential_reader_exists() {
        let file_data = (0..8192)
            .map(|i| (i as u8).wrapping_mul(17).wrapping_add(11))
            .collect::<Vec<_>>();
        let (set, access, file_id) = setup_test_set(&file_data, 1024);
        let access = ReadIntoOnlyAccess {
            files: access.files,
        };

        assert_eq!(verify_full_hash(&set, &file_id, &access), Some(true));
        assert_eq!(verify_slices(&set, &file_id, &access), Some(vec![true; 8]));
    }

    #[test]
    fn verify_selected_file_ids_reuses_quick_check_scratch_without_allocating_reads() {
        let file_a = vec![0x11u8; 2048];
        let file_b = (0..(QUICK_CHECK_16K_BYTES + 257))
            .map(|i| (i % 239) as u8)
            .collect::<Vec<_>>();
        let files = [
            (file_a.as_slice(), "alpha.bin"),
            (file_b.as_slice(), "beta.bin"),
        ];
        let (set, access, file_ids) = setup_test_set_multi(&files, 1024);
        let access = ReadIntoOnlyAccess {
            files: access.files,
        };

        // Pinned strict: this asserts the strict pipeline's quick-check
        // scratch reuse (see the sequential-pass test above for why pinning
        // beats inheriting ambient WEAVER_PAR2_FAST_VERIFY).
        let verification = verify_selected_file_ids_resolved(
            &set,
            &access,
            None,
            &file_ids,
            &VerifyOptions::default(),
            false,
        );

        assert_eq!(verification.files.len(), 2);
        assert!(
            verification
                .files
                .iter()
                .all(|file| matches!(file.status, FileStatus::Complete))
        );
        assert_eq!(verification.total_missing_blocks, 0);
    }

    #[test]
    fn verify_repairable_assessment() {
        let file_data = vec![0xABu8; 4096];
        let (mut set, mut access, file_id) = setup_test_set(&file_data, 1024);

        // Corrupt data
        let mut corrupted = file_data.clone();
        corrupted[0] ^= 0xFF;
        access.add_file(file_id, corrupted);

        // Add some recovery blocks
        use crate::par2_set::RecoverySlice;
        use bytes::Bytes;
        set.recovery_slices.insert(
            0,
            RecoverySlice {
                exponent: 0,
                data: Bytes::from(vec![0u8; 1024]).into(),
            },
        );
        set.recovery_slices.insert(
            1,
            RecoverySlice {
                exponent: 1,
                data: Bytes::from(vec![0u8; 1024]).into(),
            },
        );

        let result = verify_all(&set, &access);
        // 1 damaged slice, 2 recovery blocks available
        assert!(matches!(
            result.repairable,
            Repairability::Repairable {
                blocks_needed: 1,
                blocks_available: 2
            }
        ));
    }

    #[test]
    fn verify_insufficient_recovery() {
        let file_data = vec![0xABu8; 4096];
        let (set, _, _file_id) = setup_test_set(&file_data, 1024);
        let access = MemoryFileAccess::new(); // file missing = 4 slices needed, 0 available

        let result = verify_all(&set, &access);
        assert!(matches!(
            result.repairable,
            Repairability::Insufficient { .. }
        ));
    }

    #[test]
    fn verify_resource_limited_file_does_not_inflate_missing_blocks() {
        let set = setup_oversized_file_set();
        let access = MemoryFileAccess::new();

        let result = verify_all(&set, &access);

        assert_eq!(result.total_missing_blocks, 0);
        assert_eq!(result.files.len(), 1);
        assert_eq!(result.files[0].missing_slice_count, 0);
        assert!(result.files[0].valid_slices.is_empty());
        assert!(matches!(result.files[0].status, FileStatus::Damaged(0)));
        match result.repairable {
            Repairability::ResourceLimited { reason } => {
                assert!(reason.contains("huge.dat"));
            }
            other => panic!("expected resource-limited repairability, got {other:?}"),
        }
    }

    #[test]
    fn verify_large_file_multiple_slices() {
        // 10 slices worth of data
        let file_data: Vec<u8> = (0..10240u32).map(|i| (i % 256) as u8).collect();
        let (set, access, file_id) = setup_test_set(&file_data, 1024);

        let slices = verify_slices(&set, &file_id, &access).unwrap();
        assert_eq!(slices.len(), 10);
        assert!(slices.iter().all(|&v| v));
    }

    #[test]
    fn verify_corrupted_first_slice_16k_check_fails() {
        let file_data = vec![0xABu8; 4096];
        let (set, mut access, file_id) = setup_test_set(&file_data, 1024);

        // Corrupt the first byte
        let mut corrupted = file_data.clone();
        corrupted[0] ^= 0xFF;
        access.add_file(file_id, corrupted);

        assert_eq!(quick_check_16k(&set, &file_id, &access), Some(false));
    }

    #[test]
    fn verify_selected_file_ids_only_counts_requested_files() {
        let good = vec![0x11u8; 2048];
        let damaged = vec![0x22u8; 2048];
        let (set, mut access, file_ids) =
            setup_test_set_multi(&[(&good, "good.rar"), (&damaged, "damaged.rar")], 1024);

        let mut corrupted = damaged.clone();
        corrupted[0] ^= 0xFF;
        access.add_file(file_ids[1], corrupted);

        let selected = verify_selected_file_ids(&set, &access, &[file_ids[0]]);
        assert_eq!(selected.files.len(), 1);
        assert_eq!(selected.total_missing_blocks, 0);
        assert!(matches!(selected.repairable, Repairability::NotNeeded));

        let damaged_only = verify_selected_file_ids(&set, &access, &[file_ids[1]]);
        assert_eq!(damaged_only.files.len(), 1);
        assert_eq!(damaged_only.total_missing_blocks, 1);
        assert!(matches!(
            damaged_only.files[0].status,
            FileStatus::Damaged(1)
        ));
    }

    #[test]
    fn parallel_selected_verify_matches_serial() {
        let intact = vec![0x11u8; 4096];
        let damaged = vec![0x22u8; 4096];
        let truncated = vec![0x33u8; 4096];
        let (set, mut access, file_ids) = setup_test_set_multi(
            &[
                (&intact, "intact.rar"),
                (&damaged, "damaged.rar"),
                (&truncated, "truncated.rar"),
            ],
            1024,
        );

        let mut corrupted = damaged.clone();
        corrupted[1500] ^= 0xFF;
        corrupted[3000] ^= 0xFF;
        access.add_file(file_ids[1], corrupted);
        access.add_file(file_ids[2], truncated[..2500].to_vec());

        let serial = verify_selected_file_ids(&set, &access, &file_ids);
        let parallel = verify_selected_file_ids_parallel(&set, &access, &file_ids);
        assert!(serial.total_missing_blocks > 0, "fixture must have damage");
        assert_eq!(
            format!("{serial:?}"),
            format!("{parallel:?}"),
            "parallel selected verify must match the serial pipeline exactly"
        );

        // Merging the post-repair way: overlaying a subset onto a base
        // keeps untouched entries and recomputes the totals.
        let base = verify_selected_file_ids(&set, &access, &file_ids);
        let fixed_subset = {
            let mut fixed = MemoryFileAccess::new();
            fixed.add_file(file_ids[0], intact.clone());
            fixed.add_file(file_ids[1], damaged.clone());
            fixed.add_file(file_ids[2], truncated.clone());
            verify_selected_file_ids_parallel(&set, &fixed, &file_ids[1..])
        };
        let merged = merge_verification_results(&set, &base, fixed_subset);
        assert_eq!(merged.files.len(), file_ids.len());
        assert_eq!(merged.total_missing_blocks, 0);
        assert!(
            merged
                .files
                .iter()
                .all(|file| matches!(file.status, FileStatus::Complete)),
            "merged result must show all files complete: {merged:?}"
        );
        assert!(matches!(merged.repairable, Repairability::NotNeeded));
    }

    /// Like [`setup_test_set`] but omits the IFSC packet, so the resulting set
    /// has no per-slice checksums (`file_checksums` returns `None`). Used to
    /// prove fast verify falls back to the serial full-hash pipeline.
    fn setup_test_set_no_ifsc(
        file_data: &[u8],
        slice_size: u64,
    ) -> (Par2FileSet, MemoryFileAccess, FileId) {
        let file_length = file_data.len() as u64;
        let hash_full = checksum::md5(file_data);
        let hash_16k = checksum::md5(&file_data[..file_data.len().min(16384)]);

        let filename = b"testfile.dat";
        let mut id_input = Vec::new();
        id_input.extend_from_slice(&hash_16k);
        id_input.extend_from_slice(&file_length.to_le_bytes());
        id_input.extend_from_slice(filename);
        let file_id_bytes: [u8; 16] = Md5::digest(&id_input).into();
        let file_id = FileId::from_bytes(file_id_bytes);

        let mut main_body = Vec::new();
        main_body.extend_from_slice(&slice_size.to_le_bytes());
        main_body.extend_from_slice(&1u32.to_le_bytes());
        main_body.extend_from_slice(&file_id_bytes);
        let rsid: [u8; 16] = Md5::digest(&main_body).into();

        let mut fd_body = Vec::new();
        fd_body.extend_from_slice(&file_id_bytes);
        fd_body.extend_from_slice(&hash_full);
        fd_body.extend_from_slice(&hash_16k);
        fd_body.extend_from_slice(&file_length.to_le_bytes());
        fd_body.extend_from_slice(filename);
        while fd_body.len() % 4 != 0 {
            fd_body.push(0);
        }

        let mut stream = Vec::new();
        stream.extend_from_slice(&make_full_packet(header::TYPE_MAIN, &main_body, rsid));
        stream.extend_from_slice(&make_full_packet(header::TYPE_FILE_DESC, &fd_body, rsid));

        let set = Par2FileSet::from_files(&[&stream]).unwrap();
        let mut access = MemoryFileAccess::new();
        access.add_file(file_id, file_data.to_vec());
        (set, access, file_id)
    }

    fn strict_opts() -> VerifyOptions {
        VerifyOptions::default()
    }

    fn fast_opts() -> VerifyOptions {
        VerifyOptions {
            fast_verify: true,
            ..Default::default()
        }
    }

    #[test]
    fn resolve_fast_verify_precedence() {
        // Env override wins over the flag: "1" forces on, "0" forces off.
        assert!(resolve_fast_verify(Some("1"), false));
        assert!(!resolve_fast_verify(Some("0"), true));
        // Absent or unrecognized env falls back to the flag.
        assert!(resolve_fast_verify(None, true));
        assert!(!resolve_fast_verify(None, false));
        assert!(resolve_fast_verify(Some("yes"), true));
        assert!(!resolve_fast_verify(Some(""), false));
    }

    #[test]
    fn fast_verify_intact_matches_strict() {
        // Partial last slice (1500 % 1024) exercises tail zero-padding.
        let file_data = vec![0xA7u8; 1500];
        let (set, access, file_id) = setup_test_set(&file_data, 1024);

        let strict =
            verify_selected_file_ids_with_options(&set, &access, &[file_id], &strict_opts());
        let fast = verify_selected_file_ids_with_options(&set, &access, &[file_id], &fast_opts());

        assert!(matches!(fast.files[0].status, FileStatus::Complete));
        assert_eq!(fast.files[0].valid_slices, vec![true; 2]);
        assert_eq!(fast.total_missing_blocks, 0);
        assert!(matches!(fast.repairable, Repairability::NotNeeded));
        assert_eq!(
            format!("{strict:?}"),
            format!("{fast:?}"),
            "fast intact result must match strict byte for byte"
        );
    }

    #[test]
    fn fast_verify_uses_only_non_allocating_reads() {
        // The span reader must go through `read_file_slice_into` (looping
        // `read_file_range_into` on a reused scratch buffer), never the
        // allocating `read_file_range` this mock forbids. Pinned fast so an
        // ambient WEAVER_PAR2_FAST_VERIFY=0 cannot skip the path under test.
        // 17 full slices + a 460-byte tail exercises several SIMD MD5 lane
        // batches within one span plus the padded-tail lane.
        let file_data = (0..(QUICK_CHECK_16K_BYTES + 1500))
            .map(|i| (i % 251) as u8)
            .collect::<Vec<_>>();
        let (set, access, file_id) = setup_test_set(&file_data, 1024);
        let access = ReadIntoOnlyAccess {
            files: access.files,
        };

        let result = verify_selected_file_ids_resolved(
            &set,
            &access,
            None,
            &[file_id],
            &VerifyOptions::default(),
            true,
        );

        assert!(matches!(result.files[0].status, FileStatus::Complete));
        assert_eq!(result.total_missing_blocks, 0);
        assert_eq!(result.files[0].valid_slices, vec![true; 18]);
    }

    #[test]
    fn fast_verify_damaged_matches_strict() {
        // 20 full 1024-byte slices. The 16k quick check covers the first 16
        // slices, so damage at offset 100 trips the gate (strict reaches its
        // slice scan via the quick-hash-fail route) while damage at offset
        // 18_000 does not (strict reaches it via the full-hash-fail route).
        // Fast and strict must agree in both cases.
        let base: Vec<u8> = (0..20480u32).map(|i| (i % 256) as u8).collect();

        for corrupt_offset in [100usize, 18_000usize] {
            let (set, mut access, file_id) = setup_test_set(&base, 1024);
            let mut corrupted = base.clone();
            corrupted[corrupt_offset] ^= 0xFF;
            access.add_file(file_id, corrupted);

            let strict =
                verify_selected_file_ids_with_options(&set, &access, &[file_id], &strict_opts());
            let fast =
                verify_selected_file_ids_with_options(&set, &access, &[file_id], &fast_opts());

            assert!(
                matches!(fast.files[0].status, FileStatus::Damaged(1)),
                "offset {corrupt_offset}: unexpected status {:?}",
                fast.files[0].status
            );
            assert_eq!(fast.total_missing_blocks, 1, "offset {corrupt_offset}");
            let damaged_slice = corrupt_offset / 1024;
            assert!(
                !fast.files[0].valid_slices[damaged_slice],
                "offset {corrupt_offset}: slice {damaged_slice} should be damaged"
            );
            assert_eq!(
                fast.files[0].valid_slices.iter().filter(|v| !**v).count(),
                1,
                "offset {corrupt_offset}: exactly one slice damaged"
            );
            assert_eq!(
                format!("{strict:?}"),
                format!("{fast:?}"),
                "fast and strict must agree for corruption at offset {corrupt_offset}"
            );
        }
    }

    #[test]
    fn fast_verify_falls_back_without_ifsc() {
        let file_data = vec![0x33u8; 4096];
        let (set, access, file_id) = setup_test_set_no_ifsc(&file_data, 1024);

        let strict =
            verify_selected_file_ids_with_options(&set, &access, &[file_id], &strict_opts());
        let fast = verify_selected_file_ids_with_options(&set, &access, &[file_id], &fast_opts());

        assert!(matches!(fast.files[0].status, FileStatus::Complete));
        assert_eq!(fast.total_missing_blocks, 0);
        assert_eq!(
            format!("{strict:?}"),
            format!("{fast:?}"),
            "without IFSC, fast verify must fall back to the serial pipeline"
        );
    }

    #[test]
    fn fast_verify_length_mismatch_falls_back() {
        let file_data = vec![0x5Cu8; 4096];
        let (set, mut access, file_id) = setup_test_set(&file_data, 1024);
        // Truncate on disk so the on-disk length no longer matches the
        // description: the fast path is ineligible and both modes take the
        // serial pipeline.
        access.add_file(file_id, file_data[..2500].to_vec());

        let strict =
            verify_selected_file_ids_with_options(&set, &access, &[file_id], &strict_opts());
        let fast = verify_selected_file_ids_with_options(&set, &access, &[file_id], &fast_opts());

        assert!(matches!(fast.files[0].status, FileStatus::Damaged(_)));
        assert_eq!(
            format!("{strict:?}"),
            format!("{fast:?}"),
            "length mismatch must fall back identically to strict"
        );
    }

    #[test]
    fn fast_verify_flag_off_matches_strict_expectation() {
        // One intact + one damaged file, flag off, must reproduce the
        // pre-change strict pipeline outcome exactly.
        let intact = vec![0x11u8; 4096];
        let damaged = vec![0x22u8; 4096];
        let (set, mut access, file_ids) =
            setup_test_set_multi(&[(&intact, "intact.rar"), (&damaged, "damaged.rar")], 1024);
        let mut corrupted = damaged.clone();
        corrupted[0] ^= 0xFF;
        access.add_file(file_ids[1], corrupted);

        let off = verify_selected_file_ids_with_options(
            &set,
            &access,
            &file_ids,
            &VerifyOptions::default(),
        );
        assert!(matches!(off.files[0].status, FileStatus::Complete));
        assert!(matches!(off.files[1].status, FileStatus::Damaged(1)));
        assert_eq!(off.total_missing_blocks, 1);

        // Flag-off must equal the dedicated strict entry point byte for byte.
        let strict = verify_selected_file_ids(&set, &access, &file_ids);
        assert_eq!(format!("{off:?}"), format!("{strict:?}"));
    }

    #[test]
    fn fast_verify_parallel_matches_strict() {
        let intact = vec![0x11u8; 4096];
        let damaged = vec![0x22u8; 4096];
        let truncated = vec![0x33u8; 4096];
        let (set, mut access, file_ids) = setup_test_set_multi(
            &[
                (&intact, "intact.rar"),
                (&damaged, "damaged.rar"),
                (&truncated, "truncated.rar"),
            ],
            1024,
        );
        let mut corrupted = damaged.clone();
        corrupted[1500] ^= 0xFF;
        access.add_file(file_ids[1], corrupted);
        access.add_file(file_ids[2], truncated[..2500].to_vec());

        // Multi-file: file-parallel, each file's spans serial.
        let strict = verify_selected_file_ids_parallel(&set, &access, &file_ids);
        let fast =
            verify_selected_file_ids_parallel_with_options(&set, &access, &file_ids, &fast_opts());
        assert!(fast.total_missing_blocks > 0, "fixture must have damage");
        assert_eq!(
            format!("{strict:?}"),
            format!("{fast:?}"),
            "parallel fast verify must match parallel strict verify"
        );

        // Single file: exercises the span-parallel fast path.
        for id in &file_ids {
            let strict_one =
                verify_selected_file_ids_parallel(&set, &access, std::slice::from_ref(id));
            let fast_one = verify_selected_file_ids_parallel_with_options(
                &set,
                &access,
                std::slice::from_ref(id),
                &fast_opts(),
            );
            assert_eq!(
                format!("{strict_one:?}"),
                format!("{fast_one:?}"),
                "single-file span-parallel fast verify must match strict"
            );
        }
    }

    #[test]
    fn refresh_repairability_recomputes_after_missing_count_changes() {
        let mut result = VerificationResult {
            files: Vec::new(),
            recovery_blocks_available: 40,
            total_missing_blocks: 1380,
            repairable: Repairability::Insufficient {
                blocks_needed: 1380,
                blocks_available: 40,
                deficit: 1340,
            },
        };

        result.total_missing_blocks = 40;
        result.refresh_repairability();

        assert!(matches!(
            result.repairable,
            Repairability::Repairable {
                blocks_needed: 40,
                blocks_available: 40
            }
        ));
    }

    fn deterministic_file(len: usize) -> Vec<u8> {
        (0..len).map(|i| (i % 251) as u8).collect()
    }

    /// The strict pipeline's damaged verdict must be exactly the per-slice
    /// vector `verify_slices` computes, wherever the damage sits — including
    /// the zero-padded tail slice, and including a slice size too large for
    /// the fused stream's slice-aligned chunk (which disables the CRC32
    /// ride-along and runs the whole-file hash to completion).
    #[test]
    fn strict_verify_reports_verify_slices_vector_for_every_damage_position() {
        for slice_size in [1024u64, (VERIFY_FULL_HASH_CHUNK_BYTES as u64) + 512] {
            let len = (slice_size as usize) * 5 + 300;
            let pristine = deterministic_file(len);
            let slice_count = (len as u64).div_ceil(slice_size) as usize;
            for damaged_slice in 0..slice_count {
                let mut data = pristine.clone();
                let at = (damaged_slice as u64 * slice_size) as usize + 7;
                data[at] ^= 0xff;
                let (set, _, file_id) = setup_test_set(&pristine, slice_size);
                let mut access = MemoryFileAccess::new();
                access.add_file(file_id, data);

                let expected = verify_slices(&set, &file_id, &access).expect("slice vector");
                assert!(!expected[damaged_slice], "slice {damaged_slice} must fail");
                let expected_damaged = expected.iter().filter(|valid| !**valid).count() as u32;

                for span_access in [None, Some(&access as &(dyn FileAccess + Sync))] {
                    let result = verify_selected_file_ids_resolved(
                        &set,
                        &access,
                        span_access,
                        &[file_id],
                        &VerifyOptions::default(),
                        false,
                    );
                    let file = &result.files[0];
                    assert!(
                        matches!(file.status, FileStatus::Damaged(count) if count == expected_damaged),
                        "slice_size {slice_size} damaged slice {damaged_slice}: {:?}",
                        file.status
                    );
                    assert_eq!(file.valid_slices, expected);
                    assert_eq!(file.missing_slice_count, expected_damaged);
                    assert_eq!(result.total_missing_blocks, expected_damaged);
                }
            }
        }
    }

    /// An intact file still needs the whole-file MD5 to be called complete,
    /// and the fused stream must run it to the end to say so.
    #[test]
    fn strict_verify_still_requires_the_whole_file_hash_to_report_complete() {
        let slice_size = 1024u64;
        let data = deterministic_file((slice_size as usize) * 4 + 11);
        let (set, access, file_id) = setup_test_set(&data, slice_size);

        for span_access in [None, Some(&access as &(dyn FileAccess + Sync))] {
            let result = verify_selected_file_ids_resolved(
                &set,
                &access,
                span_access,
                &[file_id],
                &VerifyOptions::default(),
                false,
            );
            assert!(matches!(result.files[0].status, FileStatus::Complete));
            assert_eq!(result.files[0].valid_slices, vec![true; 5]);
            assert_eq!(result.total_missing_blocks, 0);
        }
    }

    /// Every slice checks out but the described whole-file MD5 does not: the
    /// strict rule is that only a full length+MD5 match may mark a file
    /// complete, so this reports `Damaged(0)` with an all-valid slice vector.
    /// The fused stream must not short-circuit its way past that.
    #[test]
    fn strict_verify_reports_damaged_zero_when_only_the_whole_file_hash_disagrees() {
        // Over 16 KiB: below that the format requires hash_16k == hash_full,
        // so the two hashes cannot be made to disagree.
        let slice_size = 4096u64;
        let data = deterministic_file((slice_size as usize) * 6);
        let (set, access, file_id) =
            setup_test_set_with_full_hash(&data, slice_size, Some([0x5a; 16]));

        for span_access in [None, Some(&access as &(dyn FileAccess + Sync))] {
            let result = verify_selected_file_ids_resolved(
                &set,
                &access,
                span_access,
                &[file_id],
                &VerifyOptions::default(),
                false,
            );
            assert!(matches!(result.files[0].status, FileStatus::Damaged(0)));
            assert_eq!(result.files[0].valid_slices, vec![true; 6]);
            assert_eq!(result.total_missing_blocks, 0);
        }
    }

    /// `strict_slice_validity` may pick either scanner, so the span-parallel
    /// scanner and `verify_slices` must agree slice for slice.
    #[test]
    fn span_parallel_slice_scan_matches_verify_slices() {
        let slice_size = 1024u64;
        let pristine = deterministic_file((slice_size as usize) * 9 + 617);
        let mut data = pristine.clone();
        for damaged_slice in [0usize, 3, 9] {
            let at = (damaged_slice as u64 * slice_size) as usize + 1;
            data[at] ^= 0xff;
        }
        let (set, _, file_id) = setup_test_set(&pristine, slice_size);
        let mut access = MemoryFileAccess::new();
        access.add_file(file_id, data);

        let serial = verify_slices(&set, &file_id, &access).expect("slice vector");
        let spanned = verify_file_sliced(&set, &access, &file_id, true).expect("span scan");
        assert_eq!(spanned.valid_slices, serial);
        assert_eq!(
            strict_slice_validity(&set, &access, Some(&access), &file_id, serial.len()),
            serial
        );
    }

    /// Measures what the CRC32 ride-along costs an intact file — the one case
    /// where its work buys nothing (every slice matches, so the early exit
    /// never fires). In-memory access, so the arithmetic has no I/O to hide
    /// behind: this is the overhead's worst case. Run by hand:
    ///
    /// ```sh
    /// cargo test --release -p par2-rs --lib -- --ignored perf_intact_ride_along
    /// ```
    #[test]
    #[ignore = "perf measurement, run by hand in release mode"]
    fn perf_intact_ride_along_overhead() {
        let slice_size = 512 * 1024u64;
        let len = 512 * 1024 * 1024usize;
        let data = deterministic_file(len);
        let (set, access, file_id) = setup_test_set_with_full_hash(&data, slice_size, None);
        let checksums = set.file_checksums(&file_id).expect("checksums");

        let mut timings: [Vec<f64>; 2] = [Vec::new(), Vec::new()];
        // Alternate arms so thermal drift hits both equally.
        for _round in 0..5 {
            for (arm, slice_checksums) in [(0usize, None), (1usize, Some(checksums))] {
                let started = std::time::Instant::now();
                let outcome = stream_strict_hashes(&set, &file_id, &access, slice_checksums)
                    .expect("stream outcome");
                let elapsed = started.elapsed().as_secs_f64();
                assert!(matches!(
                    outcome,
                    StrictStreamOutcome::Hashes {
                        quick_ok: true,
                        full_ok: true
                    }
                ));
                timings[arm].push(elapsed);
            }
        }
        let best = |samples: &[f64]| samples.iter().copied().fold(f64::INFINITY, f64::min);
        let plain = best(&timings[0]);
        let fused = best(&timings[1]);
        let mb = len as f64 / (1024.0 * 1024.0);
        println!(
            "intact strict stream, {len} bytes, slice {slice_size}: \
             plain {:.0} MB/s, ride-along {:.0} MB/s, overhead {:+.2}%",
            mb / plain,
            mb / fused,
            (fused / plain - 1.0) * 100.0
        );
    }

    /// The fused stream must not read past the described length, and must
    /// still notice a file that grew behind the length it was told.
    #[test]
    fn fused_chunk_size_is_slice_aligned_and_bounded() {
        assert_eq!(fused_chunk_bytes(0), None);
        assert_eq!(
            fused_chunk_bytes((VERIFY_FULL_HASH_CHUNK_BYTES as u64) + 1),
            None
        );
        for slice_size in [2u64, 1024, 65536, VERIFY_FULL_HASH_CHUNK_BYTES as u64] {
            let chunk = fused_chunk_bytes(slice_size).expect("chunk");
            assert!(chunk > 0);
            assert!(chunk <= VERIFY_FULL_HASH_CHUNK_BYTES);
            assert_eq!(chunk as u64 % slice_size, 0);
        }
    }

    // --- Per-slice evidence intake ----------------------------------------

    /// A [`FileAccess`] that counts the bytes its reads actually deliver, so a
    /// test can assert on what a pipeline *read* and not only on what it
    /// concluded. It changes nothing else about the access it wraps: the
    /// evidence path's whole claim is that it reads less for the same verdict,
    /// and only a counter can hold it to that.
    struct CountingAccess {
        inner: MemoryFileAccess,
        bytes_read: AtomicUsize,
    }

    impl CountingAccess {
        fn new(inner: MemoryFileAccess) -> Self {
            Self {
                inner,
                bytes_read: AtomicUsize::new(0),
            }
        }

        fn take_bytes_read(&self) -> usize {
            self.bytes_read.swap(0, Ordering::Relaxed)
        }
    }

    impl FileAccess for CountingAccess {
        fn read_file_range(&self, file_id: &FileId, offset: u64, len: u64) -> io::Result<Vec<u8>> {
            let data = self.inner.read_file_range(file_id, offset, len)?;
            self.bytes_read.fetch_add(data.len(), Ordering::Relaxed);
            Ok(data)
        }

        fn read_file_range_into(
            &self,
            file_id: &FileId,
            offset: u64,
            dst: &mut [u8],
        ) -> io::Result<usize> {
            let read = self.inner.read_file_range_into(file_id, offset, dst)?;
            self.bytes_read.fetch_add(read, Ordering::Relaxed);
            Ok(read)
        }

        fn file_exists(&self, file_id: &FileId) -> bool {
            self.inner.file_exists(file_id)
        }

        fn file_length(&self, file_id: &FileId) -> Option<u64> {
            self.inner.file_length(file_id)
        }

        fn read_file(&self, file_id: &FileId) -> io::Result<Vec<u8>> {
            let data = self.inner.read_file(file_id)?;
            self.bytes_read.fetch_add(data.len(), Ordering::Relaxed);
            Ok(data)
        }

        fn write_file_range(
            &mut self,
            file_id: &FileId,
            offset: u64,
            data: &[u8],
        ) -> io::Result<()> {
            self.inner.write_file_range(file_id, offset, data)
        }
    }

    fn evidence_opts(proven: HashMap<FileId, Vec<bool>>) -> VerifyOptions {
        VerifyOptions {
            proven_slices: proven,
            ..Default::default()
        }
    }

    fn deterministic_bytes(len: usize, seed: u8) -> Vec<u8> {
        (0..len)
            .map(|index| ((index as u8).wrapping_mul(31)).wrapping_add(seed))
            .collect()
    }

    /// One corpus covering the shapes a verify has to get right: an intact
    /// file, a damaged one, one with a short tail slice, and one that is not
    /// there at all. Returns the set, an access holding only the files that
    /// exist, and the ids in corpus order.
    fn evidence_corpus(slice_size: u64) -> (Par2FileSet, MemoryFileAccess, Vec<FileId>) {
        let intact = deterministic_bytes(4 * slice_size as usize, 1);
        let damaged_source = deterministic_bytes(4 * slice_size as usize, 2);
        let short_tail = deterministic_bytes(3 * slice_size as usize - 7, 3);
        let absent = deterministic_bytes(2 * slice_size as usize, 4);

        let (set, _, ids) = setup_test_set_multi(
            &[
                (&intact, "intact.bin"),
                (&damaged_source, "damaged.bin"),
                (&short_tail, "short-tail.bin"),
                (&absent, "absent.bin"),
            ],
            slice_size,
        );

        // Damage slice 2 of `damaged.bin` and slice 0 of `short-tail.bin`, and
        // never add `absent.bin` at all.
        let mut damaged = damaged_source.clone();
        let start = 2 * slice_size as usize;
        damaged[start..start + slice_size as usize].fill(0);
        let mut short_tail_damaged = short_tail.clone();
        short_tail_damaged[..slice_size as usize].fill(0xFF);

        let mut access = MemoryFileAccess::new();
        access.add_file(ids[0], intact);
        access.add_file(ids[1], damaged);
        access.add_file(ids[2], short_tail_damaged);

        (set, access, ids)
    }

    /// Per-slice evidence for every slice a full read found valid — the
    /// attestation a host that had genuinely proven those slices would make.
    fn evidence_from(result: &VerificationResult) -> HashMap<FileId, Vec<bool>> {
        result
            .files
            .iter()
            .map(|file| (file.file_id, file.valid_slices.clone()))
            .collect()
    }

    /// Property (a): with nothing proven, verification is what it always was.
    /// An absent map, an empty map, an all-`false` map and a map whose vectors
    /// are the wrong shape must every one of them reach the same branch and
    /// produce the same result, byte for byte.
    #[test]
    fn absent_evidence_leaves_verification_unchanged() {
        let slice_size = 512u64;
        let (set, access, ids) = evidence_corpus(slice_size);
        let baseline = verify_selected_file_ids_with_options(&set, &access, &ids, &strict_opts());
        assert!(baseline.total_missing_blocks > 0, "{baseline:#?}");

        let nothing_proven: HashMap<FileId, Vec<bool>> = baseline
            .files
            .iter()
            .map(|file| (file.file_id, vec![false; file.valid_slices.len()]))
            .collect();
        let wrong_shape: HashMap<FileId, Vec<bool>> = baseline
            .files
            .iter()
            .map(|file| (file.file_id, vec![true; file.valid_slices.len() + 1]))
            .collect();

        for (label, proven) in [
            ("empty map", HashMap::new()),
            ("all-false vectors", nothing_proven),
            ("wrong-length vectors", wrong_shape),
        ] {
            let with_evidence =
                verify_selected_file_ids_with_options(&set, &access, &ids, &evidence_opts(proven));
            assert_eq!(
                format!("{with_evidence:?}"),
                format!("{baseline:?}"),
                "{label} must leave verification unchanged"
            );
        }
    }

    /// Property (b): correct evidence for the slices a full read would have
    /// found valid produces the same verdicts from strictly fewer bytes read.
    #[test]
    fn correct_evidence_keeps_the_verdicts_and_reads_fewer_bytes() {
        let slice_size = 512u64;
        let (set, memory, ids) = evidence_corpus(slice_size);
        let access = CountingAccess::new(memory);

        let baseline = verify_selected_file_ids_with_options(&set, &access, &ids, &strict_opts());
        let baseline_bytes = access.take_bytes_read();
        assert!(baseline_bytes > 0, "the unevidenced pass reads the corpus");

        let evidenced = verify_selected_file_ids_with_options(
            &set,
            &access,
            &ids,
            &evidence_opts(evidence_from(&baseline)),
        );
        let evidenced_bytes = access.take_bytes_read();

        assert_eq!(
            format!("{evidenced:?}"),
            format!("{baseline:?}"),
            "correct evidence must not change a single verdict"
        );
        assert!(
            evidenced_bytes < baseline_bytes,
            "evidence must save reads: {evidenced_bytes} vs {baseline_bytes} bytes"
        );
    }

    /// Property (c): damage in a slice the evidence says nothing about is
    /// still found, and the whole result — statuses, per-slice vectors, totals
    /// and repairability — matches the full read exactly. Every damage
    /// position is walked, because an off-by-one in span reassembly would only
    /// show at some of them.
    #[test]
    fn damage_in_an_unproven_slice_matches_the_full_read_at_every_position() {
        let slice_size = 256u64;
        let slice_count = 6usize;
        let clean = deterministic_bytes(slice_count * slice_size as usize, 9);
        let (set, _, ids) = setup_test_set_multi(&[(&clean, "walked.bin")], slice_size);
        let file_id = ids[0];

        for damaged_index in 0..slice_count {
            let mut damaged = clean.clone();
            let start = damaged_index * slice_size as usize;
            damaged[start..start + slice_size as usize].fill(0x5A);
            let mut memory = MemoryFileAccess::new();
            memory.add_file(file_id, damaged);
            let access = CountingAccess::new(memory);

            let baseline =
                verify_selected_file_ids_with_options(&set, &access, &ids, &strict_opts());
            let baseline_bytes = access.take_bytes_read();
            assert_eq!(
                baseline.files[0].missing_slice_count, 1,
                "slice {damaged_index} must read as damaged"
            );

            // Prove everything the full read found valid — that is, every
            // slice except the damaged one — and leave the damaged one to be
            // read.
            let evidenced = verify_selected_file_ids_with_options(
                &set,
                &access,
                &ids,
                &evidence_opts(evidence_from(&baseline)),
            );
            let evidenced_bytes = access.take_bytes_read();

            assert_eq!(
                format!("{evidenced:?}"),
                format!("{baseline:?}"),
                "slice {damaged_index}: the evidenced result must match the full read exactly"
            );
            assert_eq!(
                evidenced_bytes, slice_size as usize,
                "slice {damaged_index}: only the unproven slice should be read \
                 (full read was {baseline_bytes} bytes)"
            );
        }
    }

    /// Property (d): a file that is not there cannot be attested into
    /// existence. The existence check runs before evidence is consulted, so
    /// all-`true` evidence for a missing file leaves it `Missing` with every
    /// slice invalid, exactly as with no evidence at all.
    #[test]
    fn evidence_cannot_resurrect_a_missing_file() {
        let slice_size = 512u64;
        let data = deterministic_bytes(3 * slice_size as usize, 5);
        let (set, _, ids) = setup_test_set_multi(&[(&data, "gone.bin")], slice_size);
        let access = CountingAccess::new(MemoryFileAccess::new());

        let baseline = verify_selected_file_ids_with_options(&set, &access, &ids, &strict_opts());
        let proven: HashMap<FileId, Vec<bool>> = [(ids[0], vec![true; 3])].into_iter().collect();
        let evidenced =
            verify_selected_file_ids_with_options(&set, &access, &ids, &evidence_opts(proven));

        assert!(matches!(evidenced.files[0].status, FileStatus::Missing));
        assert_eq!(evidenced.files[0].valid_slices, vec![false; 3]);
        assert_eq!(evidenced.total_missing_blocks, 3);
        assert_eq!(format!("{evidenced:?}"), format!("{baseline:?}"));
        assert_eq!(access.take_bytes_read(), 0);
    }

    /// A file whose on-disk length no longer matches the description is not
    /// the file those per-slice verdicts were about — the offsets have moved
    /// under them — so the evidence is discarded and the file takes the
    /// pipeline it always took.
    #[test]
    fn evidence_for_a_length_mismatched_file_is_ignored() {
        let slice_size = 512u64;
        let data = deterministic_bytes(3 * slice_size as usize, 6);
        let (set, _, ids) = setup_test_set_multi(&[(&data, "truncated.bin")], slice_size);
        let mut memory = MemoryFileAccess::new();
        memory.add_file(ids[0], data[..data.len() - 100].to_vec());
        let access = CountingAccess::new(memory);

        let baseline = verify_selected_file_ids_with_options(&set, &access, &ids, &strict_opts());
        let proven: HashMap<FileId, Vec<bool>> = [(ids[0], vec![true; 3])].into_iter().collect();
        let evidenced =
            verify_selected_file_ids_with_options(&set, &access, &ids, &evidence_opts(proven));

        assert!(matches!(baseline.files[0].status, FileStatus::Damaged(_)));
        assert_eq!(
            format!("{evidenced:?}"),
            format!("{baseline:?}"),
            "a length mismatch must discard the evidence, not believe it"
        );
    }

    /// The degenerate end of the range: a file whose every slice is proven is
    /// `Complete` without a byte being read. The whole-file MD5 arm cannot run
    /// once any slice is skipped, so completeness comes from slice proof — the
    /// same shape fast verify already reports.
    #[test]
    fn a_fully_proven_file_is_complete_without_a_read() {
        let slice_size = 512u64;
        let data = deterministic_bytes(4 * slice_size as usize, 7);
        let (set, memory, ids) = setup_test_set_multi(&[(&data, "proven.bin")], slice_size);
        let access = CountingAccess::new(memory);

        let proven: HashMap<FileId, Vec<bool>> = [(ids[0], vec![true; 4])].into_iter().collect();
        let evidenced =
            verify_selected_file_ids_with_options(&set, &access, &ids, &evidence_opts(proven));

        assert!(matches!(evidenced.files[0].status, FileStatus::Complete));
        assert_eq!(evidenced.files[0].valid_slices, vec![true; 4]);
        assert_eq!(evidenced.total_missing_blocks, 0);
        assert!(matches!(evidenced.repairable, Repairability::NotNeeded));
        assert_eq!(
            access.take_bytes_read(),
            0,
            "a fully proven file must cost no reads"
        );
    }

    /// Evidence composes with the fast arm the way it composes with the strict
    /// one: the proven slices are skipped, the rest go through the same
    /// per-slice checks, and the verdict is the same either way. It also
    /// composes with the file-parallel driver, which consults the same option.
    #[test]
    fn evidence_composes_with_fast_verify_and_the_parallel_driver() {
        let slice_size = 512u64;
        let (set, memory, ids) = evidence_corpus(slice_size);
        let baseline = verify_selected_file_ids_with_options(&set, &memory, &ids, &strict_opts());
        let proven = evidence_from(&baseline);

        let mut fast_with_evidence = fast_opts();
        fast_with_evidence.proven_slices = proven.clone();
        let fast = verify_selected_file_ids_with_options(&set, &memory, &ids, &fast_with_evidence);
        assert_eq!(
            format!("{fast:?}"),
            format!("{baseline:?}"),
            "evidence under fast verify must reach the same verdicts"
        );

        let access = CountingAccess::new(memory);
        let parallel_baseline =
            verify_selected_file_ids_parallel_with_options(&set, &access, &ids, &strict_opts());
        let parallel_bytes = access.take_bytes_read();
        let parallel = verify_selected_file_ids_parallel_with_options(
            &set,
            &access,
            &ids,
            &evidence_opts(proven),
        );
        let parallel_evidenced_bytes = access.take_bytes_read();
        assert_eq!(
            format!("{parallel:?}"),
            format!("{parallel_baseline:?}"),
            "the parallel driver must honour evidence without changing verdicts"
        );
        assert!(
            parallel_evidenced_bytes < parallel_bytes,
            "the parallel driver must also save reads: \
             {parallel_evidenced_bytes} vs {parallel_bytes}"
        );
    }

    /// A plan built with evidence must name exactly the unproven slices, in
    /// order, and a plan built without it must still tile the whole file in
    /// spans — the shape every existing caller depends on.
    #[test]
    fn the_plan_names_exactly_the_unproven_slices() {
        let slice_size = 512u64;
        let data = deterministic_bytes(6 * slice_size as usize, 11);
        let (set, access, ids) = setup_test_set_multi(&[(&data, "planned.bin")], slice_size);
        let file_id = ids[0];

        let full = sliced_verify_plan(&set, &access, &file_id, None).expect("plan");
        match full.work {
            SlicedVerifyWork::Spans(spans) => {
                let covered: usize = spans.iter().map(|(_, count)| count).sum();
                assert_eq!(covered, 6, "spans must tile every slice: {spans:?}");
                assert_eq!(spans[0].0, 0);
            }
            SlicedVerifyWork::Slices(_) => panic!("a plan with no evidence reads spans"),
        }
        assert!(full.proven.is_none());

        for mask in [
            vec![false; 6],
            vec![true, false, true, false, true, false],
            vec![true, true, false, false, false, true],
            vec![true; 6],
        ] {
            let plan = sliced_verify_plan(&set, &access, &file_id, Some(&mask)).expect("plan");
            let expected: Vec<usize> = mask
                .iter()
                .enumerate()
                .filter_map(|(index, proven)| (!*proven).then_some(index))
                .collect();
            match plan.work {
                SlicedVerifyWork::Slices(indices) => assert_eq!(indices, expected, "{mask:?}"),
                SlicedVerifyWork::Spans(_) => panic!("an evidenced plan reads named slices"),
            }
            assert_eq!(plan.proven.as_deref(), Some(mask.as_slice()));
        }
    }

    /// A lane batch is an arbitrary grouping, so a slice whose read fails must
    /// fail alone. Reading past the end of a truncated file is the reachable
    /// version of that, and the slices batched alongside it must keep the
    /// verdicts their own bytes earned.
    #[test]
    fn a_failed_read_fails_only_its_own_slice_in_a_batch() {
        let slice_size = 512u64;
        let slice_count = 6usize;
        let data = deterministic_bytes(slice_count * slice_size as usize, 13);
        let (set, _, ids) = setup_test_set_multi(&[(&data, "cut-short.bin")], slice_size);
        let file_id = ids[0];

        // The access reports the described length but only holds the first
        // four slices, so reads of slices 4 and 5 come up short.
        struct ShortAccess {
            data: Vec<u8>,
            reported_len: u64,
            file_id: FileId,
        }
        impl FileAccess for ShortAccess {
            fn read_file_range(
                &self,
                _file_id: &FileId,
                offset: u64,
                len: u64,
            ) -> io::Result<Vec<u8>> {
                let start = (offset as usize).min(self.data.len());
                let end = (start + len as usize).min(self.data.len());
                Ok(self.data[start..end].to_vec())
            }
            fn read_file_range_into(
                &self,
                _file_id: &FileId,
                offset: u64,
                dst: &mut [u8],
            ) -> io::Result<usize> {
                let start = (offset as usize).min(self.data.len());
                let end = (start + dst.len()).min(self.data.len());
                dst[..end - start].copy_from_slice(&self.data[start..end]);
                Ok(end - start)
            }
            fn file_exists(&self, file_id: &FileId) -> bool {
                *file_id == self.file_id
            }
            fn file_length(&self, _file_id: &FileId) -> Option<u64> {
                Some(self.reported_len)
            }
            fn read_file(&self, _file_id: &FileId) -> io::Result<Vec<u8>> {
                Ok(self.data.clone())
            }
            fn write_file_range(
                &mut self,
                _file_id: &FileId,
                _offset: u64,
                _data: &[u8],
            ) -> io::Result<()> {
                Err(io::Error::new(io::ErrorKind::Unsupported, "read-only"))
            }
        }

        let access = ShortAccess {
            data: data[..4 * slice_size as usize].to_vec(),
            reported_len: data.len() as u64,
            file_id,
        };
        let proven: HashMap<FileId, Vec<bool>> =
            [(file_id, vec![false; slice_count])].into_iter().collect();
        let result =
            verify_selected_file_ids_with_options(&set, &access, &ids, &evidence_opts(proven));

        assert_eq!(
            result.files[0].valid_slices,
            vec![true, true, true, true, false, false],
            "a short read must condemn only the slices it actually cut short"
        );
        assert_eq!(result.files[0].missing_slice_count, 2);
    }
}