ethl 0.1.30

use std::collections::VecDeque;
use std::future::Future;
use std::time::Duration;

use alloy::{
    network::Ethereum,
    primitives::B256,
    providers::{DynProvider, Provider},
    rpc::types::{BlockNumberOrTag, Filter, Header, Log},
};
use async_stream::stream;
use futures_util::{StreamExt, pin_mut, stream::Stream};
use tokio::time::{Instant, sleep, sleep_until};
use tracing::{debug, warn};

use crate::rpc::cursor::{TipGate, TipWaitOutcome, next_cursor_after_batch, wait_for_tip};
use crate::rpc::error_tracking::{
    ErrorCategory, ProviderErrorTracker, find_active_provider, truncate_error_short,
};
use crate::rpc::{
    LogRange, RpcError,
    config::{CaptureTarget, ProviderSettings},
};

/// Maximum block range per WS request to prevent "max results" errors
const MAX_BLOCKS_PER_WS_REQUEST: u64 = 300;

#[derive(Debug, Clone)]
struct HeadCoalescer {
    block_cursor: Option<u64>,
    latest_seen_head: Option<u64>,
    flush_deadline: Option<Instant>,
    /// Coalesce rapid head updates into a bounded polling cadence.
    window: Duration,
}

impl HeadCoalescer {
    fn new(block_cursor: Option<u64>, window: Duration) -> Self {
        Self {
            block_cursor,
            latest_seen_head: None,
            flush_deadline: None,
            window,
        }
    }

    fn block_cursor(&self) -> Option<u64> {
        self.block_cursor
    }

    fn observe_head(&mut self, head_number: u64, now: Instant) -> bool {
        let from_block = self.block_cursor.unwrap_or(head_number);
        if head_number < from_block {
            return false;
        }

        self.latest_seen_head = Some(
            self.latest_seen_head
                .map_or(head_number, |latest| latest.max(head_number)),
        );

        if self.flush_deadline.is_none() {
            self.flush_deadline = Some(now + self.window);
        }

        true
    }

    fn pending_range(&self) -> Option<(u64, u64)> {
        let to_block = self.latest_seen_head?;
        let from_block = self.block_cursor.unwrap_or(to_block);
        (to_block >= from_block).then_some((from_block, to_block))
    }

    fn flush_deadline(&self) -> Option<Instant> {
        self.pending_range()?;
        self.flush_deadline
    }

    fn should_flush(&self, now: Instant) -> bool {
        let Some((from_block, to_block)) = self.pending_range() else {
            return false;
        };

        if to_block - from_block + 1 >= MAX_BLOCKS_PER_WS_REQUEST {
            return true;
        }

        self.flush_deadline.is_some_and(|deadline| now >= deadline)
    }

    fn advance_cursor(&mut self, next_block: u64) {
        self.block_cursor = Some(next_block);

        if self
            .latest_seen_head
            .is_some_and(|latest| next_block > latest)
        {
            self.latest_seen_head = None;
            self.flush_deadline = None;
        }
    }

    /// Called after an empty, non-capped flush so the stream waits for the next
    /// head instead of busy-looping the same range. Clears `flush_deadline`
    /// only — the cursor and `latest_seen_head` are untouched, so the next
    /// `observe_head` re-arms the deadline while the span-cap arm of
    /// `should_flush` still fires deadline-free.
    fn defer_flush(&mut self) {
        self.flush_deadline = None;
    }
}

/// Outcome of feeding one head into the live reorg linkage check.
#[derive(Debug, Clone, PartialEq, Eq)]
enum LinkOutcome {
    /// The head extends (or bootstraps) the canonical chain — no reorg.
    Linked,
    /// A reorg whose common ancestor was resolved within the ring.
    Reorg { common_ancestor: u64 },
    /// The head's parent is not the recorded canonical block and the ancestor
    /// could not be resolved from the ring alone (the provider skipped the
    /// intermediate new-chain heads). The caller must fetch the parent block by
    /// hash and walk the new chain back — bounded RPC, reorg branch only — feeding
    /// the fetched `(number, hash)` back through `matches`.
    Detached {
        parent_hash: B256,
        child_number: u64,
    },
    /// Divergence older than the ring floor (`> finality`): catastrophe.
    BeyondFinality,
}

/// A finality-deep ring of canonical `(number, hash)` taken from the head stream,
/// plus a `parent_hash` linkage check. Pure and synchronous — the network is thin
/// glue around it, mirroring `HeadCoalescer`. `depth == 0` disables it: `observe`
/// is then inert and returns `Linked` for every head (zero overhead, the L2 /
/// snapshot default).
#[derive(Debug, Clone)]
struct HeadLinkage {
    ring: VecDeque<(u64, B256)>,
    depth: u64,
}

impl HeadLinkage {
    fn new(depth: u64) -> Self {
        Self {
            ring: VecDeque::new(),
            depth,
        }
    }

    fn enabled(&self) -> bool {
        self.depth > 0
    }

    /// Lowest block number still recorded in the ring (the resolution floor).
    fn floor(&self) -> Option<u64> {
        self.ring.front().map(|&(number, _)| number)
    }

    /// Recorded canonical hash at `number`, if it is within the ring.
    fn hash_at(&self, number: u64) -> Option<B256> {
        let &(front, _) = self.ring.front()?;
        let offset = number.checked_sub(front)? as usize;
        self.ring.get(offset).map(|&(_, hash)| hash)
    }

    /// True if the ring records `hash` as canonical at `number`.
    fn matches(&self, number: u64, hash: B256) -> bool {
        self.hash_at(number) == Some(hash)
    }

    /// Append `(number, hash)` as the new tip, evicting the oldest beyond `depth`.
    fn push(&mut self, number: u64, hash: B256) {
        self.ring.push_back((number, hash));
        while self.ring.len() as u64 > self.depth {
            self.ring.pop_front();
        }
    }

    /// The most recently recorded `(number, hash)` entry (the ring's canonical tip).
    ///
    /// Returns `None` only when the ring is completely empty (before the first
    /// `observe`). When `linkage.enabled()` and a flush fires, this is always `Some`
    /// because `observe` bootstraps or re-anchors the ring at the newest head before
    /// `should_flush` can return true.
    fn latest(&self) -> Option<(u64, B256)> {
        self.ring.back().copied()
    }

    /// Re-anchor the ring at `ancestor` (drop everything above it), then record
    /// the new-chain head `(number, hash)`.
    fn reanchor(&mut self, ancestor: u64, number: u64, hash: B256) {
        while self.ring.back().is_some_and(|&(n, _)| n > ancestor) {
            self.ring.pop_back();
        }
        self.push(number, hash);
    }

    /// Observe one head's `(number, hash, parent_hash)` against the recorded
    /// chain. See `LinkOutcome`. Disabled (`depth == 0`) ⇒ always `Linked`.
    fn observe(&mut self, number: u64, hash: B256, parent_hash: B256) -> LinkOutcome {
        if !self.enabled() {
            return LinkOutcome::Linked;
        }

        let Some(&(tip_number, tip_hash)) = self.ring.back() else {
            // Bootstrap: the first head seeds the ring.
            self.push(number, hash);
            return LinkOutcome::Linked;
        };

        // The common case: a clean forward extension of the tip.
        if number == tip_number + 1 && parent_hash == tip_hash {
            self.push(number, hash);
            return LinkOutcome::Linked;
        }

        // A forward gap (heads skipped ahead with no verifiable reorg signal —
        // e.g. across a reconnect): linkage cannot be asserted over the gap, so
        // re-anchor here rather than risk a false reorg.
        if number > tip_number + 1 {
            self.ring.clear();
            self.push(number, hash);
            return LinkOutcome::Linked;
        }

        // `number <= tip_number + 1` and the parent did not extend the tip:
        // a reorg, a duplicate, or a detachment that needs an RPC walk.
        let Some(parent_number) = number.checked_sub(1) else {
            return LinkOutcome::BeyondFinality;
        };
        match self.hash_at(parent_number) {
            // Parent links to a known canonical block.
            Some(canonical_parent) if canonical_parent == parent_hash => {
                // Already recorded exactly → duplicate, no-op.
                if self.hash_at(number) == Some(hash) {
                    return LinkOutcome::Linked;
                }
                // Block `number` was replaced on a chain sharing canonical
                // `parent_number` → that is the common ancestor.
                self.reanchor(parent_number, number, hash);
                LinkOutcome::Reorg {
                    common_ancestor: parent_number,
                }
            }
            // Parent is within the ring but differs → diverges deeper; the caller
            // must walk the new chain back to find the ancestor.
            Some(_) => LinkOutcome::Detached {
                parent_hash,
                child_number: number,
            },
            // Parent is below the ring floor → beyond finality.
            None => LinkOutcome::BeyondFinality,
        }
    }
}

/// Walk backward from `tip_hash` at `tip_number` down to `from_block`, collecting
/// canonical `(number, hash)` pairs in descending order (tip → from_block).
///
/// Used by the backfill boot path to resolve hashes for the reorg-exposed window
/// before handing off to `stream_heads_with_logs`. Does not touch any ring.
///
/// Error semantics: null block → `ReorgBeyondFinality`; transport fault → `ConnectionError`.
pub(crate) async fn collect_window_hashes(
    provider: &DynProvider<Ethereum>,
    tip_hash: B256,
    tip_number: u64,
    from_block: u64,
) -> Result<Vec<(u64, B256)>, RpcError> {
    if tip_number < from_block {
        return Ok(Vec::new());
    }
    let mut collected: Vec<(u64, B256)> = Vec::new();
    let mut current_hash = tip_hash;
    let mut current_number = tip_number;

    loop {
        collected.push((current_number, current_hash));
        if current_number == from_block {
            break;
        }
        let block = match provider.get_block_by_hash(current_hash).await {
            Ok(Some(block)) => block,
            Ok(None) => return Err(RpcError::ReorgBeyondFinality),
            Err(e) => {
                return Err(RpcError::ConnectionError(format!(
                    "window hash walk: block fetch failed: {}",
                    truncate_error_short(e.to_string(), 100)
                )));
            }
        };
        // A by-hash query returning a block with the wrong number is a provider
        // integrity violation — fail closed rather than walking past it.
        if block.header.number != current_number {
            return Err(RpcError::ReorgBeyondFinality);
        }
        let parent_hash = block.header.parent_hash;
        let Some(parent_number) = current_number.checked_sub(1) else {
            return Err(RpcError::ReorgBeyondFinality);
        };
        if parent_number < from_block {
            break;
        }
        current_hash = parent_hash;
        current_number = parent_number;
    }

    Ok(collected)
}

/// Fetch logs for a slice of canonical `(number, hash)` pairs using
/// `at_block_hash` filters — one request per block, ascending.
///
/// `at_block_hash` pins the query to the exact canonical block identified by
/// `hash`, so an orphaned block at that number can never contaminate the result.
/// The caller passes `pairs` in ascending order (from_block → tip).
pub(crate) async fn fetch_logs_by_hash_range(
    providers: &ProviderSettings,
    pairs: &[(u64, B256)],
    base_filter: Option<&Filter>,
) -> Result<Vec<LogRange>, RpcError> {
    let mut ranges: Vec<LogRange> = Vec::with_capacity(pairs.len());

    for &(number, hash) in pairs {
        let query = base_filter
            .cloned()
            .unwrap_or_else(Filter::new)
            .at_block_hash(hash);

        let (_, logs) =
            get_logs_with_http_fallback_retry(providers, &query, 0, number, number).await?;

        ranges.push((number, number, logs));
    }

    Ok(ranges)
}

/// Flush one pending window range from the coalescer by fetching all logs by
/// canonical block hash.
///
/// When the ring is missing hashes for some blocks (boot gap or post-forward-gap
/// reconnect), walks backward from `linkage.latest()` to collect canonical
/// `(number, hash)` pairs and uses them directly. Never falls back to a by-number
/// fetch — fail-closed is the invariant. Only called when `linkage.enabled()`.
async fn flush_window_range(
    linkage: &mut HeadLinkage,
    coalescer: &mut HeadCoalescer,
    providers: &ProviderSettings,
    filter: Option<&Filter>,
    provider_index: usize,
) -> Result<Vec<LogRange>, RpcError> {
    let Some((from_block, pending_to_block)) = coalescer.pending_range() else {
        return Ok(vec![]);
    };
    let to_block = std::cmp::min(from_block + MAX_BLOCKS_PER_WS_REQUEST - 1, pending_to_block);

    // Build the (number, hash) pairs for [from_block, to_block].
    // If any hash is absent from the ring, self-heal: walk backward from the
    // latest known entry to fill the gap. `latest()` is Some whenever the flush
    // fires — `observe` bootstraps or re-anchors at the newest head before
    // `should_flush` can return true.
    let pairs: Vec<(u64, B256)> = if (from_block..=to_block).all(|n| linkage.hash_at(n).is_some()) {
        (from_block..=to_block)
            .map(|n| (n, linkage.hash_at(n).expect("just verified")))
            .collect()
    } else {
        let Some((top_n, top_hash)) = linkage.latest() else {
            // Ring is completely empty — no anchor for the walk.
            return Err(RpcError::ReorgBeyondFinality);
        };
        let walk_provider = providers.connect_http(provider_index);
        let mut pairs_desc =
            collect_window_hashes(&walk_provider, top_hash, top_n, from_block).await?;
        // Reverse to ascending (from_block → top_n), then restrict to [from_block, to_block].
        pairs_desc.reverse();
        pairs_desc
            .into_iter()
            .filter(|&(n, _)| n <= to_block)
            .collect()
    };

    // Fetch the entire batch BEFORE advancing the cursor. On any block's failure,
    // return Err with the cursor UNTOUCHED so the whole window range retries on the
    // next provider — never advance past a block whose logs were not delivered to the
    // consumer (fail-closed; no silent gap). Re-fetch on retry is idempotent.
    let mut ranges = Vec::with_capacity(pairs.len());
    for &(n, hash) in &pairs {
        let query = filter
            .cloned()
            .unwrap_or_else(Filter::new)
            .at_block_hash(hash);
        let (_, logs) =
            get_logs_with_http_fallback_retry(providers, &query, provider_index, n, n).await?;
        ranges.push((n, n, logs));
    }
    if let Some(&(last, _)) = pairs.last() {
        coalescer.advance_cursor(last + 1);
    }
    Ok(ranges)
}

