tensogram 0.22.0

// (C) Copyright 2026- ECMWF and individual contributors.
//
// This software is licensed under the terms of the Apache Licence Version 2.0
// which can be obtained at http://www.apache.org/licenses/LICENSE-2.0.
// In applying this licence, ECMWF does not waive the privileges and immunities
// granted to it by virtue of its status as an intergovernmental organisation nor
// does it submit to any jurisdiction.

//! Remote object store backend for `TensogramFile`.
//!
//! Enables reading `.tgm` files from S3, GCS, Azure, and HTTP(S) via
//! the `object_store` crate. Feature-gated behind `remote`.

use std::collections::BTreeMap;
use std::ops::Range;
#[cfg(feature = "async")]
use std::sync::MutexGuard;
use std::sync::{Arc, Mutex, OnceLock};

use bytes::Bytes;
use object_store::path::Path as ObjectPath;
use object_store::{ObjectStore, ObjectStoreExt};
use url::Url;

use crate::decode::DecodeOptions;
use crate::error::{Result, TensogramError};
use crate::framing;
use crate::metadata;
use crate::remote_scan_parse::footer_region_present;
use crate::scan_opts::RemoteScanOptions;
use crate::types::{DataObjectDescriptor, GlobalMetadata, IndexFrame};
use crate::wire::{
    DATA_OBJECT_FOOTER_SIZE, DataObjectFlags, FRAME_COMMON_FOOTER_SIZE, FRAME_END,
    FRAME_HEADER_SIZE, FrameHeader, FrameType, MAGIC, MessageFlags, POSTAMBLE_SIZE, PREAMBLE_SIZE,
    Postamble, Preamble,
};

// ── URL scheme detection ─────────────────────────────────────────────────────

const REMOTE_SCHEMES: &[&str] = &["s3", "s3a", "gs", "az", "azure", "http", "https"];

// ── Shared tokio runtime ─────────────────────────────────────────────────────

/// Process-wide shared tokio runtime for remote I/O.
///
/// Wrapped in `Result` so that `get_or_init` (which runs the closure
/// exactly once, no races) can cache a build failure without panicking.
static SHARED_RUNTIME: OnceLock<std::result::Result<tokio::runtime::Runtime, String>> =
    OnceLock::new();

fn shared_runtime() -> Result<&'static tokio::runtime::Runtime> {
    SHARED_RUNTIME
        .get_or_init(|| {
            tokio::runtime::Builder::new_multi_thread()
                .worker_threads(2)
                .enable_all()
                .thread_name("tensogram-remote-io")
                .build()
                .map_err(|e| format!("tokio runtime: {e}"))
        })
        .as_ref()
        .map_err(|e| TensogramError::Remote(e.clone()))
}

/// Run an async operation synchronously using the shared runtime.
///
/// Three strategies based on the calling context:
///
/// - **Not in async context** (Python, CLI): direct `handle.block_on()`,
///   no extra thread creation.
/// - **Inside multi-thread tokio runtime** (`#[tokio::test]`, server
///   handler): `block_in_place` + `handle.block_on()`, which tells
///   tokio to spawn a replacement worker so the current one can block
///   without causing runtime starvation.
/// - **Inside current-thread tokio runtime**: scoped thread fallback,
///   since `block_in_place` is not supported on single-threaded
///   runtimes.
///
/// In all cases the shared runtime's event loop and connection pool
/// are reused, eliminating the per-call runtime creation overhead of
/// the old `block_on_thread` pattern.
fn block_on_shared<T, Fut>(future: Fut) -> Result<T>
where
    T: Send,
    Fut: std::future::Future<Output = std::result::Result<T, object_store::Error>> + Send,
{
    let rt = shared_runtime()?;
    let handle = rt.handle().clone();

    match tokio::runtime::Handle::try_current() {
        Err(_) => {
            // Not in an async context — drive the future directly.
            handle
                .block_on(future)
                .map_err(|e| TensogramError::Remote(e.to_string()))
        }
        Ok(current) if current.runtime_flavor() == tokio::runtime::RuntimeFlavor::MultiThread => {
            // Multi-thread runtime: block_in_place lets tokio spawn a
            // replacement worker so this one can block safely.
            tokio::task::block_in_place(|| {
                handle
                    .block_on(future)
                    .map_err(|e| TensogramError::Remote(e.to_string()))
            })
        }
        Ok(_) => {
            // Current-thread runtime: block_in_place is unsupported,
            // fall back to a scoped thread.
            std::thread::scope(|s| {
                match s
                    .spawn(|| {
                        handle
                            .block_on(future)
                            .map_err(|e| TensogramError::Remote(e.to_string()))
                    })
                    .join()
                {
                    Ok(result) => result,
                    Err(_) => Err(TensogramError::Remote(
                        "remote I/O thread panicked".to_string(),
                    )),
                }
            })
        }
    }
}

pub fn is_remote_url(source: &str) -> bool {
    match source.find("://") {
        Some(pos) => {
            let scheme = &source[..pos];
            REMOTE_SCHEMES
                .iter()
                .any(|s| s.eq_ignore_ascii_case(scheme))
        }
        None => false,
    }
}

// ── Cached per-message layout ────────────────────────────────────────────────

#[derive(Debug, Clone)]
struct CachedLayout {
    offset: u64,
    length: u64,
    preamble: Preamble,
    index: Option<IndexFrame>,
    global_metadata: Option<GlobalMetadata>,
}

// ── Backward hop primitives ──────────────────────────────────────────────────

/// Outcome of parsing the backward postamble fetch.  Pure — does not
/// touch state.
///
/// The private mirror of [`crate::remote_scan_parse::BackwardOutcome`]
/// — kept separate because this carries `Preamble` / `CachedLayout`
/// types that cannot cross the WASM boundary.  Both must stay in
/// parity; the parity harness fixtures surface any drift as a
/// Rust↔TS layout divergence.
#[derive(Debug)]
enum BackwardOutcome {
    /// Postamble parsed cleanly.  `msg_start` is the candidate
    /// message-start offset; the caller still has to validate the
    /// preamble at that offset before committing.  `first_footer_offset`
    /// is surfaced so the dispatcher can fold an eager footer-region
    /// fetch into the same paired round when the message is
    /// footer-indexed (see [`crate::remote_scan_parse::footer_region_present`]).
    NeedPreambleValidation {
        msg_start: u64,
        length: u64,
        first_footer_offset: u64,
    },
    /// Format error — the backward walker yields and any provisional
    /// suffix layouts are discarded.  The string identifies which
    /// taxonomy row triggered.
    Format(&'static str),
    /// Streaming non-seekable producer (`postamble.total_length == 0`).
    /// Backward yields; forward continues.
    Streaming,
}

/// Parse a backward postamble fetch and decide what (if anything) to
/// validate next.  Pure function — independent of `RemoteState` and
/// `RemoteBackend`.
///
/// `pa_bytes` is expected to be the 24-byte postamble at
/// `[snap.prev - POSTAMBLE_SIZE, snap.prev)`.  Implausible
/// `total_length` values are rejected by the arithmetic underflow
/// (`total > snap.prev`) and forward-overlap (`msg_start < snap.next`)
/// guards; the candidate-preamble validation step catches anything
/// that slips through, at the cost of a single 24-byte GET.
fn parse_backward_postamble(pa_bytes: &[u8], snap: &ScanSnapshot) -> BackwardOutcome {
    let min_message_size = (PREAMBLE_SIZE + POSTAMBLE_SIZE) as u64;

    if pa_bytes.len() < POSTAMBLE_SIZE {
        return BackwardOutcome::Format("short-fetch-bwd");
    }
    if &pa_bytes[POSTAMBLE_SIZE - crate::wire::END_MAGIC.len()..] != crate::wire::END_MAGIC {
        return BackwardOutcome::Format("bad-end-magic-bwd");
    }
    let postamble = match Postamble::read_from(pa_bytes) {
        Ok(p) => p,
        Err(_) => return BackwardOutcome::Format("postamble-parse-error"),
    };
    let total = postamble.total_length;
    if total == 0 {
        return BackwardOutcome::Streaming;
    }
    if total < min_message_size {
        return BackwardOutcome::Format("length-below-minimum-bwd");
    }
    let msg_start = match snap.prev.checked_sub(total) {
        Some(s) => s,
        None => return BackwardOutcome::Format("backward-arith-underflow"),
    };
    if msg_start < snap.next {
        return BackwardOutcome::Format("backward-overlaps-forward");
    }
    BackwardOutcome::NeedPreambleValidation {
        msg_start,
        length: total,
        first_footer_offset: postamble.first_footer_offset,
    }
}

/// Outcome of validating the backward preamble fetched at the
/// candidate `msg_start` produced by [`parse_backward_postamble`].
#[derive(Debug)]
enum BackwardCommit {
    Format(&'static str),
    Layout(CachedLayout),
}

/// Validate the backward candidate preamble and produce a
/// commit-or-yield decision.  `msg_start` and `length` are the
/// candidate values from the matching [`BackwardOutcome`].
fn validate_backward_preamble(
    preamble_bytes: &[u8],
    msg_start: u64,
    length: u64,
) -> BackwardCommit {
    if preamble_bytes.len() < PREAMBLE_SIZE {
        return BackwardCommit::Format("short-fetch-bwd");
    }
    if &preamble_bytes[..MAGIC.len()] != MAGIC {
        return BackwardCommit::Format("bad-magic-bwd");
    }
    let preamble = match Preamble::read_from(preamble_bytes) {
        Ok(p) => p,
        Err(_) => return BackwardCommit::Format("preamble-parse-error-bwd"),
    };
    if preamble.total_length == 0 {
        return BackwardCommit::Format("streaming-preamble-non-tail");
    }
    if preamble.total_length != length {
        return BackwardCommit::Format("preamble-postamble-length-mismatch");
    }
    BackwardCommit::Layout(CachedLayout {
        offset: msg_start,
        length,
        preamble,
        index: None,
        global_metadata: None,
    })
}

/// Next step the eager-discovery accessors should take.  Computed
/// while holding the lock, then dispatched without it.
#[cfg(feature = "async")]
#[derive(Debug)]
enum EagerAction {
    /// Forward-only mode: combined chunk fetch (preamble + frames in
    /// one round trip) for the next message.
    ScanForwardEager,
    /// Bidirectional mode: paired preamble fetch.  Per-message
    /// frame discovery happens later via [`RemoteBackend::ensure_layout_async`].
    ScanBidir,
    /// Layout exists but metadata/index frames not yet fetched.
    Discover,
}

/// Outcome of parsing a forward preamble fetch.  Pure — does not
/// touch state.  Mirrors the format-error taxonomy of
/// [`scan_fwd_step_locked`] / [`scan_fwd_step_async`] so the
/// bidirectional round can apply the commit decision before
/// reacquiring the lock.
#[derive(Debug)]
enum ForwardOutcome {
    Hit {
        offset: u64,
        length: u64,
        preamble: Preamble,
        msg_end: u64,
    },
    /// Forward parsed a clean preamble whose `msg_end` exceeds the
    /// bidirectional dispatcher's `bound` (= `prev_scan_offset`) while
    /// still fitting within `file_size`.  Reachable only when
    /// `suffix_rev` already holds a backward-committed layout with a
    /// corrupt offset — forward's reading of the message length is
    /// canonical, so the commit decision disables backward (clearing
    /// `suffix_rev`) and records the forward hop.  Forward-only mode
    /// never produces this variant because the dispatcher passes
    /// `bound = file_size`.
    ///
    /// Renamed from `ExceedsBound` for clarity (Wave 5.1.3): the
    /// variant carries a *valid* hit whose endpoint just lands
    /// beyond the bidirectional bound, not a generic out-of-range
    /// error.
    HitBeyondBound {
        offset: u64,
        length: u64,
        preamble: Preamble,
        msg_end: u64,
    },
    /// Streaming preamble (`total_length == 0`).  Caller decides
    /// whether to record a streaming-tail layout (forward-only) or
    /// disable backward (bidirectional, see commit decision table).
    Streaming(u64),
    Terminate(&'static str),
}

/// `true` iff a successful forward Hit and a backward-discovered
/// layout describe exactly the same message.  Triggers on the
/// 1-message file (forward and backward both identify the only
/// message) and on odd-count crossovers where the meet-in-the-middle
/// lands on a single shared middle message.  In both cases the
/// commit decision table commits the forward layout once and skips
/// the backward record without clearing `suffix_rev` from earlier
/// rounds.
fn same_message_as_forward(fwd: &ForwardOutcome, layout: &CachedLayout) -> bool {
    matches!(
        fwd,
        ForwardOutcome::Hit { offset, length, .. }
            if *offset == layout.offset && *length == layout.length
    )
}

fn parse_forward_preamble(
    preamble_bytes: &[u8],
    pos: u64,
    file_size: u64,
    bound: u64,
) -> ForwardOutcome {
    let min_message_size = (PREAMBLE_SIZE + POSTAMBLE_SIZE) as u64;

    if preamble_bytes.len() < PREAMBLE_SIZE {
        return ForwardOutcome::Terminate("short-fetch-fwd");
    }
    if &preamble_bytes[..MAGIC.len()] != MAGIC {
        return ForwardOutcome::Terminate("bad-magic-fwd");
    }
    let preamble = match Preamble::read_from(preamble_bytes) {
        Ok(p) => p,
        Err(_) => return ForwardOutcome::Terminate("preamble-parse-error-fwd"),
    };
    let msg_len = preamble.total_length;
    if msg_len == 0 {
        let remaining = file_size - pos;
        return ForwardOutcome::Streaming(remaining);
    }
    let end = match pos.checked_add(msg_len) {
        Some(e) => e,
        None => return ForwardOutcome::Terminate("length-out-of-range-fwd"),
    };
    if msg_len < min_message_size || end > file_size {
        return ForwardOutcome::Terminate("length-out-of-range-fwd");
    }
    if end > bound {
        return ForwardOutcome::HitBeyondBound {
            offset: pos,
            length: msg_len,
            preamble,
            msg_end: end,
        };
    }
    ForwardOutcome::Hit {
        offset: pos,
        length: msg_len,
        preamble,
        msg_end: end,
    }
}

// ── Remote backend ───────────────────────────────────────────────────────────

pub(crate) struct RemoteBackend {
    source_url: String,
    store: Arc<dyn ObjectStore>,
    path: ObjectPath,
    file_size: u64,
    state: Mutex<RemoteState>,
    scan_opts: RemoteScanOptions,
}

/// Two-cursor scan state for the bidirectional walker.
///
/// The forward cursor (`next_scan_offset`) and the backward cursor
/// (`prev_scan_offset`) bound the unknown gap **while bidirectional
/// is active**.  After `disable_backward`, `prev_scan_offset` is
/// intentionally left stale — `forward_bound()` ignores it once
/// `suffix_rev` is empty and reverts to `file_size`, so subsequent
/// forward-only steps cover the full file.
///
/// Invariants pinned by the helpers below and the unit tests:
///
/// 1. While `bwd_active || !suffix_rev.is_empty()`:
///    `next_scan_offset <= prev_scan_offset <= file_size`.
///    Once backward is disabled and the suffix is cleared,
///    `prev_scan_offset` is meaningless and never read.
/// 2. `layouts` is a contiguous forward prefix; `next_scan_offset`
///    is the end of that prefix.
/// 3. While `bwd_active`, `suffix_rev` is an EOF-first stack of
///    backward-discovered layouts (last push is the **leftmost**
///    layout, closest to the forward walker), and
///    `prev_scan_offset` is the offset of that leftmost / most
///    recently pushed entry.
/// 4. `bwd_active => !gap_closed && !fwd_terminated`.
/// 5. `gap_closed => suffix_rev.is_empty() && !bwd_active && scan_complete()`.
/// 6. `fwd_terminated => suffix_rev.is_empty() && !bwd_active`
///    (bidirectional is never recovery — `terminate_forward` cascades
///    to `disable_backward`).
#[derive(Debug, Default)]
struct RemoteState {
    /// Forward-discovered layouts.  `layouts[i]` always has stable
    /// absolute index `i` (forward indices grow contiguously from 0).
    layouts: Vec<CachedLayout>,
    /// Start of the unknown gap.  Forward walker advances this.
    next_scan_offset: u64,

    /// EOF-first backward-discovered layouts.  Always empty when
    /// `bwd_active = false`.  Reversed and merged into `layouts` on
    /// `gap_closed`.
    suffix_rev: Vec<CachedLayout>,
    /// End of the unknown gap.  Initialised to `file_size` in
    /// [`RemoteBackend::open`] / [`RemoteBackend::open_async`].
    prev_scan_offset: u64,
    /// `true` iff bidirectional mode is requested AND the backward
    /// walker has not yielded.  Once `false`, stays `false`.
    bwd_active: bool,

    /// Forward walker reached EOF or hit a format error and cannot
    /// advance further.
    fwd_terminated: bool,
    /// Forward and backward walkers met (`next == prev`); `suffix_rev`
    /// has been merged into `layouts`.
    gap_closed: bool,

    /// Monotone counter incremented on every state-machine transition.
    /// The lock-around-await protocol snapshots this together with
    /// `(next, prev)` and validates equality on reacquire so a stale
    /// paired fetch cannot commit after a fallback or termination has
    /// already mutated state.
    scan_epoch: u64,
}

/// Snapshot of the scan cursors taken before a network round-trip.
/// Validated for equality on lock reacquire — any mismatch means
/// another caller mutated state in the meantime, and the in-flight
/// work must be discarded.
#[derive(Debug, Clone, Copy)]
struct ScanSnapshot {
    next: u64,
    prev: u64,
    epoch: u64,
}

/// Emit a one-shot tracing event identifying which scan mode the
/// freshly-opened backend is using.  Filterable via
/// `RUST_LOG=tensogram::remote_scan=debug`.
fn emit_scan_mode(scan_opts: &RemoteScanOptions) {
    let mode = if scan_opts.bidirectional {
        "bidirectional"
    } else {
        "forward-only"
    };
    tracing::debug!(
        target: "tensogram::remote_scan",
        mode = mode,
        "remote scan mode",
    );
}

impl RemoteState {
    /// Replaces the old `scan_complete: bool` field.  When forward-only
    /// is active (`bwd_active = false` and `suffix_rev.is_empty()`)
    /// this collapses to `fwd_terminated`, byte-identical to the
    /// previous semantics.
    fn scan_complete(&self) -> bool {
        self.gap_closed || (self.fwd_terminated && self.suffix_rev.is_empty())
    }

    /// Append a forward-discovered layout, advance the cursor, bump
    /// the epoch.
    fn record_forward_hop(&mut self, layout: CachedLayout) {
        debug_assert!(
            layout.offset.checked_add(layout.length).is_some(),
            "forward end must fit in u64; validated by scan_next before recording",
        );
        let end = layout.offset + layout.length;
        // `prev_scan_offset` is only a meaningful upper bound while
        // backward is live (invariant 1).  Post-`disable_backward`,
        // `prev_scan_offset` is stale and the forward walker may
        // legitimately advance past it on its way to `file_size`.
        let backward_live = self.bwd_active || !self.suffix_rev.is_empty();
        debug_assert!(!backward_live || end <= self.prev_scan_offset);
        let offset = layout.offset;
        let length = layout.length;
        self.layouts.push(layout);
        self.next_scan_offset = end;
        self.scan_epoch = self.scan_epoch.wrapping_add(1);
        tracing::debug!(
            target: "tensogram::remote_scan",
            direction = "fwd",
            offset = offset,
            length = length,
            "scan hop",
        );
    }

    /// Append a backward-discovered layout, retreat the cursor, bump
    /// the epoch.  Pre: `bwd_active`, layout extends from
    /// `prev_scan_offset` exactly back to `layout.offset`.
    fn record_backward_hop(&mut self, layout: CachedLayout) {
        debug_assert!(self.bwd_active);
        debug_assert!(layout.offset >= self.next_scan_offset);
        debug_assert_eq!(layout.offset + layout.length, self.prev_scan_offset);
        let offset = layout.offset;
        let length = layout.length;
        self.prev_scan_offset = layout.offset;
        self.suffix_rev.push(layout);
        self.scan_epoch = self.scan_epoch.wrapping_add(1);
        tracing::debug!(
            target: "tensogram::remote_scan",
            direction = "bwd",
            offset = offset,
            length = length,
            "scan hop",
        );
    }

    /// Snapshot for the lock-around-await protocol.
    fn snapshot(&self) -> ScanSnapshot {
        ScanSnapshot {
            next: self.next_scan_offset,
            prev: self.prev_scan_offset,
            epoch: self.scan_epoch,
        }
    }

    /// `true` iff the state still matches the snapshot taken before
    /// a network round-trip.  Used at commit time to detect any
    /// concurrent mutation.
    fn matches(&self, snap: &ScanSnapshot) -> bool {
        self.next_scan_offset == snap.next
            && self.prev_scan_offset == snap.prev
            && self.scan_epoch == snap.epoch
    }

    /// Disable the backward walker and discard provisional layouts.
    /// Bumps the epoch so any in-flight paired fetch fails its
    /// snapshot validation.  Idempotent — no-op when the walker is
    /// already inactive and `suffix_rev` is empty.
    fn disable_backward(&mut self, reason: &'static str) {
        if !self.bwd_active && self.suffix_rev.is_empty() {
            return;
        }
        self.bwd_active = false;
        self.suffix_rev.clear();
        self.scan_epoch = self.scan_epoch.wrapping_add(1);
        tracing::debug!(
            target: "tensogram::remote_scan",
            reason = reason,
            "backward walker disabled",
        );
    }

    /// Terminate the forward walker.  Per the state-machine
    /// invariant `fwd_terminated => suffix_rev.is_empty() &&
    /// !bwd_active`, this also disables the backward walker and
    /// discards any provisional suffix layouts — bidirectional is
    /// an optimisation, never a recovery mode.
    fn terminate_forward(&mut self, reason: &'static str) {
        if !self.fwd_terminated {
            self.fwd_terminated = true;
            self.scan_epoch = self.scan_epoch.wrapping_add(1);
            tracing::debug!(
                target: "tensogram::remote_scan",
                reason = reason,
                "forward walker terminated",
            );
        }
        self.disable_backward(reason);
    }

    /// Merge `suffix_rev` into `layouts` (reversed) and mark the scan
    /// complete.  Pre: walkers met cleanly (`next_scan_offset ==
    /// prev_scan_offset`); `bwd_active` was true before the call.
    fn close_gap(&mut self) {
        debug_assert!(self.bwd_active);
        debug_assert_eq!(self.next_scan_offset, self.prev_scan_offset);
        let mut tail = std::mem::take(&mut self.suffix_rev);
        tail.reverse();
        self.layouts.extend(tail);
        self.gap_closed = true;
        self.bwd_active = false;
        self.scan_epoch = self.scan_epoch.wrapping_add(1);
        tracing::debug!(
            target: "tensogram::remote_scan",
            messages = self.layouts.len(),
            "gap closed",
        );
    }
}

impl std::fmt::Debug for RemoteBackend {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("RemoteBackend")
            .field("source", &self.source_url)
            .field("file_size", &self.file_size)
            .field("scan_opts", &self.scan_opts)
            .field(
                "messages",
                &self
                    .state
                    .lock()
                    .map(|state| state.layouts.len())
                    .unwrap_or(0),
            )
            .finish()
    }
}

impl RemoteBackend {
    pub(crate) fn source_url(&self) -> &str {
        &self.source_url
    }

    #[cfg(feature = "async")]
    fn lock_state(&self) -> Result<MutexGuard<'_, RemoteState>> {
        self.state
            .lock()
            .map_err(|_| TensogramError::Remote("remote state lock poisoned".to_string()))
    }

    pub(crate) fn open_with_scan_opts(
        source: &str,
        storage_options: &BTreeMap<String, String>,
        scan_opts: RemoteScanOptions,
    ) -> Result<Self> {
        emit_scan_mode(&scan_opts);
        let url = Url::parse(source)
            .map_err(|e| TensogramError::Remote(format!("invalid URL '{source}': {e}")))?;

        let mut opts: Vec<(String, String)> = storage_options
            .iter()
            .map(|(k, v)| (k.clone(), v.clone()))
            .collect();
        if url.scheme() == "http" && !opts.iter().any(|(k, _)| k == "allow_http") {
            opts.push(("allow_http".to_string(), "true".to_string()));
        }
        let (store, path) = object_store::parse_url_opts(&url, opts)
            .map_err(|e| TensogramError::Remote(format!("cannot open '{source}': {e}")))?;

        let store: Arc<dyn ObjectStore> = Arc::from(store);

        let head_store = store.clone();
        let head_path = path.clone();
        let meta = block_on_shared(async move { head_store.head(&head_path).await })?;

        let file_size = meta.size;
        if file_size < (PREAMBLE_SIZE + POSTAMBLE_SIZE) as u64 {
            return Err(TensogramError::Remote(format!(
                "remote file too small ({file_size} bytes)"
            )));
        }

        let backend = RemoteBackend {
            source_url: source.to_string(),
            store,
            path,
            file_size,
            state: Mutex::new(RemoteState {
                prev_scan_offset: file_size,
                bwd_active: scan_opts.bidirectional,
                ..RemoteState::default()
            }),
            scan_opts,
        };
        {
            let mut state = backend
                .state
                .lock()
                .map_err(|_| TensogramError::Remote("remote state lock poisoned".to_string()))?;
            backend.scan_next_locked(&mut state)?;
            if state.layouts.is_empty() {
                return Err(TensogramError::Remote(
                    "no valid messages found in remote file".to_string(),
                ));
            }
        }
        Ok(backend)
    }

    // ── Range reads ──────────────────────────────────────────────────────

    fn get_range(&self, range: Range<u64>) -> Result<Bytes> {
        let store = self.store.clone();
        let path = self.path.clone();
        block_on_shared(async move { store.get_range(&path, range).await })
    }

    // ── Message scanning ─────────────────────────────────────────────────

    fn scan_next_locked(&self, state: &mut RemoteState) -> Result<()> {
        self.scan_fwd_step_locked(state, self.file_size)
    }

    /// One forward hop bounded by `bound` (exclusive upper offset).
    ///
    /// In forward-only mode `bound == self.file_size`.  In
    /// bidirectional mode the dispatcher passes
    /// `state.prev_scan_offset` so the forward walker cannot overrun
    /// into the region already claimed by the backward walker.
    /// Streaming preambles (`total_length == 0`) are tail-only by
    /// spec; the streaming END_MAGIC is therefore read at
    /// `self.file_size - 8` regardless of `bound`.
    fn scan_fwd_step_locked(&self, state: &mut RemoteState, bound: u64) -> Result<()> {
        if state.scan_complete() {
            return Ok(());
        }
        let min_message_size = (PREAMBLE_SIZE + POSTAMBLE_SIZE) as u64;
        let pos = state.next_scan_offset;

        // Saturating addition keeps the EOF check correct even when
        // `pos` is so close to `u64::MAX` that an unchecked sum would
        // wrap.  Anything that saturates to `u64::MAX` is by definition
        // beyond `bound` (which is at most `file_size`), so the EOF
        // branch fires.  This means the downstream `pos + PREAMBLE_SIZE`
        // range below cannot wrap either: we've already established
        // `pos + min_message_size <= bound <= file_size`.
        if pos.saturating_add(min_message_size) > bound {
            state.terminate_forward("eof");
            return Ok(());
        }

        let preamble_bytes = self.get_range(pos..pos + PREAMBLE_SIZE as u64)?;
        if &preamble_bytes[..MAGIC.len()] != MAGIC {
            state.terminate_forward("bad-magic-fwd");
            return Ok(());
        }

        let preamble = match Preamble::read_from(&preamble_bytes) {
            Ok(p) => p,
            Err(_) => {
                state.terminate_forward("preamble-parse-error-fwd");
                return Ok(());
            }
        };

        let msg_len = preamble.total_length;

        if msg_len == 0 {
            let remaining = self.file_size - pos;
            if remaining < min_message_size {
                state.terminate_forward("streaming-tail-too-small");
                return Ok(());
            }
            let end_magic_pos = self.file_size - crate::wire::END_MAGIC.len() as u64;
            let end_bytes =
                self.get_range(end_magic_pos..end_magic_pos + crate::wire::END_MAGIC.len() as u64)?;
            if &end_bytes[..] != crate::wire::END_MAGIC {
                state.terminate_forward("streaming-end-magic-mismatch");
                return Ok(());
            }
            state.record_forward_hop(CachedLayout {
                offset: pos,
                length: remaining,
                preamble,
                index: None,
                global_metadata: None,
            });
            state.terminate_forward("streaming-tail");
            return Ok(());
        }

        match pos.checked_add(msg_len) {
            Some(end) if msg_len >= min_message_size && end <= bound => {}
            _ => {
                state.terminate_forward("length-out-of-range-fwd");
                return Ok(());
            }
        }

        state.record_forward_hop(CachedLayout {
            offset: pos,
            length: msg_len,
            preamble,
            index: None,
            global_metadata: None,
        });
        Ok(())
    }

