hf_fetch_model/

http_range.rs

// SPDX-License-Identifier: MIT OR Apache-2.0

//! HTTP Range `Read + Seek` substrate for remote inspection.
//!
//! [`RangeReader`] adapts a byte-range transport into [`std::io::Read`] +
//! [`std::io::Seek`], so format parsers that consume a reader (e.g.
//! `anamnesis::inspect_npz_from_reader`) can inspect a remote file's
//! metadata without downloading it. The transport is abstracted behind
//! [`RangeFetcher`]; production code uses [`HttpRangeFetcher`] (HTTP Range
//! requests against the `HuggingFace` CDN), and unit tests use an in-memory
//! fetcher — the buffering, budgeting, and `Seek` semantics are exercised
//! entirely offline.
//!
//! # Access pattern optimisations
//!
//! Tuned for archive-directory reads (`ZIP` central directory + per-entry
//! headers — the `NPZ` / `PTH` layout):
//!
//! - **Tail prefetch** — the first read landing in the last
//!   [`TAIL_PREFETCH_BYTES`] of the file fetches that whole tail once and
//!   caches it. The `ZIP` end-of-central-directory scan and the central
//!   directory itself are then served from memory.
//! - **Read-ahead** — short sequential reads are coalesced into
//!   [`READAHEAD_BYTES`]-sized window fetches, so parsing a 30-byte local
//!   header followed by a ~120-byte `NPY` header costs one request, not two.
//! - **No re-fetch** — reads covered by the cached tail or the current
//!   window never re-issue a request.
//!
//! # Safety budgets
//!
//! The reader enforces two hard caps — [`MAX_RANGE_REQUESTS`] requests and
//! [`MAX_TRANSFER_BUDGET`] bytes fetched — so a pathological or adversarial
//! archive layout (a seek storm, an absurdly large central directory) fails
//! fast with a clear error instead of degenerating into a full download.
//! [`RangeReader::with_limits`] overrides the defaults for callers with
//! different budgets.
//!
//! # Error channel
//!
//! `Read` / `Seek` must return [`std::io::Error`], which flattens the crate's
//! typed [`FetchError`]. The reader therefore stores the most recent fetch
//! failure; after a failed parse, [`RangeReader::take_last_error`] recovers
//! the typed error (e.g. an HTTP `403` that the CLI upgrades into a
//! gated-repo diagnosis).

use std::io::{self, Read, Seek, SeekFrom};
use std::time::Duration;

use serde::Serialize;

use crate::chunked;
use crate::error::FetchError;

// -----------------------------------------------------------------------
// Tuning constants
// -----------------------------------------------------------------------

/// Size of the cached file tail fetched by the first read near end-of-file.
///
/// 64 KiB covers the `ZIP` end-of-central-directory record (max comment
/// length 65 535 bytes) plus the central directory of typical `NPZ`
/// archives, so the whole directory scan costs one request.
pub const TAIL_PREFETCH_BYTES: u64 = 64 * 1024;

/// Minimum window size for a fetch serving a sequential read.
///
/// Short header-sized reads (tens of bytes) are widened to this size so
/// adjacent structures (`ZIP` local header + `NPY` header) arrive in one
/// request.
pub const READAHEAD_BYTES: u64 = 4 * 1024;

/// Default cap on total bytes fetched over the reader's lifetime.
///
/// Remote inspection of a well-formed archive transfers well under 1 MiB;
/// 32 MiB leaves two orders of magnitude of headroom (e.g. a many-thousand
/// entry central directory) while guaranteeing that no inspect silently
/// degenerates into a full-file download.
pub const MAX_TRANSFER_BUDGET: u64 = 32 * 1024 * 1024;

/// Default cap on the number of range fetches over the reader's lifetime.
///
/// A well-formed inspect needs a handful of requests; 256 tolerates a
/// fragmented directory layout while stopping request storms from
/// adversarial seek patterns.
pub const MAX_RANGE_REQUESTS: u32 = 256;

/// Wall-clock budget for a single HTTP range request (headers + body).
///
/// Bounds a stalled response body — the connect phase is separately bounded
/// by the client's TCP connect timeout.
const RANGE_REQUEST_TIMEOUT: Duration = Duration::from_secs(60);

// -----------------------------------------------------------------------
// Transport abstraction
// -----------------------------------------------------------------------

/// A transport that serves byte ranges of one remote file.
///
/// Implemented by [`HttpRangeFetcher`] for production and by in-memory
/// fetchers in tests. The contract: `fetch(start, end_inclusive)` returns
/// exactly `end_inclusive - start + 1` bytes of the file's content at
/// `start`, and `total_size` is the file's full length in bytes, known
/// up front.
pub trait RangeFetcher {
    /// Fetches the inclusive byte range `start..=end_inclusive`.
    ///
    /// # Errors
    ///
    /// Returns [`FetchError::Http`] when the transport fails (network
    /// error, unexpected HTTP status, validation mismatch).
    fn fetch(&mut self, start: u64, end_inclusive: u64) -> Result<Vec<u8>, FetchError>;

    /// Total size of the remote file in bytes.
    fn total_size(&self) -> u64;

    /// Requests spent by the transport outside [`RangeFetcher::fetch`]
    /// (e.g. the eager probe that resolved the file size and classified
    /// access). Included in [`RangeReader::stats`] so the reported request
    /// count is honest.
    fn extra_requests(&self) -> u32 {
        0
    }
}

/// Transfer statistics for one [`RangeReader`] lifetime.
///
/// Rendered by the CLI as provenance (e.g. `remote (6 range requests,
/// 136.0 KiB fetched)` — the live-measured cost of inspecting a 72 MiB
/// `NPZ`) — the on-screen proof that an inspect read metadata, not weights.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[non_exhaustive]
pub struct RangeStats {
    /// Total HTTP requests issued: range fetches (attempts, including
    /// failed ones) plus the transport's probe requests.
    pub requests: u32,
    /// Total content bytes fetched across all successful range requests.
    /// Excludes HTTP header overhead and probe responses (single-byte
    /// probes; negligible).
    pub bytes_fetched: u64,
}

// -----------------------------------------------------------------------
// RangeReader core
// -----------------------------------------------------------------------

/// A cached contiguous extent of the remote file.
struct Extent {
    /// Absolute file offset of `data[0]`.
    start: u64,
    /// The fetched bytes.
    data: Vec<u8>,
}

impl Extent {
    /// One past the last absolute offset covered by this extent.
    fn end(&self) -> u64 {
        // CAST: usize → u64, buffer length always fits (usize ≤ 64 bits)
        #[allow(clippy::as_conversions)]
        self.start.saturating_add(self.data.len() as u64)
    }

    /// Whether `pos` falls inside this extent.
    fn contains(&self, pos: u64) -> bool {
        pos >= self.start && pos < self.end()
    }

    /// Copies bytes starting at absolute offset `pos` into `buf`, returning
    /// the count copied (`None` if `pos` is outside the extent).
    fn copy_at(&self, pos: u64, buf: &mut [u8]) -> Option<usize> {
        if !self.contains(pos) {
            return None;
        }
        let offset = usize::try_from(pos.checked_sub(self.start)?).ok()?;
        let available = self.data.len().checked_sub(offset)?;
        let n = buf.len().min(available);
        let src = self.data.get(offset..offset.checked_add(n)?)?;
        let dst = buf.get_mut(..n)?;
        dst.copy_from_slice(src);
        Some(n)
    }
}

/// `Read + Seek` over a [`RangeFetcher`], with tail caching, read-ahead,
/// and hard safety budgets.
///
/// See the [module docs](self) for the access-pattern and budget design.
pub struct RangeReader<F: RangeFetcher> {
    /// The transport serving byte ranges.
    fetcher: F,
    /// Current logical read position (may exceed the file size after a
    /// permissive seek; reads there return `Ok(0)`).
    pos: u64,
    /// Most recent read-ahead window.
    window: Option<Extent>,
    /// Cached file tail (`ZIP` end-of-central-directory region).
    tail: Option<Extent>,
    /// Range fetches attempted (successful or failed).
    requests: u32,
    /// Content bytes fetched across successful range fetches.
    bytes_fetched: u64,
    /// Lifetime cap on `requests`.
    max_requests: u32,
    /// Lifetime cap on `bytes_fetched`.
    max_transfer_bytes: u64,
    /// Most recent typed transport error (see [`Self::take_last_error`]).
    last_error: Option<FetchError>,
}