/// Walk the new chain backward by hash until a block re-links to the recorded
/// canonical ring, returning the resolved `RpcError::Reorg { common_ancestor }`.
///
/// Invoked only on `LinkOutcome::Detached`, when the provider delivered a new-tip
/// head without re-emitting the intermediate new-chain blocks. Bounded to `depth`
/// `get_block_by_hash` fetches (the ring's own span); exhausting them, or walking
/// below the ring floor, means the divergence is older than finality →
/// `ReorgBeyondFinality`. The pure ring is untouched; only this glue does I/O.
async fn resolve_reorg_ancestor(
    provider: &DynProvider<Ethereum>,
    linkage: &HeadLinkage,
    detached_parent: B256,
    child_number: u64,
    depth: u64,
) -> RpcError {
    // `parent_hash` is the new chain's hash at height `parent_number`.
    let mut parent_hash = detached_parent;
    let Some(mut parent_number) = child_number.checked_sub(1) else {
        return RpcError::ReorgBeyondFinality;
    };
    let floor = linkage.floor().unwrap_or(u64::MAX);

    for _ in 0..depth {
        if linkage.matches(parent_number, parent_hash) {
            return RpcError::Reorg {
                common_ancestor: parent_number,
            };
        }
        if parent_number <= floor {
            // Cannot descend below the ring floor without leaving the finality window.
            return RpcError::ReorgBeyondFinality;
        }
        let block = match provider.get_block_by_hash(parent_hash).await {
            Ok(Some(block)) => block,
            Ok(None) => return RpcError::ReorgBeyondFinality,
            Err(e) => {
                return RpcError::ConnectionError(format!(
                    "reorg ancestor walk: block-by-hash fetch failed: {}",
                    truncate_error_short(e.to_string(), 100)
                ));
            }
        };
        parent_hash = block.header.parent_hash;
        let Some(next) = parent_number.checked_sub(1) else {
            return RpcError::ReorgBeyondFinality;
        };
        parent_number = next;
    }
    RpcError::ReorgBeyondFinality
}

fn fallback_http_index(start_index: usize, attempt: usize, num_http_providers: usize) -> usize {
    (start_index + attempt) % num_http_providers
}

fn fallback_http_exhausted_error(
    from_block: u64,
    to_block: u64,
    attempted_hosts: &[String],
    last_err: &str,
) -> RpcError {
    RpcError::LogFetchError(format!(
        "HTTP fallback exhausted for blocks {}-{} after trying providers [{}]: {}",
        from_block,
        to_block,
        attempted_hosts.join(", "),
        last_err
    ))
}

async fn get_logs_with_http_fallback_retry(
    providers: &ProviderSettings,
    query_filter: &Filter,
    start_index: usize,
    from_block: u64,
    to_block: u64,
) -> Result<(usize, Vec<alloy::rpc::types::Log>), RpcError> {
    let num_http_providers = providers.http_providers.len();
    let mut attempted_hosts: Vec<String> = Vec::with_capacity(num_http_providers);
    let mut last_err = String::new();

    for attempt in 0..num_http_providers {
        let provider_idx = fallback_http_index(start_index, attempt, num_http_providers);
        let provider_host = providers.http_settings(provider_idx).host();
        attempted_hosts.push(provider_host.clone());

        let http_provider = providers.connect_http(provider_idx);
        match http_provider.get_logs(query_filter).await {
            Ok(logs) => return Ok((provider_idx, logs)),
            Err(err) => {
                last_err = err.to_string();
                warn!(
                    "HTTP fallback {} failed for blocks {}-{}: {}",
                    provider_host,
                    from_block,
                    to_block,
                    truncate_error_short(&last_err, 100)
                );
            }
        }
    }

    Err(fallback_http_exhausted_error(
        from_block,
        to_block,
        &attempted_hosts,
        &last_err,
    ))
}

/// How far a non-empty `getLogs` response actually proves the queried range
/// was covered, decided by `yield_bounds`.
enum YieldBounds {
    /// Watermark == to_block: today's behavior verbatim.
    Full,
    /// Watermark B < to_block: yield (from, B), advance B+1, defer the tail.
    Partial(u64),
    /// No returned log carries a block number: the response proves nothing —
    /// treat it exactly as an empty response (do NOT yield these logs).
    Unattributable,
}

/// Decides how much of `[.., to_block]` a non-empty log batch proves covered.
///
/// The watermark is the highest `block_number` carried by any returned log —
/// a log at block B proves the provider's index reached >= B, which also
/// vouches for the emptiness of every logless block <= B in the range. Only
/// the tail strictly above the watermark is unproven.
fn yield_bounds(logs: &[Log], to_block: u64) -> YieldBounds {
    let Some(watermark) = logs.iter().filter_map(|log| log.block_number).max() else {
        return YieldBounds::Unattributable;
    };

    // A range query cannot legally return a block above what it asked for; a
    // watermark past to_block is a provider integrity violation. Clamp rather
    // than advance the cursor past the requested range.
    if watermark >= to_block {
        YieldBounds::Full
    } else {
        YieldBounds::Partial(watermark)
    }
}

/// Outcome of the empty-batch guard for one flushed chunk, decided by
/// `handle_empty_flush`. Callers map each variant onto their existing control
/// flow (mirrored across the WS and HTTP-fallback sites in
/// `stream_heads_with_logs`).
enum EmptyFlushOutcome {
    /// Non-capped: `defer_flush()` applied. Caller breaks the `'flush` loop.
    Deferred,
    /// Capped, tip >= to_block: caller yields the verified-empty range,
    /// advances the cursor, resets the gate.
    VerifiedEmpty,
    /// Capped, `wait_for_tip` caught up: caller re-queries (continue `'flush`).
    CaughtUp,
    /// Capped, terminal stall: caller yields `ProviderStalled` and returns.
    Stalled {
        to_block: u64,
        reported_tip: u64,
        stalled_cycles: u32,
    },
    /// Tip re-poll failed inside `wait_for_tip`: caller logs and switches provider.
    RepollFailed(String),
    /// Initial tip fetch failed: caller records the error and switches provider.
    TipFetchFailed(String),
}

/// Decides how to handle an empty log batch for one flushed chunk.
///
/// A non-capped chunk defers rather than confirming: a rare false-empty
/// (provider serving `[]` past its own tip) self-heals when the next head's
/// flush re-covers the same span from the unmoved cursor, so confirming every
/// quiet window buys nothing. A span-capped chunk (`to_block - from_block + 1
/// == MAX_BLOCKS_PER_WS_REQUEST`) always confirms against the provider's
/// reported tip — this bounds how far the cursor can advance unverified and is
/// what still catches a frozen or lying backend (the Base-outage failure
/// class), now at the span cap instead of per window.
async fn handle_empty_flush<F, Fut, E>(
    coalescer: &mut HeadCoalescer,
    gate: &mut TipGate,
    providers: &ProviderSettings,
    host: &str,
    to_block: u64,
    span_capped: bool,
    get_tip: F,
) -> EmptyFlushOutcome
where
    F: Fn() -> Fut,
    Fut: Future<Output = Result<u64, E>>,
    E: std::fmt::Display,
{
    if !span_capped {
        coalescer.defer_flush();
        return EmptyFlushOutcome::Deferred;
    }

    let reported_tip = match get_tip().await {
        Ok(tip) => tip,
        Err(e) => return EmptyFlushOutcome::TipFetchFailed(e.to_string()),
    };

    if next_cursor_after_batch(to_block, true, reported_tip).is_err() {
        match wait_for_tip(to_block, reported_tip, gate, providers, host, get_tip).await {
            TipWaitOutcome::CaughtUp => EmptyFlushOutcome::CaughtUp,
            TipWaitOutcome::Terminal {
                to_block,
                reported_tip,
                stalled_cycles,
            } => EmptyFlushOutcome::Stalled {
                to_block,
                reported_tip,
                stalled_cycles,
            },
            TipWaitOutcome::RepollFailed(err) => EmptyFlushOutcome::RepollFailed(err),
        }
    } else {
        EmptyFlushOutcome::VerifiedEmpty
    }
}

async fn stream_heads_provider(
    provider: &DynProvider<Ethereum>,
    provider_host: &str,
) -> impl Stream<Item = Header> {
    stream! {
        let head_stream = provider.subscribe_blocks().await;

        match head_stream {
            Ok(heads) => {
                debug!("WebSocket connected, streaming headers...");
                let mut s = heads.into_stream();
                while let Some(header) = s.next().await {
                    yield header;
                }
            }
            Err(e) => {
                warn!(
                    "Failed to subscribe to blocks on {}: {} - retrying",
                    provider_host,
                    truncate_error_short(e, 100)
                );
            }
        }
    }
}