    /// One step of progress in whichever scan mode is active.  When
    /// `bwd_active`, dispatches to the bidirectional paired round;
    /// otherwise falls back to bounded forward-only.  This is the
    /// single funnel through which every read accessor scans, so
    /// adding new state-machine steps doesn't require touching the
    /// accessors a second time.
    fn scan_step_locked(&self, state: &mut RemoteState) -> Result<()> {
        if state.scan_complete() {
            return Ok(());
        }
        if state.bwd_active && !state.fwd_terminated {
            self.scan_bidir_round_locked(state)
        } else {
            let bound = self.forward_bound(state);
            self.scan_fwd_step_locked(state, bound)
        }
    }

    /// Upper bound for the next forward hop.  In bidirectional mode,
    /// caps at `prev_scan_offset` so the forward walker cannot
    /// overrun into the suffix already claimed by backward.  Once
    /// backward has yielded and `suffix_rev` is empty, the bound
    /// reverts to `file_size`.
    fn forward_bound(&self, state: &RemoteState) -> u64 {
        if state.suffix_rev.is_empty() {
            self.file_size
        } else {
            state.prev_scan_offset
        }
    }

    /// One bidirectional scan round: paired forward preamble + backward
    /// postamble fetch, followed by a backward-preamble validation
    /// fetch when the postamble parses cleanly, optionally followed
    /// by an eager footer-region fetch when the postamble's
    /// `first_footer_offset` indicates a non-empty footer.  Commits
    /// the parsed outcomes atomically through the [`RemoteState`]
    /// decision table.
    ///
    /// Fetch failure semantics:
    ///
    /// - **Paired forward preamble + backward postamble**: required.
    ///   Any transport error (timeout, 503, abort) leaves state
    ///   unchanged and propagates so the caller can retry.
    /// - **Backward candidate preamble**: required when the
    ///   postamble parsed cleanly.  Same error semantics as the
    ///   paired fetch.
    /// - **Eager footer chunk**: best-effort.  Transport errors
    ///   are silently dropped via `.ok()`; the layout still commits
    ///   via the validated preamble alone, and the lazy
    ///   `ensure_layout` path picks up footer discovery on the next
    ///   metadata access.  See [`Self::try_populate_eager_footer`].
    ///
    /// *Format* errors are absorbed locally regardless of which
    /// fetch surfaces them — backward yields, forward keeps going.
    fn scan_bidir_round_locked(&self, state: &mut RemoteState) -> Result<()> {
        let snap = state.snapshot();
        // In bidirectional mode `prev_scan_offset` IS the forward
        // bound: it equals `file_size` while `suffix_rev` is empty
        // and the start of the leftmost suffix layout otherwise.  In
        // forward-only mode this function is never reached (the
        // dispatcher routes to `scan_fwd_step_locked` directly).
        let bound = snap.prev;
        let min_message_size = (PREAMBLE_SIZE + POSTAMBLE_SIZE) as u64;

        // The paired postamble fetch needs a non-overlapping pair of
        // 24-byte ranges: the forward preamble at `[next, next+24)`
        // and the backward postamble at `[prev-24, prev)`.  Those two
        // ranges are disjoint iff `prev >= next + min_message_size`
        // (= `next + 48`), which is exactly the smallest gap in which
        // a single well-formed message can fit.  When the gap holds
        // exactly one message, same-message detection in
        // `commit_or_yield_backward` handles the meet correctly and
        // `close_gap` fires from `apply_round_outcomes`.
        //
        // Saturating addition handles the (astronomical) case where
        // `snap.next` is so close to `u64::MAX` that the unchecked
        // sum would wrap; saturation to `u64::MAX` correctly forces
        // the gap-too-small branch.  Past this guard we know
        // `snap.prev >= snap.next + min_message_size`, so the
        // unchecked range constructions below cannot wrap.
        if snap.prev < snap.next.saturating_add(min_message_size) {
            // Gap can't fit a paired fetch.  Two sub-cases:
            //   • `next == prev`: walkers met; `close_gap` should have
            //     fired already, but assert-and-fix-up defensively.
            //   • otherwise: the gap is shorter than `min_message_size`,
            //     which in a valid file is unreachable (every message
            //     is ≥ 48 bytes).  Treat as suspected backward
            //     corruption — disable backward (clearing `suffix_rev`)
            //     and retry forward-only with the full file as the
            //     bound, matching what forward-only would have done.
            if snap.next == snap.prev {
                state.close_gap();
                return Ok(());
            }
            state.disable_backward("gap-below-min-message-size");
            let recovery_bound = self.forward_bound(state);
            return self.scan_fwd_step_locked(state, recovery_bound);
        }

        let fwd_r = snap.next..snap.next + PREAMBLE_SIZE as u64;
        let bwd_r = snap.prev - POSTAMBLE_SIZE as u64..snap.prev;

        // Two independent `get_range` calls run in parallel via
        // `tokio::join!` instead of `get_ranges(&[fwd, bwd])`.  The
        // latter runs the paired ranges through `object_store`'s
        // coalescer (default 1 MiB merge gap), which collapses
        // discovery-time forward/backward probes into one HTTP Range
        // covering the entire span between the two cursors —
        // catastrophic for tightly-packed files where the cursors
        // stay well within the merge window for the whole walk.  Two
        // independent requests give us the round-trip parallelism
        // without the bandwidth blow-up.
        let (fwd_bytes, bwd_bytes) = block_on_shared({
            let store = self.store.clone();
            let path = self.path.clone();
            let fwd_r2 = fwd_r.clone();
            let bwd_r2 = bwd_r.clone();
            async move {
                let (f, b) = tokio::join!(
                    store.get_range(&path, fwd_r2),
                    store.get_range(&path, bwd_r2),
                );
                Ok::<_, object_store::Error>((f?, b?))
            }
        })?;

        let bwd_outcome = parse_backward_postamble(&bwd_bytes, &snap);

        let (candidate_preamble_bytes, candidate_footer_bytes) = match &bwd_outcome {
            BackwardOutcome::NeedPreambleValidation {
                msg_start,
                length,
                first_footer_offset,
            } => {
                let preamble = self.get_range(*msg_start..*msg_start + PREAMBLE_SIZE as u64)?;
                let footer = if footer_region_present(*first_footer_offset, *length) {
                    let footer_start = msg_start.saturating_add(*first_footer_offset);
                    let footer_end = msg_start
                        .saturating_add(*length)
                        .saturating_sub(POSTAMBLE_SIZE as u64);
                    if footer_start < footer_end {
                        // Best-effort: a transport failure here means
                        // the lazy `ensure_layout` path picks up footer
                        // discovery later.  We still emit a debug
                        // tracing event so a flapping endpoint is
                        // visible to operators rather than appearing
                        // as a silent perf regression.
                        match self.get_range(footer_start..footer_end) {
                            Ok(b) => Some(b),
                            Err(e) => {
                                tracing::debug!(
                                    target: "tensogram::remote_scan",
                                    error = %e,
                                    msg_start = *msg_start,
                                    footer_start = footer_start,
                                    footer_end = footer_end,
                                    "eager footer fetch failed; falling back to lazy"
                                );
                                None
                            }
                        }
                    } else {
                        None
                    }
                } else {
                    None
                };
                (Some(preamble), footer)
            }
            _ => (None, None),
        };

        // No `state.matches(&snap)` recheck here: the sync path holds
        // `&mut state` (and therefore the mutex) across `block_on_shared`,
        // so the snapshot is guaranteed to still match.  The async sibling
        // does drop and re-acquire the lock around the await, which is
        // why it needs the explicit `state.matches` check.
        let fwd_kind = parse_forward_preamble(&fwd_bytes, snap.next, self.file_size, bound);
        self.apply_round_outcomes(
            state,
            fwd_kind,
            bwd_outcome,
            candidate_preamble_bytes.as_deref(),
            candidate_footer_bytes.as_deref(),
        );
        Ok(())
    }

    /// Apply the forward, backward, and (optional) candidate-preamble
    /// outcomes of a single bidirectional round to the locked state.
    /// Backward is processed first so a format/streaming yield can
    /// fire before the forward commit; this keeps the "forward
    /// terminate clears suffix" path simple.
    fn apply_round_outcomes(
        &self,
        state: &mut RemoteState,
        fwd: ForwardOutcome,
        bwd: BackwardOutcome,
        candidate_preamble_bytes: Option<&[u8]>,
        candidate_footer_bytes: Option<&[u8]>,
    ) {
        match bwd {
            BackwardOutcome::Format(reason) => state.disable_backward(reason),
            BackwardOutcome::Streaming => state.disable_backward("streaming-zero-bwd"),
            BackwardOutcome::NeedPreambleValidation {
                msg_start,
                length,
                first_footer_offset: _,
            } => {
                let validation = candidate_preamble_bytes
                    .map(|bytes| validate_backward_preamble(bytes, msg_start, length))
                    .unwrap_or(BackwardCommit::Format("missing-candidate-preamble"));
                Self::commit_or_yield_backward(state, &fwd, validation, candidate_footer_bytes);
            }
        }

        match fwd {
            ForwardOutcome::Hit {
                offset,
                length,
                preamble,
                msg_end,
            } => {
                debug_assert_eq!(offset, state.next_scan_offset);
                debug_assert_eq!(msg_end, offset + length);
                state.record_forward_hop(CachedLayout {
                    offset,
                    length,
                    preamble,
                    index: None,
                    global_metadata: None,
                });
            }
            ForwardOutcome::HitBeyondBound {
                offset,
                length,
                preamble,
                msg_end,
            } => {
                // Forward parsed a message whose end exceeds the
                // bidirectional dispatcher's `bound` (= `prev_scan_offset`)
                // while still fitting inside `file_size`.  That can only
                // happen if `suffix_rev` already holds a backward layout
                // with a corrupt offset — forward's reading is canonical,
                // so disable backward (clearing `suffix_rev`) before
                // committing the forward hop.  Subsequent steps revert
                // to forward-only with `bound = file_size`.
                state.disable_backward("forward-exceeds-backward-bound");
                debug_assert_eq!(offset, state.next_scan_offset);
                debug_assert_eq!(msg_end, offset + length);
                state.record_forward_hop(CachedLayout {
                    offset,
                    length,
                    preamble,
                    index: None,
                    global_metadata: None,
                });
            }
            ForwardOutcome::Streaming(_remaining) => {
                // Streaming preamble in bidirectional mode is a
                // contradiction: backward has discovered tail
                // content, so the streaming "tail" can't actually
                // extend to file end.  Yield backward and let
                // forward-only logic re-handle this position on the
                // next step (which will set fwd_terminated on the
                // streaming branch).
                state.disable_backward("streaming-fwd-non-tail");
            }
            ForwardOutcome::Terminate(reason) => state.terminate_forward(reason),
        }

        if state.bwd_active && state.next_scan_offset == state.prev_scan_offset {
            state.close_gap();
        }
    }

    /// Decide what to do with a backward candidate layout once the
    /// candidate-preamble validation has run, before the forward
    /// commit fires.  Five mutually-exclusive outcomes:
    ///
    /// - validation reported `Format(reason)`: disable backward with
    ///   that reason.
    /// - `bwd_active == false`: a concurrent caller disabled the
    ///   walker between the postamble fetch and the candidate-preamble
    ///   fetch; silently drop the layout.
    /// - `same_message_as_forward`: the 1-message file and odd-count
    ///   middle-meet case — forward will commit this exact layout
    ///   itself.  Yield silently (do NOT call `disable_backward` —
    ///   that would clear `suffix_rev` from earlier hops).
    /// - Forward Hit overlapping the candidate (`fwd.msg_end >
    ///   layout.offset`) or forward HitBeyondBound: backward's claim
    ///   conflicts with a canonical forward reading; disable backward
    ///   with the matching reason and let the forward branch commit.
    /// - Otherwise: record the backward hop.  When
    ///   `candidate_footer_bytes` is `Some` and the message is
    ///   footer-indexed, the eager-footer apply runs BEFORE
    ///   `record_backward_hop` so `metadata` and `index` land in
    ///   `suffix_rev` already populated; lazy `ensure_layout` then
    ///   short-circuits.
    fn commit_or_yield_backward(
        state: &mut RemoteState,
        fwd: &ForwardOutcome,
        validation: BackwardCommit,
        candidate_footer_bytes: Option<&[u8]>,
    ) {
        let mut layout = match validation {
            BackwardCommit::Format(reason) => {
                state.disable_backward(reason);
                return;
            }
            BackwardCommit::Layout(layout) => layout,
        };
        if !state.bwd_active {
            return;
        }
        if same_message_as_forward(fwd, &layout) {
            return;
        }
        match fwd {
            ForwardOutcome::Hit { msg_end, .. } if *msg_end > layout.offset => {
                state.disable_backward("backward-overlaps-forward");
            }
            ForwardOutcome::HitBeyondBound { .. } => {
                state.disable_backward("forward-exceeds-backward-bound");
            }
            _ => {
                if let Some(footer_bytes) = candidate_footer_bytes {
                    Self::try_populate_eager_footer(&mut layout, footer_bytes);
                }
                state.record_backward_hop(layout);
            }
        }
    }

    /// Populate `layout.global_metadata` and `layout.index` from a
    /// pre-fetched footer chunk when (and only when) the just-validated
    /// preamble's flags carry `FOOTER_METADATA | FOOTER_INDEX`.
    /// Best-effort: any parse failure is silently swallowed and the
    /// lazy `ensure_layout` path picks up the layout later.
    ///
    /// Header-indexed messages with footer hash frames have a non-empty
    /// footer region whose bytes the dispatcher fetched speculatively;
    /// for those, the FOOTER_INDEX flag check below short-circuits and
    /// the bytes are discarded harmlessly.
    fn try_populate_eager_footer(layout: &mut CachedLayout, footer_bytes: &[u8]) {
        let flags = layout.preamble.flags;
        if !(flags.has(MessageFlags::FOOTER_METADATA) && flags.has(MessageFlags::FOOTER_INDEX)) {
            return;
        }
        if let Ok((metadata, index)) = Self::parse_footer_frames_into(footer_bytes) {
            if let (Some(m), Some(i)) = (metadata, index) {
                layout.global_metadata = Some(m);
                layout.index = Some(i);
                tracing::debug!(
                    target: "tensogram::remote_scan",
                    action = "footer_eager",
                    offset = layout.offset,
                    footer_bytes = footer_bytes.len(),
                );
            }
        }
    }

    fn ensure_message_locked(&self, state: &mut RemoteState, msg_idx: usize) -> Result<()> {
        while msg_idx >= state.layouts.len() && !state.scan_complete() {
            self.scan_step_locked(state)?;
        }
        if msg_idx >= state.layouts.len() {
            return Err(TensogramError::Framing(format!(
                "message index {} out of range (count={})",
                msg_idx,
                state.layouts.len()
            )));
        }
        Ok(())
    }

    fn scan_all_locked(&self, state: &mut RemoteState) -> Result<()> {
        while !state.scan_complete() {
            self.scan_step_locked(state)?;
        }
        Ok(())
    }

    fn scan_and_discover_next_locked(&self, state: &mut RemoteState) -> Result<()> {
        if state.scan_complete() {
            return Ok(());
        }
        let min_message_size = (PREAMBLE_SIZE + POSTAMBLE_SIZE) as u64;
        let pos = state.next_scan_offset;

        // Saturating addition: see `scan_fwd_step_locked` for the
        // proof that this guard is sufficient to keep the unchecked
        // `pos + chunk_size` range below from wrapping.
        if pos.saturating_add(min_message_size) > self.file_size {
            state.terminate_forward("eof");
            return Ok(());
        }

        let chunk_size = (self.file_size - pos).min(256 * 1024);
        let chunk = self.get_range(pos..pos + chunk_size)?;

        if chunk.len() < PREAMBLE_SIZE || &chunk[..MAGIC.len()] != MAGIC {
            state.terminate_forward("bad-magic-fwd");
            return Ok(());
        }

        let preamble = match Preamble::read_from(&chunk[..PREAMBLE_SIZE]) {
            Ok(p) => p,
            Err(_) => {
                state.terminate_forward("preamble-parse-error-fwd");
                return Ok(());
            }
        };

        let msg_len = preamble.total_length;

        if msg_len == 0 {
            let remaining = self.file_size - pos;
            if remaining < min_message_size {
                state.terminate_forward("streaming-tail-too-small");
                return Ok(());
            }
            let end_magic_pos = self.file_size - crate::wire::END_MAGIC.len() as u64;
            let end_bytes =
                self.get_range(end_magic_pos..end_magic_pos + crate::wire::END_MAGIC.len() as u64)?;
            if &end_bytes[..] != crate::wire::END_MAGIC {
                state.terminate_forward("streaming-end-magic-mismatch");
                return Ok(());
            }
            state.record_forward_hop(CachedLayout {
                offset: pos,
                length: remaining,
                preamble,
                index: None,
                global_metadata: None,
            });
            state.terminate_forward("streaming-tail");
            return Ok(());
        }

        match pos.checked_add(msg_len) {
            Some(end) if msg_len >= min_message_size && end <= self.file_size => {}
            _ => {
                state.terminate_forward("length-out-of-range-fwd");
                return Ok(());
            }
        }

        let flags = preamble.flags;
        let msg_idx = state.layouts.len();

        state.record_forward_hop(CachedLayout {
            offset: pos,
            length: msg_len,
            preamble,
            index: None,
            global_metadata: None,
        });

        if flags.has(MessageFlags::HEADER_METADATA) && flags.has(MessageFlags::HEADER_INDEX) {
            let chunk_end = (msg_len as usize).min(chunk.len());
            Self::parse_header_frames(state, msg_idx, &chunk[..chunk_end])?;
        } else if flags.has(MessageFlags::FOOTER_METADATA) && flags.has(MessageFlags::FOOTER_INDEX)
        {
            self.discover_footer_layout_from_suffix_locked(state, msg_idx)?;
        }

        Ok(())
    }

    /// Discover footer layout by reading a single suffix chunk from the
    /// message end.  The suffix covers both the postamble and the footer
    /// region (metadata + index frames) in one GET, halving the round
    /// trips compared to separate postamble + footer reads.
    ///
    /// The suffix size is capped at 256 KB — footer regions are typically
    /// a few KB.  If `first_footer_offset` points outside the suffix,
    /// falls back to a separate read for the footer region.
    fn discover_footer_layout_from_suffix_locked(
        &self,
        state: &mut RemoteState,
        msg_idx: usize,
    ) -> Result<()> {
        let msg_offset = state.layouts[msg_idx].offset;
        let msg_len = state.layouts[msg_idx].length;

        let suffix_size = msg_len.min(256 * 1024);
        let msg_end = msg_offset
            .checked_add(msg_len)
            .ok_or_else(|| TensogramError::Remote("message end overflow".to_string()))?;
        let suffix_start = msg_end - suffix_size;
        let suffix = self.get_range(suffix_start..msg_end)?;

        if suffix.len() < POSTAMBLE_SIZE {
            return Err(TensogramError::Remote(
                "suffix too short for postamble".to_string(),
            ));
        }

        let pa_bytes = &suffix[suffix.len() - POSTAMBLE_SIZE..];
        let postamble = Postamble::read_from(pa_bytes)?;

        if postamble.first_footer_offset < PREAMBLE_SIZE as u64 {
            return Err(TensogramError::Remote(format!(
                "first_footer_offset ({}) is before preamble end ({PREAMBLE_SIZE})",
                postamble.first_footer_offset
            )));
        }

        let footer_abs_start = msg_offset
            .checked_add(postamble.first_footer_offset)
            .ok_or_else(|| TensogramError::Remote("footer offset overflow".to_string()))?;
        let footer_abs_end = msg_end - POSTAMBLE_SIZE as u64;

        if footer_abs_start >= footer_abs_end {
            return Err(TensogramError::Remote(
                "first_footer_offset points at or past postamble".to_string(),
            ));
        }

        if footer_abs_start >= suffix_start {
            let local_start = (footer_abs_start - suffix_start) as usize;
            let local_end = suffix.len() - POSTAMBLE_SIZE;
            Self::parse_footer_frames(state, msg_idx, &suffix[local_start..local_end])
        } else {
            let footer_bytes = self.get_range(footer_abs_start..footer_abs_end)?;
            Self::parse_footer_frames(state, msg_idx, &footer_bytes)
        }
    }

    fn ensure_layout_eager_locked(&self, state: &mut RemoteState, msg_idx: usize) -> Result<()> {
        while msg_idx >= state.layouts.len() && !state.scan_complete() {
            if state.bwd_active && !state.fwd_terminated {
                self.scan_bidir_round_locked(state)?;
            } else {
                self.scan_and_discover_next_locked(state)?;
            }
        }
        if msg_idx >= state.layouts.len() {
            return Err(TensogramError::Framing(format!(
                "message index {} out of range (count={})",
                msg_idx,
                state.layouts.len()
            )));
        }
        if state.layouts[msg_idx].global_metadata.is_some()
            && state.layouts[msg_idx].index.is_some()
        {
            return Ok(());
        }
        self.ensure_layout_locked(state, msg_idx)
    }

    // ── Layout discovery (metadata + index for a single message) ─────────

    fn ensure_layout_locked(&self, state: &mut RemoteState, msg_idx: usize) -> Result<()> {
        self.ensure_message_locked(state, msg_idx)?;
        if state.layouts[msg_idx].global_metadata.is_some()
            && state.layouts[msg_idx].index.is_some()
        {
            return Ok(());
        }

        let flags = state.layouts[msg_idx].preamble.flags;

        if flags.has(MessageFlags::HEADER_METADATA) && flags.has(MessageFlags::HEADER_INDEX) {
            self.discover_header_layout_locked(state, msg_idx)?;
        } else if flags.has(MessageFlags::FOOTER_METADATA) && flags.has(MessageFlags::FOOTER_INDEX)
        {
            self.discover_footer_layout_locked(state, msg_idx)?;
        } else {
            return Err(TensogramError::Remote(
                "remote access requires header-indexed or footer-indexed messages".to_string(),
            ));
        }

        Ok(())
    }

    fn discover_footer_layout_locked(&self, state: &mut RemoteState, msg_idx: usize) -> Result<()> {
        let msg_offset = state.layouts[msg_idx].offset;
        let msg_len = state.layouts[msg_idx].length;

        let pa_offset = msg_offset
            .checked_add(msg_len)
            .and_then(|end| end.checked_sub(POSTAMBLE_SIZE as u64))
            .ok_or_else(|| TensogramError::Remote("postamble offset overflow".to_string()))?;
        let pa_bytes = self.get_range(pa_offset..pa_offset + POSTAMBLE_SIZE as u64)?;
        let postamble = Postamble::read_from(&pa_bytes)?;

        if postamble.first_footer_offset < PREAMBLE_SIZE as u64 {
            return Err(TensogramError::Remote(format!(
                "first_footer_offset ({}) is before preamble end ({})",
                postamble.first_footer_offset, PREAMBLE_SIZE
            )));
        }
        let footer_start = msg_offset
            .checked_add(postamble.first_footer_offset)
            .ok_or_else(|| TensogramError::Remote("footer offset overflow".to_string()))?;
        let footer_end = pa_offset;
        if footer_start >= footer_end {
            return Err(TensogramError::Remote(
                "first_footer_offset points at or past postamble".to_string(),
            ));
        }
        let footer_bytes = self.get_range(footer_start..footer_end)?;

        Self::parse_footer_frames(state, msg_idx, &footer_bytes)
    }

    fn discover_header_layout_locked(&self, state: &mut RemoteState, msg_idx: usize) -> Result<()> {
        let layout = &state.layouts[msg_idx];
        let msg_offset = layout.offset;
        let msg_len = layout.length;

        // Read a generous initial chunk: up to 256KB or the message size.
        // Header metadata + index are typically a few KB.
        let chunk_size = msg_len.min(256 * 1024);
        let header_bytes = self.get_range(msg_offset..msg_offset + chunk_size)?;

        Self::parse_header_frames(state, msg_idx, &header_bytes)
    }

    fn parse_header_frames(state: &mut RemoteState, msg_idx: usize, buf: &[u8]) -> Result<()> {
        let min_frame_size = FRAME_HEADER_SIZE + FRAME_END.len();
        let mut pos = PREAMBLE_SIZE;

        while pos + FRAME_HEADER_SIZE <= buf.len() {
            if &buf[pos..pos + 2] != b"FR" {
                pos += 1;
                continue;
            }
            let fh = FrameHeader::read_from(&buf[pos..])?;
            let frame_total = usize::try_from(fh.total_length).map_err(|_| {
                TensogramError::Remote("frame total_length does not fit in usize".to_string())
            })?;

            if frame_total < min_frame_size {
                return Err(TensogramError::Remote(format!(
                    "frame total_length ({frame_total}) smaller than minimum ({min_frame_size})"
                )));
            }
            let frame_end = match pos.checked_add(frame_total) {
                Some(end) if end <= buf.len() => end,
                _ => break,
            };

            if &buf[frame_end - FRAME_END.len()..frame_end] != FRAME_END {
                return Err(TensogramError::Remote(
                    "frame missing ENDF trailer".to_string(),
                ));
            }

            let payload = &buf[pos + FRAME_HEADER_SIZE..frame_end - FRAME_COMMON_FOOTER_SIZE];

            match fh.frame_type {
                FrameType::HeaderMetadata => {
                    let meta = metadata::cbor_to_global_metadata(payload)?;
                    state.layouts[msg_idx].global_metadata = Some(meta);
                }
                FrameType::HeaderIndex => {
                    let idx = metadata::cbor_to_index(payload)?;
                    state.layouts[msg_idx].index = Some(idx);
                }
                FrameType::NTensorFrame | FrameType::PrecederMetadata => {
                    break;
                }
                _ => {}
            }

            let aligned = (frame_end.saturating_add(7)) & !7;
            pos = aligned.min(buf.len());
        }

        if state.layouts[msg_idx].global_metadata.is_none() {
            return Err(TensogramError::Remote(
                "header region did not contain a metadata frame".to_string(),
            ));
        }
        if state.layouts[msg_idx].index.is_none() {
            return Err(TensogramError::Remote(
                "header region did not contain an index frame (header chunk may be too small)"
                    .to_string(),
            ));
        }

        Ok(())
    }