impl<F: RangeFetcher> RangeReader<F> {
    /// Creates a reader with the default safety budgets
    /// ([`MAX_RANGE_REQUESTS`], [`MAX_TRANSFER_BUDGET`]).
    #[must_use]
    pub const fn new(fetcher: F) -> Self {
        Self::with_limits(fetcher, MAX_RANGE_REQUESTS, MAX_TRANSFER_BUDGET)
    }

    /// Creates a reader with explicit safety budgets.
    ///
    /// `max_requests` caps the number of range fetches; `max_transfer_bytes`
    /// caps the total content bytes fetched. Exceeding either fails the
    /// offending read with a [`std::io::Error`] naming the cap.
    #[must_use]
    pub const fn with_limits(fetcher: F, max_requests: u32, max_transfer_bytes: u64) -> Self {
        Self {
            fetcher,
            pos: 0,
            window: None,
            tail: None,
            requests: 0,
            bytes_fetched: 0,
            max_requests,
            max_transfer_bytes,
            last_error: None,
        }
    }

    /// Transfer statistics so far (range fetches + transport probes).
    #[must_use]
    pub fn stats(&self) -> RangeStats {
        RangeStats {
            requests: self.requests.saturating_add(self.fetcher.extra_requests()),
            bytes_fetched: self.bytes_fetched,
        }
    }

    /// Total size of the remote file in bytes.
    #[must_use]
    pub fn total_size(&self) -> u64 {
        self.fetcher.total_size()
    }

    /// Takes the most recent typed transport error, if any.
    ///
    /// `Read` flattens [`FetchError`] into [`std::io::Error`] strings; after
    /// a parse fails, callers use this to recover the typed error (e.g. to
    /// let the CLI upgrade an HTTP `403` into a gated-repo diagnosis).
    #[must_use]
    pub const fn take_last_error(&mut self) -> Option<FetchError> {
        self.last_error.take()
    }

    /// Fetches `start..=end_inclusive` through the budget checks, recording
    /// stats and stashing typed errors.
    fn checked_fetch(&mut self, start: u64, end_inclusive: u64) -> io::Result<Vec<u8>> {
        let len = end_inclusive
            .checked_sub(start)
            .and_then(|d| d.checked_add(1))
            .ok_or_else(|| {
                io::Error::new(
                    io::ErrorKind::InvalidInput,
                    format!("invalid range {start}..={end_inclusive}"),
                )
            })?;

        if self.requests >= self.max_requests {
            return Err(io::Error::other(format!(
                "range request cap exceeded ({} requests): pathological archive \
                 layout or seek storm; refusing further fetches",
                self.max_requests
            )));
        }
        if self.bytes_fetched.saturating_add(len) > self.max_transfer_bytes {
            return Err(io::Error::other(format!(
                "range transfer budget exceeded ({} bytes fetched, {len} more \
                 requested, cap {}): metadata inspection should never read \
                 this much; refusing further fetches",
                self.bytes_fetched, self.max_transfer_bytes
            )));
        }

        self.requests = self.requests.saturating_add(1);
        match self.fetcher.fetch(start, end_inclusive) {
            Ok(data) => {
                // CAST: usize → u64, buffer length always fits (usize ≤ 64 bits)
                #[allow(clippy::as_conversions)]
                let got = data.len() as u64;
                if got != len {
                    return Err(io::Error::other(format!(
                        "range fetcher returned {got} bytes for a {len}-byte \
                         range ({start}..={end_inclusive})"
                    )));
                }
                self.bytes_fetched = self.bytes_fetched.saturating_add(len);
                Ok(data)
            }
            Err(fetch_err) => {
                let io_err = io::Error::other(fetch_err.to_string());
                self.last_error = Some(fetch_err);
                Err(io_err)
            }
        }
    }

    /// Serves `buf` from the cached tail or window, if `pos` is covered.
    fn copy_cached(&self, buf: &mut [u8]) -> Option<usize> {
        if let Some(w) = &self.window
            && let Some(n) = w.copy_at(self.pos, buf)
        {
            return Some(n);
        }
        if let Some(t) = &self.tail
            && let Some(n) = t.copy_at(self.pos, buf)
        {
            return Some(n);
        }
        None
    }

    /// Advances the read position by `n` copied bytes.
    fn advance_pos(&mut self, n: usize) {
        // CAST: usize → u64, copy count bounded by buf length
        #[allow(clippy::as_conversions)]
        {
            self.pos = self.pos.saturating_add(n as u64);
        }
    }

    /// Copies from cache immediately after a fetch that must cover `pos`.
    ///
    /// A miss here means the fetch/cache bookkeeping is internally
    /// inconsistent; surfaced as an error rather than a panic.
    fn serve_from_cache_after_fetch(&mut self, buf: &mut [u8]) -> io::Result<usize> {
        let n = self.copy_cached(buf).ok_or_else(|| {
            io::Error::other(format!(
                "internal range cache inconsistency at offset {}",
                self.pos
            ))
        })?;
        self.advance_pos(n);
        Ok(n)
    }
}

impl<F: RangeFetcher> Read for RangeReader<F> {
    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
        if buf.is_empty() {
            return Ok(0);
        }
        let total = self.fetcher.total_size();
        if self.pos >= total {
            return Ok(0); // EOF (also covers permissive past-end seeks)
        }

        // 1. Serve from cached extents (window, then tail) — no request.
        if let Some(n) = self.copy_cached(buf) {
            self.advance_pos(n);
            return Ok(n);
        }

        // 2. First read near end-of-file: prefetch and cache the tail.
        let tail_start = total.saturating_sub(TAIL_PREFETCH_BYTES);
        if self.tail.is_none() && self.pos >= tail_start {
            let data = self.checked_fetch(tail_start, total.saturating_sub(1))?;
            self.tail = Some(Extent {
                start: tail_start,
                data,
            });
            return self.serve_from_cache_after_fetch(buf);
        }

        // 3. Read-ahead window fetch at the current position.
        // CAST: usize → u64, buffer length always fits (usize ≤ 64 bits)
        #[allow(clippy::as_conversions)]
        let want = (buf.len() as u64).max(READAHEAD_BYTES);
        let mut end = self
            .pos
            .saturating_add(want)
            .saturating_sub(1)
            .min(total.saturating_sub(1));
        // Never overlap the cached tail — those bytes are already paid for.
        if let Some(t) = &self.tail
            && self.pos < t.start
        {
            end = end.min(t.start.saturating_sub(1));
        }
        let data = self.checked_fetch(self.pos, end)?;
        self.window = Some(Extent {
            start: self.pos,
            data,
        });
        self.serve_from_cache_after_fetch(buf)
    }
}

impl<F: RangeFetcher> Seek for RangeReader<F> {
    fn seek(&mut self, pos: SeekFrom) -> io::Result<u64> {
        let total = self.fetcher.total_size();
        let target: i128 = match pos {
            SeekFrom::Start(p) => i128::from(p),
            SeekFrom::End(delta) => i128::from(total).saturating_add(i128::from(delta)),
            SeekFrom::Current(delta) => i128::from(self.pos).saturating_add(i128::from(delta)),
        };
        if target < 0 {
            return Err(io::Error::new(
                io::ErrorKind::InvalidInput,
                format!("seek to negative offset {target}"),
            ));
        }
        let new_pos = u64::try_from(target).map_err(|_| {
            io::Error::new(
                io::ErrorKind::InvalidInput,
                format!("seek offset {target} exceeds u64::MAX"),
            )
        })?;
        self.pos = new_pos;
        Ok(new_pos)
    }
}

// -----------------------------------------------------------------------
// HTTP transport
// -----------------------------------------------------------------------