/// Yield contract: empty ranges are not yielded per window. A quiet stretch
/// yields nothing until either a non-empty flush (which covers the whole span
/// back to the unmoved cursor) or a span-cap tip-verification (which yields
/// the verified-empty `MAX_BLOCKS_PER_WS_REQUEST`-block chunk). The
/// RangeTooWide HTTP-fallback path is the exception — it still confirms every
/// empty response, per-window. Consumers' cursor persistence coarsens
/// accordingly: expect progress in bursts, not one update per coalesce window.
///
/// A yielded range's `block_end` is the highest block PROVEN by the response
/// — a returned log's block number — which may be below the queried
/// `to_block` when the serving backend's index lagged behind the requested
/// range. The unproven remainder above `block_end` is re-covered like a
/// deferred empty flush: it stays in front of the cursor until the next
/// head's flush picks it up.
pub async fn stream_heads_with_logs(
    providers: &ProviderSettings,
    filter: Option<&Filter>,
) -> impl Stream<Item = Result<LogRange, RpcError>> {
    let num_providers = providers.wss_endpoints.len();

    stream! {
        // Initialize per-provider error tracking (same pattern as backfill)
        let mut provider_trackers: Vec<ProviderErrorTracker> = (0..num_providers)
            .map(|i| {
                let host = providers.wss_endpoints[i]
                    .host_str()
                    .unwrap_or("unknown")
                    .to_string();
                ProviderErrorTracker::new(host)
            })
            .collect();

        let mut provider_index = 0;
        let mut coalescer = HeadCoalescer::new(
            filter.and_then(|f| f.get_from_block()),
            Duration::from_millis(providers.live_head_coalesce_window_ms),
        );
        // Live-tip reorg detector. `reorg_detect_depth == 0` (every non-mainnet-live
        // consumer) leaves `observe` inert, so this is a no-op on those paths.
        let mut linkage = HeadLinkage::new(providers.reorg_detect_depth);
        let mut gate = TipGate::new();

        loop {
            // Find the next non-suspended provider (same pattern as backfill)
            let active_provider = find_active_provider(&mut provider_trackers, provider_index);
            let Some(idx) = active_provider else {
                yield Err(RpcError::AllProvidersSuspended(format!(
                    "All {} WebSocket providers suspended. Unable to continue streaming.",
                    num_providers
                )));
                return;
            };
            provider_index = idx;

            let provider = match providers.connect_ws(provider_index).await {
                Ok(p) => {
                    provider_trackers[provider_index].record_success();
                    p
                }
                Err(e) => {
                    let suspended = provider_trackers[provider_index].record_error();
                    let backoff = provider_trackers[provider_index].backoff_duration();

                    if suspended {
                        warn!(
                            "WebSocket provider {} suspended after connection failure: {}",
                            provider_trackers[provider_index].identifier(),
                            truncate_error_short(&e, 100)
                        );
                    } else {
                        warn!(
                            "Failed to connect to WebSocket {}: {} - retrying in {:?}",
                            provider_trackers[provider_index].identifier(),
                            truncate_error_short(&e, 100),
                            backoff
                        );
                        sleep(backoff).await;
                    }
                    provider_index = (provider_index + 1) % num_providers;
                    continue;
                }
            };

            // A new provider session must not inherit the previous session's stall
            // state: a stale last_tip makes the new provider's first observation count
            // as a stall instead of a seed (audit F1, early ProviderStalled).
            gate.reset();

            let provider_host = provider_trackers[provider_index].identifier().to_string();
            let s = stream_heads_provider(&provider, &provider_host).await;
            pin_mut!(s);
            let mut should_switch_provider = false;

            loop {
                if coalescer.should_flush(Instant::now()) {
                    // Flush from the cursor through the latest seen head in bounded chunks.
                    'flush: while let Some((from_block, pending_to_block)) = coalescer.pending_range() {
                        let to_block =
                            std::cmp::min(from_block + MAX_BLOCKS_PER_WS_REQUEST - 1, pending_to_block);

                        // When reorg detection is active fetch logs pinned to the canonical
                        // hash so a same-number orphaned block on a stale provider cannot
                        // contaminate the result. When the ring is missing a hash (boot gap
                        // or post-forward-gap reconnect), `flush_window_range` self-heals by
                        // walking backward from the latest ring entry — never spinning, never
                        // falling back to a by-number fetch inside the window.
                        if linkage.enabled() {
                            match flush_window_range(
                                &mut linkage,
                                &mut coalescer,
                                providers,
                                filter,
                                provider_index,
                            )
                            .await
                            {
                                Ok(ranges) => {
                                    for range in ranges {
                                        provider_trackers[provider_index].record_success();
                                        yield Ok(range);
                                    }
                                }
                                Err(e) => {
                                    warn!(
                                        "by-hash window flush failed: {}",
                                        truncate_error_short(&e, 100)
                                    );
                                    should_switch_provider = true;
                                    break 'flush;
                                }
                            }
                            continue;
                        }

                        let query_filter = filter
                            .cloned()
                            .unwrap_or_else(Filter::new)
                            .from_block(from_block)
                            .to_block(to_block);

                        let logs = provider.get_logs(&query_filter).await;

                        match logs {
                            Ok(logs) => {
                                provider_trackers[provider_index].record_success();

                                // A non-empty response only proves coverage up to the highest
                                // block number any returned log carries (the watermark); Partial
                                // means the tail above it is unproven and must not be crossed.
                                let watermark_bounds =
                                    (!logs.is_empty()).then(|| yield_bounds(&logs, to_block));

                                if let Some(YieldBounds::Partial(watermark)) = watermark_bounds {
                                    yield Ok((from_block, watermark, logs));
                                    coalescer.advance_cursor(watermark + 1);
                                    gate.reset();
                                    // Unproven tail [watermark+1, to_block] stays in front of the
                                    // cursor; the next head's flush re-covers it, same as an
                                    // empty flush's deferred-empty pattern.
                                    coalescer.defer_flush();
                                    break 'flush;
                                }

                                // Empty batches are ambiguous: they look identical whether the
                                // range is genuinely event-free or the provider silently returned
                                // [] for blocks past its actual tip. A non-capped chunk defers
                                // instead of confirming (see `handle_empty_flush`); a span-capped
                                // chunk still verifies against the WS provider's reported tip
                                // before advancing the cursor. Unattributable responses (logs
                                // present but none carry a block number) prove nothing and enter
                                // this same guard; the received logs are dropped, not yielded.
                                let treat_as_empty =
                                    !matches!(watermark_bounds, Some(YieldBounds::Full));
                                if treat_as_empty {
                                    let span_capped =
                                        to_block - from_block + 1 == MAX_BLOCKS_PER_WS_REQUEST;
                                    match handle_empty_flush(
                                        &mut coalescer,
                                        &mut gate,
                                        providers,
                                        &provider_host,
                                        to_block,
                                        span_capped,
                                        || provider.get_block_number(),
                                    )
                                    .await
                                    {
                                        EmptyFlushOutcome::Deferred => break 'flush,
                                        EmptyFlushOutcome::VerifiedEmpty => {}
                                        EmptyFlushOutcome::CaughtUp => continue 'flush,
                                        EmptyFlushOutcome::Stalled {
                                            to_block,
                                            reported_tip,
                                            stalled_cycles,
                                        } => {
                                            yield Err(RpcError::ProviderStalled {
                                                to_block,
                                                reported_tip,
                                                stalled_cycles,
                                            });
                                            return;
                                        }
                                        EmptyFlushOutcome::RepollFailed(err) => {
                                            let suspended = provider_trackers
                                                [provider_index]
                                                .record_error();
                                            if suspended {
                                                warn!(
                                                    "WebSocket provider {} suspended after tip re-poll failed during tip-wait for blocks {}-{}: {}",
                                                    provider_trackers[provider_index].identifier(),
                                                    from_block,
                                                    to_block,
                                                    truncate_error_short(&err, 100),
                                                );
                                            } else {
                                                let backoff = provider_trackers
                                                    [provider_index]
                                                    .backoff_duration();
                                                warn!(
                                                    "Tip re-poll failed during wait for blocks {}-{} from {}: {} - backing off {:?}",
                                                    from_block,
                                                    to_block,
                                                    provider_trackers[provider_index].identifier(),
                                                    truncate_error_short(&err, 100),
                                                    backoff
                                                );
                                                sleep(backoff).await;
                                            }
                                            should_switch_provider = true;
                                            break 'flush;
                                        }
                                        EmptyFlushOutcome::TipFetchFailed(err_str) => {
                                            let suspended =
                                                provider_trackers[provider_index].record_error();
                                            if suspended {
                                                warn!(
                                                    "WebSocket provider {} suspended after tip fetch failed during empty-batch guard for blocks {}-{}: {}",
                                                    provider_trackers[provider_index].identifier(),
                                                    from_block, to_block,
                                                    truncate_error_short(&err_str, 100)
                                                );
                                            } else {
                                                let backoff = provider_trackers[provider_index].backoff_duration();
                                                warn!(
                                                    "Tip fetch failed during empty-batch guard for blocks {}-{} from {}: {} - backing off {:?}",
                                                    from_block, to_block,
                                                    provider_trackers[provider_index].identifier(),
                                                    truncate_error_short(&err_str, 100),
                                                    backoff
                                                );
                                                sleep(backoff).await;
                                            }
                                            should_switch_provider = true;
                                            break 'flush;
                                        }
                                    }
                                }

                                let logs = if treat_as_empty { Vec::new() } else { logs };
                                yield Ok((from_block, to_block, logs));
                                coalescer.advance_cursor(to_block + 1);
                                gate.reset();
                            }
                            Err(e) => {
                                let error_msg = e.to_string();
                                let error_category = ErrorCategory::from_error_msg(&error_msg);

                                match error_category {
                                    ErrorCategory::RangeTooWide => {
                                        // Fall back to HTTP provider for this batch
                                        warn!(
                                            "Range too wide for blocks {}-{}, falling back to HTTP RPC: {}",
                                            from_block, to_block, truncate_error_short(&error_msg, 100)
                                        );

                                        match get_logs_with_http_fallback_retry(
                                            providers,
                                            &query_filter,
                                            provider_index,
                                            from_block,
                                            to_block,
                                        )
                                        .await
                                        {
                                            Ok((http_provider_idx, logs)) => {
                                                provider_trackers[provider_index].record_success();

                                                // Same watermark bound as the WS site: a non-empty
                                                // response only proves coverage up to the highest
                                                // block number a returned log carries. Unlike the
                                                // empty case below, Partial DOES defer here — it
                                                // already advanced the cursor to watermark+1 before
                                                // deferring, so each retry makes forward progress
                                                // and the WS-error -> fallback double-call objection
                                                // (see the comment below) does not apply.
                                                let watermark_bounds = (!logs.is_empty())
                                                    .then(|| yield_bounds(&logs, to_block));

                                                if let Some(YieldBounds::Partial(watermark)) =
                                                    watermark_bounds
                                                {
                                                    yield Ok((from_block, watermark, logs));
                                                    coalescer.advance_cursor(watermark + 1);
                                                    gate.reset();
                                                    coalescer.defer_flush();
                                                    break 'flush;
                                                }

                                                // Empty fallback responses are ambiguous — a lagging
                                                // HTTP provider can legitimately return [] for blocks
                                                // past its own tip. Verify against the HTTP provider
                                                // that served this response, not the WS provider.
                                                // Lazy does NOT apply here: a RangeTooWide error
                                                // followed by an empty fallback is already
                                                // contradictory (WS claimed too many results, HTTP
                                                // returned zero), and deferring would re-trigger the
                                                // WS-error -> fallback double-call on every head
                                                // until the span cap — the opposite of the saving.
                                                // Unattributable responses (logs present but none
                                                // carry a block number) prove nothing and enter this
                                                // same eager guard; the received logs are dropped.
                                                let treat_as_empty = !matches!(
                                                    watermark_bounds,
                                                    Some(YieldBounds::Full)
                                                );
                                                if treat_as_empty {
                                                    let http_provider = providers.connect_http(http_provider_idx);
                                                    let http_host = providers.http_settings(http_provider_idx).host();
                                                    match handle_empty_flush(
                                                        &mut coalescer,
                                                        &mut gate,
                                                        providers,
                                                        &http_host,
                                                        to_block,
                                                        true,
                                                        || http_provider.get_block_number(),
                                                    )
                                                    .await
                                                    {
                                                        EmptyFlushOutcome::Deferred => {
                                                            unreachable!(
                                                                "span_capped: true never defers"
                                                            )
                                                        }
                                                        EmptyFlushOutcome::VerifiedEmpty => {}
                                                        EmptyFlushOutcome::CaughtUp => {
                                                            continue 'flush
                                                        }
                                                        EmptyFlushOutcome::Stalled {
                                                            to_block,
                                                            reported_tip,
                                                            stalled_cycles,
                                                        } => {
                                                            yield Err(
                                                                RpcError::ProviderStalled {
                                                                    to_block,
                                                                    reported_tip,
                                                                    stalled_cycles,
                                                                },
                                                            );
                                                            return;
                                                        }
                                                        EmptyFlushOutcome::RepollFailed(err) => {
                                                            warn!(
                                                                "HTTP fallback {} tip re-poll failed during tip-wait for blocks {}-{}: {}",
                                                                http_host,
                                                                from_block,
                                                                to_block,
                                                                truncate_error_short(&err, 100),
                                                            );
                                                            should_switch_provider = true;
                                                            break 'flush;
                                                        }
                                                        EmptyFlushOutcome::TipFetchFailed(err) => {
                                                            warn!(
                                                                "HTTP fallback {} tip fetch failed during empty-batch guard for blocks {}-{}: {}",
                                                                http_host, from_block, to_block,
                                                                truncate_error_short(&err, 100)
                                                            );
                                                            should_switch_provider = true;
                                                            break 'flush;
                                                        }
                                                    }
                                                }

                                                let logs = if treat_as_empty {
                                                    Vec::new()
                                                } else {
                                                    logs
                                                };
                                                yield Ok((from_block, to_block, logs));
                                                coalescer.advance_cursor(to_block + 1);
                                                gate.reset();
                                            }
                                            Err(http_err) => {
                                                warn!(
                                                    "HTTP fallback retry exhausted for blocks {}-{}: {}",
                                                    from_block,
                                                    to_block,
                                                    truncate_error_short(&http_err, 100)
                                                );
                                                should_switch_provider = true;
                                                break;
                                            }
                                        }
                                    }
                                    ErrorCategory::RateLimit => {
                                        // Rate limit: apply backoff but stay on same provider
                                        provider_trackers[provider_index].record_error();
                                        let backoff = provider_trackers[provider_index].backoff_duration();
                                        warn!(
                                            "Rate limit hit on WS provider {} for blocks {}-{}: {} - backing off {:?}",
                                            provider_trackers[provider_index].identifier(),
                                            from_block, to_block,
                                            truncate_error_short(&error_msg, 100),
                                            backoff
                                        );
                                        sleep(backoff).await;
                                        // Retry same block range after backoff
                                        continue;
                                    }
                                    ErrorCategory::Connection | ErrorCategory::Other => {
                                        // Connection error or unknown: record error and switch provider
                                        let suspended = provider_trackers[provider_index].record_error();

                                        if suspended {
                                            warn!(
                                                "WebSocket provider {} suspended after error on blocks {}-{}: {}",
                                                provider_trackers[provider_index].identifier(),
                                                from_block, to_block,
                                                truncate_error_short(&error_msg, 100)
                                            );
                                        } else {
                                            let backoff = provider_trackers[provider_index].backoff_duration();
                                            warn!(
                                                "Error fetching logs for blocks {}-{} from {}: {} - backing off {:?}",
                                                from_block, to_block,
                                                provider_trackers[provider_index].identifier(),
                                                truncate_error_short(&error_msg, 100),
                                                backoff
                                            );
                                            sleep(backoff).await;
                                        }
                                        should_switch_provider = true;
                                        break;
                                    }
                                }
                            }
                        }
                    }

                    if should_switch_provider {
                        break;
                    }

                    continue;
                }

                let maybe_header = if let Some(deadline) = coalescer.flush_deadline() {
                    tokio::select! {
                        maybe_header = s.next() => maybe_header,
                        _ = sleep_until(deadline) => {
                            continue;
                        }
                    }
                } else {
                    s.next().await
                };

                match maybe_header {
                    Some(header) => {
                        let now = Instant::now();
                        let head_number = header.number;
                        if !coalescer.observe_head(head_number, now) {
                            let from_block = coalescer.block_cursor().unwrap_or(head_number);
                            warn!(
                                "Skipping header {} as it's before the cursor {}",
                                head_number, from_block
                            );
                        }

                        // Reorg detection runs alongside the coalescer's number-based
                        // flush — a parallel `(number, hash, parent_hash)` linkage check.
                        // Disabled (depth 0) ⇒ `observe` is inert and this is a no-op.
                        // On a reorg we emit and end the stream; the consumer decides
                        // resume-from-ancestor (P3 rollback) vs terminate (P2).
                        match linkage.observe(head_number, header.hash, header.parent_hash) {
                            LinkOutcome::Linked => {}
                            LinkOutcome::Reorg { common_ancestor } => {
                                warn!(
                                    "reorg at head {}: common ancestor {} — ending stream for consumer recovery",
                                    head_number, common_ancestor
                                );
                                yield Err(RpcError::Reorg { common_ancestor });
                                return;
                            }
                            LinkOutcome::Detached {
                                parent_hash,
                                child_number,
                            } => {
                                let err = resolve_reorg_ancestor(
                                    &provider,
                                    &linkage,
                                    parent_hash,
                                    child_number,
                                    providers.reorg_detect_depth,
                                )
                                .await;
                                match &err {
                                    RpcError::Reorg { common_ancestor } => warn!(
                                        "reorg at head {} (walked new chain): common ancestor {} — ending stream",
                                        head_number, common_ancestor
                                    ),
                                    _ => warn!(
                                        "reorg at head {} beyond the detector ring — ending stream",
                                        head_number
                                    ),
                                }
                                yield Err(err);
                                return;
                            }
                            LinkOutcome::BeyondFinality => {
                                warn!(
                                    "reorg beyond finality at head {}: no common ancestor within the detector ring — ending stream",
                                    head_number
                                );
                                yield Err(RpcError::ReorgBeyondFinality);
                                return;
                            }
                        }
                    }
                    None => break,
                }
            }

            // Stream ended (connection lost) - record as error and switch provider
            if !should_switch_provider {
                // Stream ended naturally without explicit error
                provider_trackers[provider_index].record_error();
                let backoff = provider_trackers[provider_index].backoff_duration();
                warn!(
                    "WS connection to {} lost or stream ended - reconnecting in {:?}",
                    provider_trackers[provider_index].identifier(),
                    backoff
                );
                sleep(backoff).await;
            }

            // Move to next provider for round-robin (same pattern as backfill)
            provider_index = (provider_index + 1) % num_providers;
        }
    }
}