    /// Walk a footer-region byte buffer and extract the
    /// `FooterMetadata` + `FooterIndex` frames (when present), without
    /// touching any backend state.  Used by both the lazy
    /// [`Self::parse_footer_frames`] populator and the eager-footer
    /// fast path on the bidirectional walker, which discards the
    /// returned options when the message turns out to be header-indexed.
    fn parse_footer_frames_into(
        buf: &[u8],
    ) -> Result<(Option<GlobalMetadata>, Option<IndexFrame>)> {
        let min_frame_size = FRAME_HEADER_SIZE + FRAME_END.len();
        let mut metadata: Option<GlobalMetadata> = None;
        let mut index: Option<IndexFrame> = None;
        let mut pos = 0;

        while pos + FRAME_HEADER_SIZE <= buf.len() {
            if &buf[pos..pos + 2] != b"FR" {
                pos += 1;
                continue;
            }
            let fh = FrameHeader::read_from(&buf[pos..])?;
            let frame_total = usize::try_from(fh.total_length).map_err(|_| {
                TensogramError::Remote(
                    "footer frame total_length does not fit in usize".to_string(),
                )
            })?;

            if frame_total < min_frame_size {
                return Err(TensogramError::Remote(format!(
                    "footer frame total_length ({frame_total}) smaller than minimum ({min_frame_size})"
                )));
            }
            let frame_end = match pos.checked_add(frame_total) {
                Some(end) if end <= buf.len() => end,
                _ => break,
            };

            if &buf[frame_end - FRAME_END.len()..frame_end] != FRAME_END {
                return Err(TensogramError::Remote(
                    "footer frame missing ENDF trailer".to_string(),
                ));
            }

            let payload = &buf[pos + FRAME_HEADER_SIZE..frame_end - FRAME_COMMON_FOOTER_SIZE];

            match fh.frame_type {
                FrameType::FooterMetadata => {
                    metadata = Some(metadata::cbor_to_global_metadata(payload)?);
                }
                FrameType::FooterIndex => {
                    index = Some(metadata::cbor_to_index(payload)?);
                }
                _ => {}
            }

            let aligned = (frame_end.saturating_add(7)) & !7;
            pos = aligned.min(buf.len());
        }

        Ok((metadata, index))
    }

    fn parse_footer_frames(state: &mut RemoteState, msg_idx: usize, buf: &[u8]) -> Result<()> {
        let (metadata, index) = Self::parse_footer_frames_into(buf)?;
        if let Some(m) = metadata {
            state.layouts[msg_idx].global_metadata = Some(m);
        }
        if let Some(i) = index {
            state.layouts[msg_idx].index = Some(i);
        }

        if state.layouts[msg_idx].global_metadata.is_none() {
            return Err(TensogramError::Remote(
                "footer region did not contain a metadata frame".to_string(),
            ));
        }
        if state.layouts[msg_idx].index.is_none() {
            return Err(TensogramError::Remote(
                "footer region did not contain an index frame".to_string(),
            ));
        }

        Ok(())
    }

    // ── Public API used by TensogramFile ─────────────────────────────────

    pub(crate) fn message_count(&self) -> Result<usize> {
        let mut state = self
            .state
            .lock()
            .map_err(|_| TensogramError::Remote("remote state lock poisoned".to_string()))?;
        self.scan_all_locked(&mut state)?;
        Ok(state.layouts.len())
    }

    pub(crate) fn message_layouts(&self) -> Result<Vec<crate::MessageLayout>> {
        let mut state = self
            .state
            .lock()
            .map_err(|_| TensogramError::Remote("remote state lock poisoned".to_string()))?;
        self.scan_all_locked(&mut state)?;
        Ok(state
            .layouts
            .iter()
            .map(|l| crate::MessageLayout {
                offset: l.offset,
                length: l.length,
            })
            .collect())
    }

    pub(crate) fn read_message(&self, msg_idx: usize) -> Result<Vec<u8>> {
        let (offset, length) = {
            let mut state = self
                .state
                .lock()
                .map_err(|_| TensogramError::Remote("remote state lock poisoned".to_string()))?;
            self.ensure_message_locked(&mut state, msg_idx)?;
            let layout = &state.layouts[msg_idx];
            (layout.offset, layout.length)
        };
        let bytes = self.get_range(offset..offset + length)?;
        Ok(bytes.to_vec())
    }

    pub(crate) fn read_metadata(&self, msg_idx: usize) -> Result<GlobalMetadata> {
        let mut state = self
            .state
            .lock()
            .map_err(|_| TensogramError::Remote("remote state lock poisoned".to_string()))?;
        self.ensure_layout_eager_locked(&mut state, msg_idx)?;
        state.layouts[msg_idx]
            .global_metadata
            .clone()
            .ok_or_else(|| TensogramError::Remote("metadata not found".to_string()))
    }

    pub(crate) fn read_descriptors(
        &self,
        msg_idx: usize,
    ) -> Result<(GlobalMetadata, Vec<DataObjectDescriptor>)> {
        let layout = {
            let mut state = self
                .state
                .lock()
                .map_err(|_| TensogramError::Remote("remote state lock poisoned".to_string()))?;
            self.ensure_layout_eager_locked(&mut state, msg_idx)?;
            state.layouts[msg_idx].clone()
        };
        let msg_offset = layout.offset;

        if let Some(ref index) = layout.index {
            if index.offsets.len() != index.lengths.len() {
                return Err(TensogramError::Remote(format!(
                    "corrupt index: offsets.len()={} != lengths.len()={}",
                    index.offsets.len(),
                    index.lengths.len()
                )));
            }

            let meta = layout
                .global_metadata
                .clone()
                .ok_or_else(|| TensogramError::Remote("metadata not cached".to_string()))?;

            let msg_length = layout.length;
            let mut descriptors = Vec::with_capacity(index.offsets.len());
            for i in 0..index.offsets.len() {
                let desc = self.read_descriptor_only(
                    msg_offset,
                    msg_length,
                    index.offsets[i],
                    index.lengths[i],
                )?;
                descriptors.push(desc);
            }
            Ok((meta, descriptors))
        } else {
            let msg_bytes = self.read_message(msg_idx)?;
            crate::decode::decode_descriptors(&msg_bytes)
        }
    }

    /// Read only the CBOR descriptor from a data object frame, without
    /// downloading the full payload.
    ///
    /// For frames below `DESCRIPTOR_PREFIX_THRESHOLD` bytes, falls back
    /// to reading the entire frame (fewer round-trips).  For large frames,
    /// reads just the header, footer, and CBOR region — typically < 10 KB
    /// even when the payload is hundreds of megabytes.
    fn read_descriptor_only(
        &self,
        msg_offset: u64,
        msg_length: u64,
        frame_offset_in_msg: u64,
        frame_length: u64,
    ) -> Result<DataObjectDescriptor> {
        const DESCRIPTOR_PREFIX_THRESHOLD: u64 = 64 * 1024;

        let range =
            Self::checked_frame_range(msg_offset, msg_length, frame_offset_in_msg, frame_length)?;

        if frame_length <= DESCRIPTOR_PREFIX_THRESHOLD {
            let frame_bytes = self.get_range(range.clone())?;
            let (desc, _payload, _mask_region, _consumed) =
                framing::decode_data_object_frame(&frame_bytes)?;
            return Ok(desc);
        }

        let frame_start = range.start;
        let frame_end = range.end;

        let header_bytes = self.get_range(frame_start..frame_start + FRAME_HEADER_SIZE as u64)?;
        let fh = FrameHeader::read_from(&header_bytes)?;

        if !fh.frame_type.is_data_object() {
            return Err(TensogramError::Remote(format!(
                "expected DataObject frame, got {:?}",
                fh.frame_type
            )));
        }

        let footer_start = frame_end - DATA_OBJECT_FOOTER_SIZE as u64;
        let footer_bytes = self.get_range(footer_start..frame_end)?;

        if footer_bytes.len() < DATA_OBJECT_FOOTER_SIZE {
            return Err(TensogramError::Remote("frame footer too short".to_string()));
        }
        // v3 data-object footer: [cbor_offset u64][hash u64][ENDF 4]
        // ENDF sits in the last 4 bytes.
        let endf_pos = footer_bytes.len() - FRAME_END.len();
        if &footer_bytes[endf_pos..] != FRAME_END {
            return Err(TensogramError::Remote(
                "frame missing ENDF trailer".to_string(),
            ));
        }

        let cbor_offset = u64::from_be_bytes(
            footer_bytes[..8]
                .try_into()
                .map_err(|_| TensogramError::Remote("footer cbor_offset truncated".to_string()))?,
        );

        if cbor_offset < FRAME_HEADER_SIZE as u64 {
            return Err(TensogramError::Remote(format!(
                "cbor_offset ({cbor_offset}) below frame header size ({FRAME_HEADER_SIZE})"
            )));
        }

        let cbor_after = fh.flags & DataObjectFlags::CBOR_AFTER_PAYLOAD != 0;
        let cbor_start = frame_start
            .checked_add(cbor_offset)
            .ok_or_else(|| TensogramError::Remote("cbor_start overflow".to_string()))?;

        if cbor_after {
            if cbor_start >= footer_start {
                return Err(TensogramError::Remote(
                    "cbor_offset points at or past footer".to_string(),
                ));
            }
            let cbor_bytes = self.get_range(cbor_start..footer_start)?;
            metadata::cbor_to_object_descriptor(&cbor_bytes)
        } else {
            if cbor_start >= footer_start {
                return Err(TensogramError::Remote(
                    "cbor_offset beyond frame body".to_string(),
                ));
            }
            let max_cbor_len = footer_start - cbor_start;
            let mut prefix_size: u64 = 8192;
            loop {
                let read_end = (cbor_start + prefix_size).min(footer_start);
                let prefix_bytes = self.get_range(cbor_start..read_end)?;
                match metadata::cbor_to_object_descriptor(&prefix_bytes) {
                    Ok(desc) => return Ok(desc),
                    Err(_) if prefix_size < max_cbor_len => {
                        prefix_size = (prefix_size * 2).min(max_cbor_len);
                    }
                    Err(e) => return Err(e),
                }
            }
        }
    }

    pub(crate) fn read_object(
        &self,
        msg_idx: usize,
        obj_idx: usize,
        options: &DecodeOptions,
    ) -> Result<(GlobalMetadata, DataObjectDescriptor, Vec<u8>)> {
        let layout = {
            let mut state = self
                .state
                .lock()
                .map_err(|_| TensogramError::Remote("remote state lock poisoned".to_string()))?;
            self.ensure_layout_eager_locked(&mut state, msg_idx)?;
            state.layouts[msg_idx].clone()
        };
        let msg_offset = layout.offset;

        if let Some(ref index) = layout.index {
            Self::validate_index_access(index, obj_idx)?;

            let meta = layout
                .global_metadata
                .clone()
                .ok_or_else(|| TensogramError::Remote("metadata not cached".to_string()))?;

            let range = Self::checked_frame_range(
                msg_offset,
                layout.length,
                index.offsets[obj_idx],
                index.lengths[obj_idx],
            )?;
            let frame_bytes = self.get_range(range)?;

            // Delegate to the high-level single-frame helper so
            // `options.verify_hash` is honoured on the remote
            // indexed fast path; the helper raises MissingHash /
            // HashMismatch with `object_index = 0` (it has no
            // surrounding context), which we repackage with the
            // real `obj_idx` so callers get an actionable error.
            let (desc, decoded) = crate::decode::decode_object_from_frame(&frame_bytes, options)
                .map_err(|e| crate::error::with_object_index(e, obj_idx))?;

            Ok((meta, desc, decoded))
        } else {
            // No index — fall back to full message download
            let msg_bytes = self.read_message(msg_idx)?;
            crate::decode::decode_object(&msg_bytes, obj_idx, options)
        }
    }

    pub(crate) fn read_range_batch(
        &self,
        msg_indices: &[usize],
        obj_idx: usize,
        ranges: &[(u64, u64)],
        options: &DecodeOptions,
    ) -> Result<Vec<(DataObjectDescriptor, Vec<Vec<u8>>)>> {
        let mut byte_ranges = Vec::with_capacity(msg_indices.len());
        {
            let mut state = self
                .state
                .lock()
                .map_err(|_| TensogramError::Remote("remote state lock poisoned".to_string()))?;
            for &msg_idx in msg_indices {
                self.ensure_layout_eager_locked(&mut state, msg_idx)?;
            }
            for &msg_idx in msg_indices {
                let layout = &state.layouts[msg_idx];
                if let Some(ref index) = layout.index {
                    Self::validate_index_access(index, obj_idx)?;
                    byte_ranges.push(Self::checked_frame_range(
                        layout.offset,
                        layout.length,
                        index.offsets[obj_idx],
                        index.lengths[obj_idx],
                    )?);
                } else {
                    return Err(TensogramError::Remote(format!(
                        "message {} has no index frame; batch requires indexed messages",
                        msg_idx
                    )));
                }
            }
        }

        let store = self.store.clone();
        let path = self.path.clone();
        let all_bytes =
            block_on_shared(async move { store.get_ranges(&path, &byte_ranges).await })?;

        let mut results = Vec::with_capacity(msg_indices.len());
        for frame_bytes in &all_bytes {
            let (desc, payload, _mask_region, _consumed) =
                framing::decode_data_object_frame(frame_bytes)?;
            let parts = crate::decode::decode_range_from_payload(&desc, payload, ranges, options)?;
            results.push((desc, parts));
        }
        Ok(results)
    }

    pub(crate) fn read_object_batch(
        &self,
        msg_indices: &[usize],
        obj_idx: usize,
        options: &DecodeOptions,
    ) -> Result<Vec<(GlobalMetadata, DataObjectDescriptor, Vec<u8>)>> {
        let mut byte_ranges = Vec::with_capacity(msg_indices.len());
        let mut metas = Vec::with_capacity(msg_indices.len());
        {
            let mut state = self
                .state
                .lock()
                .map_err(|_| TensogramError::Remote("remote state lock poisoned".to_string()))?;
            for &msg_idx in msg_indices {
                self.ensure_layout_eager_locked(&mut state, msg_idx)?;
            }
            for &msg_idx in msg_indices {
                let layout = &state.layouts[msg_idx];
                if let Some(ref index) = layout.index {
                    Self::validate_index_access(index, obj_idx)?;
                    byte_ranges.push(Self::checked_frame_range(
                        layout.offset,
                        layout.length,
                        index.offsets[obj_idx],
                        index.lengths[obj_idx],
                    )?);
                    metas.push(layout.global_metadata.clone().ok_or_else(|| {
                        TensogramError::Remote("metadata not cached".to_string())
                    })?);
                } else {
                    return Err(TensogramError::Remote(format!(
                        "message {} has no index frame; batch decode requires indexed messages",
                        msg_idx
                    )));
                }
            }
        }

        let store = self.store.clone();
        let path = self.path.clone();
        let all_bytes =
            block_on_shared(async move { store.get_ranges(&path, &byte_ranges).await })?;

        let mut results = Vec::with_capacity(msg_indices.len());
        for (frame_bytes, meta) in all_bytes.iter().zip(metas) {
            // Use the verifying single-frame helper; remap the
            // placeholder object_index = 0 to the batched obj_idx
            // so the failure names the offending object.
            let (desc, decoded) = crate::decode::decode_object_from_frame(frame_bytes, options)
                .map_err(|e| crate::error::with_object_index(e, obj_idx))?;
            results.push((meta, desc, decoded));
        }
        Ok(results)
    }

    pub(crate) fn read_range(
        &self,
        msg_idx: usize,
        obj_idx: usize,
        ranges: &[(u64, u64)],
        options: &DecodeOptions,
    ) -> Result<(DataObjectDescriptor, Vec<Vec<u8>>)> {
        let layout = {
            let mut state = self
                .state
                .lock()
                .map_err(|_| TensogramError::Remote("remote state lock poisoned".to_string()))?;
            self.ensure_layout_eager_locked(&mut state, msg_idx)?;
            state.layouts[msg_idx].clone()
        };
        let msg_offset = layout.offset;

        if let Some(ref index) = layout.index {
            Self::validate_index_access(index, obj_idx)?;

            let range = Self::checked_frame_range(
                msg_offset,
                layout.length,
                index.offsets[obj_idx],
                index.lengths[obj_idx],
            )?;
            let frame_bytes = self.get_range(range)?;
            let (desc, payload, _mask_region, _consumed) =
                framing::decode_data_object_frame(&frame_bytes)?;
            let parts = crate::decode::decode_range_from_payload(&desc, payload, ranges, options)?;
            Ok((desc, parts))
        } else {
            let msg_bytes = self.read_message(msg_idx)?;
            crate::decode::decode_range(&msg_bytes, obj_idx, ranges, options)
        }
    }

    fn validate_index_access(index: &IndexFrame, obj_idx: usize) -> Result<()> {
        if index.offsets.len() != index.lengths.len() {
            return Err(TensogramError::Remote(format!(
                "corrupt index: offsets.len()={} != lengths.len()={}",
                index.offsets.len(),
                index.lengths.len()
            )));
        }
        if obj_idx >= index.offsets.len() {
            return Err(TensogramError::Object(format!(
                "object index {} out of range (count={})",
                obj_idx,
                index.offsets.len()
            )));
        }
        Ok(())
    }

    fn checked_frame_range(
        msg_offset: u64,
        msg_length: u64,
        frame_offset_in_msg: u64,
        frame_length: u64,
    ) -> Result<Range<u64>> {
        let start = msg_offset
            .checked_add(frame_offset_in_msg)
            .ok_or_else(|| TensogramError::Remote("frame offset overflow".to_string()))?;
        let end = start
            .checked_add(frame_length)
            .ok_or_else(|| TensogramError::Remote("frame end overflow".to_string()))?;
        let msg_end = msg_offset
            .checked_add(msg_length)
            .ok_or_else(|| TensogramError::Remote("message end overflow".to_string()))?;
        if end > msg_end {
            return Err(TensogramError::Remote(format!(
                "indexed frame {start}..{end} exceeds message boundary {msg_end}"
            )));
        }
        Ok(start..end)
    }
}

// ── Native async path (remote + async) ───────────────────────────────────────

#[cfg(feature = "async")]
impl RemoteBackend {
    async fn get_range_async(&self, range: Range<u64>) -> Result<Bytes> {
        self.store
            .get_range(&self.path, range)
            .await
            .map_err(|e| TensogramError::Remote(e.to_string()))
    }

    pub(crate) async fn open_async_with_scan_opts(
        source: &str,
        storage_options: &BTreeMap<String, String>,
        scan_opts: RemoteScanOptions,
    ) -> Result<Self> {
        emit_scan_mode(&scan_opts);
        let url = Url::parse(source)
            .map_err(|e| TensogramError::Remote(format!("invalid URL '{source}': {e}")))?;

        let mut opts: Vec<(String, String)> = storage_options
            .iter()
            .map(|(k, v)| (k.clone(), v.clone()))
            .collect();
        if url.scheme() == "http" && !opts.iter().any(|(k, _)| k == "allow_http") {
            opts.push(("allow_http".to_string(), "true".to_string()));
        }
        let (store, path) = object_store::parse_url_opts(&url, opts)
            .map_err(|e| TensogramError::Remote(format!("cannot open '{source}': {e}")))?;

        let store: Arc<dyn ObjectStore> = Arc::from(store);
        let meta = store
            .head(&path)
            .await
            .map_err(|e| TensogramError::Remote(e.to_string()))?;

        let file_size = meta.size;
        if file_size < (PREAMBLE_SIZE + POSTAMBLE_SIZE) as u64 {
            return Err(TensogramError::Remote(format!(
                "remote file too small ({file_size} bytes)"
            )));
        }

        let backend = RemoteBackend {
            source_url: source.to_string(),
            store,
            path,
            file_size,
            state: Mutex::new(RemoteState {
                prev_scan_offset: file_size,
                bwd_active: scan_opts.bidirectional,
                ..RemoteState::default()
            }),
            scan_opts,
        };
        backend.scan_next_async().await?;
        {
            let state = backend.lock_state()?;
            if state.layouts.is_empty() {
                return Err(TensogramError::Remote(
                    "no valid messages found in remote file".to_string(),
                ));
            }
        }
        Ok(backend)
    }

    async fn scan_next_async(&self) -> Result<()> {
        self.scan_fwd_step_async(self.file_size).await
    }

    /// Async counterpart of [`Self::scan_fwd_step_locked`].  See that
    /// method's docstring for the `bound` parameter contract and for
    /// the saturating-arithmetic invariant that keeps the downstream
    /// `pos + PREAMBLE_SIZE` range from wrapping.
    async fn scan_fwd_step_async(&self, bound: u64) -> Result<()> {
        let min_message_size = (PREAMBLE_SIZE + POSTAMBLE_SIZE) as u64;
        let pos = {
            let state = self.lock_state()?;
            if state.scan_complete() {
                return Ok(());
            }
            state.next_scan_offset
        };

        if pos.saturating_add(min_message_size) > bound {
            let mut state = self.lock_state()?;
            if state.next_scan_offset == pos {
                state.terminate_forward("eof");
            }
            return Ok(());
        }

        let preamble_bytes = self
            .get_range_async(pos..pos + PREAMBLE_SIZE as u64)
            .await?;
        if &preamble_bytes[..MAGIC.len()] != MAGIC {
            let mut state = self.lock_state()?;
            if state.next_scan_offset == pos {
                state.terminate_forward("bad-magic-fwd");
            }
            return Ok(());
        }

        let preamble = match Preamble::read_from(&preamble_bytes) {
            Ok(preamble) => preamble,
            Err(_) => {
                let mut state = self.lock_state()?;
                if state.next_scan_offset == pos {
                    state.terminate_forward("preamble-parse-error-fwd");
                }
                return Ok(());
            }
        };

        let msg_len = preamble.total_length;

        if msg_len == 0 {
            let remaining = self.file_size - pos;
            if remaining < min_message_size {
                let mut state = self.lock_state()?;
                if state.next_scan_offset == pos {
                    state.terminate_forward("streaming-tail-too-small");
                }
                return Ok(());
            }

            let end_magic_pos = self.file_size - crate::wire::END_MAGIC.len() as u64;
            let end_bytes = self
                .get_range_async(end_magic_pos..end_magic_pos + crate::wire::END_MAGIC.len() as u64)
                .await?;
            if &end_bytes[..] != crate::wire::END_MAGIC {
                let mut state = self.lock_state()?;
                if state.next_scan_offset == pos {
                    state.terminate_forward("streaming-end-magic-mismatch");
                }
                return Ok(());
            }

            let mut state = self.lock_state()?;
            if state.scan_complete() || state.next_scan_offset != pos {
                return Ok(());
            }
            state.record_forward_hop(CachedLayout {
                offset: pos,
                length: remaining,
                preamble,
                index: None,
                global_metadata: None,
            });
            state.terminate_forward("streaming-tail");
            return Ok(());
        }

        match pos.checked_add(msg_len) {
            Some(end) if msg_len >= min_message_size && end <= bound => {}
            _ => {
                let mut state = self.lock_state()?;
                if state.next_scan_offset == pos {
                    state.terminate_forward("length-out-of-range-fwd");
                }
                return Ok(());
            }
        }

        let mut state = self.lock_state()?;
        if state.scan_complete() || state.next_scan_offset != pos {
            return Ok(());
        }
        state.record_forward_hop(CachedLayout {
            offset: pos,
            length: msg_len,
            preamble,
            index: None,
            global_metadata: None,
        });
        Ok(())
    }

    /// Async sibling of [`Self::scan_step_locked`]; see that method's
    /// docstring.  Returns immediately when the scan is already
    /// complete; otherwise dispatches to either the bidirectional
    /// paired round or a single bounded forward hop.  `None`
    /// signals bidirectional, `Some(bound)` signals forward-only.
    async fn scan_step_async(&self) -> Result<()> {
        let fwd_bound: Option<u64> = {
            let state = self.lock_state()?;
            if state.scan_complete() {
                return Ok(());
            }
            if state.bwd_active && !state.fwd_terminated {
                None
            } else {
                Some(self.forward_bound(&state))
            }
        };
        match fwd_bound {
            Some(bound) => self.scan_fwd_step_async(bound).await,
            None => self.scan_bidir_round_async().await,
        }
    }

    /// Async sibling of [`Self::scan_bidir_round_locked`]; see that
    /// method's docstring.  Lock-around-await: snapshot before the
    /// paired fetch, validate the same snapshot on reacquire, then
    /// commit through [`Self::apply_round_outcomes`].
    ///
    /// Re-checks `bwd_active && !fwd_terminated` under the initial
    /// lock — `scan_step_async` already made that decision before
    /// dropping the lock, but a concurrent caller may have disabled
    /// the backward walker (or terminated forward) in the meantime.
    /// Without the recheck, a stale dispatch would snapshot the
    /// post-mutation `prev_scan_offset` and use it as the forward
    /// bound, terminating forward at a stale offset rather than at
    /// `file_size`.
    async fn scan_bidir_round_async(&self) -> Result<()> {
        let snap_opt: Option<ScanSnapshot> = {
            let state = self.lock_state()?;
            if state.scan_complete() {
                return Ok(());
            }
            if state.bwd_active && !state.fwd_terminated {
                Some(state.snapshot())
            } else {
                None
            }
        };
        let snap = match snap_opt {
            Some(snap) => snap,
            None => return self.scan_fwd_step_async(self.file_size).await,
        };
        let bound = snap.prev;
        let min_message_size = (PREAMBLE_SIZE + POSTAMBLE_SIZE) as u64;

        // See `scan_bidir_round_locked` for why a `min_message_size`
        // threshold (the smallest disjoint pair of preamble/postamble
        // ranges) is the correct floor, why the unchecked range
        // constructions immediately below cannot wrap once this guard
        // passes, and why a gap below the threshold is treated as
        // suspected backward corruption rather than as forward EOF.
        if snap.prev < snap.next.saturating_add(min_message_size) {
            let recovery_bound = {
                let mut state = self.lock_state()?;
                if !state.matches(&snap) {
                    return Ok(());
                }
                if snap.next == snap.prev {
                    state.close_gap();
                    return Ok(());
                }
                state.disable_backward("gap-below-min-message-size");
                self.forward_bound(&state)
            };
            return self.scan_fwd_step_async(recovery_bound).await;
        }

        let fwd_r = snap.next..snap.next + PREAMBLE_SIZE as u64;
        let bwd_r = snap.prev - POSTAMBLE_SIZE as u64..snap.prev;
        // Two independent `get_range` futures joined in parallel —
        // see `scan_bidir_round_locked` for why we deliberately bypass
        // `get_ranges`'s coalescer.
        let (fwd_res, bwd_res) = tokio::join!(
            self.get_range_async(fwd_r.clone()),
            self.get_range_async(bwd_r.clone())
        );
        let fwd_bytes = fwd_res?;
        let bwd_bytes = bwd_res?;

        let bwd_outcome = parse_backward_postamble(&bwd_bytes, &snap);
        let (candidate_preamble_bytes, candidate_footer_bytes) = match &bwd_outcome {
            BackwardOutcome::NeedPreambleValidation {
                msg_start,
                length,
                first_footer_offset,
            } => {
                let preamble_fut =
                    self.get_range_async(*msg_start..*msg_start + PREAMBLE_SIZE as u64);
                if footer_region_present(*first_footer_offset, *length) {
                    let footer_start = msg_start.saturating_add(*first_footer_offset);
                    let footer_end = msg_start
                        .saturating_add(*length)
                        .saturating_sub(POSTAMBLE_SIZE as u64);
                    if footer_start < footer_end {
                        let footer_fut = self.get_range_async(footer_start..footer_end);
                        // Parallelise: preamble is required, footer is best-effort.
                        // `tokio::join!` lets one future fail without aborting the other.
                        let (preamble_res, footer_res) = tokio::join!(preamble_fut, footer_fut);
                        // Surface the best-effort failure as a tracing
                        // event so a flapping endpoint isn't invisible
                        // (parity with the sync path).
                        let footer = match footer_res {
                            Ok(b) => Some(b),
                            Err(e) => {
                                tracing::debug!(
                                    target: "tensogram::remote_scan",
                                    error = %e,
                                    msg_start = *msg_start,
                                    footer_start = footer_start,
                                    footer_end = footer_end,
                                    "eager footer fetch failed; falling back to lazy"
                                );
                                None
                            }
                        };
                        (Some(preamble_res?), footer)
                    } else {
                        (Some(preamble_fut.await?), None)
                    }
                } else {
                    (Some(preamble_fut.await?), None)
                }
            }
            _ => (None, None),
        };

        let mut state = self.lock_state()?;
        if !state.matches(&snap) {
            return Ok(());
        }
        let fwd_outcome = parse_forward_preamble(&fwd_bytes, snap.next, self.file_size, bound);
        self.apply_round_outcomes(
            &mut state,
            fwd_outcome,
            bwd_outcome,
            candidate_preamble_bytes.as_deref(),
            candidate_footer_bytes.as_deref(),
        );
        Ok(())
    }