/// [`RangeReader`] over [`HttpRangeFetcher`] — the production remote-inspect
/// substrate.
pub type HttpRangeReader = RangeReader<HttpRangeFetcher>;

/// HTTP Range transport against a `HuggingFace`-hosted file.
///
/// Opened via [`HttpRangeReader::open`]. One authenticated, redirect-free
/// probe (reusing the chunked-download probe) resolves the file size and
/// classifies access failures eagerly (gated repos, no Range support).
/// **Every range request then targets the HF `/resolve` URL** and follows
/// the redirect to a freshly-signed CDN URL; `reqwest` strips the
/// `Authorization` header on the cross-host redirect, so the `Bearer`
/// token is never sent to the CDN host.
///
/// Per-request re-resolution is **required**, not just convenient:
/// Xet-backed repos sign each CDN URL for the *exact* `Range` header that
/// minted it (a `ByteRange` policy condition in the signed URL), so a CDN
/// URL stored from the probe can never serve any other range. Re-resolving
/// also makes signed-URL expiry a non-issue — every request gets a fresh
/// signature. Discovered live against `google/gemma-scope-2b-pt-res`
/// (v0.11.0 dogfooding).
///
/// # Response validation (every request)
///
/// - Status must be `206 Partial Content`. A `200` (server ignoring the
///   `Range` header) is rejected **without reading the body** — the guard
///   against silently pulling the full file.
/// - `Content-Range` must match the requested range and the probed total
///   size.
/// - The body is streamed with a hard cap of the requested length;
///   over- or under-delivery is an error.
/// - The response `ETag` (when present) must stay identical across
///   requests — a mid-inspect change upstream aborts with a clear error.
pub struct HttpRangeFetcher {
    /// Runtime handle used to drive async `reqwest` calls from the
    /// blocking `Read` context (`spawn_blocking` thread).
    handle: tokio::runtime::Handle,
    /// Authenticated, redirect-following client. The auth header rides
    /// only to the HF host — `reqwest` removes sensitive headers when a
    /// redirect crosses hosts.
    client: reqwest::Client,
    /// The `HF` `/resolve` URL every range request targets.
    hf_url: String,
    /// Filename for error messages.
    filename: String,
    /// First `ETag` observed on a range response — later responses must
    /// match it (self-consistency across requests).
    response_etag: Option<String>,
    /// The etag observed at probe time, before any range request — the
    /// same value `chunked::probe_range_support` uses for the blob path in
    /// the `hf-hub` cache. Read by `inspect --cache-headers` to key its
    /// header cache without an extra round trip; distinct from
    /// `response_etag`, which validates self-consistency across the range
    /// requests this reader itself issues.
    probe_etag: String,
    /// Total file size from the probe's `Content-Range`.
    total_size: u64,
    /// Probe requests spent (no-redirect probe + CDN size fetch).
    extra: u32,
}

impl HttpRangeReader {
    /// Opens a range reader over `filename` in `repo_id` at `revision`
    /// (default `main`).
    ///
    /// Performs the probe eagerly, so failures (missing file, gated repo,
    /// no Range support) surface here as typed errors rather than
    /// mid-parse `io::Error`s.
    ///
    /// Must be called from within a `tokio` runtime; the returned reader is
    /// then handed to a blocking context (`tokio::task::spawn_blocking`),
    /// where its `Read` / `Seek` impls drive async requests via the
    /// captured runtime handle.
    ///
    /// # Errors
    ///
    /// Returns [`FetchError::Http`] if the probe fails, the repository or
    /// file is inaccessible (including gated repos: `returned status
    /// 401/403` errors, which the CLI upgrades into a gated-repo
    /// diagnosis), or the server does not support Range requests.
    pub async fn open(
        repo_id: &str,
        revision: Option<&str>,
        filename: &str,
        token: Option<&str>,
    ) -> Result<Self, FetchError> {
        Self::open_with_limits(
            repo_id,
            revision,
            filename,
            token,
            MAX_RANGE_REQUESTS,
            MAX_TRANSFER_BUDGET,
        )
        .await
    }

    /// Opens a range reader with explicit safety budgets (see
    /// [`RangeReader::with_limits`]) instead of the [`MAX_RANGE_REQUESTS`] /
    /// [`MAX_TRANSFER_BUDGET`] defaults [`HttpRangeReader::open`] uses.
    ///
    /// The defaults are tuned for archive-header parsing (`inspect`'s
    /// `.safetensors` / `.gguf` / `.npz` / `.pth` paths) — a handful of
    /// requests, well under 1 MiB. A caller with its own, larger,
    /// user-facing byte budget (`hf-fm peek`'s `--max`) needs its transport
    /// budget sized to match, or safety limits tuned for header parsing cut
    /// it off first with a misleading "pathological archive layout" error.
    ///
    /// Same preconditions as [`HttpRangeReader::open`] (must be called from
    /// within a `tokio` runtime; the reader is then driven from a blocking
    /// context).
    ///
    /// # Errors
    ///
    /// Same as [`HttpRangeReader::open`].
    pub async fn open_with_limits(
        repo_id: &str,
        revision: Option<&str>,
        filename: &str,
        token: Option<&str>,
        max_requests: u32,
        max_transfer_bytes: u64,
    ) -> Result<Self, FetchError> {
        let fetcher = HttpRangeFetcher::open(repo_id, revision, filename, token).await?;
        Ok(RangeReader::with_limits(
            fetcher,
            max_requests,
            max_transfer_bytes,
        ))
    }

    /// The etag observed at probe time, before any range request.
    ///
    /// Lets a caller key a cache entry (`inspect --cache-headers`) on the
    /// same etag `hf-hub`'s own blob layout uses, without an extra round
    /// trip beyond the probe [`HttpRangeReader::open`] already made.
    #[must_use]
    pub fn probe_etag(&self) -> &str {
        self.fetcher.probe_etag()
    }
}

impl HttpRangeFetcher {
    /// The etag observed at probe time, before any range request.
    #[must_use]
    fn probe_etag(&self) -> &str {
        // BORROW: explicit .as_str() instead of Deref coercion
        self.probe_etag.as_str()
    }

    /// Probes `filename` and constructs the transport (see
    /// [`HttpRangeReader::open`]).
    ///
    /// # Errors
    ///
    /// Returns [`FetchError::Http`] if the probe fails, the file is
    /// inaccessible, or the server does not support Range requests.
    async fn open(
        repo_id: &str,
        revision: Option<&str>,
        filename: &str,
        token: Option<&str>,
    ) -> Result<Self, FetchError> {
        let rev = revision.unwrap_or("main");
        let hf_url = chunked::build_download_url(repo_id, rev, filename);
        let client = chunked::build_client(token)?;

        let info = chunked::probe_range_support(
            client.clone(),
            hf_url.clone(),
            // BORROW: explicit String::from for Option<&str> → Option<String>
            token.map(String::from),
        )
        .await?;
        let Some(info) = info else {
            return Err(Self::classify_no_range_support(&client, &hf_url, filename).await);
        };

        Ok(Self {
            handle: tokio::runtime::Handle::current(),
            client,
            hf_url,
            // BORROW: explicit .to_owned() for the owned field
            filename: filename.to_owned(),
            response_etag: None,
            probe_etag: info.etag,
            total_size: info.content_length,
            extra: 2, // no-redirect probe + CDN size fetch
        })
    }

    /// Distinguishes "no Range support" from an access failure after the
    /// probe declined.
    ///
    /// The chunked probe returns `None` for **any** non-redirect,
    /// non-`206` response — including a gated repo's `401`/`403`. One
    /// follow-up request recovers the real status so gated repos get the
    /// actionable diagnosis instead of a misleading "no Range support".
    async fn classify_no_range_support(
        client: &reqwest::Client,
        url: &str,
        filename: &str,
    ) -> FetchError {
        let result = client
            .get(url)
            .header(reqwest::header::RANGE, "bytes=0-0")
            .timeout(RANGE_REQUEST_TIMEOUT)
            .send()
            .await;
        match result {
            Ok(resp) => {
                let status = resp.status();
                if status.is_client_error() || status.is_server_error() {
                    FetchError::Http(format!(
                        "Range request for {filename} returned status {status}"
                    ))
                } else {
                    FetchError::Http(format!(
                        "server does not support Range requests for {filename}"
                    ))
                }
            }
            Err(e) => FetchError::Http(format!("failed to probe {filename}: {e}")),
        }
    }