pub async fn last_block(providers: &ProviderSettings) -> Result<u64, RpcError> {
    let num_providers = providers.http_providers.len();
    if num_providers == 0 {
        return Err(RpcError::ConnectionError(
            "No HTTP providers configured for latest block fetch".to_string(),
        ));
    }

    let mut index: usize = 0;
    loop {
        // Round-robin provider selection so latest-block fetch can fail over.
        let provider_index = index % num_providers;
        let provider = providers.connect_http(provider_index);
        match provider.get_block_number().await {
            Ok(block_number) => return Ok(block_number),
            Err(e) => {
                let provider_host = providers.http_settings(provider_index).host();
                if index >= 10 {
                    return Err(RpcError::ConnectionError(format!(
                        "Failed to fetch latest block number after {} attempts from HTTP provider {}: {}",
                        index, provider_host, e
                    )));
                }
                index += 1;
                warn!(
                    "Failed to fetch latest block number from HTTP provider {} (index {}) - retrying",
                    provider_host, provider_index
                );
                let progressive_sleep = (5 * index).min(50) as u64;
                sleep(Duration::from_secs(progressive_sleep)).await;
            }
        }
    }
}

/// Resolve the block number to use as the capture target based on `settings.capture_target`.
///
/// - `Latest` → current chain tip (existing behavior).
/// - `Lag(n)` → tip saturating_sub n.
/// - `Finalized` → the `finalized` tagged block; returns `Err` if the provider returns none.
///
/// Emits a one-time `warn!` when a deep-lag policy is active so operators notice
/// the intentional delay in their logs.
pub async fn resolve_capture_target(settings: &ProviderSettings) -> Result<u64, RpcError> {
    match settings.capture_target {
        CaptureTarget::Latest => {
            let tip = last_block(settings).await?;
            warn!(
                "ethl capture target is `Latest` (block {}): capturing to the raw chain \
                 tip with NO reorg protection — a reorg can rewrite recently captured \
                 blocks. Reorg-prone chains should opt into `Lag`/`Finalized`.",
                tip,
            );
            Ok(tip)
        }
        CaptureTarget::Lag(n) => {
            let tip = last_block(settings).await?;
            warn!(
                "ethl DEEP LAG ACTIVE (reorg-safety): capturing to tip - {} = block {}.\n\
                 This build AVOIDS reorgs by lagging; events within {} blocks of tip are DELAYED.",
                n,
                tip.saturating_sub(n),
                n,
            );
            Ok(tip.saturating_sub(n))
        }
        CaptureTarget::Finalized => {
            let provider = settings.connect_http(0);
            let result = provider
                .get_block_by_number(BlockNumberOrTag::Finalized)
                .await
                .map_err(|e| {
                    RpcError::ConnectionError(format!("finalized block fetch failed: {e}"))
                })?;
            match result {
                Some(block) => {
                    let finalized = block.header.number;
                    warn!(
                        "ethl DEEP LAG ACTIVE (reorg-safety): capturing to `finalized` block {}.\n\
                         (~2 epochs ≈ 64 blocks ≈ 13 min behind chain tip).\n\
                         This build AVOIDS reorgs by lagging; events within ~64 blocks of tip are DELAYED.",
                        finalized
                    );
                    Ok(finalized)
                }
                None => Err(RpcError::Other(
                    "Provider returned no finalized block; cannot determine safe capture target. \
                     Ensure the provider supports the `finalized` tag (mainnet only)."
                        .to_string(),
                )),
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::{
        EmptyFlushOutcome, HeadCoalescer, HeadLinkage, LinkOutcome, MAX_BLOCKS_PER_WS_REQUEST,
        RpcError, TipGate, YieldBounds, collect_window_hashes, fallback_http_exhausted_error,
        fallback_http_index, fetch_logs_by_hash_range, flush_window_range, handle_empty_flush,
        resolve_reorg_ancestor, yield_bounds,
    };
    use crate::rpc::config::ProviderSettings;
    use alloy::primitives::B256;
    use alloy::providers::{Provider, ProviderBuilder};
    use alloy::rpc::types::Log;
    use alloy::transports::mock::Asserter;
    use std::time::Duration;
    use tokio::time::Instant;

    /// Coalesce window for tests that previously anchored timing assertions on
    /// the module's `LIVE_HEAD_COALESCE_WINDOW` const, before it became the
    /// per-instance `HeadCoalescer::window` field.
    const WINDOW: Duration = Duration::from_millis(500);

    /// Distinct synthetic block hash; `byte` keeps old/new chains separable.
    fn hb(byte: u8) -> B256 {
        B256::repeat_byte(byte)
    }

    #[test]
    fn fallback_http_index_wraps_round_robin() {
        assert_eq!(fallback_http_index(0, 0, 3), 0);
        assert_eq!(fallback_http_index(0, 1, 3), 1);
        assert_eq!(fallback_http_index(0, 2, 3), 2);
        assert_eq!(fallback_http_index(0, 3, 3), 0);
        assert_eq!(fallback_http_index(2, 1, 3), 0);
    }

    #[test]
    fn fallback_http_exhausted_error_includes_hosts() {
        let err = fallback_http_exhausted_error(
            100,
            120,
            &["rpc.ankr.com".to_string(), "mainnet.infura.io".to_string()],
            "timeout",
        )
        .to_string();
        assert!(err.contains("rpc.ankr.com"));
        assert!(err.contains("mainnet.infura.io"));
        assert!(err.contains("timeout"));
    }

    #[test]
    fn head_coalescer_ignores_stale_heads_before_cursor() {
        let now = Instant::now();
        let mut coalescer = HeadCoalescer::new(Some(42), WINDOW);

        assert!(!coalescer.observe_head(41, now));
        assert_eq!(coalescer.pending_range(), None);
        assert_eq!(coalescer.flush_deadline(), None);
    }

    #[test]
    fn head_coalescer_keeps_first_flush_deadline_while_heads_accumulate() {
        let now = Instant::now();
        let mut coalescer = HeadCoalescer::new(Some(100), WINDOW);

        assert!(coalescer.observe_head(100, now));
        let deadline = coalescer.flush_deadline().unwrap();

        assert!(coalescer.observe_head(105, now + Duration::from_millis(50)));
        assert_eq!(coalescer.pending_range(), Some((100, 105)));
        assert_eq!(coalescer.flush_deadline(), Some(deadline));
        assert!(!coalescer.should_flush(now + WINDOW - Duration::from_millis(1)));
        assert!(coalescer.should_flush(now + WINDOW));
    }

    #[test]
    fn head_coalescer_flushes_immediately_when_pending_range_reaches_chunk_limit() {
        let now = Instant::now();
        let mut coalescer = HeadCoalescer::new(Some(1), WINDOW);

        assert!(coalescer.observe_head(MAX_BLOCKS_PER_WS_REQUEST, now));
        assert_eq!(
            coalescer.pending_range(),
            Some((1, MAX_BLOCKS_PER_WS_REQUEST))
        );
        assert!(coalescer.should_flush(now));
    }

    #[test]
    fn head_coalescer_preserves_pending_range_across_partial_cursor_advances() {
        let now = Instant::now();
        let mut coalescer = HeadCoalescer::new(Some(10), WINDOW);

        assert!(coalescer.observe_head(25, now));
        coalescer.advance_cursor(20);
        assert_eq!(coalescer.pending_range(), Some((20, 25)));
        assert!(coalescer.flush_deadline().is_some());

        coalescer.advance_cursor(26);
        assert_eq!(coalescer.pending_range(), None);
        assert_eq!(coalescer.flush_deadline(), None);
        assert_eq!(coalescer.block_cursor(), Some(26));
    }

    #[test]
    fn head_coalescer_handles_mainnet_slow_blocks() {
        // Ethereum mainnet: ~12 s/block. Each head must flush in a single 500 ms coalesce window;
        // the chunk-limit guard (MAX_BLOCKS_PER_WS_REQUEST) must never trigger for live cadence.
        let now = Instant::now();
        let mut coalescer = HeadCoalescer::new(Some(1_000_000), WINDOW);

        // First mainnet head observed.
        assert!(coalescer.observe_head(1_000_000, now));
        assert_eq!(coalescer.pending_range(), Some((1_000_000, 1_000_000)));

        // Must not flush before the coalesce window expires.
        assert!(!coalescer.should_flush(now + Duration::from_millis(499)));
        // Must flush once the window expires.
        assert!(coalescer.should_flush(now + WINDOW));

        // Advance cursor; simulate the next block arriving ~12 s later.
        coalescer.advance_cursor(1_000_001);
        assert_eq!(coalescer.pending_range(), None);

        let t12s = now + Duration::from_secs(12);
        assert!(coalescer.observe_head(1_000_001, t12s));
        assert_eq!(coalescer.pending_range(), Some((1_000_001, 1_000_001)));

        // One block in range: chunk limit (300) must not fire.
        assert!(!coalescer.should_flush(t12s));
        assert!(coalescer.should_flush(t12s + WINDOW));
    }

    /// The coalesce window is read from the constructed instance, not a fixed
    /// constant: two coalescers built with different windows must each flush at
    /// their own deadline, not a shared hard-coded value.
    #[test]
    fn head_coalescer_honors_its_own_window() {
        let now = Instant::now();
        let short_window = Duration::from_millis(100);
        let long_window = Duration::from_secs(5);

        let mut short = HeadCoalescer::new(Some(1), short_window);
        assert!(short.observe_head(1, now));
        assert!(!short.should_flush(now + short_window - Duration::from_millis(1)));
        assert!(short.should_flush(now + short_window));

        let mut long = HeadCoalescer::new(Some(1), long_window);
        assert!(long.observe_head(1, now));
        assert!(!long.should_flush(now + long_window - Duration::from_millis(1)));
        assert!(long.should_flush(now + long_window));
    }

    #[test]
    fn head_coalescer_defer_flush_clears_deadline_only() {
        let now = Instant::now();
        let mut coalescer = HeadCoalescer::new(Some(1), WINDOW);
        assert!(coalescer.observe_head(300, now));
        assert_eq!(coalescer.pending_range(), Some((1, 300)));
        assert!(coalescer.flush_deadline().is_some());

        coalescer.defer_flush();
        assert_eq!(
            coalescer.flush_deadline(),
            None,
            "defer_flush must clear the deadline"
        );
        assert_eq!(
            coalescer.pending_range(),
            Some((1, 300)),
            "defer_flush must not touch the pending range"
        );

        // A 300-wide span still should_flush()es with no deadline armed — the
        // span-cap arm of should_flush is deadline-free.
        assert!(coalescer.should_flush(now));
    }

    /// After a `Partial(b)` yield advances the cursor to `b + 1` and defers,
    /// the deferred-empty pattern applies verbatim: the deadline clears, the
    /// pending range starts at the unproven tail, and the next head re-arms.
    #[test]
    fn coalescer_partial_defer_reserves_unproven_tail_and_rearms() {
        let now = Instant::now();
        let mut coalescer = HeadCoalescer::new(Some(100), WINDOW);
        coalescer.observe_head(105, now);
        assert_eq!(coalescer.pending_range(), Some((100, 105)));

        // Simulate a Partial(102) yield: cursor advances only through the
        // proven watermark, then the unproven tail defers.
        coalescer.advance_cursor(103);
        coalescer.defer_flush();

        assert_eq!(
            coalescer.flush_deadline(),
            None,
            "defer_flush must clear the deadline after a Partial yield"
        );
        assert_eq!(
            coalescer.pending_range(),
            Some((103, 105)),
            "pending range must start at watermark+1, the unproven tail"
        );

        // The next head's flush re-arms the deadline over the still-unproven tail.
        assert!(coalescer.observe_head(105, now + Duration::from_millis(1)));
        assert!(coalescer.flush_deadline().is_some());
    }

    // ==================== log-watermark bound (yield_bounds) ====================

    /// Builds a `Log` fixture via the JSON shape already used for the
    /// `fetch_logs_by_hash_range` test (`"blockNumber": format!("0x{:x}", n)`),
    /// nulling `blockNumber` for the `None` case (the field is required by
    /// `Log`'s deserializer, so it cannot simply be omitted).
    fn log_fixture(block_number: Option<u64>) -> Log {
        let value = serde_json::json!({
            "address": "0x0000000000000000000000000000000000000001",
            "topics": [],
            "data": "0x",
            "blockHash": hb(0).to_string(),
            "blockNumber": block_number.map(|n| format!("0x{:x}", n)),
            "transactionHash": hb(0).to_string(),
            "transactionIndex": "0x0",
            "logIndex": "0x0",
            "removed": false
        });
        serde_json::from_value(value).expect("valid Log fixture")
    }

    #[test]
    fn yield_bounds_full_when_tail_has_logs() {
        let logs = vec![log_fixture(Some(101)), log_fixture(Some(102))];
        assert!(matches!(yield_bounds(&logs, 102), YieldBounds::Full));
    }

    #[test]
    fn yield_bounds_partial_stops_at_watermark() {
        let logs = vec![log_fixture(Some(101))];
        assert!(matches!(
            yield_bounds(&logs, 102),
            YieldBounds::Partial(101)
        ));
    }

    /// Logs at 101 and 104 with `to = 105`: blocks 102-103 are vouched by the
    /// 104 log (a log at block B proves the provider's index reached >= B);
    /// only 105, above the watermark, is unproven.
    #[test]
    fn yield_bounds_interior_gaps_are_proven() {
        let logs = vec![log_fixture(Some(101)), log_fixture(Some(104))];
        assert!(matches!(
            yield_bounds(&logs, 105),
            YieldBounds::Partial(104)
        ));
    }

    #[test]
    fn yield_bounds_mixed_none_uses_some_only() {
        let logs = vec![log_fixture(Some(101)), log_fixture(None)];
        assert!(matches!(
            yield_bounds(&logs, 102),
            YieldBounds::Partial(101)
        ));
    }

    #[test]
    fn yield_bounds_all_none_is_unattributable() {
        let logs = vec![log_fixture(None), log_fixture(None)];
        assert!(matches!(
            yield_bounds(&logs, 999),
            YieldBounds::Unattributable
        ));
    }

    /// A log claiming a block above `to_block` is a provider integrity
    /// violation (a range query cannot legally return higher blocks); clamp
    /// to `Full` rather than advance the cursor past the requested range.
    #[test]
    fn yield_bounds_watermark_above_range_clamps() {
        let logs = vec![log_fixture(Some(103))];
        assert!(matches!(yield_bounds(&logs, 102), YieldBounds::Full));
    }

    // ==================== lazy empty-verification (handle_empty_flush) ====================

    /// Non-capped empty flush defers without calling the tip at all. A reverted
    /// per-window confirm would call `get_block_number` and, against an asserter
    /// with zero queued responses, panic on the exhausted queue instead of
    /// returning `Deferred`.
    #[tokio::test]
    async fn lazy_empty_non_capped_defers_without_tip_call() {
        let asserter = Asserter::new(); // zero queued responses
        let settings = ProviderSettings::from_mock(
            ProviderBuilder::new()
                .disable_recommended_fillers()
                .connect_mocked_client(asserter)
                .erased(),
        );
        let http_provider = settings.connect_http(0);

        let mut coalescer = HeadCoalescer::new(Some(100), WINDOW);
        coalescer.observe_head(105, Instant::now());
        assert_eq!(coalescer.pending_range(), Some((100, 105)));

        let mut gate = TipGate::new();
        let outcome = handle_empty_flush(
            &mut coalescer,
            &mut gate,
            &settings,
            "host",
            105,
            false,
            || http_provider.get_block_number(),
        )
        .await;

        assert!(
            matches!(outcome, EmptyFlushOutcome::Deferred),
            "expected Deferred"
        );
        assert_eq!(coalescer.flush_deadline(), None);
        assert_eq!(
            coalescer.pending_range(),
            Some((100, 105)),
            "cursor must stay unmoved by a deferred flush"
        );
    }

    /// After a deferred flush, the next `observe_head` re-arms the deadline and
    /// the pending range still starts at the ORIGINAL (unmoved) cursor — the
    /// self-heal shape: the next flush re-covers the deferred blocks.
    #[test]
    fn lazy_deferred_flush_rearms_and_spans_from_unmoved_cursor() {
        let now = Instant::now();
        let mut coalescer = HeadCoalescer::new(Some(100), WINDOW);
        coalescer.observe_head(105, now);
        assert_eq!(coalescer.pending_range(), Some((100, 105)));

        coalescer.defer_flush();
        assert_eq!(coalescer.flush_deadline(), None);

        coalescer.observe_head(110, now + Duration::from_millis(10));
        assert_eq!(
            coalescer.pending_range(),
            Some((100, 110)),
            "deferred span must extend from the original unmoved cursor"
        );
        assert!(
            coalescer.flush_deadline().is_some(),
            "the next observe_head must re-arm the deadline"
        );
    }

    /// Span-capped chunk, tip already at or past `to_block`: verified empty on
    /// the first tip check, no `wait_for_tip` episode.
    #[tokio::test]
    async fn span_cap_empty_tip_current_is_verified_empty() {
        let asserter = Asserter::new();
        asserter.push_success(&serde_json::json!("0x64")); // get_block_number → 100
        let settings = ProviderSettings::from_mock(
            ProviderBuilder::new()
                .disable_recommended_fillers()
                .connect_mocked_client(asserter)
                .erased(),
        );
        let http_provider = settings.connect_http(0);

        let mut coalescer = HeadCoalescer::new(Some(1), WINDOW);
        coalescer.observe_head(100, Instant::now());
        let mut gate = TipGate::new();

        let outcome = handle_empty_flush(
            &mut coalescer,
            &mut gate,
            &settings,
            "host",
            100,
            true,
            || http_provider.get_block_number(),
        )
        .await;

        assert!(
            matches!(outcome, EmptyFlushOutcome::VerifiedEmpty),
            "expected VerifiedEmpty"
        );
    }

    /// Span-capped chunk, tip behind `to_block`: enters `wait_for_tip`, which
    /// catches up on the first re-poll. `from_mock`'s `tip_wait_backoff_secs: 0`
    /// keeps the wait sleepless.
    #[tokio::test]
    async fn span_cap_empty_tip_behind_enters_tip_wait() {
        let asserter = Asserter::new();
        asserter.push_success(&serde_json::json!("0x63")); // initial tip → 99 (< to_block)
        asserter.push_success(&serde_json::json!("0x64")); // re-poll → 100 (caught up)
        let settings = ProviderSettings::from_mock(
            ProviderBuilder::new()
                .disable_recommended_fillers()
                .connect_mocked_client(asserter)
                .erased(),
        );
        let http_provider = settings.connect_http(0);

        let mut coalescer = HeadCoalescer::new(Some(1), WINDOW);
        coalescer.observe_head(100, Instant::now());
        let mut gate = TipGate::new();

        let outcome = handle_empty_flush(
            &mut coalescer,
            &mut gate,
            &settings,
            "host",
            100,
            true,
            || http_provider.get_block_number(),
        )
        .await;

        assert!(
            matches!(outcome, EmptyFlushOutcome::CaughtUp),
            "expected CaughtUp"
        );
    }

    /// Span-capped chunk, tip flat across every poll: `wait_for_tip` terminates
    /// via the stall gate. This plus the previous test is the Base-outage
    /// invariant — reverting the cap guard (making the span-capped arm defer
    /// too) fails both.
    #[tokio::test]
    async fn span_cap_empty_flat_tip_stalls() {
        let asserter = Asserter::new();
        asserter.push_success(&serde_json::json!("0x5a")); // initial tip → 90
        asserter.push_success(&serde_json::json!("0x5a")); // re-poll cycle 1 → 90 (flat)
        asserter.push_success(&serde_json::json!("0x5a")); // re-poll cycle 2 → 90 (flat, hits cap)
        let settings = ProviderSettings::from_mock(
            ProviderBuilder::new()
                .disable_recommended_fillers()
                .connect_mocked_client(asserter)
                .erased(),
        )
        .with_tip_wait_max_stall_cycles(2);
        let http_provider = settings.connect_http(0);

        let mut coalescer = HeadCoalescer::new(Some(1), WINDOW);
        coalescer.observe_head(100, Instant::now());
        let mut gate = TipGate::new();

        let outcome = handle_empty_flush(
            &mut coalescer,
            &mut gate,
            &settings,
            "host",
            100,
            true,
            || http_provider.get_block_number(),
        )
        .await;

        assert!(
            matches!(
                outcome,
                EmptyFlushOutcome::Stalled {
                    to_block: 100,
                    reported_tip: 90,
                    stalled_cycles: 2,
                }
            ),
            "expected Stalled(stalled_cycles=2), got non-matching outcome"
        );
    }

    /// F1 (0.1.29 core soundness audit): a `TipGate` is shared across
    /// provider sessions within one stream. Phase 1 reproduces the hazard —
    /// session A leaves the gate dirty (`RepollFailed`, the provider-switch
    /// path), and session B, run against the SAME never-reset gate, has its
    /// first observation miscounted as a regression instead of a fresh seed,
    /// so it hits `Stall(5)` on only 5 total tip polls instead of the 6 a
    /// freshly-seeded gate would need — one cycle early. Phase 2 shows the
    /// fix (`gate.reset()` between sessions, exactly what the connect-site
    /// change does): session B now correctly consumes 6 polls before
    /// `Stall(5)`. Each phase queues EXACTLY the responses its session B
    /// consumes, so an extra (correct-behavior) poll fails on an unqueued
    /// response rather than silently passing with the wrong arithmetic.
    #[tokio::test]
    async fn handle_empty_flush_stale_gate_stalls_one_cycle_early_without_reset() {
        // ---- Phase 1: no reset between sessions -- reproduce the miscount.
        let asserter = Asserter::new();
        // Session A: tip 990 (below to_block 999), then a transport error on
        // the wait-for-tip re-poll -- the RepollFailed provider-switch path
        // that leaves the gate dirty (audit F1 step 4-5).
        asserter.push_success(&serde_json::json!("0x3de")); // 990
        asserter.push_failure_msg("transport blip");
        // Session B, same gate, NOT reset: flat tip 988 (below A's stale
        // last_tip=990), so B's first observation counts as a regression
        // instead of a seed. 5 total get_tip calls (1 initial + 4 re-polls)
        // reach Stall(5) here -- one call short of a fresh gate's 6.
        for _ in 0..5 {
            asserter.push_success(&serde_json::json!("0x3dc")); // 988
        }
        let settings = ProviderSettings::from_mock(
            ProviderBuilder::new()
                .disable_recommended_fillers()
                .connect_mocked_client(asserter)
                .erased(),
        );
        let http_provider = settings.connect_http(0);

        let mut coalescer = HeadCoalescer::new(Some(1), WINDOW);
        coalescer.observe_head(999, Instant::now());
        let mut gate = TipGate::new();

        let session_a = handle_empty_flush(
            &mut coalescer,
            &mut gate,
            &settings,
            "host-a",
            999,
            true,
            || http_provider.get_block_number(),
        )
        .await;
        assert!(
            matches!(session_a, EmptyFlushOutcome::RepollFailed(_)),
            "expected RepollFailed for session A"
        );

        // Provider switch happens here in production, WITHOUT gate.reset() --
        // this is the bug this fix corrects.
        let session_b = handle_empty_flush(
            &mut coalescer,
            &mut gate,
            &settings,
            "host-b",
            999,
            true,
            || http_provider.get_block_number(),
        )
        .await;
        assert!(
            matches!(
                session_b,
                EmptyFlushOutcome::Stalled {
                    to_block: 999,
                    reported_tip: 988,
                    stalled_cycles: 5,
                }
            ),
            "expected Stalled(stalled_cycles=5) one cycle early from the stale gate"
        );

        // ---- Phase 2: gate.reset() between sessions -- the fixed contract.
        let asserter = Asserter::new();
        asserter.push_success(&serde_json::json!("0x3de")); // 990
        asserter.push_failure_msg("transport blip");
        // Session B now gets a fresh seed observation, so it needs one MORE
        // flat poll (6 total: 1 initial + 5 re-polls) before Stall(5).
        for _ in 0..6 {
            asserter.push_success(&serde_json::json!("0x3dc")); // 988
        }
        let settings = ProviderSettings::from_mock(
            ProviderBuilder::new()
                .disable_recommended_fillers()
                .connect_mocked_client(asserter)
                .erased(),
        );
        let http_provider = settings.connect_http(0);

        let mut coalescer = HeadCoalescer::new(Some(1), WINDOW);
        coalescer.observe_head(999, Instant::now());
        let mut gate = TipGate::new();

        let session_a = handle_empty_flush(
            &mut coalescer,
            &mut gate,
            &settings,
            "host-a",
            999,
            true,
            || http_provider.get_block_number(),
        )
        .await;
        assert!(
            matches!(session_a, EmptyFlushOutcome::RepollFailed(_)),
            "expected RepollFailed for session A"
        );

        // The connect-site fix: reset the gate before the new session.
        gate.reset();

        let session_b = handle_empty_flush(
            &mut coalescer,
            &mut gate,
            &settings,
            "host-b",
            999,
            true,
            || http_provider.get_block_number(),
        )
        .await;
        assert!(
            matches!(
                session_b,
                EmptyFlushOutcome::Stalled {
                    to_block: 999,
                    reported_tip: 988,
                    stalled_cycles: 5,
                }
            ),
            "expected Stalled(stalled_cycles=5) reached only after the full \
             6-poll budget once the gate is reset"
        );
    }

    /// Synthetic block with explicit `number` and `parentHash`.
    ///
    /// Use this for tests that exercise `collect_window_hashes`, which now
    /// cross-checks `block.header.number == current_number` (fail-closed).
    fn block_with_parent_at(number: u64, parent: B256) -> serde_json::Value {
        let zero = hb(0).to_string();
        serde_json::json!({
            "hash":             zero,
            "parentHash":       parent.to_string(),
            "sha3Uncles":       zero,
            "miner":            "0x0000000000000000000000000000000000000000",
            "stateRoot":        zero,
            "transactionsRoot": zero,
            "receiptsRoot":     zero,
            "logsBloom":        format!("0x{}", "0".repeat(512)),
            "difficulty":       "0x0",
            "number":           format!("0x{:x}", number),
            "gasLimit":         "0x0",
            "gasUsed":          "0x0",
            "timestamp":        "0x0",
            "extraData":        "0x",
            "mixHash":          zero,
            "nonce":            "0x0000000000000000",
            "transactions":     [],
            "uncles":           []
        })
    }

    /// Synthetic block whose only meaningful field is `parentHash`.
    ///
    /// For `resolve_reorg_ancestor` tests, the block number is not checked, so
    /// passing 0 is fine. Use `block_with_parent_at` when `collect_window_hashes`
    /// is under test (it cross-checks the block number).
    fn block_with_parent(parent: B256) -> serde_json::Value {
        block_with_parent_at(0, parent)
    }

    /// Seed a linkage with a clean canonical chain `[lo, hi]` (hash = `hb(n)`).
    fn seed_chain(depth: u64, lo: u64, hi: u64) -> HeadLinkage {
        let mut linkage = HeadLinkage::new(depth);
        for n in lo..=hi {
            let parent = if n == lo {
                hb((lo - 1) as u8)
            } else {
                hb((n - 1) as u8)
            };
            assert_eq!(linkage.observe(n, hb(n as u8), parent), LinkOutcome::Linked);
        }
        linkage
    }

    #[test]
    fn head_linkage_disabled_is_inert() {
        let mut linkage = HeadLinkage::new(0);
        // A blatant parent mismatch still returns Linked when depth == 0.
        assert_eq!(linkage.observe(100, hb(1), hb(9)), LinkOutcome::Linked);
        assert_eq!(linkage.observe(101, hb(2), hb(7)), LinkOutcome::Linked);
        assert_eq!(linkage.floor(), None);
    }

    #[test]
    fn head_linkage_clean_chain_advances_and_bounds_ring() {
        let linkage = seed_chain(3, 100, 103);
        // Ring is bounded at depth = 3; the oldest entry (100) is evicted.
        assert_eq!(linkage.floor(), Some(101));
        assert_eq!(linkage.hash_at(100), None);
        assert_eq!(linkage.hash_at(103), Some(hb(103)));
    }

    #[test]
    fn head_linkage_one_block_reorg_reports_parent_as_ancestor() {
        let mut linkage = seed_chain(8, 100, 103);
        // 103' replaces 103, still building on canonical 102 → ancestor 102.
        assert_eq!(
            linkage.observe(103, hb(203), hb(102)),
            LinkOutcome::Reorg {
                common_ancestor: 102
            }
        );
        // Ring re-anchored: 103 now records the new hash, clean extension resumes.
        assert_eq!(linkage.hash_at(103), Some(hb(203)));
        assert_eq!(linkage.observe(104, hb(204), hb(203)), LinkOutcome::Linked);
    }

    #[test]
    fn head_linkage_deep_reorg_via_reemit_reports_deeper_ancestor() {
        let mut linkage = seed_chain(8, 100, 105);
        // Provider re-emits the new chain from 101 (parent = canonical 100): a
        // five-block-deep reorg, resolved in-ring at common ancestor 100.
        assert_eq!(
            linkage.observe(101, hb(201), hb(100)),
            LinkOutcome::Reorg {
                common_ancestor: 100
            }
        );
    }

    #[test]
    fn head_linkage_skipped_intermediate_heads_detach() {
        let mut linkage = seed_chain(8, 100, 103);
        // Provider jumps to new tip 104' whose parent 103' (hb(203)) we never saw;
        // the parent at 103 differs from canonical → Detached, needs an RPC walk.
        match linkage.observe(104, hb(204), hb(203)) {
            LinkOutcome::Detached {
                parent_hash,
                child_number,
            } => {
                assert_eq!(parent_hash, hb(203));
                assert_eq!(child_number, 104);
            }
            other => panic!("expected Detached, got {other:?}"),
        }
    }

    #[test]
    fn head_linkage_divergence_below_floor_is_beyond_finality() {
        // Ring covers [103, 105]; floor = 103.
        let mut linkage = seed_chain(3, 100, 105);
        assert_eq!(linkage.floor(), Some(103));
        // A head whose parent (102) is below the ring floor cannot be resolved,
        // and no ancestor is fabricated.
        assert_eq!(
            linkage.observe(103, hb(203), hb(202)),
            LinkOutcome::BeyondFinality
        );
    }

    #[tokio::test]
    async fn resolve_reorg_ancestor_walks_skipped_chain_to_common_ancestor() {
        // Canonical ring [100, 103]; new chain shares ancestor 101.
        let linkage = seed_chain(8, 100, 103);
        // New tip 104' detached at parent 103' (hb(203)); the walk fetches
        // 103' (parent 102' = hb(202)) then 102' (parent 101 = hb(101) ∈ ring).
        let asserter = Asserter::new();
        asserter.push_success(&block_with_parent(hb(202)));
        asserter.push_success(&block_with_parent(hb(101)));
        let provider = ProviderBuilder::new()
            .disable_recommended_fillers()
            .connect_mocked_client(asserter)
            .erased();

        let err = resolve_reorg_ancestor(&provider, &linkage, hb(203), 104, 8).await;
        assert!(
            matches!(
                err,
                RpcError::Reorg {
                    common_ancestor: 101
                }
            ),
            "got {err:?}"
        );
    }

    #[tokio::test]
    async fn resolve_reorg_ancestor_exhausts_ring_as_beyond_finality() {
        // Shallow ring [102, 103] (depth 2); floor = 102.
        let linkage = seed_chain(2, 100, 103);
        assert_eq!(linkage.floor(), Some(102));
        // New tip 104' detached at 103' (hb(203)); the walk fetches 103' (parent
        // 102' = hb(202)), which differs from canonical 102 at the floor.
        let asserter = Asserter::new();
        asserter.push_success(&block_with_parent(hb(202)));
        let provider = ProviderBuilder::new()
            .disable_recommended_fillers()
            .connect_mocked_client(asserter)
            .erased();

        let err = resolve_reorg_ancestor(&provider, &linkage, hb(203), 104, 2).await;
        assert!(matches!(err, RpcError::ReorgBeyondFinality), "got {err:?}");
    }

    // ── Task A: pure observe shapes ──────────────────────────────────────────

    #[test]
    fn head_linkage_back_to_back_reorgs_ring_stays_coherent() {
        // Seed canonical [100, 105]; depth=8 so all fit and no eviction yet.
        let mut linkage = seed_chain(8, 100, 105);

        // ForkA: 104' replaces 104, building on canonical 103 → ancestor 103.
        assert_eq!(
            linkage.observe(104, hb(204), hb(103)),
            LinkOutcome::Reorg {
                common_ancestor: 103
            }
        );
        // Ring re-anchored at 103; 104 now records new-chain hash.
        assert_eq!(linkage.hash_at(104), Some(hb(204)));

        // Extend forkA's chain by one clean block.
        assert_eq!(linkage.observe(105, hb(205), hb(204)), LinkOutcome::Linked);

        // ForkB: a competing 105'' builds on forkA's 104 (canonical ancestor 104).
        // Use hb(49) for forkB's hash — a byte distinct from forkA's range (hb(200+)).
        assert_eq!(
            linkage.observe(105, hb(49), hb(204)),
            LinkOutcome::Reorg {
                common_ancestor: 104
            }
        );
        // Ring re-anchored at 104; 105 records forkB hash; ring stays bounded.
        assert_eq!(linkage.hash_at(105), Some(hb(49)));
        assert_eq!(linkage.hash_at(104), Some(hb(204))); // unchanged by forkB
        assert!(linkage.ring.len() as u64 <= 8);
    }

    #[test]
    fn head_linkage_multi_header_same_height_duplicate_vs_replace() {
        let mut linkage = seed_chain(8, 100, 104);
        // Re-observing the canonical tip with its exact hash → no-op (Linked).
        assert_eq!(linkage.observe(104, hb(104), hb(103)), LinkOutcome::Linked);
        assert_eq!(linkage.hash_at(104), Some(hb(104)));
        // A competing block at the same height with a different hash, same parent
        // → replace, ancestor = 103.
        assert_eq!(
            linkage.observe(104, hb(204), hb(103)),
            LinkOutcome::Reorg {
                common_ancestor: 103
            }
        );
        assert_eq!(linkage.hash_at(104), Some(hb(204)));
    }

    #[test]
    fn head_linkage_identical_hash_no_op() {
        // The `hash_at(number) == Some(hash)` branch: re-observing an already
        // recorded (number, hash) pair returns Linked and leaves the ring alone.
        let mut linkage = seed_chain(8, 100, 104);
        let floor_before = linkage.floor();
        let depth_before = linkage.ring.len();
        assert_eq!(linkage.observe(104, hb(104), hb(103)), LinkOutcome::Linked);
        assert_eq!(linkage.floor(), floor_before);
        assert_eq!(linkage.ring.len(), depth_before);
        assert_eq!(linkage.hash_at(104), Some(hb(104)));
    }

    #[test]
    fn head_linkage_forward_gap_reanchors_without_false_reorg() {
        // `number > tip + 1`: heads skipped ahead (e.g. across a reconnect).
        // The distinct arm clears the ring + re-anchors at the new head — no reorg.
        let mut linkage = seed_chain(8, 100, 103); // tip = 103
        assert_eq!(linkage.observe(107, hb(107), hb(106)), LinkOutcome::Linked);
        // Old entries gone; ring re-anchored at 107 only.
        assert_eq!(linkage.floor(), Some(107));
        assert_eq!(linkage.ring.len(), 1);
        assert_eq!(linkage.hash_at(103), None);
        // Clean extension from the new tip continues normally.
        assert_eq!(linkage.observe(108, hb(108), hb(107)), LinkOutcome::Linked);
        assert_eq!(linkage.floor(), Some(107));
        assert_eq!(linkage.ring.len(), 2);
    }

    #[test]
    fn head_linkage_idempotent_reanchor_bounds_ring() {
        // After a reorg, a clean chain links cleanly and the ring stays bounded.
        let mut linkage = seed_chain(5, 100, 106); // ring [102,103,104,105,106] (evicted 100,101)
        assert_eq!(linkage.floor(), Some(102));

        // Reorg: 105' replaces 105, ancestor 104.
        assert_eq!(
            linkage.observe(105, hb(205), hb(104)),
            LinkOutcome::Reorg {
                common_ancestor: 104
            }
        );
        assert_eq!(linkage.hash_at(105), Some(hb(205)));

        // Extend cleanly from the new chain tip.
        assert_eq!(linkage.observe(106, hb(206), hb(205)), LinkOutcome::Linked);
        assert_eq!(linkage.observe(107, hb(207), hb(206)), LinkOutcome::Linked);

        // Ring still bounded at depth 5; oldest entry evicted.
        assert!(linkage.ring.len() as u64 <= 5);
        assert_eq!(linkage.hash_at(105), Some(hb(205))); // new-chain hash retained
        assert_eq!(linkage.hash_at(102), None); // evicted
    }

    // ── Task B: resolve_reorg_ancestor walk-depth matrix ────────────────────

    // Shared ring for the in-window rows: depth=15, canonical [100,109], floor=100,
    // tip=109. Each test detaches at child=111 (parent_number=110 ∉ ring).
    // k pushes → common ancestor at 110-k. Rings are distinct from catastrophe rows.

    #[tokio::test]
    async fn resolve_reorg_ancestor_walk_depth_1() {
        let linkage = seed_chain(15, 100, 109);
        let asserter = Asserter::new();
        asserter.push_success(&block_with_parent(hb(109))); // 110'→canonical 109
        let provider = ProviderBuilder::new()
            .disable_recommended_fillers()
            .connect_mocked_client(asserter)
            .erased();
        let err = resolve_reorg_ancestor(&provider, &linkage, hb(200), 111, 15).await;
        assert!(
            matches!(
                err,
                RpcError::Reorg {
                    common_ancestor: 109
                }
            ),
            "got {err:?}"
        );
    }

    #[tokio::test]
    async fn resolve_reorg_ancestor_walk_depth_2() {
        let linkage = seed_chain(15, 100, 109);
        let asserter = Asserter::new();
        asserter.push_success(&block_with_parent(hb(201))); // 110'→109'
        asserter.push_success(&block_with_parent(hb(108))); // 109'→canonical 108
        let provider = ProviderBuilder::new()
            .disable_recommended_fillers()
            .connect_mocked_client(asserter)
            .erased();
        let err = resolve_reorg_ancestor(&provider, &linkage, hb(200), 111, 15).await;
        assert!(
            matches!(
                err,
                RpcError::Reorg {
                    common_ancestor: 108
                }
            ),
            "got {err:?}"
        );
    }

    #[tokio::test]
    async fn resolve_reorg_ancestor_walk_depth_3() {
        let linkage = seed_chain(15, 100, 109);
        let asserter = Asserter::new();
        asserter.push_success(&block_with_parent(hb(201))); // 110'→109'
        asserter.push_success(&block_with_parent(hb(202))); // 109'→108'
        asserter.push_success(&block_with_parent(hb(107))); // 108'→canonical 107
        let provider = ProviderBuilder::new()
            .disable_recommended_fillers()
            .connect_mocked_client(asserter)
            .erased();
        let err = resolve_reorg_ancestor(&provider, &linkage, hb(200), 111, 15).await;
        assert!(
            matches!(
                err,
                RpcError::Reorg {
                    common_ancestor: 107
                }
            ),
            "got {err:?}"
        );
    }

    #[tokio::test]
    async fn resolve_reorg_ancestor_walk_depth_7() {
        let linkage = seed_chain(15, 100, 109);
        let asserter = Asserter::new();
        // Six intermediate new-chain blocks (hb(201)..hb(206)), then canonical 103.
        asserter.push_success(&block_with_parent(hb(201)));
        asserter.push_success(&block_with_parent(hb(202)));
        asserter.push_success(&block_with_parent(hb(203)));
        asserter.push_success(&block_with_parent(hb(204)));
        asserter.push_success(&block_with_parent(hb(205)));
        asserter.push_success(&block_with_parent(hb(206)));
        asserter.push_success(&block_with_parent(hb(103))); // final→canonical 103
        let provider = ProviderBuilder::new()
            .disable_recommended_fillers()
            .connect_mocked_client(asserter)
            .erased();
        let err = resolve_reorg_ancestor(&provider, &linkage, hb(200), 111, 15).await;
        assert!(
            matches!(
                err,
                RpcError::Reorg {
                    common_ancestor: 103
                }
            ),
            "got {err:?}"
        );
    }

    #[tokio::test]
    async fn resolve_reorg_ancestor_deepest_in_ring_resolves() {
        // Ring depth=5, canonical [100,104], floor=100. Child=105 → 4 fetches →
        // ancestor=100 (the floor). With depth=5 the loop runs 5 iterations: the
        // match check at the start of iter 4 fires after 4 prior fetches bring
        // parent_number to 100. Using child=106 would need 6 iterations (off by one).
        let linkage = seed_chain(5, 100, 104);
        let asserter = Asserter::new();
        asserter.push_success(&block_with_parent(hb(201))); // 104'→103'
        asserter.push_success(&block_with_parent(hb(202))); // 103'→102'
        asserter.push_success(&block_with_parent(hb(203))); // 102'→101'
        asserter.push_success(&block_with_parent(hb(100))); // 101'→canonical 100
        let provider = ProviderBuilder::new()
            .disable_recommended_fillers()
            .connect_mocked_client(asserter)
            .erased();
        let err = resolve_reorg_ancestor(&provider, &linkage, hb(200), 105, 5).await;
        assert!(
            matches!(
                err,
                RpcError::Reorg {
                    common_ancestor: 100
                }
            ),
            "got {err:?}"
        );
    }

    #[tokio::test]
    async fn resolve_reorg_ancestor_loop_exhausted_is_beyond_finality() {
        // Depth cap (depth=3) prevents resolution: the loop exhausts its 3 iterations
        // without finding a matching canonical ancestor. Uses a distinct ring from the
        // in-window rows (depth=3, seed [100,109] → ring [107,108,109], floor=107).
        let linkage = seed_chain(3, 100, 109);
        assert_eq!(linkage.floor(), Some(107));
        let asserter = Asserter::new();
        asserter.push_success(&block_with_parent(hb(201)));
        asserter.push_success(&block_with_parent(hb(202)));
        asserter.push_success(&block_with_parent(hb(203))); // loop ends after this fetch
        let provider = ProviderBuilder::new()
            .disable_recommended_fillers()
            .connect_mocked_client(asserter)
            .erased();
        let err = resolve_reorg_ancestor(&provider, &linkage, hb(200), 111, 3).await;
        assert!(matches!(err, RpcError::ReorgBeyondFinality), "got {err:?}");
    }

    #[tokio::test]
    async fn resolve_reorg_ancestor_parent_at_floor_is_beyond_finality() {
        // `parent_number <= floor` fires before any fetch: detached_parent sits
        // exactly at the ring floor, which the walk cannot descend below.
        let linkage = seed_chain(10, 100, 109); // floor = 100
        let asserter = Asserter::new(); // no pushes needed
        let provider = ProviderBuilder::new()
            .disable_recommended_fillers()
            .connect_mocked_client(asserter)
            .erased();
        // child=101 → parent_number=100 = floor; first iteration hits the floor check.
        let err = resolve_reorg_ancestor(&provider, &linkage, hb(200), 101, 10).await;
        assert!(matches!(err, RpcError::ReorgBeyondFinality), "got {err:?}");
    }

    #[tokio::test]
    async fn resolve_reorg_ancestor_null_response_is_beyond_finality() {
        // `Ok(None)` from `get_block_by_hash` (provider claims it doesn't have the
        // block) is treated conservatively as `> finality`, not a transport error.
        let linkage = seed_chain(15, 100, 109);
        let asserter = Asserter::new();
        asserter.push_success(&serde_json::json!(null));
        let provider = ProviderBuilder::new()
            .disable_recommended_fillers()
            .connect_mocked_client(asserter)
            .erased();
        let err = resolve_reorg_ancestor(&provider, &linkage, hb(200), 111, 15).await;
        assert!(matches!(err, RpcError::ReorgBeyondFinality), "got {err:?}");
    }

    #[tokio::test]
    async fn resolve_reorg_ancestor_transport_error_is_connection_error() {
        // A mid-walk transport error is recoverable (the stream can restart and
        // re-attempt); it must NOT collapse into `ReorgBeyondFinality` and trigger
        // do-not-serve. Assert `ConnectionError`, not `ReorgBeyondFinality`.
        let linkage = seed_chain(15, 100, 109);
        let asserter = Asserter::new();
        asserter.push_failure_msg("transport blip");
        let provider = ProviderBuilder::new()
            .disable_recommended_fillers()
            .connect_mocked_client(asserter)
            .erased();
        let err = resolve_reorg_ancestor(&provider, &linkage, hb(200), 111, 15).await;
        assert!(matches!(err, RpcError::ConnectionError(_)), "got {err:?}");
    }

    // ── P3.4: by-hash recovery for the reorg-exposed tail ───────────────────

    /// Canonical chain [100, 103]; `collect_window_hashes` walks backward from
    /// tip=103 (hash=hb(103)) to from_block=100 using `get_block_by_hash`.
    /// Each block response carries the parent hash. Then `fetch_logs_by_hash_range`
    /// issues one `get_logs` per block pinned to the canonical hash (ascending).
    /// Assert: four LogRanges covering [100,100]..[103,103], one log each.
    #[tokio::test]
    async fn by_hash_tail_fetch_over_mock() {
        // Asserter for the walk: three get_block_by_hash calls (103→102→101; 100 == from_block,
        // loop ends without fetching 100's parent). Must use block_with_parent_at so the
        // number cross-check in collect_window_hashes passes.
        let walk_asserter = Asserter::new();
        walk_asserter.push_success(&block_with_parent_at(103, hb(102))); // block 103 → parent 102
        walk_asserter.push_success(&block_with_parent_at(102, hb(101))); // block 102 → parent 101
        walk_asserter.push_success(&block_with_parent_at(101, hb(100))); // block 101 → parent 100
        let walk_provider = ProviderBuilder::new()
            .disable_recommended_fillers()
            .connect_mocked_client(walk_asserter)
            .erased();

        let pairs_desc = collect_window_hashes(&walk_provider, hb(103), 103, 100)
            .await
            .expect("walk should succeed");

        // Pairs are in descending order: [(103,hb(103)), (102,hb(102)), (101,hb(101)), (100,hb(100))].
        assert_eq!(pairs_desc.len(), 4);
        assert_eq!(pairs_desc[0], (103, hb(103)));
        assert_eq!(pairs_desc[1], (102, hb(102)));
        assert_eq!(pairs_desc[2], (101, hb(101)));
        assert_eq!(pairs_desc[3], (100, hb(100)));

        // Reverse to ascending for the log fetch.
        let mut pairs = pairs_desc;
        pairs.reverse();

        // Asserter for the log fetches: four get_logs calls, one per block.
        // Each returns a single synthetic log tagged with the block number.
        let log_asserter = Asserter::new();
        for n in 100u64..=103 {
            log_asserter.push_success(&serde_json::json!([{
                "address": "0x0000000000000000000000000000000000000001",
                "topics": [],
                "data": "0x",
                "blockHash": hb(n as u8).to_string(),
                "blockNumber": format!("0x{:x}", n),
                "transactionHash": hb(0).to_string(),
                "transactionIndex": "0x0",
                "logIndex": "0x0",
                "removed": false
            }]));
        }
        let log_provider = ProviderBuilder::new()
            .disable_recommended_fillers()
            .connect_mocked_client(log_asserter)
            .erased();
        let settings = ProviderSettings::from_mock(log_provider);

        let ranges = fetch_logs_by_hash_range(&settings, &pairs, None)
            .await
            .expect("log fetch should succeed");

        // Four LogRanges, each a single block with one log.
        assert_eq!(ranges.len(), 4, "expected one range per block");
        for (i, (from, to, logs)) in ranges.iter().enumerate() {
            let block_n = 100u64 + i as u64;
            assert_eq!(*from, block_n, "from_block mismatch at index {i}");
            assert_eq!(*to, block_n, "to_block mismatch at index {i}");
            assert_eq!(logs.len(), 1, "expected one log at block {block_n}");
            assert_eq!(
                logs[0].block_number,
                Some(block_n),
                "log block number mismatch"
            );
        }
    }

    /// `flush_window_range` fetches in-window blocks by canonical hash only.
    ///
    /// Setup: ring holds canonical hb(200) for block 200. The mock has exactly one
    /// log response — the canonical log tagged with hb(200). The mock also has an
    /// "orphan" log queued second (tagged with a different hash). If by-number were
    /// used, the asserter would return responses in order and the test would get the
    /// canonical log (first) for the by-number call — so instead we prove by-number
    /// is never invoked structurally: `flush_window_range` is the by-hash-only path;
    /// the mock has only ONE response queued; if a second call (by-number fallback)
    /// happened, `flush_window_range` would panic on an exhausted asserter.
    #[tokio::test]
    async fn orphan_rejection() {
        let canonical_log = serde_json::json!([{
            "address": "0xCAFECAFECAFECAFECAFECAFECAFECAFECAFECAFE",
            "topics": [],
            "data": "0x01",
            "blockHash": hb(200).to_string(),
            "blockNumber": "0xc8",
            "transactionHash": hb(1).to_string(),
            "transactionIndex": "0x0",
            "logIndex": "0x0",
            "removed": false
        }]);
        // One response only. A by-number fallback would exhaust the asserter and panic.
        let asserter = Asserter::new();
        asserter.push_success(&canonical_log);
        let settings = ProviderSettings::from_mock(
            ProviderBuilder::new()
                .disable_recommended_fillers()
                .connect_mocked_client(asserter)
                .erased(),
        );

        // Ring holds canonical hash for block 200; coalescer has [200, 200] pending.
        let mut linkage = HeadLinkage::new(8);
        linkage.observe(200, hb(200), hb(199)); // bootstraps ring with (200, hb(200))
        let mut coalescer = HeadCoalescer::new(Some(200), WINDOW);
        coalescer.observe_head(200, Instant::now());

        let ranges = flush_window_range(&mut linkage, &mut coalescer, &settings, None, 0)
            .await
            .expect("flush should succeed");

        assert_eq!(ranges.len(), 1, "one range per block");
        let (from, to, logs) = &ranges[0];
        assert_eq!((*from, *to), (200, 200));
        assert_eq!(logs.len(), 1, "canonical log returned");
        assert_eq!(
            logs[0].block_hash,
            Some(hb(200)),
            "log must carry the canonical block hash, not an orphan's"
        );
        // Cursor advanced past block 200.
        assert_eq!(coalescer.pending_range(), None);
    }

    /// A null block response from `get_block_by_hash` during the hash walk is
    /// classified as `ReorgBeyondFinality` (fail-closed). A transport error is
    /// classified as `ConnectionError` (not `ReorgBeyondFinality`) so the caller
    /// can distinguish a recoverable network fault from a genuine finality violation.
    #[tokio::test]
    async fn hash_mismatch_null_block_fail_closed() {
        // Null block → ReorgBeyondFinality.
        {
            let asserter = Asserter::new();
            asserter.push_success(&serde_json::json!(null));
            let provider = ProviderBuilder::new()
                .disable_recommended_fillers()
                .connect_mocked_client(asserter)
                .erased();
            // Walk from block 5 (hash hb(5)) down to block 1; the first fetch returns null.
            let err = collect_window_hashes(&provider, hb(5), 5, 1)
                .await
                .expect_err("null block must be an error");
            assert!(
                matches!(err, RpcError::ReorgBeyondFinality),
                "null block must yield ReorgBeyondFinality, got {err:?}"
            );
        }
        // Transport error → ConnectionError (not ReorgBeyondFinality).
        {
            let asserter = Asserter::new();
            asserter.push_failure_msg("connection reset");
            let provider = ProviderBuilder::new()
                .disable_recommended_fillers()
                .connect_mocked_client(asserter)
                .erased();
            let err = collect_window_hashes(&provider, hb(5), 5, 1)
                .await
                .expect_err("transport error must be an error");
            assert!(
                matches!(err, RpcError::ConnectionError(_)),
                "transport fault must yield ConnectionError, got {err:?}"
            );
        }
    }

    /// When `depth == 0` the by-hash path is never entered.
    /// `HeadLinkage::enabled()` returns false; `observe` returns `Linked` for every
    /// head regardless of parent mismatch. No `get_block_by_hash` call is issued.
    #[test]
    fn dormancy_at_depth_zero() {
        let mut linkage = HeadLinkage::new(0);
        assert!(!linkage.enabled(), "depth=0 must be disabled");

        // Any head — even one with a bogus parent — must return Linked.
        assert_eq!(
            linkage.observe(100, hb(1), hb(99)),
            LinkOutcome::Linked,
            "depth=0 must not trigger reorg detection"
        );
        assert_eq!(
            linkage.observe(101, hb(2), hb(55)), // obviously wrong parent
            LinkOutcome::Linked,
            "depth=0 must not trigger reorg detection on bad parent"
        );
        // Ring stays empty: no state accrues when disabled.
        assert_eq!(
            linkage.floor(),
            None,
            "disabled linkage must not record state"
        );
    }

    /// After the boot gap is flushed by `flush_window_range`, the ring holds the
    /// first observed head. The production path for the NEXT head is a clean
    /// `observe` extension — no false reorg fires.
    ///
    /// In production: `stream_heads_with_logs` creates a fresh empty ring, then
    /// `observe(first_head)` bootstraps it. `flush_window_range` handles the gap
    /// [cursor, first_head] using the walked pairs directly (no ring seeding).
    /// The second head (first_head + 1) then links cleanly via the normal parent
    /// check — because `observe(first_head+1)` checks ring.back() = first_head.
    #[test]
    fn boot_tail_seeds_ring() {
        // Production state after first head 105 arrives: ring bootstrapped to {105}.
        let mut linkage = HeadLinkage::new(8);
        assert_eq!(
            linkage.observe(105, hb(105), hb(104)),
            LinkOutcome::Linked,
            "bootstrap observe"
        );
        assert_eq!(linkage.hash_at(105), Some(hb(105)));

        // After flush_window_range handles [cursor=101, 105], the cursor moves to 106.
        // The next WS head is 106 with correct parent hb(105) — must link cleanly.
        assert_eq!(
            linkage.observe(106, hb(106), hb(105)),
            LinkOutcome::Linked,
            "first post-flush head must link without false reorg"
        );
        assert_eq!(linkage.hash_at(106), Some(hb(106)));

        // Subsequent heads also link cleanly.
        assert_eq!(linkage.observe(107, hb(107), hb(106)), LinkOutcome::Linked);
    }

    // ── P3.4 addendum: live flush self-heal + orphan rejection ──────────────

    /// REGRESSION — live flush with a head gap must not wedge.
    ///
    /// Scenario: depth=8, cursor=101, first WS head is 105 (gap [101,104]).
    /// Before the fix the flush found `hash_at(101) = None`, broke out of the for
    /// loop, and `continue`d the `'flush while` with the cursor unmoved — an
    /// infinite spin. After the fix `flush_window_range` self-heals by walking from
    /// ring.latest() (105) back to 101, then fetches logs for each block by hash.
    ///
    /// The test is bounded by tokio's async executor; a spin would deadlock it.
    #[tokio::test]
    async fn live_flush_gap_self_heals() {
        // Ring: observe(105) bootstraps to {105}. Cursor: 101.
        let mut linkage = HeadLinkage::new(8);
        assert_eq!(linkage.observe(105, hb(105), hb(104)), LinkOutcome::Linked);
        let mut coalescer = HeadCoalescer::new(Some(101), WINDOW);
        coalescer.observe_head(105, Instant::now());
        assert_eq!(coalescer.pending_range(), Some((101, 105)));

        // Mock: walk 105→104→103→102 (4 get_block_by_hash calls; 101==from_block so no
        // fetch for 101 itself). Then 5 get_logs (at_block_hash) for blocks 101..=105.
        let asserter = Asserter::new();
        asserter.push_success(&block_with_parent_at(105, hb(104)));
        asserter.push_success(&block_with_parent_at(104, hb(103)));
        asserter.push_success(&block_with_parent_at(103, hb(102)));
        asserter.push_success(&block_with_parent_at(102, hb(101)));
        for _ in 101u64..=105 {
            asserter.push_success(&serde_json::json!([]));
        }
        let settings = ProviderSettings::from_mock(
            ProviderBuilder::new()
                .disable_recommended_fillers()
                .connect_mocked_client(asserter)
                .erased(),
        );

        let ranges = flush_window_range(&mut linkage, &mut coalescer, &settings, None, 0)
            .await
            .expect("flush must succeed, not spin");

        assert_eq!(ranges.len(), 5, "one LogRange per block 101..=105");
        for (i, (from, to, _)) in ranges.iter().enumerate() {
            let n = 101u64 + i as u64;
            assert_eq!((*from, *to), (n, n));
        }
        // Cursor advanced past all flushed blocks.
        assert_eq!(coalescer.pending_range(), None);
    }

    /// After a forward-gap reconnect the ring is cleared to just the new head.
    /// The next flush must self-heal the gap without spinning.
    #[tokio::test]
    async fn forward_gap_mid_stream_self_heals() {
        // Steady state: ring covers [100, 103] after 4 clean heads.
        let mut linkage = HeadLinkage::new(8);
        for n in 100u64..=103 {
            let parent = hb((n - 1) as u8);
            assert_eq!(linkage.observe(n, hb(n as u8), parent), LinkOutcome::Linked);
        }

        // Forward gap: head 107 arrives (skipping 104-106) → ring cleared + re-anchored.
        assert_eq!(
            linkage.observe(107, hb(107), hb(106)),
            LinkOutcome::Linked,
            "forward gap re-anchors without false reorg"
        );
        assert_eq!(linkage.ring.len(), 1, "ring holds only the new head");
        assert_eq!(linkage.hash_at(107), Some(hb(107)));
        assert_eq!(linkage.hash_at(103), None, "old entries cleared");

        // Coalescer: cursor stayed at 104 (the flush hadn't advanced it yet).
        let mut coalescer = HeadCoalescer::new(Some(104), WINDOW);
        coalescer.observe_head(107, Instant::now());
        assert_eq!(coalescer.pending_range(), Some((104, 107)));

        // Mock: walk 107→106→105→104 (3 block fetches; 104==from_block, stops there).
        // Then 4 get_logs at_block_hash for 104..=107.
        let asserter = Asserter::new();
        asserter.push_success(&block_with_parent_at(107, hb(106)));
        asserter.push_success(&block_with_parent_at(106, hb(105)));
        asserter.push_success(&block_with_parent_at(105, hb(104)));
        for _ in 104u64..=107 {
            asserter.push_success(&serde_json::json!([]));
        }
        let settings = ProviderSettings::from_mock(
            ProviderBuilder::new()
                .disable_recommended_fillers()
                .connect_mocked_client(asserter)
                .erased(),
        );

        let ranges = flush_window_range(&mut linkage, &mut coalescer, &settings, None, 0)
            .await
            .expect("mid-stream gap flush must not spin");

        assert_eq!(ranges.len(), 4, "blocks 104..=107");
        assert_eq!(coalescer.pending_range(), None);
    }

    /// `collect_window_hashes`: a by-hash response whose `header.number` does not
    /// match the expected number is a provider integrity violation — fail closed.
    #[tokio::test]
    async fn collect_window_hashes_number_mismatch() {
        // Walk from tip=5 to from_block=1; mock returns a block claiming number=99.
        let asserter = Asserter::new();
        asserter.push_success(&block_with_parent_at(99, hb(4))); // wrong number
        let provider = ProviderBuilder::new()
            .disable_recommended_fillers()
            .connect_mocked_client(asserter)
            .erased();

        let err = collect_window_hashes(&provider, hb(5), 5, 1)
            .await
            .expect_err("number mismatch must be an error");
        assert!(
            matches!(err, RpcError::ReorgBeyondFinality),
            "number mismatch must yield ReorgBeyondFinality, got {err:?}"
        );
    }

    /// `flush_window_range`: when `latest()` is None (ring completely empty),
    /// the function returns `ReorgBeyondFinality` rather than panicking.
    #[tokio::test]
    async fn flush_window_range_empty_ring_is_fail_closed() {
        let asserter = Asserter::new(); // no responses needed
        let settings = ProviderSettings::from_mock(
            ProviderBuilder::new()
                .disable_recommended_fillers()
                .connect_mocked_client(asserter)
                .erased(),
        );
        let mut linkage = HeadLinkage::new(8); // empty ring (no observe yet)
        let mut coalescer = HeadCoalescer::new(Some(100), WINDOW);
        coalescer.observe_head(100, Instant::now());

        let err = flush_window_range(&mut linkage, &mut coalescer, &settings, None, 0)
            .await
            .expect_err("empty ring must be fail-closed");
        assert!(matches!(err, RpcError::ReorgBeyondFinality), "got {err:?}");
    }

    /// The backfill window boundary: blocks strictly below `tip - depth` are
    /// fetched by number (the normal backfill path); blocks `>= tip - depth` (within
    /// the reorg-exposed window) are fetched by hash.
    ///
    /// This test exercises the predicate logic directly: given a specific tip and
    /// depth, verify that `from_block >= tip.saturating_sub(depth)` correctly
    /// partitions the range.
    #[test]
    fn window_boundary() {
        // tip = 1000, depth = 50: window starts at 950.
        let tip: u64 = 1000;
        let depth: u64 = 50;
        let window_floor = tip.saturating_sub(depth); // 950

        // Blocks below the window floor are served by the backfill-by-number path.
        for n in [0u64, 500, 949] {
            assert!(
                n < window_floor,
                "block {n} should be below the window floor {window_floor}"
            );
        }

        // The window floor itself and everything above it is in the by-hash zone.
        for n in [950u64, 975, 1000] {
            assert!(
                n >= window_floor,
                "block {n} should be inside the by-hash window (floor {window_floor})"
            );
        }

        // Saturating_sub at depth == 0 yields u64::MAX (no window) — the by-hash
        // path is never entered, matching the dormancy_at_depth_zero invariant.
        let no_window_floor = tip.saturating_sub(0);
        assert_eq!(no_window_floor, tip, "depth=0 must not create a window");

        // Depth larger than tip saturates at 0: the entire history is in the window.
        let huge_depth_floor = tip.saturating_sub(tip + 1);
        assert_eq!(
            huge_depth_floor, 0,
            "depth > tip must floor at 0 (all blocks in window)"
        );
    }

    /// REGRESSION — a partial failure in a multi-block flush must not silently
    /// skip earlier blocks.
    ///
    /// Before the fix, `flush_window_range` advanced `coalescer.advance_cursor`
    /// inside the fetch loop. If block 201's fetch failed after 200 succeeded,
    /// the cursor advanced past 200 and its logs were discarded. On reconnect the
    /// stream resumed from 201, silently skipping block 200.
    ///
    /// After the fix the cursor is only advanced once the ENTIRE batch succeeds.
    /// This test fails against the pre-fix code (cursor at 201) and passes after
    /// (cursor stays at 200 — the whole range retries on the next provider).
    #[tokio::test]
    async fn flush_window_range_partial_failure_leaves_cursor_unmoved() {
        // Ring: blocks 200 and 201 observed (clean chain).
        let mut linkage = HeadLinkage::new(8);
        assert_eq!(linkage.observe(200, hb(200), hb(199)), LinkOutcome::Linked);
        assert_eq!(linkage.observe(201, hb(201), hb(200)), LinkOutcome::Linked);

        // Coalescer: cursor at 200, head 201 seen → pending [200, 201].
        let mut coalescer = HeadCoalescer::new(Some(200), WINDOW);
        coalescer.observe_head(201, Instant::now());
        assert_eq!(coalescer.pending_range(), Some((200, 201)));

        // Single mocked HTTP provider (from_mock): one queued success (block 200's
        // get_logs), then one queued failure (block 201's get_logs). With a single
        // provider, one failure exhausts all retry attempts.
        let asserter = Asserter::new();
        asserter.push_success(&serde_json::json!([])); // block 200 logs succeed
        asserter.push_failure_msg("transient error"); // block 201 logs fail

        let settings = ProviderSettings::from_mock(
            ProviderBuilder::new()
                .disable_recommended_fillers()
                .connect_mocked_client(asserter)
                .erased(),
        );

        let result = flush_window_range(&mut linkage, &mut coalescer, &settings, None, 0).await;

        // Must return Err — the batch did not complete.
        assert!(result.is_err(), "partial failure must return Err");

        // The cursor must NOT have advanced past block 200 — the whole range must
        // retry on the next provider so block 200 is not silently skipped.
        assert_eq!(
            coalescer.pending_range(),
            Some((200, 201)),
            "cursor must stay at 200 so block 200 is not silently skipped on reconnect"
        );
    }

    /// Hermetic wiring proof for the live consumption loop:
    ///   observe_head → observe → should_flush → flush_window_range → [ranges]
    ///   → observe_reorg → LinkOutcome::Reorg
    ///
    /// This directly sequences the operations `stream_heads_with_logs` performs
    /// in its inner loop and asserts the correct outcomes at each step. It covers
    /// the two behaviors that had no hermetic test: the gap self-heal path and the
    /// reorg-terminate path.
    ///
    /// Residual: the outer WS-reconnect/failover loop in `stream_heads_with_logs`
    /// is exercised only by the `#[ignore]`d network test (`test_event_streamer_head`
    /// in `tests/events_test.rs`). A full seam extraction was evaluated and deferred
    /// to avoid restructuring the core stream at a release gate (tokio test-util feature
    /// is not enabled in this crate, so time-paused tests are not available).
    #[tokio::test]
    async fn live_loop_wiring_gap_then_reorg() {
        // Coalescer: cursor at 101 (live filter from_block = 101).
        // Linkage: depth 8.
        let now = Instant::now();
        let mut coalescer = HeadCoalescer::new(Some(101), WINDOW);
        let mut linkage = HeadLinkage::new(8);

        // Step 1: head 105 arrives with a gap (101-104 skipped).
        // observe_head extends the coalescer's pending range to [101, 105].
        assert!(coalescer.observe_head(105, now));
        assert_eq!(coalescer.pending_range(), Some((101, 105)));

        // observe bootstraps the ring to {105} (forward-gap path: ring.clear() + push).
        assert_eq!(linkage.observe(105, hb(105), hb(104)), LinkOutcome::Linked);
        assert_eq!(linkage.latest(), Some((105, hb(105))));

        // Step 2: simulate time advancing past the coalesce window by querying
        // should_flush with a synthetic instant past the deadline.
        // (tokio test-util / time::pause is not available in this crate's feature set.)
        let after_window = now + WINDOW + Duration::from_millis(1);
        assert!(
            coalescer.should_flush(after_window),
            "coalesce window expired — flush must fire"
        );

        // Step 3: flush_window_range self-heals the gap [101, 105].
        // Walk: get_block_by_hash for hb(105)→hb(104)→hb(103)→hb(102) (4 fetches;
        // 101==from_block so the walk stops after pushing block 101).
        // Logs: 5 get_logs at_block_hash for blocks 101..=105.
        // Both walk and log calls use the same mocked HTTP provider.
        let asserter = Asserter::new();
        asserter.push_success(&block_with_parent_at(105, hb(104)));
        asserter.push_success(&block_with_parent_at(104, hb(103)));
        asserter.push_success(&block_with_parent_at(103, hb(102)));
        asserter.push_success(&block_with_parent_at(102, hb(101)));
        for _ in 101u64..=105 {
            asserter.push_success(&serde_json::json!([]));
        }
        let settings = ProviderSettings::from_mock(
            ProviderBuilder::new()
                .disable_recommended_fillers()
                .connect_mocked_client(asserter)
                .erased(),
        );

        let ranges = flush_window_range(&mut linkage, &mut coalescer, &settings, None, 0)
            .await
            .expect("gap self-heal must succeed");

        assert_eq!(ranges.len(), 5, "one range per block 101..=105");
        for (i, &(from, to, _)) in ranges.iter().enumerate() {
            let n = 101u64 + i as u64;
            assert_eq!((from, to), (n, n), "range at index {i} must be block {n}");
        }
        // Cursor advanced past the full batch — no silent gap.
        assert_eq!(
            coalescer.pending_range(),
            None,
            "cursor must be at 106 after flush"
        );

        // Step 4: two more clean heads arrive and link.
        assert_eq!(linkage.observe(106, hb(106), hb(105)), LinkOutcome::Linked);
        assert_eq!(linkage.observe(107, hb(107), hb(106)), LinkOutcome::Linked);

        // Step 5: a competing block 107 with the same parent (hb(106)) but a
        // different hash (hb(250)) triggers a reorg. The ring has canonical hb(107)
        // at 107; a new block with parent matching canonical hb(106) at 106 produces
        // Reorg { common_ancestor: 106 }. `stream_heads_with_logs` would yield
        // Err(RpcError::Reorg { common_ancestor: 106 }) and return here.
        let outcome = linkage.observe(107, hb(250), hb(106));
        assert!(
            matches!(
                outcome,
                LinkOutcome::Reorg {
                    common_ancestor: 106
                }
            ),
            "competing block at 107 with canonical parent must produce Reorg{{106}}, got {outcome:?}"
        );
    }
}