    /// Pipelined bidirectional scan with optional forward-side target.
    ///
    /// Each iteration fetches three ranges in parallel: the next forward
    /// preamble, the next backward postamble, AND the validation
    /// preamble for the previous iteration's backward-discovered
    /// candidate.  That overlap collapses the per-round critical path
    /// from 2 RTTs (paired primary then candidate validation) to 1 RTT,
    /// halving wall-clock for full-discovery operations like
    /// `message_count_async` and `message_layouts_async`.
    ///
    /// `target_fwd_count = Some(n)` stops the loop as soon as the
    /// forward cursor has discovered at least `n` messages
    /// (committed + locally-accumulated).  This eliminates the
    /// "full-discovery overhead for early-access targets" regression:
    /// for `decode_metadata(idx)` the caller passes `Some(idx + 1)`
    /// so the walker stops walking once the target message's layout
    /// is known on the forward side, rather than continuing all the
    /// way to the cursor meeting point.  `None` requests full
    /// discovery (cursors meet).
    ///
    /// Snapshot/commit pattern: the function runs entirely off the
    /// state mutex.  Layouts accumulate in local `Vec`s.  At the end,
    /// the lock is reacquired; if the snapshot still matches, all
    /// layouts commit through `record_forward_hop` / `record_backward_hop`.
    /// If the snapshot is stale, the work is silently dropped and the
    /// caller's outer loop drives discovery via `scan_step_async`.
    ///
    /// Returns `Ok(true)` on a successful pipelined run (state now
    /// reflects discovery up to the target or to gap-closed) or
    /// `Ok(false)` to signal the caller must fall back to the
    /// per-round walker — fired on any anomaly (Format error,
    /// HitBeyondBound, Streaming, gap-below-min, snapshot mismatch).
    /// The fallback path preserves all the edge-case handling in
    /// `apply_round_outcomes`.
    ///
    /// Eager footer fetches are deliberately skipped on this fast
    /// path; callers that need metadata/index frames trigger the
    /// lazy `ensure_layout` fetch on first access.
    async fn scan_pipelined_async(&self, target_fwd_count: Option<usize>) -> Result<bool> {
        let (snap, committed_fwd_at_start) = {
            let state = self.lock_state()?;
            if state.scan_complete() {
                return Ok(true);
            }
            if let Some(target) = target_fwd_count {
                if state.layouts.len() >= target {
                    return Ok(true);
                }
            }
            if !state.bwd_active || state.fwd_terminated {
                return Ok(false);
            }
            (state.snapshot(), state.layouts.len())
        };

        let min_message_size = (PREAMBLE_SIZE + POSTAMBLE_SIZE) as u64;
        let mut fwd_cursor = snap.next;
        let mut bwd_cursor = snap.prev;
        let mut pending: Option<(u64, u64)> = None;
        let mut local_fwd: Vec<CachedLayout> = Vec::new();
        let mut local_bwd: Vec<CachedLayout> = Vec::new();

        loop {
            if fwd_cursor == bwd_cursor {
                break;
            }
            if let Some(target) = target_fwd_count {
                if committed_fwd_at_start + local_fwd.len() >= target {
                    break;
                }
            }
            if bwd_cursor < fwd_cursor.saturating_add(min_message_size) {
                return Ok(false);
            }

            let fwd_r = fwd_cursor..fwd_cursor + PREAMBLE_SIZE as u64;
            let bwd_r = bwd_cursor - POSTAMBLE_SIZE as u64..bwd_cursor;
            let val_r_opt = pending.map(|(ms, _)| ms..ms + PREAMBLE_SIZE as u64);

            let (fwd_bytes, bwd_postamble_bytes, val_bytes_opt) = match val_r_opt {
                Some(val_r) => {
                    let (f, b, v) = tokio::join!(
                        self.get_range_async(fwd_r),
                        self.get_range_async(bwd_r),
                        self.get_range_async(val_r),
                    );
                    (f?, b?, Some(v?))
                }
                None => {
                    let (f, b) =
                        tokio::join!(self.get_range_async(fwd_r), self.get_range_async(bwd_r));
                    (f?, b?, None)
                }
            };

            if let Some(val_bytes) = val_bytes_opt {
                let (msg_start, length) = pending.take().expect("pending matched val_r_opt");
                match validate_backward_preamble(&val_bytes, msg_start, length) {
                    BackwardCommit::Layout(layout) => local_bwd.push(layout),
                    BackwardCommit::Format(_) => return Ok(false),
                }
            }

            let pre_iter_fwd_cursor = fwd_cursor;
            let fwd_outcome =
                parse_forward_preamble(&fwd_bytes, pre_iter_fwd_cursor, self.file_size, bwd_cursor);
            let (fwd_offset, fwd_length) = match fwd_outcome {
                ForwardOutcome::Hit {
                    offset,
                    length,
                    preamble,
                    msg_end,
                } => {
                    local_fwd.push(CachedLayout {
                        offset,
                        length,
                        preamble,
                        index: None,
                        global_metadata: None,
                    });
                    fwd_cursor = msg_end;
                    (offset, length)
                }
                _ => return Ok(false),
            };

            // `parse_backward_postamble` rejects `msg_start < snap.next`
            // as `backward-overlaps-forward`, so we must pass the
            // pre-iteration forward cursor — the post-hop cursor
            // would falsely reject the same-message-meet case where
            // backward identifies the message forward just committed
            // (1-msg files, odd-count crossover).
            let snap_local = ScanSnapshot {
                next: pre_iter_fwd_cursor,
                prev: bwd_cursor,
                epoch: snap.epoch,
            };
            match parse_backward_postamble(&bwd_postamble_bytes, &snap_local) {
                BackwardOutcome::NeedPreambleValidation {
                    msg_start,
                    length,
                    first_footer_offset: _,
                } => {
                    if msg_start == fwd_offset && length == fwd_length {
                        // Same-message meet: the backward postamble
                        // points at the message forward just committed.
                        // Skip the backward record (forward already
                        // produced an authoritative layout) and leave
                        // `bwd_cursor` where it is so the next
                        // iteration's `fwd_cursor == bwd_cursor` check
                        // terminates the walk via gap-closed.
                        continue;
                    }
                    if msg_start < fwd_cursor {
                        // Backward candidate overlaps the
                        // just-committed forward message.
                        // `parse_backward_postamble` only saw the
                        // pre-iteration cursor and accepted the
                        // candidate; without this guard a
                        // target-driven walk would break out of the
                        // loop before the next-iteration gap check
                        // fires and the post-loop drain would commit
                        // overlapping layouts (validation only checks
                        // that the preamble bytes parse as a preamble
                        // with a length matching the postamble's
                        // claim, which is satisfied by an attacker
                        // who plants a fake preamble inside the
                        // forward message).  Bail to the per-round
                        // walker, which catches the overlap via
                        // `apply_round_outcomes`'
                        // `msg_end > layout.offset` rejection and
                        // continues forward-only.
                        return Ok(false);
                    }
                    pending = Some((msg_start, length));
                    bwd_cursor = msg_start;
                }
                _ => return Ok(false),
            }
        }

        if let Some((msg_start, length)) = pending {
            let val_bytes = self
                .get_range_async(msg_start..msg_start + PREAMBLE_SIZE as u64)
                .await?;
            match validate_backward_preamble(&val_bytes, msg_start, length) {
                BackwardCommit::Layout(layout) => local_bwd.push(layout),
                BackwardCommit::Format(_) => return Ok(false),
            }
        }

        let mut state = self.lock_state()?;
        if !state.matches(&snap) {
            // Snapshot drift: another caller advanced state under us
            // while we ran the pipeline off-lock.  Discard our local
            // accumulators and tell the caller to drive the fallback
            // loop — its termination check (`scan_complete()` or
            // `layouts.len() >= target`) sees the post-drift state and
            // either returns immediately (no further work needed) or
            // dispatches per-round walks for whatever remains.
            // Returning Ok(true) here would mislead `message_count_async`
            // / `message_layouts_async` into treating the partial post-
            // drift state as the final result.
            return Ok(false);
        }
        for layout in local_fwd {
            state.record_forward_hop(layout);
        }
        for layout in local_bwd {
            state.record_backward_hop(layout);
        }
        if state.next_scan_offset == state.prev_scan_offset {
            state.close_gap();
        }
        Ok(true)
    }

    async fn ensure_message_async(&self, msg_idx: usize) -> Result<()> {
        let needs_layout = {
            let state = self.lock_state()?;
            msg_idx >= state.layouts.len()
                && !state.scan_complete()
                && state.bwd_active
                && !state.fwd_terminated
        };
        if needs_layout {
            let _ = self.scan_pipelined_async(Some(msg_idx + 1)).await?;
        }
        loop {
            let ready = {
                let state = self.lock_state()?;
                if msg_idx < state.layouts.len() {
                    return Ok(());
                }
                if state.scan_complete() {
                    return Err(TensogramError::Framing(format!(
                        "message index {} out of range (count={})",
                        msg_idx,
                        state.layouts.len()
                    )));
                }
                false
            };

            if !ready {
                self.scan_step_async().await?;
            }
        }
    }

    async fn scan_and_discover_next_async(&self) -> Result<()> {
        let min_message_size = (PREAMBLE_SIZE + POSTAMBLE_SIZE) as u64;
        let pos = {
            let state = self.lock_state()?;
            if state.scan_complete() {
                return Ok(());
            }
            state.next_scan_offset
        };

        // Saturating addition: see `scan_fwd_step_locked` for the
        // proof that this guard is sufficient to keep the unchecked
        // `pos + chunk_size` range below from wrapping.
        if pos.saturating_add(min_message_size) > self.file_size {
            let mut state = self.lock_state()?;
            if state.next_scan_offset == pos {
                state.terminate_forward("eof");
            }
            return Ok(());
        }

        let chunk_size = (self.file_size - pos).min(256 * 1024);
        let chunk = self.get_range_async(pos..pos + chunk_size).await?;

        if chunk.len() < PREAMBLE_SIZE || &chunk[..MAGIC.len()] != MAGIC {
            let mut state = self.lock_state()?;
            if state.next_scan_offset == pos {
                state.terminate_forward("bad-magic-fwd");
            }
            return Ok(());
        }

        let preamble = match Preamble::read_from(&chunk[..PREAMBLE_SIZE]) {
            Ok(preamble) => preamble,
            Err(_) => {
                let mut state = self.lock_state()?;
                if state.next_scan_offset == pos {
                    state.terminate_forward("preamble-parse-error-fwd");
                }
                return Ok(());
            }
        };

        let msg_len = preamble.total_length;

        if msg_len == 0 {
            let remaining = self.file_size - pos;
            if remaining < min_message_size {
                let mut state = self.lock_state()?;
                if state.next_scan_offset == pos {
                    state.terminate_forward("streaming-tail-too-small");
                }
                return Ok(());
            }

            let end_magic_pos = self.file_size - crate::wire::END_MAGIC.len() as u64;
            let end_bytes = self
                .get_range_async(end_magic_pos..end_magic_pos + crate::wire::END_MAGIC.len() as u64)
                .await?;
            if &end_bytes[..] != crate::wire::END_MAGIC {
                let mut state = self.lock_state()?;
                if state.next_scan_offset == pos {
                    state.terminate_forward("streaming-end-magic-mismatch");
                }
                return Ok(());
            }

            let mut state = self.lock_state()?;
            if state.scan_complete() || state.next_scan_offset != pos {
                return Ok(());
            }
            state.record_forward_hop(CachedLayout {
                offset: pos,
                length: remaining,
                preamble,
                index: None,
                global_metadata: None,
            });
            state.terminate_forward("streaming-tail");
            return Ok(());
        }

        match pos.checked_add(msg_len) {
            Some(end) if msg_len >= min_message_size && end <= self.file_size => {}
            _ => {
                let mut state = self.lock_state()?;
                if state.next_scan_offset == pos {
                    state.terminate_forward("length-out-of-range-fwd");
                }
                return Ok(());
            }
        }

        let flags = preamble.flags;
        let msg_idx = {
            let mut state = self.lock_state()?;
            if state.scan_complete() || state.next_scan_offset != pos {
                return Ok(());
            }
            let msg_idx = state.layouts.len();
            state.record_forward_hop(CachedLayout {
                offset: pos,
                length: msg_len,
                preamble,
                index: None,
                global_metadata: None,
            });
            msg_idx
        };

        if flags.has(MessageFlags::HEADER_METADATA) && flags.has(MessageFlags::HEADER_INDEX) {
            let chunk_end = (msg_len as usize).min(chunk.len());
            let mut state = self.lock_state()?;
            if msg_idx < state.layouts.len()
                && state.layouts[msg_idx].offset == pos
                && state.layouts[msg_idx].global_metadata.is_none()
                && state.layouts[msg_idx].index.is_none()
            {
                Self::parse_header_frames(&mut state, msg_idx, &chunk[..chunk_end])?;
            }
        } else if flags.has(MessageFlags::FOOTER_METADATA) && flags.has(MessageFlags::FOOTER_INDEX)
        {
            self.discover_footer_layout_from_suffix_async(msg_idx)
                .await?;
        }

        Ok(())
    }

    async fn discover_footer_layout_from_suffix_async(&self, msg_idx: usize) -> Result<()> {
        let (msg_offset, msg_len) = {
            let state = self.lock_state()?;
            let layout = state.layouts.get(msg_idx).ok_or_else(|| {
                TensogramError::Framing(format!(
                    "message index {} out of range (count={})",
                    msg_idx,
                    state.layouts.len()
                ))
            })?;
            (layout.offset, layout.length)
        };

        let suffix_size = msg_len.min(256 * 1024);
        let msg_end = msg_offset
            .checked_add(msg_len)
            .ok_or_else(|| TensogramError::Remote("message end overflow".to_string()))?;
        let suffix_start = msg_end - suffix_size;
        let suffix = self.get_range_async(suffix_start..msg_end).await?;

        if suffix.len() < POSTAMBLE_SIZE {
            return Err(TensogramError::Remote(
                "suffix too short for postamble".to_string(),
            ));
        }

        let pa_bytes = &suffix[suffix.len() - POSTAMBLE_SIZE..];
        let postamble = Postamble::read_from(pa_bytes)?;

        if postamble.first_footer_offset < PREAMBLE_SIZE as u64 {
            return Err(TensogramError::Remote(format!(
                "first_footer_offset ({}) is before preamble end ({PREAMBLE_SIZE})",
                postamble.first_footer_offset
            )));
        }

        let footer_abs_start = msg_offset
            .checked_add(postamble.first_footer_offset)
            .ok_or_else(|| TensogramError::Remote("footer offset overflow".to_string()))?;
        let footer_abs_end = msg_end - POSTAMBLE_SIZE as u64;

        if footer_abs_start >= footer_abs_end {
            return Err(TensogramError::Remote(
                "first_footer_offset points at or past postamble".to_string(),
            ));
        }

        if footer_abs_start >= suffix_start {
            let local_start = (footer_abs_start - suffix_start) as usize;
            let local_end = suffix.len() - POSTAMBLE_SIZE;
            let mut state = self.lock_state()?;
            if state.layouts[msg_idx].global_metadata.is_some()
                && state.layouts[msg_idx].index.is_some()
            {
                return Ok(());
            }
            Self::parse_footer_frames(&mut state, msg_idx, &suffix[local_start..local_end])
        } else {
            let footer_bytes = self
                .get_range_async(footer_abs_start..footer_abs_end)
                .await?;
            let mut state = self.lock_state()?;
            if state.layouts[msg_idx].global_metadata.is_some()
                && state.layouts[msg_idx].index.is_some()
            {
                return Ok(());
            }
            Self::parse_footer_frames(&mut state, msg_idx, &footer_bytes)
        }
    }

    async fn ensure_layout_eager_async(&self, msg_idx: usize) -> Result<()> {
        let needs_layout = {
            let state = self.lock_state()?;
            msg_idx >= state.layouts.len()
                && !state.scan_complete()
                && state.bwd_active
                && !state.fwd_terminated
        };
        if needs_layout {
            let _ = self.scan_pipelined_async(Some(msg_idx + 1)).await?;
        }
        loop {
            let action = {
                let state = self.lock_state()?;
                if let Some(layout) = state.layouts.get(msg_idx) {
                    if layout.global_metadata.is_some() && layout.index.is_some() {
                        return Ok(());
                    }
                    EagerAction::Discover
                } else if state.scan_complete() {
                    return Err(TensogramError::Framing(format!(
                        "message index {} out of range (count={})",
                        msg_idx,
                        state.layouts.len()
                    )));
                } else if state.bwd_active && !state.fwd_terminated {
                    EagerAction::ScanBidir
                } else {
                    EagerAction::ScanForwardEager
                }
            };

            match action {
                EagerAction::ScanBidir => {
                    self.scan_bidir_round_async().await?;
                }
                EagerAction::ScanForwardEager => {
                    self.scan_and_discover_next_async().await?;
                }
                EagerAction::Discover => {
                    return self.ensure_layout_async(msg_idx).await;
                }
            }
        }
    }

    async fn ensure_layout_async(&self, msg_idx: usize) -> Result<()> {
        self.ensure_message_async(msg_idx).await?;

        let flags = {
            let state = self.lock_state()?;
            let layout = &state.layouts[msg_idx];
            if layout.global_metadata.is_some() && layout.index.is_some() {
                return Ok(());
            }
            layout.preamble.flags
        };

        if flags.has(MessageFlags::HEADER_METADATA) && flags.has(MessageFlags::HEADER_INDEX) {
            self.discover_header_layout_async(msg_idx).await?;
        } else if flags.has(MessageFlags::FOOTER_METADATA) && flags.has(MessageFlags::FOOTER_INDEX)
        {
            self.discover_footer_layout_async(msg_idx).await?;
        } else {
            return Err(TensogramError::Remote(
                "remote access requires header-indexed or footer-indexed messages".to_string(),
            ));
        }

        Ok(())
    }

    async fn discover_header_layout_async(&self, msg_idx: usize) -> Result<()> {
        let (msg_offset, msg_len) = {
            let state = self.lock_state()?;
            let layout = state.layouts.get(msg_idx).ok_or_else(|| {
                TensogramError::Framing(format!(
                    "message index {} out of range (count={})",
                    msg_idx,
                    state.layouts.len()
                ))
            })?;
            (layout.offset, layout.length)
        };

        let chunk_size = msg_len.min(256 * 1024);
        let header_bytes = self
            .get_range_async(msg_offset..msg_offset + chunk_size)
            .await?;

        let mut state = self.lock_state()?;
        if state.layouts[msg_idx].global_metadata.is_some()
            && state.layouts[msg_idx].index.is_some()
        {
            return Ok(());
        }
        Self::parse_header_frames(&mut state, msg_idx, &header_bytes)
    }

    async fn discover_footer_layout_async(&self, msg_idx: usize) -> Result<()> {
        let (msg_offset, msg_len) = {
            let state = self.lock_state()?;
            let layout = state.layouts.get(msg_idx).ok_or_else(|| {
                TensogramError::Framing(format!(
                    "message index {} out of range (count={})",
                    msg_idx,
                    state.layouts.len()
                ))
            })?;
            (layout.offset, layout.length)
        };

        let pa_offset = msg_offset
            .checked_add(msg_len)
            .and_then(|end| end.checked_sub(POSTAMBLE_SIZE as u64))
            .ok_or_else(|| TensogramError::Remote("postamble offset overflow".to_string()))?;
        let pa_bytes = self
            .get_range_async(pa_offset..pa_offset + POSTAMBLE_SIZE as u64)
            .await?;
        let postamble = Postamble::read_from(&pa_bytes)?;

        if postamble.first_footer_offset < PREAMBLE_SIZE as u64 {
            return Err(TensogramError::Remote(format!(
                "first_footer_offset ({}) is before preamble end ({})",
                postamble.first_footer_offset, PREAMBLE_SIZE
            )));
        }
        let footer_start = msg_offset
            .checked_add(postamble.first_footer_offset)
            .ok_or_else(|| TensogramError::Remote("footer offset overflow".to_string()))?;
        let footer_end = pa_offset;
        if footer_start >= footer_end {
            return Err(TensogramError::Remote(
                "first_footer_offset points at or past postamble".to_string(),
            ));
        }

        let footer_bytes = self.get_range_async(footer_start..footer_end).await?;
        let mut state = self.lock_state()?;
        if state.layouts[msg_idx].global_metadata.is_some()
            && state.layouts[msg_idx].index.is_some()
        {
            return Ok(());
        }
        Self::parse_footer_frames(&mut state, msg_idx, &footer_bytes)
    }

    pub(crate) async fn message_count_async(&self) -> Result<usize> {
        if self.scan_pipelined_async(None).await? {
            let state = self.lock_state()?;
            return Ok(state.layouts.len());
        }
        loop {
            let done = {
                let state = self.lock_state()?;
                state.scan_complete()
            };
            if done {
                break;
            }
            self.scan_step_async().await?;
        }
        let state = self.lock_state()?;
        Ok(state.layouts.len())
    }

    pub(crate) async fn message_layouts_async(&self) -> Result<Vec<crate::MessageLayout>> {
        if !self.scan_pipelined_async(None).await? {
            loop {
                let done = {
                    let state = self.lock_state()?;
                    state.scan_complete()
                };
                if done {
                    break;
                }
                self.scan_step_async().await?;
            }
        }
        let state = self.lock_state()?;
        Ok(state
            .layouts
            .iter()
            .map(|l| crate::MessageLayout {
                offset: l.offset,
                length: l.length,
            })
            .collect())
    }

    pub(crate) async fn read_message_async(&self, msg_idx: usize) -> Result<Vec<u8>> {
        self.ensure_message_async(msg_idx).await?;
        let (offset, length) = {
            let state = self.lock_state()?;
            let layout = &state.layouts[msg_idx];
            (layout.offset, layout.length)
        };
        let bytes = self.get_range_async(offset..offset + length).await?;
        Ok(bytes.to_vec())
    }

    pub(crate) async fn read_metadata_async(&self, msg_idx: usize) -> Result<GlobalMetadata> {
        self.ensure_layout_eager_async(msg_idx).await?;
        let state = self.lock_state()?;
        state.layouts[msg_idx]
            .global_metadata
            .clone()
            .ok_or_else(|| TensogramError::Remote("metadata not found".to_string()))
    }

    pub(crate) async fn read_descriptors_async(
        &self,
        msg_idx: usize,
    ) -> Result<(GlobalMetadata, Vec<DataObjectDescriptor>)> {
        self.ensure_layout_eager_async(msg_idx).await?;
        let layout = {
            let state = self.lock_state()?;
            state.layouts[msg_idx].clone()
        };
        let msg_offset = layout.offset;

        if let Some(ref index) = layout.index {
            if index.offsets.len() != index.lengths.len() {
                return Err(TensogramError::Remote(format!(
                    "corrupt index: offsets.len()={} != lengths.len()={}",
                    index.offsets.len(),
                    index.lengths.len()
                )));
            }

            let meta = layout
                .global_metadata
                .clone()
                .ok_or_else(|| TensogramError::Remote("metadata not cached".to_string()))?;

            let msg_length = layout.length;
            let mut descriptors = Vec::with_capacity(index.offsets.len());
            for i in 0..index.offsets.len() {
                let desc = self
                    .read_descriptor_only_async(
                        msg_offset,
                        msg_length,
                        index.offsets[i],
                        index.lengths[i],
                    )
                    .await?;
                descriptors.push(desc);
            }
            Ok((meta, descriptors))
        } else {
            let msg_bytes = self.read_message_async(msg_idx).await?;
            crate::decode::decode_descriptors(&msg_bytes)
        }
    }

    async fn read_descriptor_only_async(
        &self,
        msg_offset: u64,
        msg_length: u64,
        frame_offset_in_msg: u64,
        frame_length: u64,
    ) -> Result<DataObjectDescriptor> {
        const DESCRIPTOR_PREFIX_THRESHOLD: u64 = 64 * 1024;

        let range =
            Self::checked_frame_range(msg_offset, msg_length, frame_offset_in_msg, frame_length)?;

        if frame_length <= DESCRIPTOR_PREFIX_THRESHOLD {
            let frame_bytes = self.get_range_async(range.clone()).await?;
            let (desc, _payload, _mask_region, _consumed) =
                framing::decode_data_object_frame(&frame_bytes)?;
            return Ok(desc);
        }

        let frame_start = range.start;
        let frame_end = range.end;

        let header_bytes = self
            .get_range_async(frame_start..frame_start + FRAME_HEADER_SIZE as u64)
            .await?;
        let fh = FrameHeader::read_from(&header_bytes)?;

        if !fh.frame_type.is_data_object() {
            return Err(TensogramError::Remote(format!(
                "expected DataObject frame, got {:?}",
                fh.frame_type
            )));
        }

        let footer_start = frame_end - DATA_OBJECT_FOOTER_SIZE as u64;
        let footer_bytes = self.get_range_async(footer_start..frame_end).await?;

        if footer_bytes.len() < DATA_OBJECT_FOOTER_SIZE {
            return Err(TensogramError::Remote("frame footer too short".to_string()));
        }
        // v3 data-object footer: [cbor_offset u64][hash u64][ENDF 4]
        let endf_pos = footer_bytes.len() - FRAME_END.len();
        if &footer_bytes[endf_pos..] != FRAME_END {
            return Err(TensogramError::Remote(
                "frame missing ENDF trailer".to_string(),
            ));
        }

        let cbor_offset = u64::from_be_bytes(
            footer_bytes[..8]
                .try_into()
                .map_err(|_| TensogramError::Remote("footer cbor_offset truncated".to_string()))?,
        );

        if cbor_offset < FRAME_HEADER_SIZE as u64 {
            return Err(TensogramError::Remote(format!(
                "cbor_offset ({cbor_offset}) below frame header size ({FRAME_HEADER_SIZE})"
            )));
        }

        let cbor_after = fh.flags & DataObjectFlags::CBOR_AFTER_PAYLOAD != 0;
        let cbor_start = frame_start
            .checked_add(cbor_offset)
            .ok_or_else(|| TensogramError::Remote("cbor_start overflow".to_string()))?;

        if cbor_after {
            if cbor_start >= footer_start {
                return Err(TensogramError::Remote(
                    "cbor_offset points at or past footer".to_string(),
                ));
            }
            let cbor_bytes = self.get_range_async(cbor_start..footer_start).await?;
            metadata::cbor_to_object_descriptor(&cbor_bytes)
        } else {
            if cbor_start >= footer_start {
                return Err(TensogramError::Remote(
                    "cbor_offset beyond frame body".to_string(),
                ));
            }
            let max_cbor_len = footer_start - cbor_start;
            let mut prefix_size: u64 = 8192;
            loop {
                let read_end = (cbor_start + prefix_size).min(footer_start);
                let prefix_bytes = self.get_range_async(cbor_start..read_end).await?;
                match metadata::cbor_to_object_descriptor(&prefix_bytes) {
                    Ok(desc) => return Ok(desc),
                    Err(_) if prefix_size < max_cbor_len => {
                        prefix_size = (prefix_size * 2).min(max_cbor_len);
                    }
                    Err(e) => return Err(e),
                }
            }
        }
    }

    pub(crate) async fn read_object_async(
        &self,
        msg_idx: usize,
        obj_idx: usize,
        options: &DecodeOptions,
    ) -> Result<(GlobalMetadata, DataObjectDescriptor, Vec<u8>)> {
        self.ensure_layout_eager_async(msg_idx).await?;
        let layout = {
            let state = self.lock_state()?;
            state.layouts[msg_idx].clone()
        };
        let msg_offset = layout.offset;

        if let Some(ref index) = layout.index {
            Self::validate_index_access(index, obj_idx)?;

            let meta = layout
                .global_metadata
                .clone()
                .ok_or_else(|| TensogramError::Remote("metadata not cached".to_string()))?;

            let range = Self::checked_frame_range(
                msg_offset,
                layout.length,
                index.offsets[obj_idx],
                index.lengths[obj_idx],
            )?;
            let frame_bytes = self.get_range_async(range).await?;
            // See `read_object` for the rationale; same hash-
            // verification dispatch on the indexed fast path.
            let (desc, decoded) = crate::decode::decode_object_from_frame(&frame_bytes, options)
                .map_err(|e| crate::error::with_object_index(e, obj_idx))?;
            Ok((meta, desc, decoded))
        } else {
            let msg_bytes = self.read_message_async(msg_idx).await?;
            crate::decode::decode_object(&msg_bytes, obj_idx, options)
        }
    }