    /// Issues one validated range request against the `/resolve` URL,
    /// following the redirect to a freshly-signed CDN URL.
    ///
    /// Returns the body plus the response `ETag` (cleaned), which the
    /// caller records for cross-request consistency.
    fn fetch_once(
        &self,
        start: u64,
        end_inclusive: u64,
    ) -> Result<(Vec<u8>, Option<String>), FetchError> {
        let expected_len = end_inclusive
            .checked_sub(start)
            .and_then(|d| d.checked_add(1))
            .ok_or_else(|| {
                FetchError::Http(format!(
                    "invalid range {start}..={end_inclusive} for {}",
                    self.filename
                ))
            })?;
        let expected_usize = usize::try_from(expected_len).map_err(|_| {
            FetchError::Http(format!(
                "range length {expected_len} exceeds addressable memory for {}",
                self.filename
            ))
        })?;

        let range_value = format!("bytes={start}-{end_inclusive}");
        // BORROW: explicit .as_str() instead of Deref coercion
        let filename = self.filename.as_str();
        let total_size = self.total_size;

        self.handle.block_on(async {
            let resp = self
                .client
                .get(self.hf_url.as_str())
                // BORROW: explicit .as_str() instead of Deref coercion
                .header(reqwest::header::RANGE, range_value.as_str())
                .timeout(RANGE_REQUEST_TIMEOUT)
                .send()
                .await
                .map_err(|e| {
                    FetchError::Http(format!("failed to send Range request for {filename}: {e}"))
                })?;

            let status = resp.status();
            if status == reqwest::StatusCode::OK {
                // Body deliberately not read: a 200 means the server ignored
                // the Range header and is offering the FULL file.
                return Err(FetchError::Http(format!(
                    "server ignored the Range header for {filename} (status 200 \
                     for bytes={start}-{end_inclusive}); refusing to read the full file"
                )));
            }
            if status != reqwest::StatusCode::PARTIAL_CONTENT {
                return Err(FetchError::Http(format!(
                    "Range request for {filename} returned status {status}"
                )));
            }

            let content_range = resp
                .headers()
                .get(reqwest::header::CONTENT_RANGE)
                .and_then(|v| v.to_str().ok())
                // BORROW: explicit .to_owned() — header value outlives the response borrow
                .map(str::to_owned)
                .ok_or_else(|| {
                    FetchError::Http(format!("missing Content-Range header for {filename}"))
                })?;
            let (cr_start, cr_end, cr_total) = parse_content_range(content_range.as_str())
                .ok_or_else(|| {
                    FetchError::Http(format!(
                        "invalid Content-Range header for {filename}: {content_range}"
                    ))
                })?;
            if cr_start != start || cr_end != end_inclusive || cr_total != total_size {
                return Err(FetchError::Http(format!(
                    "Content-Range mismatch for {filename}: requested \
                     bytes={start}-{end_inclusive} of {total_size}, server answered {content_range}"
                )));
            }

            let etag = resp
                .headers()
                .get(reqwest::header::ETAG)
                .and_then(|v| v.to_str().ok())
                .map(clean_etag);

            // Stream the body with a hard cap of the requested length.
            let mut data: Vec<u8> = Vec::with_capacity(expected_usize);
            let mut resp = resp;
            while let Some(chunk) = resp.chunk().await.map_err(|e| {
                FetchError::Http(format!("failed to read Range response for {filename}: {e}"))
            })? {
                if data.len().saturating_add(chunk.len()) > expected_usize {
                    return Err(FetchError::Http(format!(
                        "server sent more than the requested {expected_len} bytes \
                         for {filename} (bytes={start}-{end_inclusive}); aborting"
                    )));
                }
                data.extend_from_slice(&chunk);
            }
            if data.len() != expected_usize {
                return Err(FetchError::Http(format!(
                    "server returned {} bytes for a {expected_len}-byte range \
                     of {filename} (bytes={start}-{end_inclusive})",
                    data.len()
                )));
            }

            Ok((data, etag))
        })
    }

    /// Records / checks the response `ETag` for cross-request consistency.
    fn check_response_etag(&mut self, etag: Option<String>) -> Result<(), FetchError> {
        if let Some(current) = etag {
            match &self.response_etag {
                Some(previous) if *previous != current => {
                    return Err(FetchError::Http(format!(
                        "{} changed upstream during inspect (etag {previous} \
                         became {current})",
                        self.filename
                    )));
                }
                Some(_) => {} // EXPLICIT: etag unchanged — nothing to record
                None => self.response_etag = Some(current),
            }
        }
        Ok(())
    }
}

impl RangeFetcher for HttpRangeFetcher {
    fn fetch(&mut self, start: u64, end_inclusive: u64) -> Result<Vec<u8>, FetchError> {
        // Each fetch re-resolves through /resolve (fresh signed CDN URL),
        // so there is no stored-signature expiry to manage; failures
        // surface directly with their real HTTP status.
        let (data, etag) = self.fetch_once(start, end_inclusive)?;
        self.check_response_etag(etag)?;
        Ok(data)
    }

    fn total_size(&self) -> u64 {
        self.total_size
    }

    fn extra_requests(&self) -> u32 {
        self.extra
    }
}

/// Parses a `Content-Range: bytes S-E/T` value into `(S, E, T)`.
///
/// Returns `None` on any deviation from that exact form (including the
/// `bytes */T` unsatisfied-range form, which is never valid for a `206`).
fn parse_content_range(value: &str) -> Option<(u64, u64, u64)> {
    let rest = value.strip_prefix("bytes ")?;
    let (range, total) = rest.split_once('/')?;
    let (start, end) = range.split_once('-')?;
    Some((
        start.trim().parse().ok()?,
        end.trim().parse().ok()?,
        total.trim().parse().ok()?,
    ))
}

/// Normalises an `ETag` value: strips the weak-validator prefix and quotes.
///
/// Matches the probe's normalisation (`etag.replace('"', "")`), so probe
/// and response etags compare in the same representation.
fn clean_etag(raw: &str) -> String {
    raw.strip_prefix("W/").unwrap_or(raw).replace('"', "")
}

// -----------------------------------------------------------------------
// Tests
// -----------------------------------------------------------------------

#[cfg(test)]
mod tests {
    #![allow(clippy::unwrap_used, clippy::expect_used, clippy::indexing_slicing)]

    use super::*;

    /// In-memory fetcher over a byte vector, logging every fetched range.
    struct InMemoryFetcher {
        data: Vec<u8>,
        calls: Vec<(u64, u64)>,
    }

    impl InMemoryFetcher {
        fn new(data: Vec<u8>) -> Self {
            Self {
                data,
                calls: Vec::new(),
            }
        }
    }

    impl RangeFetcher for InMemoryFetcher {
        fn fetch(&mut self, start: u64, end_inclusive: u64) -> Result<Vec<u8>, FetchError> {
            self.calls.push((start, end_inclusive));
            let s = usize::try_from(start).unwrap();
            let e = usize::try_from(end_inclusive).unwrap();
            self.data
                .get(s..=e)
                .map(<[u8]>::to_vec)
                .ok_or_else(|| FetchError::Http(format!("bad range {start}..={end_inclusive}")))
        }

        fn total_size(&self) -> u64 {
            u64::try_from(self.data.len()).unwrap()
        }
    }

    /// A fetcher that always fails with an HTTP-status-shaped error.
    struct FailingFetcher {
        size: u64,
    }

    impl RangeFetcher for FailingFetcher {
        fn fetch(&mut self, _start: u64, _end_inclusive: u64) -> Result<Vec<u8>, FetchError> {
            Err(FetchError::Http(
                "Range request for x.npz returned status 403 Forbidden".to_owned(),
            ))
        }