    pub(crate) async fn ensure_all_layouts_batch_async(&self, msg_indices: &[usize]) -> Result<()> {
        if msg_indices.is_empty() {
            return Ok(());
        }
        let max_idx = msg_indices.iter().copied().max().unwrap_or(0);
        let needs_discovery = {
            let state = self.lock_state()?;
            state.layouts.len() <= max_idx
                && !state.scan_complete()
                && state.bwd_active
                && !state.fwd_terminated
        };
        if needs_discovery {
            let _ = self.scan_pipelined_async(Some(max_idx + 1)).await?;
        }
        loop {
            let bidir = {
                let state = self.lock_state()?;
                if state.layouts.len() > max_idx || state.scan_complete() {
                    break;
                }
                state.bwd_active && !state.fwd_terminated
            };
            if bidir {
                self.scan_bidir_round_async().await?;
            } else {
                self.scan_and_discover_next_async().await?;
            }
        }

        {
            let state = self.lock_state()?;
            for &idx in msg_indices {
                if idx >= state.layouts.len() {
                    return Err(TensogramError::Framing(format!(
                        "message index {} out of range (count={})",
                        idx,
                        state.layouts.len()
                    )));
                }
            }
        }

        // Phase 2: find which messages still need layout discovery.
        let needs_layout: Vec<usize> = {
            let state = self.lock_state()?;
            msg_indices
                .iter()
                .copied()
                .filter(|&idx| {
                    state
                        .layouts
                        .get(idx)
                        .is_none_or(|l| l.global_metadata.is_none() || l.index.is_none())
                })
                .collect()
        };

        if needs_layout.is_empty() {
            return Ok(());
        }

        // Phase 3: compute header/footer byte ranges for all cold messages.
        let mut fetch_ranges: Vec<Range<u64>> = Vec::new();
        let mut fetch_map: Vec<(usize, bool)> = Vec::new(); // (msg_idx, is_header)
        {
            let state = self.lock_state()?;
            for &msg_idx in &needs_layout {
                let layout = &state.layouts[msg_idx];
                let flags = layout.preamble.flags;
                if flags.has(MessageFlags::HEADER_METADATA) && flags.has(MessageFlags::HEADER_INDEX)
                {
                    let chunk_size = layout.length.min(256 * 1024);
                    fetch_ranges.push(layout.offset..layout.offset + chunk_size);
                    fetch_map.push((msg_idx, true));
                } else if flags.has(MessageFlags::FOOTER_METADATA)
                    && flags.has(MessageFlags::FOOTER_INDEX)
                {
                    let pa_offset = layout
                        .offset
                        .checked_add(layout.length)
                        .and_then(|end| end.checked_sub(POSTAMBLE_SIZE as u64))
                        .ok_or_else(|| {
                            TensogramError::Remote("postamble offset overflow".to_string())
                        })?;
                    fetch_ranges.push(pa_offset..pa_offset + POSTAMBLE_SIZE as u64);
                    fetch_map.push((msg_idx, false));
                } else {
                    return Err(TensogramError::Remote(
                        "remote batch requires header-indexed or footer-indexed messages"
                            .to_string(),
                    ));
                }
            }
        }

        // Phase 4: batched HTTP fetch for all layout headers.
        let all_bytes = self
            .store
            .get_ranges(&self.path, &fetch_ranges)
            .await
            .map_err(|e| TensogramError::Remote(e.to_string()))?;

        // Phase 5: parse header layouts.
        let mut footer_fetches: Vec<(usize, Range<u64>)> = Vec::new();
        {
            let mut state = self.lock_state()?;
            for (bytes, &(msg_idx, is_header)) in all_bytes.iter().zip(fetch_map.iter()) {
                if state.layouts[msg_idx].global_metadata.is_some()
                    && state.layouts[msg_idx].index.is_some()
                {
                    continue;
                }
                if is_header {
                    Self::parse_header_frames(&mut state, msg_idx, bytes)?;
                } else {
                    let postamble = Postamble::read_from(bytes)?;
                    let layout = &state.layouts[msg_idx];
                    let footer_start = layout
                        .offset
                        .checked_add(postamble.first_footer_offset)
                        .ok_or_else(|| {
                            TensogramError::Remote("footer offset overflow".to_string())
                        })?;
                    let pa_offset = layout
                        .offset
                        .checked_add(layout.length)
                        .and_then(|end| end.checked_sub(POSTAMBLE_SIZE as u64))
                        .ok_or_else(|| {
                            TensogramError::Remote("postamble offset overflow".to_string())
                        })?;
                    footer_fetches.push((msg_idx, footer_start..pa_offset));
                }
            }
        }

        // Phase 6: batched footer fetch if any footer-indexed messages.
        if !footer_fetches.is_empty() {
            let footer_ranges: Vec<Range<u64>> =
                footer_fetches.iter().map(|(_, r)| r.clone()).collect();
            let footer_bytes = self
                .store
                .get_ranges(&self.path, &footer_ranges)
                .await
                .map_err(|e| TensogramError::Remote(e.to_string()))?;
            let mut state = self.lock_state()?;
            for (bytes, &(msg_idx, _)) in footer_bytes.iter().zip(footer_fetches.iter()) {
                if state.layouts[msg_idx].global_metadata.is_some()
                    && state.layouts[msg_idx].index.is_some()
                {
                    continue;
                }
                Self::parse_footer_frames(&mut state, msg_idx, bytes)?;
            }
        }

        Ok(())
    }

    pub(crate) async fn read_object_batch_async(
        &self,
        msg_indices: &[usize],
        obj_idx: usize,
        options: &DecodeOptions,
    ) -> Result<Vec<(GlobalMetadata, DataObjectDescriptor, Vec<u8>)>> {
        self.ensure_all_layouts_batch_async(msg_indices).await?;

        let mut byte_ranges = Vec::with_capacity(msg_indices.len());
        let mut metas = Vec::with_capacity(msg_indices.len());
        {
            let state = self.lock_state()?;
            for &msg_idx in msg_indices {
                let layout = &state.layouts[msg_idx];
                if let Some(ref index) = layout.index {
                    Self::validate_index_access(index, obj_idx)?;
                    byte_ranges.push(Self::checked_frame_range(
                        layout.offset,
                        layout.length,
                        index.offsets[obj_idx],
                        index.lengths[obj_idx],
                    )?);
                    metas.push(layout.global_metadata.clone().ok_or_else(|| {
                        TensogramError::Remote("metadata not cached".to_string())
                    })?);
                } else {
                    return Err(TensogramError::Remote(format!(
                        "message {} has no index frame; batch decode requires indexed messages",
                        msg_idx
                    )));
                }
            }
        }

        let all_bytes = self
            .store
            .get_ranges(&self.path, &byte_ranges)
            .await
            .map_err(|e| TensogramError::Remote(e.to_string()))?;

        let mut results = Vec::with_capacity(msg_indices.len());
        for (frame_bytes, meta) in all_bytes.iter().zip(metas) {
            // See `read_object_batch` (sync) for the rationale.
            let (desc, decoded) = crate::decode::decode_object_from_frame(frame_bytes, options)
                .map_err(|e| crate::error::with_object_index(e, obj_idx))?;
            results.push((meta, desc, decoded));
        }
        Ok(results)
    }

    pub(crate) async fn read_range_batch_async(
        &self,
        msg_indices: &[usize],
        obj_idx: usize,
        ranges: &[(u64, u64)],
        options: &DecodeOptions,
    ) -> Result<Vec<(DataObjectDescriptor, Vec<Vec<u8>>)>> {
        // 1. Batch-discover all layouts (scan + batched header/footer fetch).
        self.ensure_all_layouts_batch_async(msg_indices).await?;

        // 2. Compute byte ranges for each message's data object frame.
        let mut byte_ranges = Vec::with_capacity(msg_indices.len());
        {
            let state = self.lock_state()?;
            for &msg_idx in msg_indices {
                let layout = &state.layouts[msg_idx];
                if let Some(ref index) = layout.index {
                    Self::validate_index_access(index, obj_idx)?;
                    let range = Self::checked_frame_range(
                        layout.offset,
                        layout.length,
                        index.offsets[obj_idx],
                        index.lengths[obj_idx],
                    )?;
                    byte_ranges.push(range);
                } else {
                    return Err(TensogramError::Remote(format!(
                        "message {} has no index frame; batch range decode requires indexed messages",
                        msg_idx
                    )));
                }
            }
        }

        // 3. Single batched HTTP fetch for all frames.
        let all_bytes = self
            .store
            .get_ranges(&self.path, &byte_ranges)
            .await
            .map_err(|e| TensogramError::Remote(e.to_string()))?;

        // 4. Decode each frame locally.
        let mut results = Vec::with_capacity(msg_indices.len());
        for frame_bytes in &all_bytes {
            let (desc, payload, _mask_region, _consumed) =
                framing::decode_data_object_frame(frame_bytes)?;
            let parts = crate::decode::decode_range_from_payload(&desc, payload, ranges, options)?;
            results.push((desc, parts));
        }
        Ok(results)
    }

    pub(crate) async fn read_range_async(
        &self,
        msg_idx: usize,
        obj_idx: usize,
        ranges: &[(u64, u64)],
        options: &DecodeOptions,
    ) -> Result<(DataObjectDescriptor, Vec<Vec<u8>>)> {
        self.ensure_layout_eager_async(msg_idx).await?;
        let layout = {
            let state = self.lock_state()?;
            state.layouts[msg_idx].clone()
        };
        let msg_offset = layout.offset;

        if let Some(ref index) = layout.index {
            Self::validate_index_access(index, obj_idx)?;

            let range = Self::checked_frame_range(
                msg_offset,
                layout.length,
                index.offsets[obj_idx],
                index.lengths[obj_idx],
            )?;
            let frame_bytes = self.get_range_async(range).await?;
            let (desc, payload, _mask_region, _consumed) =
                framing::decode_data_object_frame(&frame_bytes)?;
            let parts = crate::decode::decode_range_from_payload(&desc, payload, ranges, options)?;
            Ok((desc, parts))
        } else {
            let msg_bytes = self.read_message_async(msg_idx).await?;
            crate::decode::decode_range(&msg_bytes, obj_idx, ranges, options)
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn dummy_layout(offset: u64, length: u64) -> CachedLayout {
        CachedLayout {
            offset,
            length,
            preamble: Preamble {
                version: 3,
                flags: MessageFlags::default(),
                reserved: 0,
                total_length: length,
            },
            index: None,
            global_metadata: None,
        }
    }

    #[test]
    fn scan_options_default_is_bidirectional() {
        let opts = RemoteScanOptions::default();
        assert!(
            opts.bidirectional,
            "default enables the pipelined bidirectional walker",
        );
    }

    #[test]
    fn scan_complete_truth_table_forward_only_equivalence() {
        for fwd_terminated in [false, true] {
            for gap_closed in [false, true] {
                let s = RemoteState {
                    fwd_terminated,
                    gap_closed,
                    ..RemoteState::default()
                };
                let expected = gap_closed || fwd_terminated;
                assert_eq!(
                    s.scan_complete(),
                    expected,
                    "fwd_terminated={fwd_terminated} gap_closed={gap_closed}",
                );
            }
        }
    }

    #[test]
    fn record_forward_hop_advances_cursor_and_bumps_epoch() {
        let mut s = RemoteState {
            prev_scan_offset: 1000,
            ..RemoteState::default()
        };
        let epoch_before = s.scan_epoch;
        s.record_forward_hop(dummy_layout(0, 100));
        assert_eq!(s.next_scan_offset, 100);
        assert_eq!(s.layouts.len(), 1);
        assert_ne!(s.scan_epoch, epoch_before, "epoch must bump on forward hop");
    }

    #[test]
    fn disable_backward_clears_suffix_and_disables() {
        let mut s = RemoteState {
            prev_scan_offset: 1000,
            bwd_active: true,
            ..RemoteState::default()
        };
        s.suffix_rev.push(dummy_layout(800, 200));
        s.disable_backward("test");
        assert!(!s.bwd_active);
        assert!(s.suffix_rev.is_empty());
    }

    #[test]
    fn disable_backward_is_idempotent_when_already_inactive() {
        let mut s = RemoteState::default();
        let epoch_before = s.scan_epoch;
        s.disable_backward("test");
        assert_eq!(
            s.scan_epoch, epoch_before,
            "no-op disable on already-inactive state must not bump epoch",
        );
    }

    #[test]
    fn terminate_forward_clears_provisional_backward_state() {
        let mut s = RemoteState {
            prev_scan_offset: 1000,
            bwd_active: true,
            ..RemoteState::default()
        };
        s.suffix_rev.push(dummy_layout(800, 200));
        s.terminate_forward("test");
        assert!(s.fwd_terminated);
        assert!(
            s.suffix_rev.is_empty(),
            "fwd_terminated => suffix_rev empty (bidirectional is never recovery)",
        );
        assert!(!s.bwd_active);
    }

    fn snap_at(prev: u64) -> ScanSnapshot {
        ScanSnapshot {
            next: 0,
            prev,
            epoch: 0,
        }
    }

    #[test]
    fn parse_backward_postamble_short_fetch_yields_format() {
        let buf = vec![0u8; POSTAMBLE_SIZE - 1];
        match parse_backward_postamble(&buf, &snap_at(POSTAMBLE_SIZE as u64)) {
            BackwardOutcome::Format("short-fetch-bwd") => {}
            other => panic!("expected short-fetch-bwd, got {other:?}"),
        }
    }

    #[test]
    fn parse_backward_postamble_bad_end_magic_yields_format() {
        let mut buf = vec![0u8; POSTAMBLE_SIZE];
        buf[POSTAMBLE_SIZE - crate::wire::END_MAGIC.len()..].copy_from_slice(b"NOTMAGIC");
        match parse_backward_postamble(&buf, &snap_at(POSTAMBLE_SIZE as u64)) {
            BackwardOutcome::Format("bad-end-magic-bwd") => {}
            other => panic!("expected bad-end-magic-bwd, got {other:?}"),
        }
    }

    #[test]
    fn parse_backward_postamble_arith_underflow_yields_format() {
        let mut buf = vec![0u8; POSTAMBLE_SIZE];
        buf[8..16].copy_from_slice(&u64::MAX.to_be_bytes());
        buf[POSTAMBLE_SIZE - crate::wire::END_MAGIC.len()..]
            .copy_from_slice(crate::wire::END_MAGIC);
        match parse_backward_postamble(&buf, &snap_at(100)) {
            BackwardOutcome::Format("backward-arith-underflow") => {}
            other => panic!("expected backward-arith-underflow, got {other:?}"),
        }
    }

    #[test]
    fn parse_backward_postamble_overlap_with_forward_yields_format() {
        let mut buf = vec![0u8; POSTAMBLE_SIZE];
        let total: u64 = 200;
        buf[8..16].copy_from_slice(&total.to_be_bytes());
        buf[POSTAMBLE_SIZE - crate::wire::END_MAGIC.len()..]
            .copy_from_slice(crate::wire::END_MAGIC);
        let snap = ScanSnapshot {
            next: 100,
            prev: 250,
            epoch: 0,
        };
        match parse_backward_postamble(&buf, &snap) {
            BackwardOutcome::Format("backward-overlaps-forward") => {}
            other => panic!("expected backward-overlaps-forward, got {other:?}"),
        }
    }

    #[test]
    fn parse_backward_postamble_propagates_first_footer_offset() {
        let mut buf = vec![0u8; POSTAMBLE_SIZE];
        let footer_offset: u64 = 96;
        let total: u64 = 200;
        buf[..8].copy_from_slice(&footer_offset.to_be_bytes());
        buf[8..16].copy_from_slice(&total.to_be_bytes());
        buf[POSTAMBLE_SIZE - crate::wire::END_MAGIC.len()..]
            .copy_from_slice(crate::wire::END_MAGIC);
        match parse_backward_postamble(&buf, &snap_at(200)) {
            BackwardOutcome::NeedPreambleValidation {
                msg_start,
                length,
                first_footer_offset,
            } => {
                assert_eq!(msg_start, 0);
                assert_eq!(length, 200);
                assert_eq!(first_footer_offset, footer_offset);
            }
            other => panic!("expected NeedPreambleValidation, got {other:?}"),
        }
    }

    #[test]
    fn validate_backward_preamble_short_fetch_yields_format() {
        let buf = vec![0u8; PREAMBLE_SIZE - 1];
        match validate_backward_preamble(&buf, 0, 100) {
            BackwardCommit::Format("short-fetch-bwd") => {}
            other => panic!("expected short-fetch-bwd, got {other:?}"),
        }
    }

    #[test]
    fn validate_backward_preamble_bad_magic_yields_format() {
        let buf = vec![0u8; PREAMBLE_SIZE];
        match validate_backward_preamble(&buf, 0, 100) {
            BackwardCommit::Format("bad-magic-bwd") => {}
            other => panic!("expected bad-magic-bwd, got {other:?}"),
        }
    }

    #[test]
    fn validate_backward_preamble_streaming_at_non_tail_yields_format() {
        let preamble = Preamble {
            version: crate::wire::WIRE_VERSION,
            flags: MessageFlags::default(),
            reserved: 0,
            total_length: 0,
        };
        let mut buf = Vec::with_capacity(PREAMBLE_SIZE);
        preamble.write_to(&mut buf);
        match validate_backward_preamble(&buf, 0, 100) {
            BackwardCommit::Format("streaming-preamble-non-tail") => {}
            other => panic!("expected streaming-preamble-non-tail, got {other:?}"),
        }
    }
}

#[cfg(all(test, feature = "async"))]
mod bidir_http_tests {
    use std::net::SocketAddr;
    use std::sync::Arc;
    use std::sync::atomic::{AtomicUsize, Ordering};

    use http_body_util::Full;
    use hyper::body::Bytes as HyperBytes;
    use hyper::server::conn::http1;
    use hyper::service::service_fn;
    use hyper::{Request, Response, StatusCode};
    use hyper_util::rt::TokioIo;
    use tokio::net::TcpListener;

    use super::*;
    use crate::Dtype;
    use crate::encode::{self, EncodeOptions};
    use crate::types::{ByteOrder, DataObjectDescriptor, GlobalMetadata};

    /// Tiny hyper-based mock object store.  Serves a fixed byte
    /// buffer with single-Range support, counts every request, and
    /// gives the test a port URL to feed `RemoteBackend::open*`.
    struct MockObjectStore {
        request_count: Arc<AtomicUsize>,
        range_request_count: Arc<AtomicUsize>,
        addr: SocketAddr,
    }

    impl MockObjectStore {
        async fn start(data: Vec<u8>) -> std::io::Result<Self> {
            let data = Arc::new(data);
            let request_count = Arc::new(AtomicUsize::new(0));
            let range_request_count = Arc::new(AtomicUsize::new(0));
            let listener = TcpListener::bind("127.0.0.1:0").await?;
            let addr = listener.local_addr()?;

            let data_clone = data.clone();
            let count_clone = request_count.clone();
            let range_count_clone = range_request_count.clone();
            tokio::spawn(async move {
                loop {
                    let (stream, _) = match listener.accept().await {
                        Ok(v) => v,
                        Err(_) => break,
                    };
                    let io = TokioIo::new(stream);
                    let data = data_clone.clone();
                    let count = count_clone.clone();
                    let range_count = range_count_clone.clone();
                    tokio::spawn(async move {
                        let _ = http1::Builder::new()
                            .serve_connection(
                                io,
                                service_fn(move |req: Request<hyper::body::Incoming>| {
                                    let data = data.clone();
                                    let count = count.clone();
                                    let range_count = range_count.clone();
                                    async move { handle(req, data, count, range_count) }
                                }),
                            )
                            .await;
                    });
                }
            });

            Ok(MockObjectStore {
                request_count,
                range_request_count,
                addr,
            })
        }

        fn url(&self) -> String {
            format!("http://127.0.0.1:{}/test.tgm", self.addr.port())
        }

        fn request_count(&self) -> usize {
            self.request_count.load(Ordering::SeqCst)
        }

        fn range_request_count(&self) -> usize {
            self.range_request_count.load(Ordering::SeqCst)
        }
    }

    fn handle(
        req: Request<hyper::body::Incoming>,
        data: Arc<Vec<u8>>,
        request_count: Arc<AtomicUsize>,
        range_request_count: Arc<AtomicUsize>,
    ) -> std::io::Result<Response<Full<HyperBytes>>> {
        request_count.fetch_add(1, Ordering::SeqCst);

        if req.method() == hyper::Method::HEAD {
            return Response::builder()
                .status(StatusCode::OK)
                .header("Content-Length", data.len())
                .header("Accept-Ranges", "bytes")
                .body(Full::new(HyperBytes::new()))
                .map_err(std::io::Error::other);
        }

        if let Some(range_header) = req.headers().get("Range") {
            range_request_count.fetch_add(1, Ordering::SeqCst);
            let range_str = range_header.to_str().unwrap_or("");
            if let Some((start, end_exclusive)) = parse_range(range_str, data.len()) {
                let slice = &data[start..end_exclusive];
                return Response::builder()
                    .status(StatusCode::PARTIAL_CONTENT)
                    .header(
                        "Content-Range",
                        format!("bytes {}-{}/{}", start, end_exclusive - 1, data.len()),
                    )
                    .header("Content-Length", slice.len())
                    .body(Full::new(HyperBytes::copy_from_slice(slice)))
                    .map_err(std::io::Error::other);
            }
            return Response::builder()
                .status(StatusCode::RANGE_NOT_SATISFIABLE)
                .header("Content-Range", format!("bytes */{}", data.len()))
                .body(Full::new(HyperBytes::new()))
                .map_err(std::io::Error::other);
        }

        Response::builder()
            .status(StatusCode::OK)
            .header("Content-Length", data.len())
            .body(Full::new(HyperBytes::copy_from_slice(&data)))
            .map_err(std::io::Error::other)
    }

    fn parse_range(header: &str, total: usize) -> Option<(usize, usize)> {
        let header = header.strip_prefix("bytes=")?;
        if total == 0 {
            return None;
        }
        if let Some(suffix) = header.strip_prefix('-') {
            let n: usize = suffix.parse().ok()?;
            if n == 0 {
                return None;
            }
            Some((total.saturating_sub(n), total))
        } else if let Some((start_s, end_s)) = header.split_once('-') {
            let start: usize = start_s.parse().ok()?;
            if start >= total {
                return None;
            }
            if end_s.is_empty() {
                Some((start, total))
            } else {
                let end: usize = end_s.parse().ok()?;
                if end < start {
                    return None;
                }
                Some((start, end.min(total - 1) + 1))
            }
        } else {
            None
        }
    }

    fn make_message(shape: Vec<u64>, fill: u8) -> Vec<u8> {
        let strides = if shape.is_empty() {
            vec![]
        } else {
            let mut s = vec![1u64; shape.len()];
            for i in (0..shape.len() - 1).rev() {
                s[i] = s[i + 1] * shape[i + 1];
            }
            s
        };
        let desc = DataObjectDescriptor {
            obj_type: "ntensor".to_string(),
            ndim: shape.len() as u64,
            shape: shape.clone(),
            strides,
            dtype: Dtype::Float32,
            byte_order: ByteOrder::native(),
            encoding: "none".to_string(),
            filter: "none".to_string(),
            compression: "none".to_string(),
            params: BTreeMap::new(),
            masks: None,
        };
        let num_bytes = shape.iter().product::<u64>() as usize * 4;
        let data = vec![fill; num_bytes];
        let meta = GlobalMetadata {
            extra: BTreeMap::new(),
            ..Default::default()
        };
        encode::encode(&meta, &[(&desc, &data)], &EncodeOptions::default())
            .expect("encode test message")
    }

    fn concat_messages(parts: Vec<Vec<u8>>) -> Vec<u8> {
        let total = parts.iter().map(|p| p.len()).sum();
        let mut out = Vec::with_capacity(total);
        for p in parts {
            out.extend_from_slice(&p);
        }
        out
    }

    fn empty_storage() -> BTreeMap<String, String> {
        BTreeMap::new()
    }

    fn forward_layouts(url: &str) -> Vec<(u64, u64)> {
        let backend =
            RemoteBackend::open_with_scan_opts(url, &empty_storage(), RemoteScanOptions::default())
                .expect("open forward-only");
        backend.message_count().expect("count");
        let state = backend.state.lock().expect("lock");
        state.layouts.iter().map(|l| (l.offset, l.length)).collect()
    }

    fn bidir_layouts(url: &str) -> Vec<(u64, u64)> {
        let backend = RemoteBackend::open_with_scan_opts(
            url,
            &empty_storage(),
            RemoteScanOptions {
                bidirectional: true,
            },
        )
        .expect("open bidirectional");
        backend.message_count().expect("count");
        let state = backend.state.lock().expect("lock");
        state.layouts.iter().map(|l| (l.offset, l.length)).collect()
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn bidir_one_message_no_duplicate() {
        let buf = make_message(vec![4], 42);
        let server = MockObjectStore::start(buf).await.expect("start");
        let layouts = bidir_layouts(&server.url());
        assert_eq!(
            layouts.len(),
            1,
            "1-message file must produce exactly one layout, got {layouts:?}",
        );
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn bidir_two_messages_match_forward_only() {
        let buf = concat_messages(vec![make_message(vec![4], 10), make_message(vec![8], 20)]);
        let server = MockObjectStore::start(buf).await.expect("start");
        let fwd = forward_layouts(&server.url());
        let bidir = bidir_layouts(&server.url());
        assert_eq!(
            fwd, bidir,
            "2-message file: bidirectional must match forward-only"
        );
        assert_eq!(bidir.len(), 2);
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn bidir_three_messages_odd_count_meet() {
        let buf = concat_messages(vec![
            make_message(vec![4], 10),
            make_message(vec![8], 20),
            make_message(vec![16], 30),
        ]);
        let server = MockObjectStore::start(buf).await.expect("start");
        let fwd = forward_layouts(&server.url());
        let bidir = bidir_layouts(&server.url());
        assert_eq!(
            fwd, bidir,
            "3-message file: bidirectional must match forward-only"
        );
        assert_eq!(bidir.len(), 3);
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn bidir_ten_messages_match_forward_only() {
        let parts: Vec<Vec<u8>> = (0..10)
            .map(|i| make_message(vec![4 + i as u64], i as u8))
            .collect();
        let buf = concat_messages(parts);
        let server = MockObjectStore::start(buf).await.expect("start");
        let fwd = forward_layouts(&server.url());
        let bidir = bidir_layouts(&server.url());
        assert_eq!(
            fwd, bidir,
            "10-message file: bidirectional must match forward-only"
        );
        assert_eq!(bidir.len(), 10);
    }

    /// Streaming-mode (`postamble.total_length == 0`) encoder output:
    /// backward must yield with `streaming-zero-bwd`; forward
    /// completes the scan via the existing streaming-tail branch.
    /// Validates the format-vs-transport taxonomy `Streaming` row.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn bidir_streaming_postamble_yields_backward() {
        let mut buf = make_message(vec![4], 99);
        let pa_start = buf.len() - POSTAMBLE_SIZE;
        let total_off = pa_start + 8;
        for byte in &mut buf[total_off..total_off + 8] {
            *byte = 0;
        }
        let preamble_total_off = 16;
        for byte in &mut buf[preamble_total_off..preamble_total_off + 8] {
            *byte = 0;
        }
        let server = MockObjectStore::start(buf).await.expect("start");
        let bidir = bidir_layouts(&server.url());
        assert_eq!(
            bidir.len(),
            1,
            "streaming-tail must still produce a single forward-discovered layout, got {bidir:?}",
        );
        assert!(server.request_count() > 0);
    }

    /// Corrupt `END_MAGIC` at file end: backward yields with
    /// `bad-end-magic-bwd`; forward parses both messages cleanly
    /// because non-streaming forward scanning never consults the
    /// file-end `END_MAGIC`.  Validates that bidirectional gracefully
    /// degrades to forward-only on backward-side corruption.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn bidir_corrupt_end_magic_yields_backward() {
        let mut buf = concat_messages(vec![make_message(vec![4], 10), make_message(vec![8], 20)]);
        let len = buf.len();
        buf[len - 8..].copy_from_slice(b"BADMAGIC");
        let server = MockObjectStore::start(buf).await.expect("start");
        let backend = RemoteBackend::open_with_scan_opts(
            &server.url(),
            &empty_storage(),
            RemoteScanOptions {
                bidirectional: true,
            },
        )
        .expect("open");
        let count = backend.message_count().expect("count");
        assert_eq!(
            count, 2,
            "corrupt END_MAGIC: forward path still finds both messages",
        );
        assert!(server.range_request_count() > 0);
    }

    /// Concurrent readers on a 10-message file with bidirectional
    /// enabled.  Each task fetches a different message; the mutex +
    /// snapshot-and-validate-on-reacquire protocol must serialise all
    /// state mutations so every task sees a consistent view.
    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
    async fn bidir_concurrent_readers_consistent() {
        let parts: Vec<Vec<u8>> = (0..10)
            .map(|i| make_message(vec![4 + i as u64], i as u8))
            .collect();
        let buf = concat_messages(parts);
        let server = MockObjectStore::start(buf).await.expect("start");
        let backend = Arc::new(
            RemoteBackend::open_with_scan_opts(
                &server.url(),
                &empty_storage(),
                RemoteScanOptions {
                    bidirectional: true,
                },
            )
            .expect("open"),
        );
        let mut handles = Vec::new();
        for idx in 0..10 {
            let backend = backend.clone();
            handles.push(tokio::spawn(async move {
                backend.read_message_async(idx).await
            }));
        }
        for h in handles {
            h.await.expect("join").expect("read_message_async");
        }
        let final_count = backend.message_count_async().await.expect("count");
        assert_eq!(
            final_count, 10,
            "concurrent readers must converge to 10 layouts"
        );
    }

    /// Concurrent `message_count_async` racing a target-driven discovery
    /// (e.g. `decode_metadata(idx)`).  The pipelined scanner runs off-lock,
    /// so a sibling caller's commit can drift the snapshot under it.  The
    /// scanner must surface that drift as `Ok(false)` so the caller's
    /// fallback loop drives any remaining discovery — returning `Ok(true)`
    /// on drift would expose the partial post-drift state as the final
    /// answer.
    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
    async fn bidir_message_count_race_with_partial_target() {
        let parts: Vec<Vec<u8>> = (0..20)
            .map(|i| make_message(vec![4 + i as u64], i as u8))
            .collect();
        let buf = concat_messages(parts);
        let server = MockObjectStore::start(buf).await.expect("start");
        let backend = Arc::new(
            RemoteBackend::open_with_scan_opts(
                &server.url(),
                &empty_storage(),
                RemoteScanOptions {
                    bidirectional: true,
                },
            )
            .expect("open"),
        );

        let mut handles = Vec::new();
        for _ in 0..8 {
            let backend = backend.clone();
            handles.push(tokio::spawn(
                async move { backend.read_metadata_async(5).await },
            ));
        }
        let count_handle = {
            let backend = backend.clone();
            tokio::spawn(async move { backend.message_count_async().await })
        };
        for h in handles {
            h.await.expect("join").expect("read_metadata_async");
        }
        let count = count_handle.await.expect("join").expect("count");
        assert_eq!(
            count, 20,
            "message_count_async must return the FULL count even when racing target-driven readers; \
             a partial result would mean the pipelined scanner returned Ok(true) on snapshot mismatch"
        );
    }

    /// 1-message file: backward and forward identify the same message
    /// in iteration 1.  The pipelined walker must record the forward
    /// hop and skip the duplicate backward record without bailing to
    /// the per-round fallback.  Asserts `Ok(true)` directly because
    /// `message_count_async`'s outer fallback would mask a `false`
    /// return as a (slower) success.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn pipelined_one_message_completes_without_fallback() {
        let buf = make_message(vec![4], 42);
        let server = MockObjectStore::start(buf).await.expect("start");
        let backend = RemoteBackend::open_with_scan_opts(
            &server.url(),
            &empty_storage(),
            RemoteScanOptions {
                bidirectional: true,
            },
        )
        .expect("open bidirectional");
        let pipelined_ok = backend.scan_pipelined_async(None).await.expect("pipelined");
        assert!(
            pipelined_ok,
            "1-message file: pipelined walker must complete without falling back \
             (same-message-meet must not be misread as backward-overlaps-forward)",
        );
        let state = backend.lock_state().expect("lock");
        assert_eq!(state.layouts.len(), 1);
    }

    /// Odd-count crossover (3 messages): forward commits msg 0 in
    /// round 1, then in round 2 forward commits msg 1 while backward
    /// postamble points at the same msg 1.  The pipelined walker must
    /// detect the same-message meet and complete without bailing.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn pipelined_three_messages_odd_count_meet_completes_without_fallback() {
        let buf = concat_messages(vec![
            make_message(vec![4], 10),
            make_message(vec![8], 20),
            make_message(vec![16], 30),
        ]);
        let server = MockObjectStore::start(buf).await.expect("start");
        let backend = RemoteBackend::open_with_scan_opts(
            &server.url(),
            &empty_storage(),
            RemoteScanOptions {
                bidirectional: true,
            },
        )
        .expect("open bidirectional");
        let pipelined_ok = backend.scan_pipelined_async(None).await.expect("pipelined");
        assert!(
            pipelined_ok,
            "3-message file: pipelined walker must complete without falling back \
             on the odd-count meet",
        );
        let state = backend.lock_state().expect("lock");
        assert_eq!(state.layouts.len(), 3);
    }

    /// Forge a file where the last message's postamble claims a
    /// `msg_start` strictly inside the message forward will commit
    /// next, AND splice a copy of a real preamble at that interior
    /// offset so backward-side validation would otherwise pass.
    /// Without the inline `msg_start < fwd_cursor` guard, a
    /// target-driven walk (`scan_pipelined_async(Some(2))` after
    /// `open_with_scan_opts` has auto-committed msg_a) would queue
    /// the candidate as `pending`, break out of the loop on the
    /// target check at the top of the next iteration, drain
    /// `pending` post-loop, validate it (succeeds because the
    /// planted preamble parses and its `total_length` matches the
    /// postamble's claim), and commit a fake layout that overlaps
    /// the forward-committed message.  The guard must bail to the
    /// per-round fallback, which detects the overlap via
    /// `apply_round_outcomes`' `msg_end > layout.offset` rejection
    /// and disables backward.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn pipelined_overlapping_backward_candidate_bails_to_fallback() {
        let msg_a = make_message(vec![10], 7);
        let msg_a_len = msg_a.len();
        let msg_b = make_message(vec![10], 8);
        let msg_b_len = msg_b.len();
        let mut buf = concat_messages(vec![msg_a, msg_b.clone()]);
        let file_len = buf.len() as u64;

        // Plant the fake preamble inside msg_b (which is what forward
        // will commit on the first pipelined iteration after open's
        // auto-commit of msg_a).  Use msg_b's own preamble bytes so
        // the splice has valid MAGIC + parsable version/flags.
        let fake_offset = (msg_a_len as u64) + (msg_b_len as u64) / 2;
        let claimed_total = file_len - fake_offset;
        let fake = fake_offset as usize;
        buf[fake..fake + PREAMBLE_SIZE].copy_from_slice(&msg_b[..PREAMBLE_SIZE]);
        // Patch the spliced preamble's `total_length` (offset 16 in
        // the 24-byte preamble) to the forged claim so
        // `validate_backward_preamble` accepts it.
        buf[fake + 16..fake + 16 + 8].copy_from_slice(&claimed_total.to_be_bytes());
        // Patch msg_b's postamble's `total_length` (offset 8 in the
        // 24-byte postamble) so backward parsing yields
        // `msg_start = fake_offset` with `length = claimed_total`.
        let pa_total_off = buf.len() - POSTAMBLE_SIZE + 8;
        buf[pa_total_off..pa_total_off + 8].copy_from_slice(&claimed_total.to_be_bytes());

        let server = MockObjectStore::start(buf).await.expect("start");
        let backend = RemoteBackend::open_with_scan_opts(
            &server.url(),
            &empty_storage(),
            RemoteScanOptions {
                bidirectional: true,
            },
        )
        .expect("open bidirectional");

        // open_with_scan_opts auto-commits msg_a (state.layouts.len() == 1).
        // Pass target = Some(2): the pipelined walker enters iter 1
        // (committed=1 < target=2), commits msg_b on the forward side
        // (advancing fwd_cursor to file_len), and then sees the corrupted
        // backward candidate at `fake_offset` strictly inside msg_b.
        let pipelined_ok = backend
            .scan_pipelined_async(Some(2))
            .await
            .expect("pipelined");
        assert!(
            !pipelined_ok,
            "pipelined walker MUST bail (Ok(false)) on a backward candidate that \
             overlaps the just-committed forward message; otherwise the post-loop \
             drain on a target-driven walk would commit overlapping layouts",
        );

        // Per-round fallback handles the corruption: forward commits
        // msg_a + msg_b, backward is disabled via apply_round_outcomes'
        // overlap rejection.  No fake-msg layout is committed.
        let count = backend.message_count_async().await.expect("count");
        assert_eq!(
            count, 2,
            "after fallback only msg_a and msg_b are committed"
        );
        let state = backend.lock_state().expect("lock");
        assert!(
            !state.bwd_active,
            "backward must be disabled after the per-round walker detects the overlap",
        );
    }

    /// Forge a postamble whose `total_length` is between 1 and
    /// `PREAMBLE_SIZE + POSTAMBLE_SIZE - 1`.  Backward must yield
    /// with `length-below-minimum-bwd`; forward completes the scan.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn bidir_postamble_length_below_minimum_yields_backward() {
        let mut buf = make_message(vec![4], 7);
        let pa_start = buf.len() - POSTAMBLE_SIZE;
        let total_off = pa_start + 8;
        buf[total_off..total_off + 8].copy_from_slice(&1u64.to_be_bytes());
        let server = MockObjectStore::start(buf).await.expect("start");
        let backend = RemoteBackend::open_with_scan_opts(
            &server.url(),
            &empty_storage(),
            RemoteScanOptions {
                bidirectional: true,
            },
        )
        .expect("open");
        let count = backend.message_count().expect("count");
        assert_eq!(
            count, 1,
            "below-min postamble: forward still finds the message"
        );
    }