        fn total_size(&self) -> u64 {
            self.size
        }
    }

    /// A fetcher that returns fewer bytes than requested.
    struct ShortFetcher {
        size: u64,
    }

    impl RangeFetcher for ShortFetcher {
        fn fetch(&mut self, _start: u64, _end_inclusive: u64) -> Result<Vec<u8>, FetchError> {
            Ok(vec![0u8; 1])
        }

        fn total_size(&self) -> u64 {
            self.size
        }
    }

    fn sample_data(len: usize) -> Vec<u8> {
        // CAST: usize → u8, deliberate wrapping for a recognisable pattern
        #[allow(clippy::as_conversions, clippy::cast_possible_truncation)]
        (0..len).map(|i| (i % 251) as u8).collect()
    }

    // ---------- Seek semantics ----------

    #[test]
    fn seek_start_end_current_semantics() {
        let mut r = RangeReader::new(InMemoryFetcher::new(sample_data(1000)));
        assert_eq!(r.seek(SeekFrom::Start(10)).unwrap(), 10);
        assert_eq!(r.seek(SeekFrom::Current(5)).unwrap(), 15);
        assert_eq!(r.seek(SeekFrom::Current(-15)).unwrap(), 0);
        assert_eq!(r.seek(SeekFrom::End(0)).unwrap(), 1000);
        assert_eq!(r.seek(SeekFrom::End(-1000)).unwrap(), 0);
        // Past-EOF seek is permitted; reads there return 0.
        assert_eq!(r.seek(SeekFrom::End(50)).unwrap(), 1050);
        let mut buf = [0u8; 4];
        assert_eq!(r.read(&mut buf).unwrap(), 0);
    }

    #[test]
    fn seek_negative_is_invalid_input() {
        let mut r = RangeReader::new(InMemoryFetcher::new(sample_data(100)));
        let err = r.seek(SeekFrom::Current(-1)).unwrap_err();
        assert_eq!(err.kind(), io::ErrorKind::InvalidInput);
        let err = r.seek(SeekFrom::End(-101)).unwrap_err();
        assert_eq!(err.kind(), io::ErrorKind::InvalidInput);
    }

    #[test]
    fn empty_file_reads_zero_with_no_requests() {
        let mut r = RangeReader::new(InMemoryFetcher::new(Vec::new()));
        let mut buf = [0u8; 8];
        assert_eq!(r.read(&mut buf).unwrap(), 0);
        assert_eq!(r.stats().requests, 0);
    }

    // ---------- Read-ahead and caching ----------

    #[test]
    fn sequential_small_reads_coalesce_into_one_window_fetch() {
        // 100 KiB file, reads far from the tail region.
        let mut r = RangeReader::new(InMemoryFetcher::new(sample_data(100 * 1024)));
        let mut buf = [0u8; 16];
        for i in 0..10 {
            r.read_exact(&mut buf).unwrap();
            assert_eq!(buf[0], sample_data(100 * 1024)[i * 16]);
        }
        // 160 bytes of sequential reads << 4 KiB read-ahead → one request.
        assert_eq!(r.stats().requests, 1);
        assert_eq!(r.fetcher.calls[0], (0, READAHEAD_BYTES - 1));
    }

    #[test]
    fn tail_region_read_prefetches_tail_once() {
        let size: u64 = 1024 * 1024; // 1 MiB
        let mut r = RangeReader::new(InMemoryFetcher::new(sample_data(
            usize::try_from(size).unwrap(),
        )));
        // Mimic the ZIP EOCD scan: seek near the end, read a few bytes.
        r.seek(SeekFrom::End(-22)).unwrap();
        let mut buf = [0u8; 22];
        r.read_exact(&mut buf).unwrap();
        assert_eq!(r.stats().requests, 1);
        assert_eq!(r.fetcher.calls[0], (size - TAIL_PREFETCH_BYTES, size - 1));
        // Every further read inside the tail is served from cache.
        r.seek(SeekFrom::End(-4096)).unwrap();
        let mut big = [0u8; 4096];
        r.read_exact(&mut big).unwrap();
        assert_eq!(r.stats().requests, 1);
        assert_eq!(r.stats().bytes_fetched, TAIL_PREFETCH_BYTES);
    }

    #[test]
    fn window_reuse_after_seek_back() {
        let mut r = RangeReader::new(InMemoryFetcher::new(sample_data(64 * 1024)));
        let mut buf = [0u8; 128];
        r.read_exact(&mut buf).unwrap();
        assert_eq!(r.stats().requests, 1);
        // Seek back inside the fetched window: no new request.
        r.seek(SeekFrom::Start(32)).unwrap();
        r.read_exact(&mut buf).unwrap();
        assert_eq!(r.stats().requests, 1);
        let expected = sample_data(64 * 1024);
        assert_eq!(&buf[..], &expected[32..160]);
    }

    #[test]
    fn window_fetch_never_overlaps_cached_tail() {
        let size: u64 = 200 * 1024;
        let mut r = RangeReader::new(InMemoryFetcher::new(sample_data(
            usize::try_from(size).unwrap(),
        )));
        // Prime the tail cache.
        r.seek(SeekFrom::End(-10)).unwrap();
        let mut small = [0u8; 10];
        r.read_exact(&mut small).unwrap();
        // Large read starting just before the tail: the window fetch must
        // stop at the tail boundary, and the rest is served from the tail.
        let tail_start = size - TAIL_PREFETCH_BYTES;
        r.seek(SeekFrom::Start(tail_start - 100)).unwrap();
        let mut big = vec![0u8; 4096];
        r.read_exact(&mut big).unwrap();
        assert_eq!(r.fetcher.calls[1], (tail_start - 100, tail_start - 1));
        let expected = sample_data(usize::try_from(size).unwrap());
        let from = usize::try_from(tail_start - 100).unwrap();
        assert_eq!(&big[..], &expected[from..from + 4096]);
    }

    // ---------- Safety budgets ----------

    #[test]
    fn request_cap_is_enforced() {
        let data = sample_data(10 * 1024 * 1024);
        let mut r = RangeReader::with_limits(InMemoryFetcher::new(data), 3, u64::MAX);
        let mut buf = [0u8; 8];
        // Three widely-spaced reads consume the three allowed requests
        // (spacing > READAHEAD_BYTES so no window reuse).
        for i in 0u64..3 {
            r.seek(SeekFrom::Start(i * 100 * 1024)).unwrap();
            r.read_exact(&mut buf).unwrap();
        }
        r.seek(SeekFrom::Start(1024 * 1024)).unwrap();
        let err = r.read(&mut buf).unwrap_err();
        assert!(
            err.to_string()
                .contains("range request cap exceeded (3 requests)"),
            "unexpected error: {err}"
        );
    }

    #[test]
    fn transfer_budget_is_enforced() {
        let data = sample_data(10 * 1024 * 1024);
        // Budget of 6 KiB: one 4 KiB window fits, the second does not.
        let mut r = RangeReader::with_limits(InMemoryFetcher::new(data), u32::MAX, 6 * 1024);
        let mut buf = [0u8; 8];
        r.read_exact(&mut buf).unwrap();
        r.seek(SeekFrom::Start(1024 * 1024)).unwrap();
        let err = r.read(&mut buf).unwrap_err();
        assert!(
            err.to_string().contains("range transfer budget exceeded"),
            "unexpected error: {err}"
        );
    }

    // ---------- Error channel ----------

    #[test]
    fn fetch_error_surfaces_as_io_and_is_recoverable_typed() {
        let mut r = RangeReader::new(FailingFetcher { size: 1024 });
        let mut buf = [0u8; 8];
        let err = r.read(&mut buf).unwrap_err();
        assert!(err.to_string().contains("returned status 403"));
        let typed = r.take_last_error().expect("typed error must be stored");
        assert!(matches!(typed, FetchError::Http(msg)
            if msg.contains("returned status 403 Forbidden")));
        // Taken once — the slot is now empty.
        assert!(r.take_last_error().is_none());
    }

    #[test]
    fn short_fetch_is_a_contract_error() {
        let mut r = RangeReader::new(ShortFetcher { size: 1024 * 1024 });
        let mut buf = [0u8; 8];
        let err = r.read(&mut buf).unwrap_err();
        assert!(
            err.to_string().contains("bytes for a"),
            "unexpected error: {err}"
        );
    }

    // ---------- Pure helpers ----------

    #[test]
    fn parse_content_range_accepts_the_exact_206_form() {
        assert_eq!(parse_content_range("bytes 0-7/1234"), Some((0, 7, 1234)));
        assert_eq!(
            parse_content_range("bytes 100-199/200"),
            Some((100, 199, 200))
        );
    }

    #[test]
    fn parse_content_range_rejects_deviant_forms() {
        assert_eq!(parse_content_range("bytes */1234"), None);
        assert_eq!(parse_content_range("bytes 0-7/*"), None);
        assert_eq!(parse_content_range("0-7/1234"), None);
        assert_eq!(parse_content_range("bytes 7/1234"), None);
        assert_eq!(parse_content_range(""), None);
    }

    #[test]
    fn clean_etag_strips_quotes_and_weak_prefix() {
        assert_eq!(clean_etag("\"abc123\""), "abc123");
        assert_eq!(clean_etag("W/\"abc123\""), "abc123");
        assert_eq!(clean_etag("abc123"), "abc123");
    }

    // ---------- NPZ end-to-end over the reader ----------

    /// Builds a minimal `NPY` v1.0 payload: magic, version, padded header
    /// dict, and zeroed data.
    fn npy_bytes(descr: &str, shape_literal: &str, data_len: usize) -> Vec<u8> {
        let dict =
            format!("{{'descr': '{descr}', 'fortran_order': False, 'shape': {shape_literal}, }}");
        // Pad so (magic 6 + version 2 + len 2 + header) % 64 == 0, per spec.
        let unpadded = 10 + dict.len() + 1; // +1 for the trailing '\n'
        let padding = (64 - unpadded % 64) % 64;
        let header_len = dict.len() + padding + 1;
        let mut out = Vec::with_capacity(10 + header_len + data_len);
        out.extend_from_slice(b"\x93NUMPY\x01\x00");
        out.extend_from_slice(&u16::try_from(header_len).unwrap().to_le_bytes());
        out.extend_from_slice(dict.as_bytes());
        out.extend(std::iter::repeat_n(b' ', padding));
        out.push(b'\n');
        out.extend(std::iter::repeat_n(0u8, data_len));
        out
    }

    /// Builds a stored (uncompressed) ZIP archive from `(name, payload)`
    /// entries: local headers, central directory, EOCD. CRC fields are
    /// zero — the inspect path never reads entry data, so they are unused.
    fn stored_zip(entries: &[(&str, Vec<u8>)]) -> Vec<u8> {
        let mut out = Vec::new();
        let mut offsets = Vec::new();
        for (name, payload) in entries {
            offsets.push(u32::try_from(out.len()).unwrap());
            let size = u32::try_from(payload.len()).unwrap();
            out.extend_from_slice(&0x0403_4b50u32.to_le_bytes()); // LFH sig
            out.extend_from_slice(&20u16.to_le_bytes()); // version needed
            out.extend_from_slice(&0u16.to_le_bytes()); // flags
            out.extend_from_slice(&0u16.to_le_bytes()); // method: stored
            out.extend_from_slice(&0u32.to_le_bytes()); // mod time + date
            out.extend_from_slice(&0u32.to_le_bytes()); // CRC-32 (unused)
            out.extend_from_slice(&size.to_le_bytes()); // compressed size
            out.extend_from_slice(&size.to_le_bytes()); // uncompressed size
            out.extend_from_slice(&u16::try_from(name.len()).unwrap().to_le_bytes());
            out.extend_from_slice(&0u16.to_le_bytes()); // extra len
            out.extend_from_slice(name.as_bytes());
            out.extend_from_slice(payload);
        }
        let cd_offset = u32::try_from(out.len()).unwrap();
        for ((name, payload), lfh_offset) in entries.iter().zip(&offsets) {
            let size = u32::try_from(payload.len()).unwrap();
            out.extend_from_slice(&0x0201_4b50u32.to_le_bytes()); // CDFH sig
            out.extend_from_slice(&20u16.to_le_bytes()); // version made by
            out.extend_from_slice(&20u16.to_le_bytes()); // version needed
            out.extend_from_slice(&0u16.to_le_bytes()); // flags
            out.extend_from_slice(&0u16.to_le_bytes()); // method: stored
            out.extend_from_slice(&0u32.to_le_bytes()); // mod time + date
            out.extend_from_slice(&0u32.to_le_bytes()); // CRC-32 (unused)
            out.extend_from_slice(&size.to_le_bytes()); // compressed size
            out.extend_from_slice(&size.to_le_bytes()); // uncompressed size
            out.extend_from_slice(&u16::try_from(name.len()).unwrap().to_le_bytes());
            out.extend_from_slice(&0u16.to_le_bytes()); // extra len
            out.extend_from_slice(&0u16.to_le_bytes()); // comment len
            out.extend_from_slice(&0u16.to_le_bytes()); // disk number
            out.extend_from_slice(&0u16.to_le_bytes()); // internal attrs
            out.extend_from_slice(&0u32.to_le_bytes()); // external attrs
            out.extend_from_slice(&lfh_offset.to_le_bytes());
            out.extend_from_slice(name.as_bytes());
        }
        let cd_size = u32::try_from(out.len()).unwrap() - cd_offset;
        let n = u16::try_from(entries.len()).unwrap();
        out.extend_from_slice(&0x0605_4b50u32.to_le_bytes()); // EOCD sig
        out.extend_from_slice(&0u16.to_le_bytes()); // this disk
        out.extend_from_slice(&0u16.to_le_bytes()); // CD start disk
        out.extend_from_slice(&n.to_le_bytes()); // entries this disk
        out.extend_from_slice(&n.to_le_bytes()); // entries total
        out.extend_from_slice(&cd_size.to_le_bytes());
        out.extend_from_slice(&cd_offset.to_le_bytes());
        out.extend_from_slice(&0u16.to_le_bytes()); // comment len
        out
    }

    #[test]
    fn npz_inspect_over_range_reader_reads_metadata_not_data() {
        // Two arrays; the big one carries 600 KiB of data the inspect
        // must never fetch.
        let npz = stored_zip(&[
            ("w_enc.npy", npy_bytes("<f4", "(2, 3)", 2 * 3 * 4)),
            ("b_dec.npy", npy_bytes("<f4", "(150, 1024)", 600 * 1024)),
        ]);
        let total = u64::try_from(npz.len()).unwrap();
        let mut reader = RangeReader::new(InMemoryFetcher::new(npz));

        let info = anamnesis::inspect_npz_from_reader(&mut reader)
            .expect("synthetic NPZ must inspect cleanly");

        assert_eq!(info.tensors.len(), 2);
        let names: Vec<&str> = info.tensors.iter().map(|t| t.name.as_str()).collect();
        assert!(names.contains(&"w_enc"), "names: {names:?}");
        assert!(names.contains(&"b_dec"), "names: {names:?}");
        let b_dec = info.tensors.iter().find(|t| t.name == "b_dec").unwrap();
        assert_eq!(b_dec.shape, vec![150, 1024]);

        // The efficiency property this module exists for: metadata-only
        // transfer, a small number of requests, no weight data fetched.
        let stats = reader.stats();
        assert!(
            stats.requests <= 8,
            "expected a handful of range requests, got {}",
            stats.requests
        );
        assert!(
            stats.bytes_fetched < total / 4,
            "fetched {} of {total} bytes — inspect must not read tensor data",
            stats.bytes_fetched
        );
    }

    // ---------- Safetensors end-to-end over the reader ----------