    /// Forge a postamble.total_length that differs from the
    /// preamble.total_length but is otherwise plausible.  The
    /// candidate-preamble cross-validator must yield with
    /// `preamble-postamble-length-mismatch`; forward completes
    /// using its own (correct) preamble.total_length reading.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn bidir_preamble_postamble_length_mismatch_yields_backward() {
        let buf_a = make_message(vec![4], 11);
        let buf_b = make_message(vec![8], 22);
        let mut buf = concat_messages(vec![buf_a.clone(), buf_b.clone()]);
        // Patch the postamble at file end to claim a smaller length
        // than the preamble actually says.  Backward computes
        // msg_start from this fake length, fetches that preamble,
        // sees a mismatch with its own claim, and yields.
        let actual_len_b = buf_b.len() as u64;
        let fake_len_b = actual_len_b - 8;
        let pa_start = buf.len() - POSTAMBLE_SIZE;
        let total_off = pa_start + 8;
        buf[total_off..total_off + 8].copy_from_slice(&fake_len_b.to_be_bytes());
        let server = MockObjectStore::start(buf).await.expect("start");
        let backend = RemoteBackend::open_with_scan_opts(
            &server.url(),
            &empty_storage(),
            RemoteScanOptions {
                bidirectional: true,
            },
        )
        .expect("open");
        let count = backend.message_count().expect("count");
        assert_eq!(
            count, 2,
            "length-mismatch backward yield: forward still finds both messages",
        );
    }

    /// `[normal, streaming, normal]` synthetic file: open consumes
    /// the leading normal message via forward-only `scan_next_locked`,
    /// then bidirectional dispatch fires.  Forward sees a streaming
    /// preamble at the second message; backward discovers the third
    /// message's postamble.  The `streaming-fwd-non-tail` branch in
    /// `apply_round_outcomes` must disable backward (clearing the
    /// freshly-discovered third-message layout); subsequent
    /// forward-only steps then match the forward-only baseline
    /// exactly.  Validates "bidirectional is never recovery": a
    /// streaming preamble in the middle of the gap cannot be used
    /// to surface tail content that forward-only scanning would
    /// otherwise swallow into the streaming layout.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn bidir_streaming_in_middle_no_recovery() {
        let normal_a = make_message(vec![4], 1);
        let mut streaming = make_message(vec![8], 2);
        let preamble_total_off = 16;
        for byte in &mut streaming[preamble_total_off..preamble_total_off + 8] {
            *byte = 0;
        }
        let pa_start = streaming.len() - POSTAMBLE_SIZE;
        let total_off = pa_start + 8;
        for byte in &mut streaming[total_off..total_off + 8] {
            *byte = 0;
        }
        let normal_b = make_message(vec![16], 3);
        let buf = concat_messages(vec![normal_a, streaming, normal_b]);

        let server = MockObjectStore::start(buf).await.expect("start");

        let fwd_backend = RemoteBackend::open_with_scan_opts(
            &server.url(),
            &empty_storage(),
            RemoteScanOptions::default(),
        )
        .expect("open forward-only");
        let fwd_count = fwd_backend.message_count().expect("count");

        let bidir_backend = RemoteBackend::open_with_scan_opts(
            &server.url(),
            &empty_storage(),
            RemoteScanOptions {
                bidirectional: true,
            },
        )
        .expect("open bidirectional");
        let bidir_count = bidir_backend.message_count().expect("count");

        assert_eq!(
            bidir_count, fwd_count,
            "streaming-in-middle: bidirectional must match forward-only ({fwd_count} layouts)",
        );
    }

    fn make_streaming_message(shape: Vec<u64>, fill: u8) -> Vec<u8> {
        use crate::streaming::StreamingEncoder;

        let strides = if shape.is_empty() {
            vec![]
        } else {
            let mut s = vec![1u64; shape.len()];
            for i in (0..shape.len() - 1).rev() {
                s[i] = s[i + 1] * shape[i + 1];
            }
            s
        };
        let desc = DataObjectDescriptor {
            obj_type: "ntensor".to_string(),
            ndim: shape.len() as u64,
            shape: shape.clone(),
            strides,
            dtype: Dtype::Float32,
            byte_order: ByteOrder::native(),
            encoding: "none".to_string(),
            filter: "none".to_string(),
            compression: "none".to_string(),
            params: BTreeMap::new(),
            masks: None,
        };
        let num_bytes = shape.iter().product::<u64>() as usize * 4;
        let data = vec![fill; num_bytes];
        let meta = GlobalMetadata {
            extra: BTreeMap::new(),
            ..Default::default()
        };
        let cursor = std::io::Cursor::new(Vec::<u8>::new());
        let mut enc = StreamingEncoder::new(cursor, &meta, &EncodeOptions::default())
            .expect("streaming encoder");
        enc.write_object(&desc, &data).expect("write object");
        let cursor = enc.finish_with_backfill().expect("finish_with_backfill");
        cursor.into_inner()
    }

    fn metadata_index_populated(backend: &RemoteBackend, msg_idx: usize) -> bool {
        let state = backend.state.lock().expect("lock");
        let layout = &state.layouts[msg_idx];
        layout.global_metadata.is_some() && layout.index.is_some()
    }

    /// Footer-indexed messages discovered backward must have
    /// `global_metadata` and `index` populated inline by the
    /// eager-footer path; the lazy `discover_footer_layout` path then
    /// short-circuits.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn bidir_eager_footer_populates_metadata_index_on_footer_indexed_file() {
        let parts: Vec<Vec<u8>> = (0..6)
            .map(|i| make_streaming_message(vec![4 + i as u64], i as u8))
            .collect();
        let buf = concat_messages(parts);
        let server = MockObjectStore::start(buf).await.expect("start");
        let backend = RemoteBackend::open_with_scan_opts(
            &server.url(),
            &empty_storage(),
            RemoteScanOptions {
                bidirectional: true,
            },
        )
        .expect("open bidirectional");
        let count = backend.message_count().expect("count");
        assert_eq!(count, 6);

        // Backward-discovered messages (the suffix half) must be
        // pre-populated; otherwise the eager-footer path is not
        // firing on footer-indexed messages.
        let any_pre_populated = (0..count).any(|i| metadata_index_populated(&backend, i));
        assert!(
            any_pre_populated,
            "footer-indexed file: at least one layout must have eager-populated metadata + index",
        );
    }

    /// Header-indexed messages on backward must NOT trigger eager
    /// footer parsing — the existing test fixture set is header-
    /// indexed and the eager-footer code path is gated on the
    /// `FOOTER_METADATA | FOOTER_INDEX` flag combination.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn bidir_header_indexed_skips_eager_footer_apply() {
        let buf = concat_messages(vec![
            make_message(vec![4], 1),
            make_message(vec![8], 2),
            make_message(vec![16], 3),
            make_message(vec![32], 4),
        ]);
        let server = MockObjectStore::start(buf).await.expect("start");
        let backend = RemoteBackend::open_with_scan_opts(
            &server.url(),
            &empty_storage(),
            RemoteScanOptions {
                bidirectional: true,
            },
        )
        .expect("open bidirectional");
        let count = backend.message_count().expect("count");
        assert_eq!(count, 4);

        // Header-indexed messages stay lazy after scan — eager-footer
        // gates on the FOOTER_INDEX flag and never fires for these.
        let state = backend.state.lock().expect("lock");
        for (i, layout) in state.layouts.iter().enumerate() {
            assert!(
                layout.global_metadata.is_none(),
                "layout[{i}]: header-indexed must remain lazy, but metadata was populated",
            );
            assert!(
                layout.index.is_none(),
                "layout[{i}]: header-indexed must remain lazy, but index was populated",
            );
        }
    }

    /// Corrupted footer bytes must NOT poison preamble validation:
    /// the layout commits regardless, and the lazy path takes over
    /// for that message on subsequent metadata access.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn bidir_corrupt_footer_falls_back_to_lazy_without_poisoning_layout() {
        let mut parts: Vec<Vec<u8>> = (0..3)
            .map(|i| make_streaming_message(vec![4 + i as u64], i as u8))
            .collect();
        // Corrupt the second message's footer region.  Each part
        // begins with a preamble; the footer region for a streaming
        // message lives between `first_footer_offset` (read from the
        // postamble) and the postamble itself.  Patching the very
        // start of the footer frames is enough to break CBOR parse.
        let target = &mut parts[1];
        let pa_start = target.len() - POSTAMBLE_SIZE;
        let first_footer_offset_bytes: [u8; 8] = target[pa_start..pa_start + 8]
            .try_into()
            .expect("postamble first_footer_offset slot");
        let first_footer_offset = u64::from_be_bytes(first_footer_offset_bytes) as usize;
        for byte in &mut target[first_footer_offset..first_footer_offset + 8] {
            *byte ^= 0xFF;
        }
        let buf = concat_messages(parts);
        let server = MockObjectStore::start(buf).await.expect("start");
        let backend = RemoteBackend::open_with_scan_opts(
            &server.url(),
            &empty_storage(),
            RemoteScanOptions {
                bidirectional: true,
            },
        )
        .expect("open bidirectional");
        let count = backend.message_count().expect("count");
        assert_eq!(
            count, 3,
            "corrupt footer must not abort layout commit: all 3 messages still discovered",
        );
    }

    /// Exercise the synchronous remote read API over HTTP on a
    /// non-indexed single-object message: message / metadata / descriptor
    /// reads plus the object- and range-decode paths.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn remote_sync_read_api_roundtrip() {
        let buf = make_message(vec![4], 0x42);
        let server = MockObjectStore::start(buf).await.expect("start");
        let backend = RemoteBackend::open_with_scan_opts(
            &server.url(),
            &empty_storage(),
            RemoteScanOptions::default(),
        )
        .expect("open");

        assert_eq!(backend.message_count().expect("count"), 1);
        assert_eq!(backend.message_layouts().expect("layouts").len(), 1);

        assert!(!backend.read_message(0).expect("read_message").is_empty());
        let _meta = backend.read_metadata(0).expect("read_metadata");

        let (_m, descs) = backend.read_descriptors(0).expect("read_descriptors");
        assert_eq!(descs.len(), 1);
        assert_eq!(descs[0].dtype, Dtype::Float32);

        let opts = crate::DecodeOptions::default();
        let (_m, desc, decoded) = backend.read_object(0, 0, &opts).expect("read_object");
        assert_eq!(desc.shape, vec![4]);
        assert_eq!(decoded.len(), 4 * 4); // 4 × float32

        let (_d, parts) = backend
            .read_range(0, 0, &[(0, 4)], &opts)
            .expect("read_range");
        assert_eq!(parts.len(), 1);
        assert_eq!(parts[0].len(), 4 * 4);

        // The encoded message carries an index frame, so the indexed
        // batch fast-paths return one result per requested message.
        let obj_batch = backend
            .read_object_batch(&[0], 0, &opts)
            .expect("read_object_batch");
        assert_eq!(obj_batch.len(), 1);
        assert_eq!(obj_batch[0].2.len(), 4 * 4);
        let range_batch = backend
            .read_range_batch(&[0], 0, &[(0, 4)], &opts)
            .expect("read_range_batch");
        assert_eq!(range_batch.len(), 1);
        assert_eq!(range_batch[0].1[0].len(), 4 * 4);
    }

    /// Async mirror of the remote read API over HTTP on a non-indexed
    /// single-object message.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn remote_async_read_api_roundtrip() {
        let buf = make_message(vec![4], 0x07);
        let server = MockObjectStore::start(buf).await.expect("start");
        let backend = RemoteBackend::open_with_scan_opts(
            &server.url(),
            &empty_storage(),
            RemoteScanOptions::default(),
        )
        .expect("open");

        assert_eq!(backend.message_count_async().await.expect("count"), 1);
        assert_eq!(
            backend
                .message_layouts_async()
                .await
                .expect("layouts")
                .len(),
            1
        );
        let _meta = backend
            .read_metadata_async(0)
            .await
            .expect("read_metadata_async");
        let (_m, descs) = backend
            .read_descriptors_async(0)
            .await
            .expect("read_descriptors_async");
        assert_eq!(descs.len(), 1);

        let opts = crate::DecodeOptions::default();
        let (_m, _d, decoded) = backend
            .read_object_async(0, 0, &opts)
            .await
            .expect("read_object_async");
        assert_eq!(decoded.len(), 4 * 4);

        let (_d, parts) = backend
            .read_range_async(0, 0, &[(0, 4)], &opts)
            .await
            .expect("read_range_async");
        assert_eq!(parts[0].len(), 4 * 4);

        let obj_batch = backend
            .read_object_batch_async(&[0], 0, &opts)
            .await
            .expect("read_object_batch_async");
        assert_eq!(obj_batch.len(), 1);
        assert_eq!(obj_batch[0].2.len(), 4 * 4);
        let range_batch = backend
            .read_range_batch_async(&[0], 0, &[(0, 4)], &opts)
            .await
            .expect("read_range_batch_async");
        assert_eq!(range_batch.len(), 1);
        assert_eq!(range_batch[0].1[0].len(), 4 * 4);
    }

    /// A remote file smaller than the minimum message size is rejected
    /// up-front by the file-size guard, before any scan runs.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn remote_open_too_small_file_errors() {
        let server = MockObjectStore::start(vec![0u8; 8]).await.expect("start");
        let result = RemoteBackend::open_with_scan_opts(
            &server.url(),
            &empty_storage(),
            RemoteScanOptions::default(),
        );
        assert!(result.is_err(), "tiny remote file must be rejected");
    }

    /// Exercise the async `open_async_with_scan_opts` constructor end
    /// to end over HTTP: URL parsing, `parse_url_opts`, the async
    /// `head` size probe and the open-time forward scan.  The other
    /// in-memory tests construct the backend directly, so this is the
    /// only coverage of the async open path's network preamble.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn remote_open_async_roundtrip() {
        let buf = concat_messages(vec![make_message(vec![4], 1), make_message(vec![8], 2)]);
        let server = MockObjectStore::start(buf).await.expect("start");
        let backend = RemoteBackend::open_async_with_scan_opts(
            &server.url(),
            &empty_storage(),
            RemoteScanOptions::default(),
        )
        .await
        .expect("open_async");
        assert_eq!(backend.message_count_async().await.expect("count"), 2);
        let _ = backend.source_url();
    }

    /// Async open must reject a file smaller than the minimum message
    /// size after the `head` probe, before any scan runs.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn remote_open_async_too_small_errors() {
        let server = MockObjectStore::start(vec![0u8; 8]).await.expect("start");
        let result = RemoteBackend::open_async_with_scan_opts(
            &server.url(),
            &empty_storage(),
            RemoteScanOptions::default(),
        )
        .await;
        assert!(result.is_err(), "tiny remote file must be rejected (async)");
    }

    /// Async open of a non-`.tgm` (all-zero) buffer large enough to
    /// pass the size guard: the open-time scan finds no valid message
    /// and the constructor surfaces the "no valid messages" error.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn remote_open_async_no_valid_messages_errors() {
        let server = MockObjectStore::start(vec![0u8; 128]).await.expect("start");
        let result = RemoteBackend::open_async_with_scan_opts(
            &server.url(),
            &empty_storage(),
            RemoteScanOptions::default(),
        )
        .await;
        assert!(
            result.is_err(),
            "no valid messages must be rejected (async)"
        );
    }
}

// ── In-memory store tests (no live network) ──────────────────────────────────
//
// These tests construct `RemoteBackend` directly over an
// `object_store::memory::InMemory` store, feeding crafted (valid and
// malformed) `.tgm` bytes via the private fields the inline test
// module can access.  This drives the scan / fetch / decode I/O paths
// without the hyper HTTP server that `bidir_http_tests` and
// `tests/remote_http.rs` rely on, letting us reach the format-error,
// EOF, streaming, oversized-length, footer-discovery and large-frame
// descriptor branches by simply mutating the in-memory buffer.
#[cfg(test)]
mod inmem_tests {
    use std::collections::BTreeMap;
    use std::sync::{Arc, Mutex};

    use object_store::memory::InMemory;
    use object_store::path::Path as ObjectPath;

    use super::*;
    use crate::Dtype;
    use crate::encode::{self, EncodeOptions};
    use crate::types::{ByteOrder, DataObjectDescriptor, GlobalMetadata};

    const TEST_PATH: &str = "file.tgm";

    /// Drive an async future to completion on the crate's shared
    /// runtime.  Used by the synchronous setup helpers (e.g. `put`)
    /// from a non-async test body.
    fn run<F: std::future::Future>(fut: F) -> F::Output {
        shared_runtime().expect("shared runtime").block_on(fut)
    }

    /// Build a header-indexed single-object message via the public
    /// encoder.  Header-indexed (`HEADER_METADATA | HEADER_INDEX`) is
    /// the default `encode::encode` layout, exercising the
    /// `discover_header_layout*` paths and the indexed fast paths.
    fn make_message(shape: Vec<u64>, fill: u8) -> Vec<u8> {
        let strides = strides_for(&shape);
        let desc = DataObjectDescriptor {
            obj_type: "ntensor".to_string(),
            ndim: shape.len() as u64,
            shape: shape.clone(),
            strides,
            dtype: Dtype::Float32,
            byte_order: ByteOrder::native(),
            encoding: "none".to_string(),
            filter: "none".to_string(),
            compression: "none".to_string(),
            params: BTreeMap::new(),
            masks: None,
        };
        let num_bytes = shape.iter().product::<u64>() as usize * 4;
        let data = vec![fill; num_bytes];
        let meta = GlobalMetadata {
            extra: BTreeMap::new(),
            ..Default::default()
        };
        encode::encode(&meta, &[(&desc, &data)], &EncodeOptions::default())
            .expect("encode test message")
    }

    /// Footer-indexed message (`FOOTER_METADATA | FOOTER_INDEX`) built
    /// via the streaming encoder with backfill.  Exercises the
    /// `discover_footer_layout*` paths.
    fn make_streaming_message(shape: Vec<u64>, fill: u8) -> Vec<u8> {
        use crate::streaming::StreamingEncoder;
        let strides = strides_for(&shape);
        let desc = DataObjectDescriptor {
            obj_type: "ntensor".to_string(),
            ndim: shape.len() as u64,
            shape: shape.clone(),
            strides,
            dtype: Dtype::Float32,
            byte_order: ByteOrder::native(),
            encoding: "none".to_string(),
            filter: "none".to_string(),
            compression: "none".to_string(),
            params: BTreeMap::new(),
            masks: None,
        };
        let num_bytes = shape.iter().product::<u64>() as usize * 4;
        let data = vec![fill; num_bytes];
        let meta = GlobalMetadata {
            extra: BTreeMap::new(),
            ..Default::default()
        };
        let cursor = std::io::Cursor::new(Vec::<u8>::new());
        let mut enc = StreamingEncoder::new(cursor, &meta, &EncodeOptions::default())
            .expect("streaming encoder");
        enc.write_object(&desc, &data).expect("write object");
        let cursor = enc.finish_with_backfill().expect("finish_with_backfill");
        cursor.into_inner()
    }

    fn strides_for(shape: &[u64]) -> Vec<u64> {
        if shape.is_empty() {
            return vec![];
        }
        let mut s = vec![1u64; shape.len()];
        for i in (0..shape.len() - 1).rev() {
            s[i] = s[i + 1] * shape[i + 1];
        }
        s
    }

    fn dummy_layout(offset: u64, length: u64) -> CachedLayout {
        CachedLayout {
            offset,
            length,
            preamble: Preamble {
                version: 3,
                flags: MessageFlags::default(),
                reserved: 0,
                total_length: length,
            },
            index: None,
            global_metadata: None,
        }
    }

    fn concat(parts: Vec<Vec<u8>>) -> Vec<u8> {
        let mut out = Vec::new();
        for p in parts {
            out.extend_from_slice(&p);
        }
        out
    }

    /// Construct a `RemoteBackend` directly over an `InMemory` store
    /// holding `buf`, WITHOUT running the open-time scan.  The caller
    /// drives whichever scan/read method it wants to exercise.  This
    /// is the key to reaching malformed-input branches that
    /// `open_with_scan_opts` would reject up-front.
    fn raw_backend(buf: Vec<u8>, scan_opts: RemoteScanOptions) -> RemoteBackend {
        let store = Arc::new(InMemory::new());
        let path = ObjectPath::from(TEST_PATH);
        let file_size = buf.len() as u64;
        {
            let store = store.clone();
            let path = path.clone();
            run(async move {
                store.put(&path, buf.into()).await.expect("put");
            });
        }
        RemoteBackend {
            source_url: "memory://file.tgm".to_string(),
            store,
            path,
            file_size,
            state: Mutex::new(RemoteState {
                prev_scan_offset: file_size,
                bwd_active: scan_opts.bidirectional,
                ..RemoteState::default()
            }),
            scan_opts,
        }
    }

    /// Same as [`raw_backend`] but mirrors `open_with_scan_opts`: runs
    /// the initial forward scan so the backend is in the same state a
    /// real open would leave it.  Returns an error if the buffer is
    /// too small or has no valid messages, matching production.
    fn open_backend(buf: Vec<u8>, scan_opts: RemoteScanOptions) -> Result<RemoteBackend> {
        if (buf.len() as u64) < (PREAMBLE_SIZE + POSTAMBLE_SIZE) as u64 {
            return Err(TensogramError::Remote("too small".to_string()));
        }
        let backend = raw_backend(buf, scan_opts);
        {
            let mut state = backend.state.lock().expect("lock");
            backend.scan_next_locked(&mut state)?;
            if state.layouts.is_empty() {
                return Err(TensogramError::Remote("no valid messages".to_string()));
            }
        }
        Ok(backend)
    }

    fn forward_opts() -> RemoteScanOptions {
        RemoteScanOptions {
            bidirectional: false,
        }
    }

    fn bidir_opts() -> RemoteScanOptions {
        RemoteScanOptions {
            bidirectional: true,
        }
    }

    fn layout_offsets(backend: &RemoteBackend) -> Vec<(u64, u64)> {
        let state = backend.state.lock().expect("lock");
        state.layouts.iter().map(|l| (l.offset, l.length)).collect()
    }

    // ── Sync forward-scan termination branches ───────────────────────────

    #[test]
    fn sync_open_too_small_buffer_rejected() {
        // file_size < PREAMBLE_SIZE + POSTAMBLE_SIZE is rejected before
        // any scan runs (mirrors open_with_scan_opts guard).
        let err = open_backend(vec![0u8; 8], forward_opts());
        assert!(err.is_err(), "tiny buffer must be rejected");
    }

    #[test]
    fn sync_scan_bad_magic_terminates_forward() {
        // A buffer large enough to pass the size guard but whose first
        // bytes are not MAGIC: forward scan terminates with bad-magic
        // and produces no layouts.
        let buf = vec![0u8; 128];
        let backend = raw_backend(buf, forward_opts());
        {
            let mut state = backend.state.lock().expect("lock");
            backend.scan_next_locked(&mut state).expect("scan");
            assert!(state.layouts.is_empty(), "bad magic => no layouts");
            assert!(state.fwd_terminated, "bad magic terminates forward");
        }
    }

    #[test]
    fn sync_scan_eof_when_remaining_below_min() {
        // Valid first message, then a trailing tail < min_message_size:
        // after the first hop, the next forward step terminates at EOF.
        let msg = make_message(vec![4], 1);
        let mut buf = msg.clone();
        buf.extend_from_slice(&[0u8; 10]); // < 48-byte min, pure tail
        let backend = raw_backend(buf, forward_opts());
        backend.message_count().expect("count");
        let state = backend.state.lock().expect("lock");
        assert_eq!(state.layouts.len(), 1, "tail too small => single message");
        assert!(state.fwd_terminated);
    }

    #[test]
    fn sync_scan_preamble_parse_error_terminates() {
        // Valid MAGIC but a preamble whose version byte is invalid so
        // Preamble::read_from fails => preamble-parse-error-fwd branch.
        let mut buf = make_message(vec![4], 1);
        // Corrupt the version field (offset 8 in the 24-byte preamble:
        // [MAGIC 8][version 1]...) to an unsupported value.
        buf[8] = 0xFF;
        let backend = raw_backend(buf, forward_opts());
        {
            let mut state = backend.state.lock().expect("lock");
            backend.scan_next_locked(&mut state).expect("scan");
            assert!(state.fwd_terminated, "bad preamble terminates forward");
            assert!(state.layouts.is_empty());
        }
    }

    #[test]
    fn sync_scan_length_out_of_range_terminates() {
        // Patch the preamble.total_length (offset 16) to claim a length
        // larger than the file => length-out-of-range-fwd.
        let mut buf = make_message(vec![4], 1);
        let huge = (buf.len() as u64) + 10_000;
        buf[16..24].copy_from_slice(&huge.to_be_bytes());
        let backend = raw_backend(buf, forward_opts());
        {
            let mut state = backend.state.lock().expect("lock");
            backend.scan_next_locked(&mut state).expect("scan");
            assert!(state.fwd_terminated);
            assert!(state.layouts.is_empty());
        }
    }

    // ── Streaming (total_length == 0) forward branches ───────────────────

    /// Turn a backfilled message into a streaming one by zeroing the
    /// preamble.total_length (offset 16) and the postamble.total_length
    /// (offset 8 within the trailing 24-byte postamble).  The file-end
    /// END_MAGIC is left intact so the streaming-tail branch records a
    /// layout spanning to EOF.
    fn make_streaming_preamble(shape: Vec<u64>, fill: u8) -> Vec<u8> {
        let mut buf = make_streaming_message(shape, fill);
        buf[16..24].copy_from_slice(&0u64.to_be_bytes());
        let pa_total = buf.len() - POSTAMBLE_SIZE + 8;
        buf[pa_total..pa_total + 8].copy_from_slice(&0u64.to_be_bytes());
        buf
    }

    #[test]
    fn sync_scan_streaming_tail_recorded() {
        // A real streaming message: preamble.total_length == 0,
        // file-end END_MAGIC present.  Forward records a single
        // streaming-tail layout spanning to EOF and terminates.
        let buf = make_streaming_preamble(vec![4], 7);
        let backend = raw_backend(buf.clone(), forward_opts());
        backend.message_count().expect("count");
        let layouts = layout_offsets(&backend);
        assert_eq!(layouts.len(), 1, "streaming tail => single layout");
        assert_eq!(layouts[0].0, 0);
        assert_eq!(layouts[0].1, buf.len() as u64, "tail spans to EOF");
    }

    #[test]
    fn sync_scan_streaming_end_magic_mismatch_terminates() {
        // Streaming preamble but corrupt file-end END_MAGIC: the
        // streaming-end-magic-mismatch branch terminates without
        // recording a layout.
        let mut buf = make_streaming_preamble(vec![4], 7);
        let len = buf.len();
        buf[len - 8..].copy_from_slice(b"BADMAGIC");
        let backend = raw_backend(buf, forward_opts());
        {
            let mut state = backend.state.lock().expect("lock");
            backend.scan_next_locked(&mut state).expect("scan");
            assert!(state.fwd_terminated);
            assert!(state.layouts.is_empty(), "mismatch => no layout");
        }
    }

    // ── scan_and_discover_next_locked (eager forward + parse) ────────────

    #[test]
    fn sync_eager_discover_bad_magic_terminates() {
        // ensure_layout_eager_locked routes forward-only discovery
        // through scan_and_discover_next_locked.  Feed a buffer whose
        // second "message" region has no MAGIC so the eager discover
        // terminates after the first message.
        let good = make_message(vec![4], 1);
        let mut buf = good.clone();
        // append junk >= min message size so the size guard passes but
        // MAGIC check fails on the second hop.
        buf.extend_from_slice(&[0u8; 64]);
        let backend = raw_backend(buf, forward_opts());
        // read_metadata(0) drives ensure_layout_eager_locked ->
        // scan_and_discover_next_locked for msg 0 (header-indexed).
        let meta = backend.read_metadata(0).expect("metadata");
        let _ = meta;
        let count = backend.message_count().expect("count");
        assert_eq!(count, 1, "junk after msg 0 => only one message");
    }

    #[test]
    fn sync_eager_discover_streaming_preamble_branch() {
        // Forward-only eager discovery (scan_and_discover_next_locked)
        // over a streaming-preamble (total_length == 0) message: the
        // streaming-tail layout spanning to EOF is recorded via the
        // streaming branch.  The body still carries footer frames
        // (the buffer was a real backfilled message before zeroing the
        // total_length), so the subsequent lazy footer discovery
        // populates metadata + index.
        let buf = make_streaming_preamble(vec![4], 5);
        let backend = raw_backend(buf.clone(), forward_opts());
        let _meta = backend
            .read_metadata(0)
            .expect("footer frames still recoverable on streaming tail");
        let state = backend.state.lock().expect("lock");
        assert_eq!(state.layouts.len(), 1, "streaming tail recorded");
        assert_eq!(
            state.layouts[0].length,
            buf.len() as u64,
            "streaming tail spans to EOF",
        );
    }

    #[test]
    fn sync_eager_discover_streaming_recorded() {
        // Footer-indexed streaming message read via the eager-discover
        // forward path: read_metadata must populate metadata + index
        // through discover_footer_layout_from_suffix_locked.
        let buf = make_streaming_message(vec![6], 3);
        let backend = open_backend(buf, forward_opts()).expect("open");
        let meta = backend.read_metadata(0).expect("metadata");
        let _ = meta;
        let (_m, descs) = backend.read_descriptors(0).expect("descriptors");
        assert_eq!(descs.len(), 1);
        let state = backend.state.lock().expect("lock");
        assert!(state.layouts[0].global_metadata.is_some());
        assert!(state.layouts[0].index.is_some());
    }

    #[test]
    fn sync_eager_discover_footer_indexed_second_message() {
        // Forward-only eager discovery of a footer-indexed message at
        // index 1 routes through scan_and_discover_next_locked ->
        // discover_footer_layout_from_suffix_locked.
        let buf = concat(vec![
            make_streaming_message(vec![4], 1),
            make_streaming_message(vec![8], 2),
        ]);
        let backend = open_backend(buf, forward_opts()).expect("open");
        let _meta = backend.read_metadata(1).expect("metadata for message 1");
        let (_m, descs) = backend.read_descriptors(1).expect("descriptors");
        assert_eq!(descs.len(), 1);
        let state = backend.state.lock().expect("lock");
        assert!(state.layouts[1].global_metadata.is_some());
        assert!(state.layouts[1].index.is_some());
    }

    #[test]
    fn sync_eager_discover_preamble_parse_error_at_second_message() {
        // Eager forward discovery of message 1 hits a corrupt preamble
        // (bad version) in scan_and_discover_next_locked => terminate;
        // requesting message 1 then errors out of range.
        let good = make_message(vec![4], 1);
        let good_len = good.len();
        let mut buf = concat(vec![good, make_message(vec![8], 2)]);
        buf[good_len + 8] = 0xFF; // corrupt version of 2nd preamble
        let backend = open_backend(buf, forward_opts()).expect("open");
        let err = backend.read_metadata(1);
        assert!(err.is_err(), "corrupt 2nd preamble => message 1 missing");
        assert_eq!(backend.message_count().expect("count"), 1);
    }

    #[test]
    fn sync_eager_discover_length_out_of_range_at_second_message() {
        // Eager forward discovery of message 1 reads a preamble whose
        // total_length overruns the file => length-out-of-range-fwd.
        let good = make_message(vec![4], 1);
        let good_len = good.len();
        let second = make_message(vec![8], 2);
        let mut buf = concat(vec![good.clone(), second]);
        let huge = (buf.len() as u64) + 50_000;
        buf[good_len + 16..good_len + 24].copy_from_slice(&huge.to_be_bytes());
        let backend = open_backend(buf, forward_opts()).expect("open");
        let err = backend.read_metadata(1);
        assert!(err.is_err(), "overrun 2nd length => message 1 missing");
        assert_eq!(backend.message_count().expect("count"), 1);
    }

    #[test]
    fn sync_debug_format_includes_message_count() {
        // Exercise the RemoteBackend Debug impl (locks state, reads
        // layouts.len()).
        let buf = make_message(vec![4], 1);
        let backend = open_backend(buf, forward_opts()).expect("open");
        backend.message_count().expect("count");
        let dbg = format!("{backend:?}");
        assert!(dbg.contains("RemoteBackend"), "debug includes type name");
        assert!(dbg.contains("messages"), "debug includes message count");
    }

    // ── Sync read API: indexed multi-message batch paths ─────────────────

    #[test]
    fn sync_read_object_and_range_batch_multi_message() {
        // Two header-indexed messages; batch APIs must return one
        // result per requested message and decode each frame.
        let buf = concat(vec![make_message(vec![4], 10), make_message(vec![8], 20)]);
        let backend = open_backend(buf, forward_opts()).expect("open");
        assert_eq!(backend.message_count().expect("count"), 2);
        let opts = crate::DecodeOptions::default();

        let obj_batch = backend
            .read_object_batch(&[0, 1], 0, &opts)
            .expect("read_object_batch");
        assert_eq!(obj_batch.len(), 2);
        assert_eq!(obj_batch[0].2.len(), 4 * 4);
        assert_eq!(obj_batch[1].2.len(), 8 * 4);

        let range_batch = backend
            .read_range_batch(&[0, 1], 0, &[(0, 4)], &opts)
            .expect("read_range_batch");
        assert_eq!(range_batch.len(), 2);
        assert_eq!(range_batch[0].1[0].len(), 4 * 4);
    }

    #[test]
    fn sync_read_object_out_of_range_obj_idx_errors() {
        // validate_index_access: obj_idx beyond the single object.
        let buf = make_message(vec![4], 1);
        let backend = open_backend(buf, forward_opts()).expect("open");
        let opts = crate::DecodeOptions::default();
        let err = backend.read_object(0, 5, &opts);
        assert!(err.is_err(), "obj idx out of range must error");
    }

    #[test]
    fn sync_read_message_index_out_of_range_errors() {
        let buf = make_message(vec![4], 1);
        let backend = open_backend(buf, forward_opts()).expect("open");
        let err = backend.read_message(99);
        assert!(err.is_err(), "msg idx out of range must error");
    }

    #[test]
    fn sync_read_descriptors_large_frame_prefix_path() {
        // A data object whose frame exceeds DESCRIPTOR_PREFIX_THRESHOLD
        // (64 KiB) forces read_descriptor_only down the header/footer/
        // CBOR-prefix path instead of the small-frame fast path.
        let n = 20_000u64; // 20k float32 = 80 KB payload > 64 KiB
        let buf = make_message(vec![n], 9);
        let backend = open_backend(buf, forward_opts()).expect("open");
        let (_m, descs) = backend
            .read_descriptors(0)
            .expect("read_descriptors large frame");
        assert_eq!(descs.len(), 1);
        assert_eq!(descs[0].shape, vec![n]);
    }

    fn one_desc(shape: Vec<u64>) -> DataObjectDescriptor {
        DataObjectDescriptor {
            obj_type: "ntensor".to_string(),
            ndim: shape.len() as u64,
            shape: shape.clone(),
            strides: strides_for(&shape),
            dtype: Dtype::Float32,
            byte_order: ByteOrder::native(),
            encoding: "none".to_string(),
            filter: "none".to_string(),
            compression: "none".to_string(),
            params: BTreeMap::new(),
            masks: None,
        }
    }

    /// Build a backend whose buffer is exactly one hand-built data
    /// object frame stored at offset 0.  The buffer is NOT a valid
    /// `.tgm` message, but `read_descriptor_only` operates purely on
    /// frame offsets, so we can drive it directly to exercise the
    /// large-frame header/footer/CBOR-prefix branches.
    fn frame_backend(frame: Vec<u8>) -> RemoteBackend {
        let store = Arc::new(InMemory::new());
        let path = ObjectPath::from(TEST_PATH);
        let file_size = frame.len() as u64;
        {
            let store = store.clone();
            let path = path.clone();
            run(async move {
                store.put(&path, frame.into()).await.expect("put");
            });
        }
        RemoteBackend {
            source_url: "memory://frame.tgm".to_string(),
            store,
            path,
            file_size,
            state: Mutex::new(RemoteState {
                prev_scan_offset: file_size,
                ..RemoteState::default()
            }),
            scan_opts: forward_opts(),
        }
    }

    #[test]
    fn read_descriptor_only_large_cbor_before_prefix_loop() {
        // cbor_before = true places the CBOR descriptor right after the
        // 16-byte header; a frame > 64 KiB drives read_descriptor_only
        // down the `else` (non-cbor-after) prefix-doubling loop.
        let desc = one_desc(vec![20_000]);
        let payload = vec![0u8; 80_000];
        let frame = framing::encode_data_object_frame(&desc, &payload, true, false)
            .expect("cbor-before frame");
        let frame_len = frame.len() as u64;
        let backend = frame_backend(frame);
        let got = backend
            .read_descriptor_only(0, frame_len, 0, frame_len)
            .expect("descriptor via prefix loop");
        assert_eq!(got.shape, vec![20_000]);
    }

    #[test]
    fn read_descriptor_only_large_cbor_after() {
        // cbor_before = false (default): the > 64 KiB branch reads the
        // CBOR region between cbor_start and footer_start in one shot.
        let desc = one_desc(vec![20_000]);
        let payload = vec![0u8; 80_000];
        let frame = framing::encode_data_object_frame(&desc, &payload, false, false)
            .expect("cbor-after frame");
        let frame_len = frame.len() as u64;
        let backend = frame_backend(frame);
        let got = backend
            .read_descriptor_only(0, frame_len, 0, frame_len)
            .expect("descriptor cbor-after");
        assert_eq!(got.shape, vec![20_000]);
    }

    #[test]
    fn read_descriptor_only_large_non_data_object_frame_errors() {
        // A > 64 KiB frame whose type is not a data object must be
        // rejected at the `is_data_object()` guard.
        let desc = one_desc(vec![20_000]);
        let payload = vec![0u8; 80_000];
        let mut frame =
            framing::encode_data_object_frame(&desc, &payload, false, false).expect("frame");
        // Frame header byte layout: [b"FR"][type u8]... Patch the frame
        // type to a non-data-object value (HeaderMetadata).
        frame[2] = FrameType::HeaderMetadata as u8;
        let frame_len = frame.len() as u64;
        let backend = frame_backend(frame);
        let err = backend.read_descriptor_only(0, frame_len, 0, frame_len);
        assert!(err.is_err(), "non-data-object large frame must error");
    }

    #[test]
    fn read_descriptor_only_large_cbor_offset_below_header_errors() {
        // Patch the footer cbor_offset to 0 (< FRAME_HEADER_SIZE) on a
        // > 64 KiB frame: the "below frame header size" guard fires.
        let desc = one_desc(vec![20_000]);
        let payload = vec![0u8; 80_000];
        let mut frame =
            framing::encode_data_object_frame(&desc, &payload, false, false).expect("frame");
        let len = frame.len();
        // cbor_offset = first 8 bytes of the 20-byte data-object footer.
        let cbor_off_pos = len - DATA_OBJECT_FOOTER_SIZE;
        frame[cbor_off_pos..cbor_off_pos + 8].copy_from_slice(&0u64.to_be_bytes());
        let frame_len = frame.len() as u64;
        let backend = frame_backend(frame);
        let err = backend.read_descriptor_only(0, frame_len, 0, frame_len);
        assert!(err.is_err(), "cbor_offset below header size must error");
    }

    #[test]
    fn read_descriptor_only_large_cbor_offset_past_footer_errors() {
        // Patch cbor_offset to land at/after the footer start: the
        // "cbor_offset points at or past footer" guard fires (cbor_after).
        let desc = one_desc(vec![20_000]);
        let payload = vec![0u8; 80_000];
        let mut frame =
            framing::encode_data_object_frame(&desc, &payload, false, false).expect("frame");
        let len = frame.len();
        let cbor_off_pos = len - DATA_OBJECT_FOOTER_SIZE;
        // frame_length covers the whole frame; setting cbor_offset to
        // (frame_len - footer) puts cbor_start exactly at footer_start.
        let footer_rel = (len - DATA_OBJECT_FOOTER_SIZE) as u64;
        frame[cbor_off_pos..cbor_off_pos + 8].copy_from_slice(&footer_rel.to_be_bytes());
        let frame_len = frame.len() as u64;
        let backend = frame_backend(frame);
        let err = backend.read_descriptor_only(0, frame_len, 0, frame_len);
        assert!(err.is_err(), "cbor_offset at/past footer must error");
    }

    #[test]
    fn read_descriptor_only_large_bad_endf_errors() {
        // Corrupt the ENDF trailer of a > 64 KiB frame: the footer
        // ENDF check fires.
        let desc = one_desc(vec![20_000]);
        let payload = vec![0u8; 80_000];
        let mut frame =
            framing::encode_data_object_frame(&desc, &payload, false, false).expect("frame");
        let len = frame.len();
        frame[len - 4..].copy_from_slice(b"XXXX");
        let frame_len = frame.len() as u64;
        let backend = frame_backend(frame);
        let err = backend.read_descriptor_only(0, frame_len, 0, frame_len);
        assert!(err.is_err(), "bad ENDF on large frame must error");
    }

    // ── parse_header_frames / parse_footer_frames(_into) units ───────────

    /// Build a single frame: `[FrameHeader 16][payload][hash 8][ENDF 4]`.
    /// `total_length` covers header..end-of-ENDF.
    fn build_frame(frame_type: FrameType, payload: &[u8]) -> Vec<u8> {
        let total = FRAME_HEADER_SIZE + payload.len() + FRAME_COMMON_FOOTER_SIZE;
        let fh = FrameHeader {
            frame_type,
            version: 1,
            flags: 0,
            total_length: total as u64,
        };
        let mut out = Vec::with_capacity(total);
        fh.write_to(&mut out);
        out.extend_from_slice(payload);
        out.extend_from_slice(&[0u8; 8]); // hash slot
        out.extend_from_slice(FRAME_END);
        out
    }

    #[test]
    fn parse_footer_frames_into_frame_too_small_errors() {
        // A frame whose declared total_length is below the minimum
        // (header + ENDF) trips the "smaller than minimum" guard.
        let mut buf = build_frame(FrameType::FooterMetadata, &[1, 2, 3, 4]);
        // Patch total_length (offset 8 in the 16-byte frame header) to 1.
        buf[8..16].copy_from_slice(&1u64.to_be_bytes());
        let err = RemoteBackend::parse_footer_frames_into(&buf);
        assert!(err.is_err(), "below-minimum frame total_length must error");
    }

    #[test]
    fn parse_footer_frames_into_missing_endf_errors() {
        // Corrupt the ENDF trailer: the frame-end check fires.
        let mut buf = build_frame(FrameType::FooterMetadata, &[1, 2, 3, 4]);
        let len = buf.len();
        buf[len - 4..].copy_from_slice(b"XXXX");
        let err = RemoteBackend::parse_footer_frames_into(&buf);
        assert!(err.is_err(), "missing ENDF trailer must error");
    }

    #[test]
    fn parse_footer_frames_into_skips_non_footer_frames() {
        // A frame whose type is neither FooterMetadata nor FooterIndex
        // is skipped without populating either option.
        let buf = build_frame(FrameType::HeaderMetadata, &[9, 9, 9, 9]);
        let (metadata, index) = RemoteBackend::parse_footer_frames_into(&buf).expect("parse ok");
        assert!(metadata.is_none() && index.is_none(), "non-footer skipped");
    }

    #[test]
    fn parse_footer_frames_missing_metadata_errors() {
        // parse_footer_frames requires BOTH metadata and index frames;
        // an empty buffer yields neither and errors.
        let mut state = RemoteState::default();
        state.layouts.push(dummy_layout(0, 100));
        let err = RemoteBackend::parse_footer_frames(&mut state, 0, &[]);
        assert!(
            err.is_err(),
            "footer region without a metadata frame must error",
        );
    }

    #[test]
    fn parse_header_frames_missing_metadata_errors() {
        // No FR magic at all in the buffer: parse_header_frames finds
        // no metadata frame and errors.
        let mut state = RemoteState::default();
        state.layouts.push(dummy_layout(0, 100));
        let buf = vec![0u8; PREAMBLE_SIZE + 32];
        let err = RemoteBackend::parse_header_frames(&mut state, 0, &buf);
        assert!(err.is_err(), "header region without metadata must error");
    }

    fn valid_metadata_cbor() -> Vec<u8> {
        let meta = GlobalMetadata {
            extra: BTreeMap::new(),
            ..Default::default()
        };
        crate::metadata::global_metadata_to_cbor(&meta).expect("meta cbor")
    }

    #[test]
    fn parse_header_frames_missing_index_errors() {
        // A header region holding only a metadata frame (no index)
        // populates metadata but errors on the missing index frame.
        let meta_frame = build_frame(FrameType::HeaderMetadata, &valid_metadata_cbor());
        let mut buf = vec![0u8; PREAMBLE_SIZE];
        buf.extend_from_slice(&meta_frame);
        let mut state = RemoteState::default();
        state.layouts.push(dummy_layout(0, buf.len() as u64));
        let err = RemoteBackend::parse_header_frames(&mut state, 0, &buf);
        assert!(err.is_err(), "header region without an index must error");
        // metadata WAS populated before the index check failed.
        assert!(state.layouts[0].global_metadata.is_some());
    }

    #[test]
    fn parse_footer_frames_missing_index_errors() {
        // A footer region holding only a metadata frame (no index)
        // populates metadata but errors on the missing index frame.
        let meta_frame = build_frame(FrameType::FooterMetadata, &valid_metadata_cbor());
        let mut state = RemoteState::default();
        state.layouts.push(dummy_layout(0, 100));
        let err = RemoteBackend::parse_footer_frames(&mut state, 0, &meta_frame);
        assert!(err.is_err(), "footer region without an index must error");
        assert!(state.layouts[0].global_metadata.is_some());
    }

    // ── checked_frame_range / validate_index_access units ────────────────