    /// Builds a synthetic safetensors buffer: an 8-byte little-endian length
    /// prefix, a JSON header with `count` tensor entries (deliberately large
    /// enough to exceed the reader's 4 KiB read-ahead window, mirroring a
    /// real many-tensor model), and a dummy data section the inspect must
    /// never fetch. Returns `(buffer, header_len)` so callers can assert on
    /// the exact header size instead of hardcoding it.
    fn synthetic_safetensors(count: usize, data_section_len: usize) -> (Vec<u8>, usize) {
        use std::fmt::Write as _;

        let mut entries = String::new();
        let mut offset: u64 = 0;
        for i in 0..count {
            if i > 0 {
                entries.push(',');
            }
            let start = offset;
            let end = offset + 16;
            offset = end;
            let _ = write!(
                entries,
                "\"tensor_{i}\":{{\"dtype\":\"F32\",\"shape\":[4],\"data_offsets\":[{start},{end}]}}"
            );
        }
        let header_bytes = format!("{{{entries}}}").into_bytes();
        let header_len = header_bytes.len();

        let mut buf = Vec::with_capacity(8 + header_len + data_section_len);
        buf.extend_from_slice(&u64::try_from(header_len).unwrap().to_le_bytes());
        buf.extend_from_slice(&header_bytes);
        buf.extend(std::iter::repeat_n(0u8, data_section_len));
        (buf, header_len)
    }

    #[test]
    fn safetensors_inspect_over_range_reader_reads_metadata_not_data() {
        // 100 tensor entries push the JSON header past the 4 KiB read-ahead
        // window, so this also exercises the second-fetch path a real
        // many-tensor model triggers. 256 KiB of dummy tensor data the
        // inspect must never touch.
        let (buf, header_len) = synthetic_safetensors(100, 256 * 1024);
        assert!(
            header_len > 4096,
            "fixture must exceed the 4 KiB read-ahead window to exercise \
             the second-fetch path, got {header_len} bytes"
        );
        let total = u64::try_from(buf.len()).unwrap();
        let mut reader = RangeReader::new(InMemoryFetcher::new(buf));

        let header = anamnesis::parse_safetensors_header_from_reader(&mut reader)
            .expect("synthetic safetensors header must parse cleanly");
        assert_eq!(header.tensors.len(), 100);

        // The efficiency property this module exists for: metadata-only
        // transfer, exactly two requests (the header exceeds the first
        // window fetch, forcing a second), no data section fetched.
        let stats = reader.stats();
        assert_eq!(
            stats.requests, 2,
            "a {header_len}-byte header should need exactly two range \
             requests (first window fetch covers 0..4096, second covers \
             the rest); got {}",
            stats.requests
        );
        assert!(
            stats.bytes_fetched < total / 4,
            "fetched {} of {total} bytes — inspect must not read tensor data",
            stats.bytes_fetched
        );
    }

    #[test]
    fn safetensors_small_header_fits_in_one_read_ahead_window() {
        // A handful of tensors keeps the header well under 4 KiB, so the
        // length prefix and the JSON header are both served by the same
        // initial window fetch — one request total.
        let (buf, header_len) = synthetic_safetensors(3, 1024);
        assert!(
            header_len < 4096,
            "fixture must stay under the 4 KiB read-ahead window, got {header_len} bytes"
        );
        let mut reader = RangeReader::new(InMemoryFetcher::new(buf));

        let header = anamnesis::parse_safetensors_header_from_reader(&mut reader)
            .expect("synthetic safetensors header must parse cleanly");
        assert_eq!(header.tensors.len(), 3);
        assert_eq!(reader.stats().requests, 1);
    }

    // ---------- GGUF end-to-end over the reader ----------

    /// In-memory `GGUF` v3 byte-stream builder — mirrors the layout
    /// anamnesis's parser expects (magic, version, counts, tensor-info
    /// table, alignment-padded data section). Minimal: no metadata KV
    /// entries, `F32` tensors only (`GGML_TYPE_F32 = 0`).
    struct GgufBuilder {
        buf: Vec<u8>,
    }

    impl GgufBuilder {
        fn new(tensor_count: u64) -> Self {
            let mut buf = Vec::new();
            buf.extend_from_slice(b"GGUF");
            buf.extend_from_slice(&3u32.to_le_bytes()); // version
            buf.extend_from_slice(&tensor_count.to_le_bytes());
            buf.extend_from_slice(&0u64.to_le_bytes()); // kv_count = 0
            Self { buf }
        }

        fn push_f32_tensor_info(&mut self, name: &str, shape: &[u64], relative_offset: u64) {
            self.buf
                .extend_from_slice(&u64::try_from(name.len()).unwrap().to_le_bytes());
            self.buf.extend_from_slice(name.as_bytes());
            self.buf
                .extend_from_slice(&u32::try_from(shape.len()).unwrap().to_le_bytes());
            for d in shape {
                self.buf.extend_from_slice(&d.to_le_bytes());
            }
            self.buf.extend_from_slice(&0u32.to_le_bytes()); // GGML_TYPE_F32 = 0
            self.buf.extend_from_slice(&relative_offset.to_le_bytes());
        }

        fn pad_to_alignment(&mut self, alignment: usize) {
            let rem = self.buf.len() % alignment;
            if rem != 0 {
                self.buf.extend(std::iter::repeat_n(0u8, alignment - rem));
            }
        }

        fn push_zeros(&mut self, n: usize) {
            self.buf.extend(std::iter::repeat_n(0u8, n));
        }

        fn finish(self) -> Vec<u8> {
            self.buf
        }
    }

    /// Builds a minimal `GGUF` v3 byte stream: two `F32` tensors, no
    /// metadata, default 32-byte alignment. `b_dec` carries 600 KiB of data
    /// the front-matter parse must never fetch — same magnitude as the NPZ
    /// fixture above, for an apples-to-apples `bytes_fetched` comparison.
    fn synthetic_gguf() -> Vec<u8> {
        let mut b = GgufBuilder::new(2);
        b.push_f32_tensor_info("w_enc", &[2, 3], 0);
        b.push_f32_tensor_info("b_dec", &[153_600], 32);
        b.pad_to_alignment(32); // end of tensor-info table -> data_section_start
        b.push_zeros(24); // w_enc data (2*3*4 bytes)
        b.push_zeros(8); // pad up to b_dec's relative offset 32
        b.push_zeros(153_600 * 4); // b_dec data (600 KiB)
        b.finish()
    }

    #[test]
    fn gguf_front_matter_over_range_reader_reads_metadata_not_data() {
        let gguf = synthetic_gguf();
        let total = u64::try_from(gguf.len()).unwrap();
        let mut reader = RangeReader::new(InMemoryFetcher::new(gguf));

        let front = anamnesis::parse_gguf_front_matter_from_reader(&mut reader)
            .expect("synthetic GGUF must parse cleanly");

        assert_eq!(front.tensor_infos.len(), 2);
        let names: Vec<&str> = front.tensor_infos.iter().map(|t| t.name.as_str()).collect();
        assert!(names.contains(&"w_enc"), "names: {names:?}");
        assert!(names.contains(&"b_dec"), "names: {names:?}");
        let b_dec = front
            .tensor_infos
            .iter()
            .find(|t| t.name == "b_dec")
            .unwrap();
        assert_eq!(b_dec.shape, vec![153_600]);

        // The efficiency property this module exists for: metadata-only
        // transfer, a small number of requests, no weight data fetched.
        // Measured: exactly 1 range request, 64 KiB fetched (bounded by the
        // reader's internal `BufReader` capacity) out of the 600 KiB payload.
        let stats = reader.stats();
        assert!(
            stats.requests <= 8,
            "expected a handful of range requests, got {}",
            stats.requests
        );
        assert!(
            stats.bytes_fetched < total / 4,
            "fetched {} of {total} bytes — front-matter parse must not read tensor data",
            stats.bytes_fetched
        );
    }

    // ---------- PTH end-to-end over the reader ----------