    #[test]
    fn checked_frame_range_overflow_rejected() {
        // frame_offset + frame_length overflows u64.
        let err = RemoteBackend::checked_frame_range(0, 100, u64::MAX, 10);
        assert!(err.is_err(), "frame offset overflow must error");
        // frame end exceeds message boundary.
        let err = RemoteBackend::checked_frame_range(0, 100, 50, 100);
        assert!(err.is_err(), "frame beyond message boundary must error");
        // valid range.
        let ok = RemoteBackend::checked_frame_range(10, 100, 20, 30).expect("valid");
        assert_eq!(ok, 30..60);
    }

    #[test]
    fn validate_index_access_mismatched_lengths_errors() {
        let index = IndexFrame {
            offsets: vec![0, 100],
            lengths: vec![50], // mismatched
        };
        let err = RemoteBackend::validate_index_access(&index, 0);
        assert!(err.is_err(), "mismatched offsets/lengths must error");
    }

    /// Inject a corrupt index (offsets/lengths length mismatch) plus a
    /// dummy metadata so `ensure_layout_eager_locked` short-circuits,
    /// then confirm `read_descriptors` surfaces the corrupt-index error
    /// before any frame fetch.
    #[test]
    fn sync_read_descriptors_corrupt_index_errors() {
        let buf = make_message(vec![4], 1);
        let backend = open_backend(buf, forward_opts()).expect("open");
        {
            let mut state = backend.state.lock().expect("lock");
            state.layouts[0].global_metadata = Some(GlobalMetadata::default());
            state.layouts[0].index = Some(IndexFrame {
                offsets: vec![0, 100],
                lengths: vec![50],
            });
        }
        let err = backend.read_descriptors(0);
        assert!(err.is_err(), "corrupt index must error in read_descriptors");
    }

    // ── Bidirectional scan over InMemory ─────────────────────────────────

    #[test]
    fn sync_bidir_matches_forward_only_multi_message() {
        let parts: Vec<Vec<u8>> = (0..7)
            .map(|i| make_message(vec![4 + i as u64], i as u8))
            .collect();
        let buf = concat(parts);
        let fwd = open_backend(buf.clone(), forward_opts()).expect("open fwd");
        fwd.message_count().expect("count");
        let bidir = open_backend(buf, bidir_opts()).expect("open bidir");
        bidir.message_count().expect("count");
        assert_eq!(
            layout_offsets(&fwd),
            layout_offsets(&bidir),
            "bidirectional must match forward-only layouts",
        );
        assert_eq!(layout_offsets(&bidir).len(), 7);
    }

    #[test]
    fn sync_bidir_one_message_no_duplicate() {
        let buf = make_message(vec![4], 42);
        let backend = open_backend(buf, bidir_opts()).expect("open");
        let count = backend.message_count().expect("count");
        assert_eq!(count, 1, "1-message bidir must not duplicate");
    }

    #[test]
    fn sync_bidir_corrupt_postamble_falls_back_to_forward() {
        // Corrupt the file-end postamble's END_MAGIC: backward yields
        // bad-end-magic-bwd, forward still finds both messages.
        let mut buf = concat(vec![make_message(vec![4], 1), make_message(vec![8], 2)]);
        let len = buf.len();
        buf[len - 8..].copy_from_slice(b"BADMAGIC");
        let backend = open_backend(buf, bidir_opts()).expect("open");
        let count = backend.message_count().expect("count");
        assert_eq!(count, 2, "backward corruption => forward still scans");
    }

    #[test]
    fn sync_bidir_footer_indexed_eager_populates() {
        // Footer-indexed multi-message file: backward eager-footer path
        // pre-populates metadata + index on at least one layout.
        let parts: Vec<Vec<u8>> = (0..6)
            .map(|i| make_streaming_message(vec![4 + i as u64], i as u8))
            .collect();
        let buf = concat(parts);
        let backend = open_backend(buf, bidir_opts()).expect("open");
        let count = backend.message_count().expect("count");
        assert_eq!(count, 6);
        let state = backend.state.lock().expect("lock");
        let any = (0..6).any(|i| {
            state.layouts[i].global_metadata.is_some() && state.layouts[i].index.is_some()
        });
        assert!(any, "eager footer must populate at least one layout");
    }

    // ── Footer discovery error branches (sync) ───────────────────────────

    #[test]
    fn sync_footer_discovery_first_footer_offset_too_small_errors() {
        // Footer-indexed message but patch the postamble's
        // first_footer_offset to a value < PREAMBLE_SIZE so
        // discover_footer_layout_locked rejects it.
        let mut buf = make_streaming_message(vec![6], 3);
        let pa_start = buf.len() - POSTAMBLE_SIZE;
        // first_footer_offset is the first u64 of the postamble.
        buf[pa_start..pa_start + 8].copy_from_slice(&1u64.to_be_bytes());
        let backend = open_backend(buf, forward_opts()).expect("open");
        // Force the lazy (non-eager) footer discovery by clearing any
        // eager population — open in forward-only never eager-populates
        // footer, so read_metadata drives discover_footer_layout.
        let err = backend.read_metadata(0);
        assert!(
            err.is_err(),
            "first_footer_offset below preamble end must error",
        );
    }

    #[test]
    fn sync_footer_discovery_first_footer_offset_past_postamble_errors() {
        // Patch first_footer_offset to a value that lands at/after the
        // postamble (footer_abs_start >= footer_abs_end), tripping the
        // "points at or past postamble" guard.
        let mut buf = make_streaming_message(vec![6], 3);
        let msg_len = buf.len() as u64;
        let pa_start = buf.len() - POSTAMBLE_SIZE;
        // first_footer_offset == msg_len puts footer_abs_start at EOF,
        // well past footer_abs_end (= msg_end - POSTAMBLE_SIZE).
        buf[pa_start..pa_start + 8].copy_from_slice(&msg_len.to_be_bytes());
        let backend = open_backend(buf, forward_opts()).expect("open");
        let err = backend.read_metadata(0);
        assert!(
            err.is_err(),
            "first_footer_offset past postamble must error",
        );
    }
}

// ── Async in-memory store tests (no live network) ────────────────────────────
//
// Mirror of `inmem_tests` for the native async path
// (`#[cfg(feature = "async")]`).  Drives the async scan / fetch /
// decode methods (`scan_*_async`, `read_*_async`, `*_batch_async`,
// `ensure_all_layouts_batch_async`, `discover_*_async`,
// `read_descriptor_only_async`) over a crafted `InMemory` buffer.
#[cfg(all(test, feature = "async"))]
mod inmem_async_tests {
    use std::collections::BTreeMap;
    use std::sync::{Arc, Mutex};

    use object_store::memory::InMemory;
    use object_store::path::Path as ObjectPath;

    use super::*;
    use crate::Dtype;
    use crate::encode::{self, EncodeOptions};
    use crate::types::{ByteOrder, DataObjectDescriptor, GlobalMetadata};

    const TEST_PATH: &str = "file.tgm";

    fn strides_for(shape: &[u64]) -> Vec<u64> {
        if shape.is_empty() {
            return vec![];
        }
        let mut s = vec![1u64; shape.len()];
        for i in (0..shape.len() - 1).rev() {
            s[i] = s[i + 1] * shape[i + 1];
        }
        s
    }

    fn make_message(shape: Vec<u64>, fill: u8) -> Vec<u8> {
        let desc = DataObjectDescriptor {
            obj_type: "ntensor".to_string(),
            ndim: shape.len() as u64,
            shape: shape.clone(),
            strides: strides_for(&shape),
            dtype: Dtype::Float32,
            byte_order: ByteOrder::native(),
            encoding: "none".to_string(),
            filter: "none".to_string(),
            compression: "none".to_string(),
            params: BTreeMap::new(),
            masks: None,
        };
        let num_bytes = shape.iter().product::<u64>() as usize * 4;
        let data = vec![fill; num_bytes];
        let meta = GlobalMetadata {
            extra: BTreeMap::new(),
            ..Default::default()
        };
        encode::encode(&meta, &[(&desc, &data)], &EncodeOptions::default())
            .expect("encode test message")
    }

    fn make_streaming_message(shape: Vec<u64>, fill: u8) -> Vec<u8> {
        use crate::streaming::StreamingEncoder;
        let desc = DataObjectDescriptor {
            obj_type: "ntensor".to_string(),
            ndim: shape.len() as u64,
            shape: shape.clone(),
            strides: strides_for(&shape),
            dtype: Dtype::Float32,
            byte_order: ByteOrder::native(),
            encoding: "none".to_string(),
            filter: "none".to_string(),
            compression: "none".to_string(),
            params: BTreeMap::new(),
            masks: None,
        };
        let num_bytes = shape.iter().product::<u64>() as usize * 4;
        let data = vec![fill; num_bytes];
        let meta = GlobalMetadata {
            extra: BTreeMap::new(),
            ..Default::default()
        };
        let cursor = std::io::Cursor::new(Vec::<u8>::new());
        let mut enc = StreamingEncoder::new(cursor, &meta, &EncodeOptions::default())
            .expect("streaming encoder");
        enc.write_object(&desc, &data).expect("write object");
        let cursor = enc.finish_with_backfill().expect("finish_with_backfill");
        cursor.into_inner()
    }

    fn concat(parts: Vec<Vec<u8>>) -> Vec<u8> {
        let mut out = Vec::new();
        for p in parts {
            out.extend_from_slice(&p);
        }
        out
    }

    async fn raw_backend(buf: Vec<u8>, scan_opts: RemoteScanOptions) -> RemoteBackend {
        let store = Arc::new(InMemory::new());
        let path = ObjectPath::from(TEST_PATH);
        let file_size = buf.len() as u64;
        store.put(&path, buf.into()).await.expect("put");
        RemoteBackend {
            source_url: "memory://file.tgm".to_string(),
            store,
            path,
            file_size,
            state: Mutex::new(RemoteState {
                prev_scan_offset: file_size,
                bwd_active: scan_opts.bidirectional,
                ..RemoteState::default()
            }),
            scan_opts,
        }
    }

    /// Build a backend and run the async open-time forward scan so it
    /// matches the state `open_async_with_scan_opts` would leave.
    async fn open_backend(buf: Vec<u8>, scan_opts: RemoteScanOptions) -> Result<RemoteBackend> {
        if (buf.len() as u64) < (PREAMBLE_SIZE + POSTAMBLE_SIZE) as u64 {
            return Err(TensogramError::Remote("too small".to_string()));
        }
        let backend = raw_backend(buf, scan_opts).await;
        backend.scan_next_async().await?;
        {
            let state = backend.lock_state()?;
            if state.layouts.is_empty() {
                return Err(TensogramError::Remote("no valid messages".to_string()));
            }
        }
        Ok(backend)
    }

    fn forward_opts() -> RemoteScanOptions {
        RemoteScanOptions {
            bidirectional: false,
        }
    }

    fn bidir_opts() -> RemoteScanOptions {
        RemoteScanOptions {
            bidirectional: true,
        }
    }

    // ── Async forward-scan termination branches ──────────────────────────

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn async_scan_bad_magic_terminates() {
        let backend = raw_backend(vec![0u8; 128], forward_opts()).await;
        backend.scan_next_async().await.expect("scan");
        let state = backend.lock_state().expect("lock");
        assert!(state.fwd_terminated, "bad magic terminates forward");
        assert!(state.layouts.is_empty());
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn async_scan_preamble_parse_error_terminates() {
        let mut buf = make_message(vec![4], 1);
        buf[8] = 0xFF; // invalid version
        let backend = raw_backend(buf, forward_opts()).await;
        backend.scan_next_async().await.expect("scan");
        let state = backend.lock_state().expect("lock");
        assert!(state.fwd_terminated);
        assert!(state.layouts.is_empty());
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn async_scan_length_out_of_range_terminates() {
        let mut buf = make_message(vec![4], 1);
        let huge = (buf.len() as u64) + 10_000;
        buf[16..24].copy_from_slice(&huge.to_be_bytes());
        let backend = raw_backend(buf, forward_opts()).await;
        backend.scan_next_async().await.expect("scan");
        let state = backend.lock_state().expect("lock");
        assert!(state.fwd_terminated);
        assert!(state.layouts.is_empty());
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn async_scan_streaming_tail_recorded() {
        let mut buf = make_streaming_message(vec![4], 7);
        buf[16..24].copy_from_slice(&0u64.to_be_bytes());
        let pa_total = buf.len() - POSTAMBLE_SIZE + 8;
        buf[pa_total..pa_total + 8].copy_from_slice(&0u64.to_be_bytes());
        let file_len = buf.len() as u64;
        let backend = raw_backend(buf, forward_opts()).await;
        backend.message_count_async().await.expect("count");
        let state = backend.lock_state().expect("lock");
        assert_eq!(state.layouts.len(), 1);
        assert_eq!(state.layouts[0].length, file_len, "tail spans to EOF");
    }

    // ── Async read API roundtrips ────────────────────────────────────────

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn async_read_api_multi_message() {
        let buf = concat(vec![make_message(vec![4], 10), make_message(vec![8], 20)]);
        let backend = open_backend(buf, forward_opts()).await.expect("open");
        assert_eq!(backend.message_count_async().await.expect("count"), 2);
        assert_eq!(
            backend
                .message_layouts_async()
                .await
                .expect("layouts")
                .len(),
            2
        );
        assert!(
            !backend
                .read_message_async(1)
                .await
                .expect("read_message")
                .is_empty()
        );
        let opts = crate::DecodeOptions::default();

        let _meta = backend.read_metadata_async(0).await.expect("metadata");
        let (_m, descs) = backend
            .read_descriptors_async(1)
            .await
            .expect("descriptors");
        assert_eq!(descs.len(), 1);

        let (_m, _d, decoded) = backend
            .read_object_async(0, 0, &opts)
            .await
            .expect("read_object_async");
        assert_eq!(decoded.len(), 4 * 4);

        let (_d, parts) = backend
            .read_range_async(1, 0, &[(0, 8)], &opts)
            .await
            .expect("read_range_async");
        assert_eq!(parts[0].len(), 8 * 4);
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn async_batch_apis_multi_message() {
        let buf = concat(vec![
            make_message(vec![4], 10),
            make_message(vec![8], 20),
            make_message(vec![16], 30),
        ]);
        let backend = open_backend(buf, forward_opts()).await.expect("open");
        let opts = crate::DecodeOptions::default();

        // ensure_all_layouts_batch_async + read_object_batch_async.
        let obj_batch = backend
            .read_object_batch_async(&[0, 1, 2], 0, &opts)
            .await
            .expect("read_object_batch_async");
        assert_eq!(obj_batch.len(), 3);
        assert_eq!(obj_batch[2].2.len(), 16 * 4);

        let range_batch = backend
            .read_range_batch_async(&[0, 2], 0, &[(0, 4)], &opts)
            .await
            .expect("read_range_batch_async");
        assert_eq!(range_batch.len(), 2);
        assert_eq!(range_batch[0].1[0].len(), 4 * 4);
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn async_batch_empty_indices_is_noop() {
        let buf = make_message(vec![4], 1);
        let backend = open_backend(buf, forward_opts()).await.expect("open");
        backend
            .ensure_all_layouts_batch_async(&[])
            .await
            .expect("empty batch is a no-op");
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn async_batch_out_of_range_index_errors() {
        let buf = make_message(vec![4], 1);
        let backend = open_backend(buf, forward_opts()).await.expect("open");
        let err = backend.ensure_all_layouts_batch_async(&[0, 9]).await;
        assert!(err.is_err(), "out-of-range batch index must error");
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn async_read_descriptors_large_frame_prefix_path() {
        // Frame > 64 KiB forces read_descriptor_only_async down the
        // header/footer/CBOR-prefix branch.
        let n = 20_000u64;
        let buf = make_message(vec![n], 9);
        let backend = open_backend(buf, forward_opts()).await.expect("open");
        let (_m, descs) = backend
            .read_descriptors_async(0)
            .await
            .expect("descriptors");
        assert_eq!(descs.len(), 1);
        assert_eq!(descs[0].shape, vec![n]);
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn async_read_message_out_of_range_errors() {
        let buf = make_message(vec![4], 1);
        let backend = open_backend(buf, forward_opts()).await.expect("open");
        let err = backend.read_message_async(42).await;
        assert!(err.is_err(), "out-of-range message index must error");
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn async_read_descriptors_corrupt_index_errors() {
        let buf = make_message(vec![4], 1);
        let backend = open_backend(buf, forward_opts()).await.expect("open");
        {
            let mut state = backend.lock_state().expect("lock");
            state.layouts[0].global_metadata = Some(GlobalMetadata::default());
            state.layouts[0].index = Some(IndexFrame {
                offsets: vec![0, 100],
                lengths: vec![50],
            });
        }
        let err = backend.read_descriptors_async(0).await;
        assert!(err.is_err(), "corrupt index must error (async)");
    }

    fn one_desc(shape: Vec<u64>) -> DataObjectDescriptor {
        DataObjectDescriptor {
            obj_type: "ntensor".to_string(),
            ndim: shape.len() as u64,
            shape: shape.clone(),
            strides: strides_for(&shape),
            dtype: Dtype::Float32,
            byte_order: ByteOrder::native(),
            encoding: "none".to_string(),
            filter: "none".to_string(),
            compression: "none".to_string(),
            params: BTreeMap::new(),
            masks: None,
        }
    }

    async fn frame_backend(frame: Vec<u8>) -> RemoteBackend {
        let store = Arc::new(InMemory::new());
        let path = ObjectPath::from(TEST_PATH);
        let file_size = frame.len() as u64;
        store.put(&path, frame.into()).await.expect("put");
        RemoteBackend {
            source_url: "memory://frame.tgm".to_string(),
            store,
            path,
            file_size,
            state: Mutex::new(RemoteState {
                prev_scan_offset: file_size,
                ..RemoteState::default()
            }),
            scan_opts: forward_opts(),
        }
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn async_read_descriptor_only_large_cbor_before_prefix_loop() {
        let desc = one_desc(vec![20_000]);
        let payload = vec![0u8; 80_000];
        let frame = framing::encode_data_object_frame(&desc, &payload, true, false)
            .expect("cbor-before frame");
        let frame_len = frame.len() as u64;
        let backend = frame_backend(frame).await;
        let got = backend
            .read_descriptor_only_async(0, frame_len, 0, frame_len)
            .await
            .expect("descriptor via prefix loop");
        assert_eq!(got.shape, vec![20_000]);
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn async_read_descriptor_only_large_cbor_after() {
        let desc = one_desc(vec![20_000]);
        let payload = vec![0u8; 80_000];
        let frame = framing::encode_data_object_frame(&desc, &payload, false, false)
            .expect("cbor-after frame");
        let frame_len = frame.len() as u64;
        let backend = frame_backend(frame).await;
        let got = backend
            .read_descriptor_only_async(0, frame_len, 0, frame_len)
            .await
            .expect("descriptor cbor-after");
        assert_eq!(got.shape, vec![20_000]);
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn async_read_descriptor_only_large_non_data_object_errors() {
        let desc = one_desc(vec![20_000]);
        let payload = vec![0u8; 80_000];
        let mut frame =
            framing::encode_data_object_frame(&desc, &payload, false, false).expect("frame");
        frame[2] = FrameType::HeaderMetadata as u8;
        let frame_len = frame.len() as u64;
        let backend = frame_backend(frame).await;
        let err = backend
            .read_descriptor_only_async(0, frame_len, 0, frame_len)
            .await;
        assert!(err.is_err(), "non-data-object large frame must error");
    }

    // ── Async footer-indexed discovery ───────────────────────────────────

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn async_footer_indexed_read_metadata() {
        let buf = make_streaming_message(vec![6], 3);
        let backend = open_backend(buf, forward_opts()).await.expect("open");
        let _meta = backend.read_metadata_async(0).await.expect("metadata");
        let (_m, descs) = backend
            .read_descriptors_async(0)
            .await
            .expect("descriptors");
        assert_eq!(descs.len(), 1);
        let state = backend.lock_state().expect("lock");
        assert!(state.layouts[0].global_metadata.is_some());
        assert!(state.layouts[0].index.is_some());
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn async_footer_first_footer_offset_too_small_errors() {
        let mut buf = make_streaming_message(vec![6], 3);
        let pa_start = buf.len() - POSTAMBLE_SIZE;
        buf[pa_start..pa_start + 8].copy_from_slice(&1u64.to_be_bytes());
        let backend = open_backend(buf, forward_opts()).await.expect("open");
        let err = backend.read_metadata_async(0).await;
        assert!(
            err.is_err(),
            "first_footer_offset below preamble must error"
        );
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn async_footer_first_footer_offset_past_postamble_errors() {
        let mut buf = make_streaming_message(vec![6], 3);
        let msg_len = buf.len() as u64;
        let pa_start = buf.len() - POSTAMBLE_SIZE;
        buf[pa_start..pa_start + 8].copy_from_slice(&msg_len.to_be_bytes());
        let backend = open_backend(buf, forward_opts()).await.expect("open");
        let err = backend.read_metadata_async(0).await;
        assert!(
            err.is_err(),
            "first_footer_offset past postamble must error"
        );
    }

    // ── Async forward-only eager discovery (scan_and_discover_next_async) ─

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn async_eager_discover_bad_magic_terminates() {
        // ensure_layout_eager_async forward-only path routes through
        // scan_and_discover_next_async; junk after msg 0 terminates it.
        let mut buf = make_message(vec![4], 1);
        buf.extend_from_slice(&[0u8; 64]);
        let backend = raw_backend(buf, forward_opts()).await;
        let _meta = backend.read_metadata_async(0).await.expect("metadata");
        let count = backend.message_count_async().await.expect("count");
        assert_eq!(count, 1, "junk after msg 0 => only one message");
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn async_eager_discover_preamble_parse_error_at_second_message() {
        let good = make_message(vec![4], 1);
        let good_len = good.len();
        let mut buf = concat(vec![good, make_message(vec![8], 2)]);
        buf[good_len + 8] = 0xFF;
        let backend = open_backend(buf, forward_opts()).await.expect("open");
        let err = backend.read_metadata_async(1).await;
        assert!(err.is_err(), "corrupt 2nd preamble => message 1 missing");
        assert_eq!(backend.message_count_async().await.expect("count"), 1);
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn async_eager_discover_length_out_of_range_at_second_message() {
        let good = make_message(vec![4], 1);
        let good_len = good.len();
        let mut buf = concat(vec![good, make_message(vec![8], 2)]);
        let huge = (buf.len() as u64) + 50_000;
        buf[good_len + 16..good_len + 24].copy_from_slice(&huge.to_be_bytes());
        let backend = open_backend(buf, forward_opts()).await.expect("open");
        let err = backend.read_metadata_async(1).await;
        assert!(err.is_err(), "overrun 2nd length => message 1 missing");
        assert_eq!(backend.message_count_async().await.expect("count"), 1);
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn async_eager_discover_streaming_preamble_branch() {
        // Streaming-preamble message read via the async eager-discover
        // forward path: scan_and_discover_next_async's streaming branch
        // records a tail layout; footer frames in the body still feed
        // the lazy footer discovery.
        let mut buf = make_streaming_message(vec![4], 5);
        buf[16..24].copy_from_slice(&0u64.to_be_bytes());
        let pa_total = buf.len() - POSTAMBLE_SIZE + 8;
        buf[pa_total..pa_total + 8].copy_from_slice(&0u64.to_be_bytes());
        let file_len = buf.len() as u64;
        let backend = raw_backend(buf, forward_opts()).await;
        let _meta = backend
            .read_metadata_async(0)
            .await
            .expect("footer frames recoverable on streaming tail");
        let state = backend.lock_state().expect("lock");
        assert_eq!(state.layouts.len(), 1);
        assert_eq!(state.layouts[0].length, file_len);
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn async_eager_discover_streaming_end_magic_mismatch() {
        // Streaming preamble but corrupt file-end END_MAGIC: the async
        // streaming-end-magic-mismatch branch terminates forward with
        // no layout, so the metadata read fails with out-of-range.
        let mut buf = make_streaming_message(vec![4], 5);
        buf[16..24].copy_from_slice(&0u64.to_be_bytes());
        let pa_total = buf.len() - POSTAMBLE_SIZE + 8;
        buf[pa_total..pa_total + 8].copy_from_slice(&0u64.to_be_bytes());
        let len = buf.len();
        buf[len - 8..].copy_from_slice(b"BADMAGIC");
        let backend = raw_backend(buf, forward_opts()).await;
        let err = backend.read_metadata_async(0).await;
        assert!(err.is_err(), "streaming end-magic mismatch => no layout");
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn async_footer_indexed_eager_discover_second_message() {
        // Forward-only eager discovery of a footer-indexed message at
        // index 1 routes through scan_and_discover_next_async ->
        // discover_footer_layout_from_suffix_async (the suffix-read
        // footer path), populating metadata + index inline.
        let buf = concat(vec![
            make_streaming_message(vec![4], 1),
            make_streaming_message(vec![8], 2),
        ]);
        let backend = open_backend(buf, forward_opts()).await.expect("open");
        let _meta = backend
            .read_metadata_async(1)
            .await
            .expect("footer metadata for message 1");
        let (_m, descs) = backend
            .read_descriptors_async(1)
            .await
            .expect("descriptors");
        assert_eq!(descs.len(), 1);
        let state = backend.lock_state().expect("lock");
        assert!(state.layouts[1].global_metadata.is_some());
        assert!(state.layouts[1].index.is_some());
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn async_footer_indexed_batch() {
        // ensure_all_layouts_batch_async footer branch (Phase 5/6).
        let buf = concat(vec![
            make_streaming_message(vec![4], 1),
            make_streaming_message(vec![8], 2),
        ]);
        let backend = open_backend(buf, forward_opts()).await.expect("open");
        let opts = crate::DecodeOptions::default();
        let obj_batch = backend
            .read_object_batch_async(&[0, 1], 0, &opts)
            .await
            .expect("footer batch decode");
        assert_eq!(obj_batch.len(), 2);
    }

    // ── Async bidirectional / pipelined ──────────────────────────────────

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn async_bidir_matches_forward_only() {
        let parts: Vec<Vec<u8>> = (0..8)
            .map(|i| make_message(vec![4 + i as u64], i as u8))
            .collect();
        let buf = concat(parts);
        let fwd = open_backend(buf.clone(), forward_opts())
            .await
            .expect("open fwd");
        let fwd_layouts = fwd.message_layouts_async().await.expect("fwd layouts");
        let bidir = open_backend(buf, bidir_opts()).await.expect("open bidir");
        let bidir_layouts = bidir.message_layouts_async().await.expect("bidir layouts");
        assert_eq!(
            fwd_layouts.len(),
            bidir_layouts.len(),
            "bidir must find the same number of messages",
        );
        for (a, b) in fwd_layouts.iter().zip(bidir_layouts.iter()) {
            assert_eq!(a.offset, b.offset);
            assert_eq!(a.length, b.length);
        }
        assert_eq!(bidir_layouts.len(), 8);
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn async_pipelined_one_message_completes_without_fallback() {
        let buf = make_message(vec![4], 42);
        let backend = open_backend(buf, bidir_opts()).await.expect("open");
        let ok = backend.scan_pipelined_async(None).await.expect("pipelined");
        assert!(ok, "1-message file: pipeline completes without fallback");
        let state = backend.lock_state().expect("lock");
        assert_eq!(state.layouts.len(), 1);
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn async_bidir_corrupt_end_magic_forward_still_scans() {
        let mut buf = concat(vec![make_message(vec![4], 1), make_message(vec![8], 2)]);
        let len = buf.len();
        buf[len - 8..].copy_from_slice(b"BADMAGIC");
        let backend = open_backend(buf, bidir_opts()).await.expect("open");
        let count = backend.message_count_async().await.expect("count");
        assert_eq!(count, 2, "backward corruption => forward still scans");
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn async_bidir_footer_indexed_eager_populates() {
        let parts: Vec<Vec<u8>> = (0..6)
            .map(|i| make_streaming_message(vec![4 + i as u64], i as u8))
            .collect();
        let buf = concat(parts);
        let backend = open_backend(buf, bidir_opts()).await.expect("open");
        let count = backend.message_count_async().await.expect("count");
        assert_eq!(count, 6);
    }
}