    /// Raw `data.pkl` bytes extracted from anamnesis's real
    /// `tests/fixtures/pth_reference/algzoo_rnn_small.pth` fixture — a
    /// genuine `torch.save()`-produced pickle stream (448 bytes), not a
    /// hand-rolled one. Describes 3 `F32` tensors: `rnn.weight_ih_l0`
    /// `[2, 1]`, `rnn.weight_hh_l0` `[2, 2]`, `linear.weight` `[2, 2]` (see
    /// that fixture's sibling `algzoo_rnn_small_reference.json`). Reusing a
    /// real pickle stream avoids hand-crafting pickle VM opcodes — the same
    /// reasoning anamnesis's own `bench_pth_inspect` fixture uses. The
    /// surrounding ZIP archive below is synthetic, with a large dummy
    /// storage entry the front-matter parse must never fetch.
    const ALGZOO_RNN_SMALL_DATA_PKL: &[u8] = &[
        0x80, 0x02, 0x63, 0x63, 0x6f, 0x6c, 0x6c, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x0a,
        0x4f, 0x72, 0x64, 0x65, 0x72, 0x65, 0x64, 0x44, 0x69, 0x63, 0x74, 0x0a, 0x71, 0x00, 0x29,
        0x52, 0x71, 0x01, 0x28, 0x58, 0x10, 0x00, 0x00, 0x00, 0x72, 0x6e, 0x6e, 0x2e, 0x77, 0x65,
        0x69, 0x67, 0x68, 0x74, 0x5f, 0x69, 0x68, 0x5f, 0x6c, 0x30, 0x71, 0x02, 0x63, 0x74, 0x6f,
        0x72, 0x63, 0x68, 0x2e, 0x5f, 0x75, 0x74, 0x69, 0x6c, 0x73, 0x0a, 0x5f, 0x72, 0x65, 0x62,
        0x75, 0x69, 0x6c, 0x64, 0x5f, 0x74, 0x65, 0x6e, 0x73, 0x6f, 0x72, 0x5f, 0x76, 0x32, 0x0a,
        0x71, 0x03, 0x28, 0x28, 0x58, 0x07, 0x00, 0x00, 0x00, 0x73, 0x74, 0x6f, 0x72, 0x61, 0x67,
        0x65, 0x71, 0x04, 0x63, 0x74, 0x6f, 0x72, 0x63, 0x68, 0x0a, 0x46, 0x6c, 0x6f, 0x61, 0x74,
        0x53, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x0a, 0x71, 0x05, 0x58, 0x01, 0x00, 0x00, 0x00,
        0x30, 0x71, 0x06, 0x58, 0x06, 0x00, 0x00, 0x00, 0x63, 0x75, 0x64, 0x61, 0x3a, 0x30, 0x71,
        0x07, 0x4b, 0x0a, 0x74, 0x71, 0x08, 0x51, 0x4b, 0x00, 0x4b, 0x02, 0x4b, 0x01, 0x86, 0x71,
        0x09, 0x4b, 0x01, 0x4b, 0x01, 0x86, 0x71, 0x0a, 0x89, 0x68, 0x00, 0x29, 0x52, 0x71, 0x0b,
        0x74, 0x71, 0x0c, 0x52, 0x71, 0x0d, 0x58, 0x10, 0x00, 0x00, 0x00, 0x72, 0x6e, 0x6e, 0x2e,
        0x77, 0x65, 0x69, 0x67, 0x68, 0x74, 0x5f, 0x68, 0x68, 0x5f, 0x6c, 0x30, 0x71, 0x0e, 0x68,
        0x03, 0x28, 0x28, 0x68, 0x04, 0x68, 0x05, 0x68, 0x06, 0x58, 0x06, 0x00, 0x00, 0x00, 0x63,
        0x75, 0x64, 0x61, 0x3a, 0x30, 0x71, 0x0f, 0x4b, 0x0a, 0x74, 0x71, 0x10, 0x51, 0x4b, 0x02,
        0x4b, 0x02, 0x4b, 0x02, 0x86, 0x71, 0x11, 0x4b, 0x02, 0x4b, 0x01, 0x86, 0x71, 0x12, 0x89,
        0x68, 0x00, 0x29, 0x52, 0x71, 0x13, 0x74, 0x71, 0x14, 0x52, 0x71, 0x15, 0x58, 0x0d, 0x00,
        0x00, 0x00, 0x6c, 0x69, 0x6e, 0x65, 0x61, 0x72, 0x2e, 0x77, 0x65, 0x69, 0x67, 0x68, 0x74,
        0x71, 0x16, 0x68, 0x03, 0x28, 0x28, 0x68, 0x04, 0x68, 0x05, 0x58, 0x01, 0x00, 0x00, 0x00,
        0x31, 0x71, 0x17, 0x58, 0x06, 0x00, 0x00, 0x00, 0x63, 0x75, 0x64, 0x61, 0x3a, 0x30, 0x71,
        0x18, 0x4b, 0x04, 0x74, 0x71, 0x19, 0x51, 0x4b, 0x00, 0x4b, 0x02, 0x4b, 0x02, 0x86, 0x71,
        0x1a, 0x4b, 0x02, 0x4b, 0x01, 0x86, 0x71, 0x1b, 0x89, 0x68, 0x00, 0x29, 0x52, 0x71, 0x1c,
        0x74, 0x71, 0x1d, 0x52, 0x71, 0x1e, 0x75, 0x7d, 0x71, 0x1f, 0x58, 0x09, 0x00, 0x00, 0x00,
        0x5f, 0x6d, 0x65, 0x74, 0x61, 0x64, 0x61, 0x74, 0x61, 0x71, 0x20, 0x68, 0x00, 0x29, 0x52,
        0x71, 0x21, 0x28, 0x58, 0x00, 0x00, 0x00, 0x00, 0x71, 0x22, 0x7d, 0x71, 0x23, 0x58, 0x07,
        0x00, 0x00, 0x00, 0x76, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x71, 0x24, 0x4b, 0x01, 0x73,
        0x58, 0x03, 0x00, 0x00, 0x00, 0x72, 0x6e, 0x6e, 0x71, 0x25, 0x7d, 0x71, 0x26, 0x68, 0x24,
        0x4b, 0x01, 0x73, 0x58, 0x06, 0x00, 0x00, 0x00, 0x6c, 0x69, 0x6e, 0x65, 0x61, 0x72, 0x71,
        0x27, 0x7d, 0x71, 0x28, 0x68, 0x24, 0x4b, 0x01, 0x73, 0x75, 0x73, 0x62, 0x2e,
    ];

    #[test]
    fn pth_front_matter_over_range_reader_reads_metadata_not_data() {
        // The real `data.pkl` plus a 600 KiB dummy storage entry the
        // front-matter parse must never fetch — same magnitude as the
        // NPZ/GGUF fixtures above, for an apples-to-apples comparison.
        let pth = stored_zip(&[
            ("archive/data.pkl", ALGZOO_RNN_SMALL_DATA_PKL.to_vec()),
            ("archive/data/bulk", vec![0u8; 600 * 1024]),
        ]);
        let total = u64::try_from(pth.len()).unwrap();
        let mut reader = RangeReader::new(InMemoryFetcher::new(pth));

        let front = anamnesis::parse_pth_front_matter_from_reader(&mut reader)
            .expect("real algzoo data.pkl must parse cleanly");

        assert_eq!(front.tensors.len(), 3);
        let names: Vec<&str> = front.tensors.iter().map(|t| t.name.as_str()).collect();
        assert!(names.contains(&"rnn.weight_ih_l0"), "names: {names:?}");
        assert!(names.contains(&"rnn.weight_hh_l0"), "names: {names:?}");
        assert!(names.contains(&"linear.weight"), "names: {names:?}");
        let hh = front
            .tensors
            .iter()
            .find(|t| t.name == "rnn.weight_hh_l0")
            .unwrap();
        assert_eq!(hh.shape, vec![2, 2]);
        assert!(!front.big_endian);

        // The efficiency property this module exists for: metadata-only
        // transfer, a small number of requests, no bulk storage data
        // fetched.
        let stats = reader.stats();
        assert!(
            stats.requests <= 8,
            "expected a handful of range requests, got {}",
            stats.requests
        );
        assert!(
            stats.bytes_fetched < total / 4,
            "fetched {} of {total} bytes — front-matter parse must not read tensor data",
            stats.bytes_fetched
        );
    }
}