//! Segment receiver: consumes WAL segments from the transport and applies
//! them to the local signal ledger.
//!
//! The receiver runs in a background thread, blocking on
//! [`Transport::recv_segment`] and replaying each batch into the shared
//! [`SignalLedger`] via `apply_replicated_events`. Idempotent replay is ensured
//! by checking the per-shard high-water-mark in [`ReplicationState`].
//!
//! # Durability (BLOCKER 6)
//!
//! Each accepted segment is applied **WAL-first** on the follower:
//! `apply_replicated_events` appends every event to the follower's *own* WAL
//! and awaits the group-commit fsync once for the whole staged batch (m11p1:
//! one shared fsync, not one solo `batch_timeout` per event) before any
//! in-memory state mutates. The
//! follower's normal WAL replay on open therefore reconstructs all replicated
//! state — without this, every replicated event between the last checkpoint and
//! a follower crash was silently lost. The leader-seqno high-water-mark is
//! persisted separately by the periodic checkpoint thread (a re-shipped segment
//! whose last seq is at or below the restored high-water-mark is an idempotent
//! no-op).
//!
//! # Error handling
//!
//! If a batch fails BLAKE3 verification or structural decode, `apply_payload`
//! returns [`WalError::Corruption`]. If the follower's own WAL append fails (a
//! disk fault), it returns [`WalError::Io`]. In both cases the receiver thread
//! propagates the error immediately (it does **not** skip the payload) so that
//! [`SegmentReceiverHandle::join`] can surface it to the operator — the
//! follower must never silently acknowledge an event it failed to durably
//! record.

use std::{
    sync::{
        Arc,
        atomic::{AtomicBool, Ordering},
    },
    thread::JoinHandle,
};

use crate::{
    replication::{
        lag::ReplicationLagGauge, shard::ShardId, state::ReplicationState, transport::Transport,
    },
    schema::{EntityId, Timestamp},
    signals::{SignalLedger, SignalTypeId},
    wal::{
        error::WalError,
        format::batch::{BatchPayload, BlobRecord, HEADER_SIZE, decode_batch_payload},
    },
};

/// Applier for replicated blob (kind-1/2) records on a follower.
///
/// Implemented by `TidalDb` (via a `Weak` adapter — see
/// `start_replication_with_blobs` — so the receiver thread's handle cannot
/// keep the database alive past its owner). Routes each record through the
/// SAME WAL-first write path the leader used: the follower re-journals the
/// blob in its own WAL (so follower recovery rebuilds it) and then applies it
/// to storage as an idempotent upsert.
pub trait ReplicatedBlobApplier: Send + Sync {
    /// Apply one apply-round's replicated blob records, in order, as a
    /// BATCH: the implementation stages every record's WAL append before
    /// waiting, so a whole round shares group-commit fsyncs (m11p3 — a
    /// record-at-a-time apply pays one solo fsync per item, capping item
    /// apply throughput at the fsync floor, ~100/s on macOS). Records are
    /// handed over by value so the engine can share them with its WAL
    /// writer by refcount instead of deep-cloning every payload.
    ///
    /// # Errors
    ///
    /// Any engine error; the receiver HALTS on it (a follower must never
    /// silently acknowledge an item it failed to durably record). Blobs are
    /// idempotent upserts, so redelivery after a mid-batch halt is safe.
    fn apply_blobs(&self, records: Vec<BlobRecord>) -> crate::Result<()>;
}

/// Handle to a running segment receiver thread.
///
/// Call [`join`](Self::join) to block until the thread exits and retrieve any
/// corruption error that caused it to stop. While the node is still open, poll
/// [`died`](Self::died) (or [`is_finished`](Self::is_finished)) to observe a
/// receiver that halted on an error mid-life — without it, a dead receiver is
/// indistinguishable from a healthy one until shutdown calls `join` (C11).
pub struct SegmentReceiverHandle {
    thread: Option<JoinHandle<Result<(), WalError>>>,
    /// Latched `true` exactly when the receiver thread halts on an apply error
    /// (corrupt segment / follower-WAL IO fault) or an unexpected exit — i.e. a
    /// silent replication stall. A clean `recv_segment() == None` shutdown leaves
    /// this `false`, so it distinguishes "the follower stopped applying because
    /// something broke" from "the follower was asked to shut down" (C11). Shared
    /// with the thread, which sets it before returning the error.
    died: Arc<AtomicBool>,
}

impl SegmentReceiverHandle {
    /// Block until the receiver thread exits.
    ///
    /// # Errors
    ///
    /// Returns `Err(WalError::Corruption { .. })` if a corrupt batch was
    /// received. Returns `Err(WalError::Corruption { .. })` if the thread
    /// panicked unexpectedly.
    pub fn join(mut self) -> Result<(), WalError> {
        self.thread.take().map_or(Ok(()), |handle| {
            handle.join().unwrap_or_else(|_| {
                Err(WalError::Corruption {
                    message: "segment receiver thread panicked".into(),
                })
            })
        })
    }

    /// Whether the receiver thread has halted on an error (corrupt segment,
    /// follower-WAL IO fault, or an unexpected exit) while the node is still
    /// open.
    ///
    /// `true` means replication has silently stopped applying and the follower
    /// is frozen — health checks must report degraded. A clean shutdown
    /// (`recv_segment() == None`) does **not** set this, so it does not
    /// false-positive on intentional teardown (C11). Cheap, lock-free, and safe
    /// to poll from `health_check` while the handle is still parked in its mutex.
    #[must_use]
    pub fn died(&self) -> bool {
        self.died.load(Ordering::Acquire)
    }

    /// Whether the receiver thread's join handle has finished.
    ///
    /// Mirrors [`std::thread::JoinHandle::is_finished`]. Unlike [`died`](Self::died)
    /// this is also `true` after a clean shutdown, so health checks should prefer
    /// `died` to avoid flagging an intentional teardown as a fault.
    #[must_use]
    pub fn is_finished(&self) -> bool {
        self.thread
            .as_ref()
            .is_some_and(std::thread::JoinHandle::is_finished)
    }
}

/// Spawn a background thread that receives WAL segments and replays them
/// into the signal ledger.
///
/// The thread exits when `transport.recv_segment()` returns `None` (transport
/// closed / shutdown) **or** when a corrupt batch is detected.
///
/// When `lag_gauge` is `Some`, every received segment advances the gauge's
/// leader high-water-mark to the highest WAL sequence the leader has shipped
/// (obs-REPL-1). The gauge's `applied_seqno` comes from the same
/// `ReplicationState` this receiver advances, so `lag_segments()` reflects real
/// follower backpressure (`leader_seqno − applied_seqno`).
///
/// # Panics
///
/// Panics if the OS fails to spawn the background thread.
/// Max segments drained into one coalesced apply round.
const MAX_COALESCED_SEGMENTS: usize = 64;
/// Max events (by payload metadata) drained into one coalesced apply round.
///
/// A SOFT cap: the drain loop checks it before accepting the next payload, so
/// the final accepted payload may overshoot it by up to one payload's events
/// (≤ the wire format's 256/batch for honest peers). The cap bounds the
/// halting blast radius of one apply round, not memory — overshooting by one
/// payload is harmless and cheaper than splitting it.
const MAX_COALESCED_EVENTS: u64 = 2048;

pub fn spawn_receiver<T: Transport + ?Sized>(
    transport: Arc<T>,
    ledger: Arc<SignalLedger>,
    replication_state: Arc<ReplicationState>,
    lag_gauge: Option<Arc<ReplicationLagGauge>>,
    blob_applier: Option<Arc<dyn ReplicatedBlobApplier>>,
) -> SegmentReceiverHandle {
    // Liveness latch (C11): set true iff the thread halts on an apply error.
    // Shared with the handle so health_check can observe a silently-dead
    // receiver while the node is still open. A clean shutdown leaves it false.
    let died = Arc::new(AtomicBool::new(false));
    let thread_died = Arc::clone(&died);
    let thread = std::thread::Builder::new()
        .name("tidaldb-segment-receiver".into())
        .spawn(move || -> Result<(), WalError> {
            // Per-source-shard floor of the last frontier handed to
            // `notify_applied`: an apply round that did not advance a shard's
            // frontier re-notifies nothing (the transport dedups too, but
            // skipping here saves its lock acquisition per quiet round).
            let mut last_notified: std::collections::HashMap<ShardId, u64> =
                std::collections::HashMap::new();
            loop {
                let Some(first) = transport.recv_segment() else {
                    // Clean shutdown: the transport closed / shutdown was
                    // requested. Leave the `died` latch false so health_check
                    // does not flag an intentional teardown as a fault (C11).
                    tracing::debug!("segment receiver: transport closed, shutting down");
                    return Ok(());
                };

                // ── Coalesce (m11p1): drain the immediately-available backlog ──
                //
                // Every applied segment pays one follower group-commit wait
                // (fsync is ms-scale on real volumes), so applying segments
                // one-at-a-time caps follower throughput at
                // events-per-segment / fsync-cost — the live ramp measured
                // ~12-event segments × ~6.5ms fsync ≈ 1.8k events/s while the
                // leader wrote 2.7k+/s, growing lag without bound. Draining
                // the backlog and staging it through ONE shared group commit
                // amortizes the fsync across everything queued.
                let mut pending = vec![first];
                let mut drained_events = pending[0].event_count;
                while pending.len() < MAX_COALESCED_SEGMENTS
                    && drained_events < MAX_COALESCED_EVENTS
                {
                    let Some(next) = transport.try_recv_segment() else {
                        break;
                    };
                    drained_events += next.event_count;
                    pending.push(next);
                }

                // Per-source-shard maxima of this round's shipped seqnos,
                // captured before `pending` moves: the gap detector below
                // compares them against the post-apply contiguous frontier.
                let mut shard_maxima: Vec<(ShardId, u64)> = Vec::new();
                for p in &pending {
                    let last = p.leader_last_seq.max(p.id.seqno);
                    if last == 0 {
                        continue;
                    }
                    match shard_maxima.iter_mut().find(|(s, _)| *s == p.id.shard_id) {
                        Some((_, m)) => *m = (*m).max(last),
                        None => shard_maxima.push((p.id.shard_id, last)),
                    }
                }

                if let Err(e) = apply_drained(
                    pending,
                    &ledger,
                    &replication_state,
                    lag_gauge.as_deref(),
                    blob_applier.as_deref(),
                ) {
                    // A corrupt segment / follower-WAL fault halts the receiver
                    // thread. Latch `died` BEFORE returning so a stalled replica
                    // is observable via SegmentReceiverHandle::died() while the
                    // node is still open — not just at shutdown via join(). Log at
                    // the failure site too -- mirrors wal::reader's per-corruption
                    // logging, except replication cannot skip-and-continue (a gap
                    // would diverge the replica), so we halt and surface.
                    thread_died.store(true, Ordering::Release);
                    tracing::error!(
                        error = %e,
                        "replication apply failed; receiver halting (health degraded)"
                    );
                    return Err(e);
                }

                // ── Gap detection → follower-pulled catch-up (m11p2) ──
                //
                // If this round left a shard's contiguous frontier short of
                // the highest seqno we were shipped, the round parked a range
                // ahead of a hole: data older than the leader's in-memory
                // ship tail. The leader cannot push it (the tail rotated);
                // the FOLLOWER pulls it via the transport's catch-up stream
                // (`StreamSegments` from the durable WAL). The transport
                // rate-limits and de-duplicates in-flight pulls per shard;
                // the default (in-process) implementation is a no-op.
                for (shard, max_last) in shard_maxima {
                    let applied = replication_state.applied_seqno(shard).unwrap_or(0);
                    // Durable-ack release (m11p3): everything this round
                    // applied is durably folded (storage upserts + this
                    // node's own WAL fsync precede the frontier advance), so
                    // ship acks parked on seqnos <= applied may now answer
                    // with a true durable mark. Only an ADVANCED frontier is
                    // worth a notification.
                    if applied > 0 {
                        let notified = last_notified.entry(shard).or_insert(0);
                        if applied > *notified {
                            *notified = applied;
                            transport.notify_applied(shard, applied);
                        }
                    }
                    if applied < max_last {
                        transport.request_catchup(shard, applied + 1);
                    }
                }
            }
        })
        .expect("failed to spawn segment receiver thread");

    SegmentReceiverHandle {
        thread: Some(thread),
        died,
    }
}

/// Whether two payloads' authoritative seqno ranges overlap (same source
/// shard). A payload with an unknown range (`id.seqno == 0` or
/// `leader_last_seq < id.seqno`) is conservatively treated as overlapping
/// everything from its shard, so it is always applied alone.
fn ranges_overlap(
    a: &crate::replication::WalSegmentPayload,
    b: &crate::replication::WalSegmentPayload,
) -> bool {
    if a.id.shard_id != b.id.shard_id {
        return false;
    }
    let known = |p: &crate::replication::WalSegmentPayload| {
        p.id.seqno > 0 && p.leader_last_seq >= p.id.seqno
    };
    if !known(a) || !known(b) {
        return true;
    }
    !(a.leader_last_seq < b.id.seqno || b.leader_last_seq < a.id.seqno)
}

/// Apply a drained backlog of payloads in range-disjoint groups.
///
/// # Why grouping is load-bearing (no within-group double-fold)
///
/// The idempotency gate (`range_applied`) reads `ReplicationState`, which only
/// advances at COMMIT time. Two payloads covering the same seqnos (a sender
/// retry duplicate; a heal single-event re-ship inside an already-queued
/// batched run) would BOTH pass a gate taken before either committed, folding
/// the shared events twice. Within one group, ranges are therefore disjoint by
/// construction; an overlapping payload is deferred to a later group, where it
/// gates against the now-advanced state exactly like the sequential path.
fn apply_drained(
    mut pending: Vec<crate::replication::WalSegmentPayload>,
    ledger: &SignalLedger,
    state: &ReplicationState,
    lag_gauge: Option<&ReplicationLagGauge>,
    blob_applier: Option<&dyn ReplicatedBlobApplier>,
) -> Result<(), WalError> {
    while !pending.is_empty() {
        let mut group: Vec<crate::replication::WalSegmentPayload> = Vec::new();
        let mut deferred: Vec<crate::replication::WalSegmentPayload> = Vec::new();
        for payload in pending {
            if group.iter().any(|g| ranges_overlap(g, &payload)) {
                deferred.push(payload);
            } else {
                group.push(payload);
            }
        }
        pending = deferred;

        // Prepare (decode + validate + gauge + gate) every payload in the
        // group, then commit the survivors through one shared group commit.
        let mut prepared = Vec::with_capacity(group.len());
        for payload in &group {
            // Stream-baseline jump (m11p2): a catch-up chunk from a promoted
            // leader announces where its stream STARTED. Seqnos at or below
            // the baseline are pre-stream history — data this follower
            // already holds via the previous leader's stream — so the
            // frontier jumps to the baseline instead of waiting forever for
            // a gap that is not data.
            if payload.stream_baseline > 0 {
                let applied = state.applied_seqno(payload.id.shard_id).unwrap_or(0);
                if applied < payload.stream_baseline {
                    tracing::info!(
                        shard = %payload.id.shard_id,
                        baseline = payload.stream_baseline,
                        applied,
                        "replication: stream baseline announced; jumping frontier \
                         past pre-stream history"
                    );
                    state.advance(payload.id.shard_id, payload.stream_baseline);
                }
            }
            let leader_seqno = (payload.leader_last_seq > 0).then_some(payload.leader_last_seq);
            let segment_first = (payload.id.seqno > 0).then_some(payload.id.seqno);
            if let Some(seg) = prepare_segment(
                &payload.bytes,
                payload.id.shard_id,
                state,
                lag_gauge,
                segment_first,
                leader_seqno,
            )? {
                prepared.push(seg);
            }
        }
        commit_segments(&mut prepared, ledger, state, blob_applier)?;
    }
    Ok(())
}

/// Apply all batches in a WAL segment payload to the signal ledger.
///
/// Idempotent: batches whose last sequence number is at or below the
/// replication state's high-water-mark for the source shard are skipped.
///
/// # Atomic payload application (WARNING fix)
///
/// The payload is applied **all-or-nothing**. We first decode and validate
/// *every* batch (BLAKE3, structural decode, and the `first_seq + event_count`
/// overflow check) into a staged list, advancing only the lag gauge's *leader*
/// high-water-mark as boundaries are observed. Only after the whole payload
/// validates do we apply staged events to the follower ledger (WAL-first) and
/// advance the *applied* high-water-mark to the payload's final boundary in one
/// step.
///
/// This means a corrupt batch *anywhere* in the payload causes the function to
/// return `Err` **before** any event is applied or any applied-seqno is
/// advanced, so the follower's ledger and idempotency boundary never reflect a
/// half-applied payload. The leader can cleanly re-ship the same un-acked
/// segment and it replays from a single consistent resume point. (Without this,
/// a later corrupt batch left earlier batches applied with the applied-seqno
/// stuck mid-payload, with no clean point to resume from.)
///
/// `leader_seqno` is the segment's authoritative last WAL seqno as known by the
/// leader *before* community-overlay filtering (see [`WalSegmentPayload`]). It
/// feeds the lag gauge's leader high-water-mark regardless of how many batches
/// survived filtering — an all-local segment ships zero batches yet still
/// advances the leader HWM so the gauge does not under-report (obs-REPL-1,
/// WARNING fix). `None` falls back to the per-batch boundaries (legacy callers
/// / tests with no authoritative value).
///
/// When `lag_gauge` is `Some`, the gauge's applied seqno tracks
/// `state.applied_seqno`, so `lag_segments()` is non-zero whenever a payload has
/// been received from the leader but not yet applied, and returns to 0 once the
/// follower catches up.
///
/// # Errors
///
/// Returns `WalError::Corruption` on any BLAKE3 or structural decode failure
/// (or a wrapped seqno boundary). The offset of the first corrupt batch is
/// included in the error message. On error, no state has been mutated.
pub fn apply_payload(
    bytes: &[u8],
    from_shard: ShardId,
    ledger: &SignalLedger,
    state: &ReplicationState,
    lag_gauge: Option<&ReplicationLagGauge>,
) -> Result<(), WalError> {
    apply_payload_with_leader_seq(bytes, from_shard, ledger, state, lag_gauge, None)
}

/// Like [`apply_payload`] but carrying the segment's authoritative leader
/// high-water-mark (the true last WAL seqno before community-overlay filtering).
///
/// See [`apply_payload`] for the atomicity and leader-seqno contract. The
/// `leader_seqno` is fed to the lag gauge independently of how many batches
/// survive filtering, so an all-local (empty) segment still advances the leader
/// HWM. The segment's FIRST seqno is taken from the decoded batches' minimum
/// `first_seq` (see [`apply_segment`] to pass the leader's authoritative,
/// pre-filter segment-first instead — required to gap-gate a filtered segment
/// whose surviving batches were re-stamped to a higher `first_seq`).
///
/// # Errors
///
/// Returns `WalError::Corruption` on any decode/checksum/overflow failure,
/// before any state is mutated.
pub fn apply_payload_with_leader_seq(
    bytes: &[u8],
    from_shard: ShardId,
    ledger: &SignalLedger,
    state: &ReplicationState,
    lag_gauge: Option<&ReplicationLagGauge>,
    leader_seqno: Option<u64>,
) -> Result<(), WalError> {
    apply_segment(
        bytes,
        from_shard,
        ledger,
        state,
        lag_gauge,
        None,
        leader_seqno,
    )
}

/// Apply a WAL segment payload, gap-gating the WHOLE segment's seqno range.
///
/// `segment_first_seq` is the segment's authoritative FIRST WAL seqno as known
/// by the leader *before* community-overlay filtering (the payload's
/// `id.seqno`); `leader_seqno` is its authoritative LAST. Together they are the
/// segment's true `[first, last]` coverage range — the unit the gap-aware
/// [`ReplicationState::apply_range`] gates on. Passing the leader's pre-filter
/// `segment_first_seq` is load-bearing for filtered segments: a filtered batch
/// is re-stamped to a higher `first_seq` (the dropped local events vacate the
/// front of the range), so the surviving batches' headers no longer reveal the
/// segment's true start; without the explicit value the receiver would mistake
/// the re-stamp for a gap and stall. When `None` (legacy/test callers) the range
/// falls back to the decoded batches' min `first_seq` / max `last_seq`.
///
/// # Why one range per SEGMENT, not per batch or per event
///
/// Within a segment the leader assigns seqnos contiguously and filtering only
/// DROPS events (never reorders), so the whole `[first, last]` span is one
/// atomic unit of seqno coverage on this follower: interior drops (filtered
/// local events) are legitimately covered, never gaps. A genuine gap is a hole
/// BETWEEN segments — an out-of-order eager ship whose `first > frontier + 1`.
/// `apply_range` refuses to advance the contiguous frontier across such a hole
/// (parking the segment instead), so heal/anti-entropy redelivery — gated on
/// `seqno > applied_seqno` (the contiguous frontier) — still backfills the
/// missing prefix in order. This is the lost-update / rolling-upgrade-divergence
/// fix: the old monotonic-max advance jumped the frontier past the gap, after
/// which redelivery saw the gap seqnos as already-applied and never re-shipped
/// them, silently dropping acknowledged events (spec §criterion 7).
///
/// # Errors
///
/// Returns `WalError::Corruption` on any decode/checksum/overflow failure,
/// before any state is mutated.
#[allow(clippy::too_many_arguments)]
pub fn apply_segment(
    bytes: &[u8],
    from_shard: ShardId,
    ledger: &SignalLedger,
    state: &ReplicationState,
    lag_gauge: Option<&ReplicationLagGauge>,
    segment_first_seq: Option<u64>,
    leader_seqno: Option<u64>,
) -> Result<(), WalError> {
    let prepared = prepare_segment(
        bytes,
        from_shard,
        state,
        lag_gauge,
        segment_first_seq,
        leader_seqno,
    )?;
    match prepared {
        // No blob applier on this legacy/test call shape: a blob batch
        // reaching it is a wiring error and halts loudly in commit_segments.
        Some(mut seg) => commit_segments(std::slice::from_mut(&mut seg), ledger, state, None),
        None => Ok(()),
    }
}

/// A validated, gated, not-yet-committed segment: the decoded events plus the
/// authoritative `[first, last]` seqno range, ready for a (possibly shared)
/// group-commit fold via [`commit_segments`].
struct PreparedSegment {
    from_shard: ShardId,
    folded: Vec<(SignalTypeId, EntityId, f64, Timestamp)>,
    /// Blob (kind-1/2) records in batch (seqno) order, applied via the
    /// [`ReplicatedBlobApplier`] BEFORE the signal fold (blobs are idempotent
    /// upserts, so a halt after blob apply but before the fold redelivers
    /// safely; the reverse order would double-fold signals on redelivery).
    blobs: Vec<BlobRecord>,
    first: u64,
    last: u64,
}

/// Phase 1 of [`apply_segment`]: decode + validate the WHOLE payload, advance
/// the lag gauge's leader high-water-mark, and gate against the idempotency
/// boundary. Returns `None` for segments with nothing to commit (empty /
/// all-filtered / already-applied). Mutates NO replication state.
fn prepare_segment(
    bytes: &[u8],
    from_shard: ShardId,
    state: &ReplicationState,
    lag_gauge: Option<&ReplicationLagGauge>,
    segment_first_seq: Option<u64>,
    leader_seqno: Option<u64>,
) -> Result<Option<PreparedSegment>, WalError> {
    // ── Decode + validate the WHOLE payload before mutating state ──
    //
    // Collect every surviving event and the segment's seqno range. If any batch
    // fails decode/checksum/overflow we return here, with neither the follower
    // ledger nor the applied high-water-mark touched.
    let mut folded: Vec<(SignalTypeId, EntityId, f64, Timestamp)> = Vec::new();
    let mut blobs: Vec<BlobRecord> = Vec::new();
    // The lowest first-seq / highest last-seq across all decoded batches, the
    // fallback segment range when no authoritative leader value is supplied.
    let mut min_batch_first_seq: Option<u64> = None;
    let mut max_batch_last_seq: u64 = 0;

    let mut offset = 0;
    while offset < bytes.len() {
        let remaining = &bytes[offset..];
        if remaining.len() < HEADER_SIZE {
            break;
        }
        let (header, payload) =
            decode_batch_payload(remaining).map_err(|e| WalError::Corruption {
                message: format!("corrupt batch at payload offset {offset}: {e}"),
            })?;

        // The last WAL seq is `first_seq + event_count - 1`. Both operands
        // originate from peer- or disk-controlled header bytes, so a
        // corrupt/hostile `first_seq` near `u64::MAX` could wrap. A wrapped
        // seqno would corrupt the idempotency boundary and the lag gauge, so we
        // reject the batch as corruption rather than applying it.
        // (`decode_batch` rejects `event_count == 0` as corruption, so the
        // zero-count arm below is defensive only — an EMPTY SEGMENT is zero
        // batches in the payload, handled by the `segment_last == 0` early
        // return after this loop.)
        let batch_last_seq = if header.event_count > 0 {
            header
                .first_seq
                .checked_add(u64::from(header.event_count) - 1)
                .ok_or_else(|| WalError::Corruption {
                    message: format!(
                        "WAL seq overflow at payload offset {offset}: \
                         first_seq={} + event_count={} would wrap u64",
                        header.first_seq, header.event_count
                    ),
                })?
        } else {
            header.first_seq
        };
        max_batch_last_seq = max_batch_last_seq.max(batch_last_seq);
        min_batch_first_seq =
            Some(min_batch_first_seq.map_or(header.first_seq, |m: u64| m.min(header.first_seq)));

        match payload {
            BatchPayload::Signals(events) => {
                for event in &events {
                    folded.push((
                        SignalTypeId::new(u16::from(event.signal_type)),
                        EntityId::new(event.entity_id),
                        f64::from(event.weight),
                        Timestamp::from_nanos(event.timestamp_nanos),
                    ));
                }
            }
            BatchPayload::ItemMetadata(record) => blobs.push(BlobRecord::ItemMetadata(record)),
            BatchPayload::Embedding(record) => blobs.push(BlobRecord::Embedding(record)),
            // m11p4: a replicated term marker rides the blob path — the
            // applier journals it into THIS node's WAL (advancing its durable
            // tail term) and folds the term-mark cell instead of storage.
            BatchPayload::TermMarker(record) => blobs.push(BlobRecord::TermMarker(record)),
            // m11p5: a replicated membership record rides the same blob path —
            // journaled into THIS node's WAL, then folded into the roster cell
            // (no storage effect).
            BatchPayload::Membership(record) => blobs.push(BlobRecord::Membership(record)),
        }

        let batch_size = HEADER_SIZE + header.payload_len as usize;
        offset += batch_size;
    }

    // Record the leader high-water-mark. Prefer the authoritative segment-level
    // value (known before filtering, so an all-local segment that decoded zero
    // batches still advances it); otherwise fall back to the highest per-batch
    // boundary we observed. The leader's seqno is known the moment the segment
    // arrives, independent of how many batches survived filtering or the
    // idempotency check, so `lag_segments() = leader_seqno − applied_seqno` is
    // the real follower lag.
    let segment_last = leader_seqno
        .filter(|&s| s > 0)
        .unwrap_or(max_batch_last_seq);
    if let Some(gauge) = lag_gauge
        && segment_last > 0
    {
        // Per-source-shard HWM (BUG 1): the status endpoint computes lag against
        // the CURRENT leader's shard, so the HWM must be keyed by the source
        // shard this segment came from, not folded into one scalar that a
        // leadership change would strand. This also advances the single-scalar
        // metric HWM when `from_shard` is this node's tracked shard.
        gauge.update_leader_seqno_for(from_shard, segment_last);
    }

    // An empty segment (all-local, filtered to nothing, or a zero-event batch)
    // covers no seqno range we can apply; it has already advanced the leader-HWM
    // gauge above, which is all an empty segment must do. Return before gating.
    if segment_last == 0 {
        return Ok(None);
    }
    // The segment's authoritative FIRST seqno (pre-filter). Prefer the explicit
    // leader value; fall back to the decoded min. Clamp to <= segment_last so a
    // single-event segment (`first == last`) and a malformed pairing are both
    // well-formed for `apply_range`'s `first <= last` contract.
    let segment_first = segment_first_seq
        .or(min_batch_first_seq)
        .unwrap_or(segment_last)
        .min(segment_last);

    // ── Phase 2: gap-gate the segment range, FOLD, then advance the frontier ──
    //
    // The order is load-bearing: the contiguous frontier (`applied_seqno`) must
    // never be advanced past a seqno whose event is not yet folded into the
    // aggregate. `applied_seqno` is the value the convergence check and the
    // replication-latency poller read to decide "this segment is applied", so an
    // advance-before-fold opens a window where a reader sees `applied >= n` while
    // the n-th event is still pending — reading a count short by one (a transient
    // same-process race, distinct from the cross-restart lost-update). We
    // therefore:
    //
    // 1. GATE with `range_applied` (does NOT mutate): skip a re-ship whose whole
    //    range is at/below the contiguous frontier or is an exact parked range.
    // 2. FOLD the surviving events WAL-first, as ONE staged batch
    //    (`apply_replicated_events`): every event is submitted to the
    //    follower's WAL first, durability is awaited once for the whole set
    //    (they share group-commit batches instead of each paying a solo
    //    batch-timeout + fsync — without this the follower's apply ceiling is
    //    ~1/batch_timeout events/s and it lags unboundedly behind an m11p1
    //    leader), and only then are the events folded into the aggregate. A
    //    WAL failure returns `Err` and halts the receiver BEFORE the frontier
    //    advances, so the follower never acknowledges (via `applied_seqno`)
    //    an event it could not durably record — the re-shipped segment
    //    replays in full from one consistent point on the next open.
    // 3. COMMIT with `apply_range`: advance the contiguous frontier (or park the
    //    range ahead of an open gap, draining any parked contiguous run). Because
    //    the gate in step 1 already excluded the already-applied cases, a fresh
    //    range is always recorded; a `false` return would mean only the
    //    bounded-buffer refusal of a never-closing gap (its WARN fires), which on
    //    the live ship path is unreachable for any real window. If it ever DOES
    //    fire, the receiver halts with an error (below) instead of tearing the
    //    fold/record invariant — a torn range would double-apply on redelivery.
    //
    // The single receiver thread is the only writer of this shard's state, so the
    // gate→fold→commit window cannot race another applier. (Coalesced commits
    // preserve this: `apply_drained` keeps every group's ranges disjoint, so a
    // gate taken here at prepare time cannot be invalidated by a groupmate.)
    if state.range_applied(from_shard, segment_first, segment_last) {
        // Already applied (re-ship at/below the frontier, or an exact parked
        // range): do NOT fold this segment again.
        return Ok(None);
    }
    // Partial-overlap guard (m11p2): a range whose PREFIX is already inside
    // the contiguous frontier would double-fold those events if applied. The
    // protocol never produces partial overlaps (every stream position —
    // applied frontier, dispatch cursor, baseline — is a batch boundary by
    // construction), so this is a defensive refusal: skip with a WARN and let
    // the follower's pull-based catch-up realign at a clean boundary, turning
    // a would-be silent double-count into a self-healing skip.
    let frontier = state.applied_seqno(from_shard).unwrap_or(0);
    if segment_first <= frontier && frontier < segment_last {
        tracing::warn!(
            shard = %from_shard,
            first = segment_first,
            last = segment_last,
            frontier,
            "replication: partially-applied segment range refused; awaiting \
             boundary-aligned redelivery (catch-up stream)"
        );
        return Ok(None);
    }
    Ok(Some(PreparedSegment {
        from_shard,
        folded,
        blobs,
        first: segment_first,
        last: segment_last,
    }))
}

/// Phase 2 of [`apply_segment`], shared by the sequential and coalesced paths:
/// fold every prepared segment's events WAL-first through ONE staged
/// group-commit batch, then advance each segment's range in order.
fn commit_segments(
    prepared: &mut [PreparedSegment],
    ledger: &SignalLedger,
    state: &ReplicationState,
    blob_applier: Option<&dyn ReplicatedBlobApplier>,
) -> Result<(), WalError> {
    let all_events: Vec<(SignalTypeId, EntityId, f64, Timestamp)> = prepared
        .iter()
        .flat_map(|seg| seg.folded.iter().copied())
        .collect();
    if all_events.is_empty() && prepared.is_empty() {
        return Ok(());
    }

    // ── Blob (item-metadata / embedding) records apply FIRST ──
    //
    // The applier routes the whole group's records through the engine's
    // WAL-first item write path as ONE batch (stage all, wait all — group-
    // commit fsyncs; m11p3), then upserts storage. Blobs are idempotent, so
    // a halt AFTER blob apply but BEFORE the signal fold redelivers safely;
    // folding signals first would double-count them when a later blob
    // failure forces redelivery of the same range. The records are DRAINED
    // out of the prepared segments (ownership moves into the applier — no
    // deep clone); on failure the receiver halts and redelivery rebuilds
    // them from the segment bytes, so nothing here needs them back.
    let group_blobs: Vec<BlobRecord> = prepared
        .iter_mut()
        .flat_map(|seg| seg.blobs.drain(..))
        .collect();
    if !group_blobs.is_empty() {
        let Some(applier) = blob_applier else {
            return Err(WalError::Io(std::io::Error::other(format!(
                "received {} blob record(s) but no blob applier is wired on \
                 this receiver; halting rather than silently dropping \
                 replicated items",
                group_blobs.len()
            ))));
        };
        let blob_count = group_blobs.len();
        if let Err(e) = applier.apply_blobs(group_blobs) {
            return Err(WalError::Io(std::io::Error::other(format!(
                "replicated blob batch apply failed ({blob_count} records): {e}"
            ))));
        }
    }

    if !all_events.is_empty() {
        ledger.apply_replicated_events(&all_events).map_err(|e| {
            WalError::Io(std::io::Error::other(format!(
                "follower WAL append failed applying replicated segment group \
                 ({} segments, {} events): {e}",
                prepared.len(),
                all_events.len()
            )))
        })?;
    }
    // Durable now: advance each contiguous frontier (or park ranges ahead of a
    // gap). The prepare-time gates excluded already-applied cases, and group
    // ranges are disjoint, so each records a genuinely new range.
    for seg in prepared {
        let committed = state.apply_range(seg.from_shard, seg.first, seg.last);
        if !committed && !state.range_applied(seg.from_shard, seg.first, seg.last) {
            // The range was refused (ahead-buffer overflow on a never-closing
            // gap, or an unknown shard) AFTER its events were durably folded.
            // Returning `Ok` here would leave a torn state: the events are in
            // the aggregate, but the range is unrecorded — the next redelivery
            // of this segment passes the `range_applied` gate and folds the
            // same events AGAIN, silently double-counting. Halt the receiver
            // instead; the `died` latch surfaces the halt, and
            // heal/anti-entropy redelivery closes the gap in order from one
            // consistent point on the next open.
            tracing::error!(
                shard = %seg.from_shard,
                first = seg.first,
                last = seg.last,
                "folded segment's range could not be recorded (ahead-of-frontier \
                 buffer overflow or unknown shard); halting receiver to prevent \
                 double-apply on redelivery — close the gap via heal/anti-entropy \
                 redelivery and reopen"
            );
            return Err(WalError::Io(std::io::Error::other(format!(
                "replication state refused range [{}, {}] for shard {} after \
                 its events were folded; receiver halted to prevent \
                 double-apply on redelivery",
                seg.first, seg.last, seg.from_shard
            ))));
        }
    }

    Ok(())
}

#[cfg(test)]
#[allow(clippy::unwrap_used)]
mod tests {
    use std::time::Duration;

    use super::*;
    use crate::{
        replication::{segment_id::WalSegmentId, shard::RegionId},
        schema::{DecaySpec, SchemaBuilder, Window},
        signals::NoopWalWriter,
        wal::format::batch::{EventRecord, encode_batch},
    };

    fn make_schema() -> crate::schema::Schema {
        let mut builder = SchemaBuilder::new();
        let _ = builder
            .signal(
                "view",
                crate::schema::EntityKind::Item,
                DecaySpec::Exponential {
                    half_life: Duration::from_secs(7 * 24 * 3600),
                },
            )
            .windows(&[Window::AllTime])
            .velocity(false)
            .add();
        builder.build().unwrap()
    }

    fn make_event(entity_id: u64, signal_type: u8, ts_ns: u64) -> EventRecord {
        EventRecord::signal(entity_id, signal_type, 1.0, ts_ns)
    }

    /// Ship one single-event segment for `entity` at `seqno` (the eager/redeliver
    /// payload shape: `id.seqno == leader_last_seq == seqno`).
    fn ship_single(
        ledger: &SignalLedger,
        state: &ReplicationState,
        type_id: u8,
        entity: u64,
        seqno: u64,
    ) -> Result<(), WalError> {
        let bytes =
            encode_batch(&[make_event(entity, type_id, seqno * 1_000_000)], seqno, 1).unwrap();
        apply_segment(
            &bytes,
            ShardId::SINGLE,
            ledger,
            state,
            None,
            Some(seqno),
            Some(seqno),
        )
    }

    /// REGRESSION (torn fold/record): when `apply_range` refuses to record a
    /// range AFTER its events were durably folded (unknown shard here; the
    /// ahead-buffer overflow takes the same path in production), `apply_segment`
    /// must halt with an error rather than return `Ok` — an `Ok` would let the
    /// next redelivery of the same segment pass the `range_applied` gate and
    /// fold the same events again, silently double-counting.
    #[test]
    fn apply_segment_halts_when_folded_range_cannot_be_recorded() {
        let schema = make_schema();
        let ledger = Arc::new(SignalLedger::new(schema, Box::new(NoopWalWriter)));
        let state = ReplicationState::single();
        let tid = ledger.resolve_signal_type("view").unwrap().as_u16() as u8;

        // A shard the state does not track: `apply_range` returns `false` and
        // `range_applied` stays `false` — the refused-after-fold torn state.
        let untracked = ShardId(99);
        let bytes = encode_batch(&[make_event(10, tid, 1_000_000)], 1, 1).unwrap();
        let err = apply_segment(&bytes, untracked, &ledger, &state, None, Some(1), Some(1))
            .expect_err("a folded range that cannot be recorded must halt the receiver");
        assert!(
            err.to_string().contains("halted"),
            "halt error must explain the refusal: {err}"
        );
    }

    /// REGRESSION (rolling-upgrade divergence, lost acknowledged events): an
    /// out-of-order eager ship that skips ahead of the contiguous frontier must
    /// NOT advance the frontier across the gap, and the gap MUST be backfillable
    /// by heal redelivery (gated on `seqno > applied_seqno`) without losing or
    /// double-counting any event.
    ///
    /// Before the fix, `apply_range` was a monotonic max: the ahead ship for
    /// seqno 8 jumped the frontier to 8, swallowing 6,7. Heal redelivery then
    /// gated on `seqno > applied` (now 8) and skipped 6,7, so those acknowledged
    /// events were lost from the follower's aggregate forever — the exact 0.5-vs-
    /// 1.0/0.0 trending divergence the m8p10 UAT surfaced (a one-event-per-item
    /// hole the rank-relative normalize blew up). This test models that interleave
    /// at the engine level (no OS processes) and proves the follower converges to
    /// the leader's per-entity counts.
    #[test]
    fn out_of_order_ship_then_heal_loses_no_events() {
        let schema = make_schema();
        let ledger = Arc::new(SignalLedger::new(schema, Box::new(NoopWalWriter)));
        let state = ReplicationState::single();
        let tid = ledger.resolve_signal_type("view").unwrap().as_u16() as u8;

        // Leader's logical stream: seqno -> entity. Each event is one `view` on
        // its entity. The leader applied all 8 (its own contiguous WAL).
        let stream: [(u64, u64); 8] = [
            (1, 10),
            (2, 11),
            (3, 12),
            (4, 13),
            (5, 14),
            (6, 12), // the events the buggy version lost (entities 12,13 lose one)
            (7, 13),
            (8, 15),
        ];

        // In-order eager ships for 1..=5 land while the follower is up.
        for &(seq, entity) in &stream[..5] {
            ship_single(&ledger, &state, tid, entity, seq).unwrap();
        }
        assert_eq!(state.applied_seqno(ShardId::SINGLE), Some(5));

        // The follower "restarts": seqnos 6,7 were shipped while it was down and
        // FAILED (breaker). After it returns, an eager ship for the NEXT write
        // (seqno 8) arrives FIRST — out of order, ahead of the 6,7 gap.
        ship_single(&ledger, &state, tid, 15, 8).unwrap();
        assert_eq!(
            state.applied_seqno(ShardId::SINGLE),
            Some(5),
            "the out-of-order ship for 8 must NOT advance the frontier across the 6,7 gap"
        );

        // Heal redelivery re-ships everything above the contiguous frontier (5) in
        // order: 6, 7, 8. (8 is a re-ship of the parked range — must not double-
        // apply.) An operator-style redelivery would gate on applied_seqno=5.
        for &(seq, entity) in &stream[5..] {
            ship_single(&ledger, &state, tid, entity, seq).unwrap();
        }
        assert_eq!(
            state.applied_seqno(ShardId::SINGLE),
            Some(8),
            "heal redelivery must close the gap and reach the leader's last seq"
        );

        // The follower's per-entity `view` counts must equal the leader's logical
        // stream: entity 12 got events at seqs 3 AND 6 (count 2), entity 13 at 4
        // AND 7 (count 2), entity 15 at 8 (count 1, NOT double-counted by the
        // re-shipped seqno 8). The lost-update bug would leave 12 and 13 at count
        // 1 each (the 6,7 events dropped).
        let count = |e: u64| {
            ledger
                .read_windowed_count(EntityId::new(e), "view", Window::AllTime)
                .unwrap()
        };
        assert_eq!(count(10), 1);
        assert_eq!(count(11), 1);
        assert_eq!(
            count(12),
            2,
            "entity 12 must keep BOTH events (seqs 3 and 6)"
        );
        assert_eq!(
            count(13),
            2,
            "entity 13 must keep BOTH events (seqs 4 and 7)"
        );
        assert_eq!(count(14), 1);
        assert_eq!(
            count(15),
            1,
            "entity 15's seqno-8 event must be applied EXACTLY once (no double-count from the re-ship)"
        );
    }

    #[test]
    fn apply_payload_updates_ledger() {
        let schema = make_schema();
        let ledger = Arc::new(SignalLedger::new(schema, Box::new(NoopWalWriter)));
        let state = Arc::new(ReplicationState::new(&[ShardId::SINGLE]));

        // Resolve the signal type for "view" to get the correct type id.
        let type_id = ledger.resolve_signal_type("view").unwrap();

        let events = vec![make_event(42, type_id.as_u16() as u8, 1_000_000_000)];
        let bytes = encode_batch(&events, 1, 1).unwrap();

        apply_payload(&bytes, ShardId::SINGLE, &ledger, &state, None).unwrap();

        // Verify the ledger was updated.
        assert!(ledger.entries().contains_key(&(EntityId::new(42), type_id)));
        assert_eq!(state.applied_seqno(ShardId::SINGLE), Some(1));
    }

    #[test]
    fn apply_payload_idempotent() {
        let schema = make_schema();
        let ledger = Arc::new(SignalLedger::new(schema, Box::new(NoopWalWriter)));
        let state = Arc::new(ReplicationState::new(&[ShardId::SINGLE]));

        let type_id = ledger.resolve_signal_type("view").unwrap();
        let events = vec![make_event(42, type_id.as_u16() as u8, 1_000_000_000)];
        let bytes = encode_batch(&events, 1, 1).unwrap();

        // Apply once.
        apply_payload(&bytes, ShardId::SINGLE, &ledger, &state, None).unwrap();
        // Apply again -- should be idempotent.
        apply_payload(&bytes, ShardId::SINGLE, &ledger, &state, None).unwrap();

        assert_eq!(state.applied_seqno(ShardId::SINGLE), Some(1));
    }

    #[test]
    fn apply_payload_multiple_batches() {
        let schema = make_schema();
        let ledger = Arc::new(SignalLedger::new(schema, Box::new(NoopWalWriter)));
        let state = Arc::new(ReplicationState::new(&[ShardId::SINGLE]));

        let type_id = ledger.resolve_signal_type("view").unwrap();
        let e1 = vec![make_event(1, type_id.as_u16() as u8, 100)];
        let e2 = vec![make_event(2, type_id.as_u16() as u8, 200)];

        let mut bytes = encode_batch(&e1, 1, 100).unwrap();
        bytes.extend(encode_batch(&e2, 2, 200).unwrap());

        apply_payload(&bytes, ShardId::SINGLE, &ledger, &state, None).unwrap();

        assert!(ledger.entries().contains_key(&(EntityId::new(1), type_id)));
        assert!(ledger.entries().contains_key(&(EntityId::new(2), type_id)));
        assert_eq!(state.applied_seqno(ShardId::SINGLE), Some(2));
    }

    /// CRITICAL (REPL): a community-overlay filter must not move the WAL
    /// high-water-mark. Applying a filtered segment must advance `applied_seqno`
    /// to the *same* boundary as applying the original (unfiltered) segment,
    /// even though the filtered batch carries fewer events. This is the
    /// load-bearing invariant for both per-segment idempotency and the
    /// `lag = leader_seqno - applied_seqno` gauge.
    #[test]
    fn filtered_segment_advances_to_same_boundary_as_unfiltered() {
        use crate::governance::{CommunityId, SignalScope};
        use crate::replication::shipper::filter_segment_drop_local;

        let schema = make_schema();

        let community = |entity: u64, ts: u64| {
            EventRecord::scoped(
                entity,
                crate::wal::format::batch::RECORD_TYPE_SIGNAL,
                1.0,
                ts,
                SignalScope::Community(CommunityId(1)).discriminant(),
                0,
                0,
                0,
            )
        };
        let local = |entity: u64, ts: u64| make_event(entity, 1, ts);

        // A single batch starting at first_seq=10 with 4 events (true last seq
        // = 13); the middle two are local and will be stripped by the filter.
        let events = vec![
            community(100, 1_000_000_000),
            local(101, 1_100_000_000),
            local(102, 1_200_000_000),
            community(103, 1_300_000_000),
        ];
        let original = encode_batch(&events, 10, 1).unwrap();
        let filtered = filter_segment_drop_local(&original);
        assert!(
            !filtered.is_empty(),
            "two community events survive the filter"
        );

        // Apply each via `apply_segment` carrying the segment's AUTHORITATIVE
        // range `[10, 13]` (the leader knows this before filtering; the real
        // receiver reads it from the payload's `id.seqno` + `leader_last_seq`).
        // Passing it is what lets the filtered segment — whose surviving batch is
        // re-stamped to a higher first_seq — gate on its TRUE range rather than
        // be mistaken for a gap. Both states start at frontier 0 and the segment
        // begins at seqno 10 > frontier+1, so this also exercises the gap-aware
        // path: the very first segment a node ever applies establishes its
        // baseline (the leader ships from the follower's applied+1, so seqno 10
        // here is the contiguous start for this follower).
        let ledger_a = Arc::new(SignalLedger::new(schema.clone(), Box::new(NoopWalWriter)));
        let state_a = ReplicationState::new(&[ShardId::SINGLE]);
        // Frontier starts at 9 so the segment `[10,13]` is contiguous (the
        // follower has applied through 9 from an earlier in-order stream).
        state_a.advance(ShardId::SINGLE, 9);
        apply_segment(
            &original,
            ShardId::SINGLE,
            &ledger_a,
            &state_a,
            None,
            Some(10),
            Some(13),
        )
        .unwrap();

        // Apply the filtered segment to a fresh state at the same frontier.
        let ledger_b = Arc::new(SignalLedger::new(schema, Box::new(NoopWalWriter)));
        let state_b = ReplicationState::new(&[ShardId::SINGLE]);
        state_b.advance(ShardId::SINGLE, 9);
        apply_segment(
            &filtered,
            ShardId::SINGLE,
            &ledger_b,
            &state_b,
            None,
            Some(10),
            Some(13),
        )
        .unwrap();

        assert_eq!(
            state_a.applied_seqno(ShardId::SINGLE),
            Some(13),
            "unfiltered segment advances to its true last seq (10 + 4 - 1)"
        );
        assert_eq!(
            state_b.applied_seqno(ShardId::SINGLE),
            state_a.applied_seqno(ShardId::SINGLE),
            "filtered segment must advance to the SAME boundary as unfiltered"
        );
    }

    /// CRITICAL (REPL): the filtered-vs-unfiltered boundary equivalence must
    /// hold for arbitrary mixes of local/community events across multiple
    /// batches — not just the single hand-picked case above.
    ///
    /// Scope of the invariant: the per-batch re-stamp guarantees a surviving
    /// batch preserves *its own* last seq. A segment's overall boundary is the
    /// last seq of its last surviving batch, so the equivalence holds whenever
    /// the trailing batch survives the filter. We therefore force the final
    /// batch to carry a community event (its boundary must survive); the
    /// interior batches are arbitrary local/community mixes. (A trailing
    /// all-local batch legitimately produces no shippable bytes, so its seqno
    /// range cannot — and should not — be reconstructed by the receiver.)
    #[test]
    fn prop_filtered_segment_preserves_boundary() {
        use proptest::prelude::*;

        use crate::governance::{CommunityId, SignalScope};
        use crate::replication::shipper::filter_segment_drop_local;

        // Each event is (is_community, entity_id). Batches are non-empty.
        let event_strategy = (any::<bool>(), 1u64..1000);
        let batch_strategy = proptest::collection::vec(event_strategy, 1..6);
        let batches_strategy = proptest::collection::vec(batch_strategy, 1..4);

        let mut runner = proptest::test_runner::TestRunner::new(proptest::test_runner::Config {
            cases: 128,
            ..Default::default()
        });
        runner
            .run(&batches_strategy, |batches| {
                let schema = make_schema();
                let community = |entity: u64| {
                    EventRecord::scoped(
                        entity,
                        crate::wal::format::batch::RECORD_TYPE_SIGNAL,
                        1.0,
                        1_000_000_000,
                        SignalScope::Community(CommunityId(1)).discriminant(),
                        0,
                        0,
                        0,
                    )
                };

                // Lay out batches with contiguous WAL seqnos: batch i gets
                // first_seq = next, occupying `len` slots.
                let mut original = Vec::new();
                let mut next_seq = 1u64;
                let last_idx = batches.len() - 1;
                for (i, batch) in batches.iter().enumerate() {
                    let mut recs: Vec<EventRecord> = batch
                        .iter()
                        .map(|&(is_comm, entity)| {
                            if is_comm {
                                community(entity)
                            } else {
                                make_event(entity, 1, 1_000_000_000)
                            }
                        })
                        .collect();
                    // Force the LAST event of the LAST batch to be community so
                    // the segment's trailing boundary always survives the filter.
                    if i == last_idx {
                        let last = recs.len() - 1;
                        recs[last] = community(9999);
                    }
                    original.extend(encode_batch(&recs, next_seq, 1).unwrap());
                    next_seq += batch.len() as u64;
                }
                let true_last_seq = next_seq - 1;

                let filtered = filter_segment_drop_local(&original);

                // Apply via `apply_segment` with the segment's authoritative range
                // `[1, true_last_seq]` (seqnos laid out contiguously from 1, so the
                // segment is frontier-contiguous on a fresh node). The filtered
                // version's surviving batches are re-stamped to higher first_seqs,
                // but the explicit segment-first keeps both gating on the SAME true
                // range — the receiver contract the real `spawn_receiver` uses.
                let ledger_a = Arc::new(SignalLedger::new(schema.clone(), Box::new(NoopWalWriter)));
                let state_a = ReplicationState::new(&[ShardId::SINGLE]);
                apply_segment(
                    &original,
                    ShardId::SINGLE,
                    &ledger_a,
                    &state_a,
                    None,
                    Some(1),
                    Some(true_last_seq),
                )
                .unwrap();

                let ledger_b = Arc::new(SignalLedger::new(schema, Box::new(NoopWalWriter)));
                let state_b = ReplicationState::new(&[ShardId::SINGLE]);
                apply_segment(
                    &filtered,
                    ShardId::SINGLE,
                    &ledger_b,
                    &state_b,
                    None,
                    Some(1),
                    Some(true_last_seq),
                )
                .unwrap();

                prop_assert_eq!(state_a.applied_seqno(ShardId::SINGLE), Some(true_last_seq));
                prop_assert_eq!(
                    state_b.applied_seqno(ShardId::SINGLE),
                    state_a.applied_seqno(ShardId::SINGLE),
                    "filtered segment must reach the same boundary as unfiltered"
                );
                Ok(())
            })
            .unwrap();
    }

    /// WARNING (REPL): a corrupt header whose `first_seq + event_count` wraps
    /// u64 must be rejected as corruption, not silently wrapped into a bogus
    /// (small) high-water-mark.
    #[test]
    fn apply_payload_seq_overflow_is_corruption() {
        let schema = make_schema();
        let ledger = Arc::new(SignalLedger::new(schema, Box::new(NoopWalWriter)));
        let state = ReplicationState::new(&[ShardId::SINGLE]);

        let type_id = ledger.resolve_signal_type("view").unwrap();
        // Encode a structurally-valid 2-event batch with first_seq = u64::MAX.
        // The checksum is computed over these bytes, so decode succeeds and the
        // batch reaches the seq arithmetic, where first_seq + (2 - 1) wraps.
        let e1 = make_event(1, type_id.as_u16() as u8, 100);
        let e2 = make_event(2, type_id.as_u16() as u8, 200);
        let bytes = encode_batch(&[e1, e2], u64::MAX, 1).unwrap();

        let result = apply_payload(&bytes, ShardId::SINGLE, &ledger, &state, None);
        assert!(
            matches!(result, Err(WalError::Corruption { .. })),
            "first_seq=u64::MAX with 2 events must overflow → Corruption, got {result:?}"
        );
        assert_eq!(
            state.applied_seqno(ShardId::SINGLE),
            Some(0),
            "overflowing batch must not advance the high-water-mark"
        );
    }

    #[test]
    fn apply_payload_corrupt_returns_error() {
        let schema = make_schema();
        let ledger = Arc::new(SignalLedger::new(schema, Box::new(NoopWalWriter)));
        let state = Arc::new(ReplicationState::new(&[ShardId::SINGLE]));

        // Build a valid batch then corrupt the checksum region.
        let type_id = ledger.resolve_signal_type("view").unwrap();
        let events = vec![make_event(7, type_id.as_u16() as u8, 500)];
        let mut bytes = encode_batch(&events, 1, 1).unwrap();
        // Flip bytes in the checksum region (bytes 32-63) to break BLAKE3.
        for b in &mut bytes[32..64] {
            *b = b.wrapping_add(1);
        }

        let result = apply_payload(&bytes, ShardId::SINGLE, &ledger, &state, None);
        assert!(
            matches!(result, Err(WalError::Corruption { .. })),
            "expected Corruption, got {result:?}"
        );
        // Ledger must NOT have been updated.
        assert!(!ledger.entries().contains_key(&(EntityId::new(7), type_id)));
    }

    /// WARNING (REPL): a payload whose LATER batch is corrupt must apply
    /// **none** of its batches and must NOT advance the applied high-water-mark.
    /// The earlier (valid) batch must not be half-applied — the leader can then
    /// cleanly re-ship the same un-acked segment and replay it from one
    /// consistent resume point.
    #[test]
    fn apply_payload_atomic_on_mid_payload_corruption() {
        let schema = make_schema();
        let ledger = Arc::new(SignalLedger::new(schema, Box::new(NoopWalWriter)));
        let state = ReplicationState::new(&[ShardId::SINGLE]);
        let type_id = ledger.resolve_signal_type("view").unwrap();

        // First batch (seqs 1..=1): valid, entity 100. Second batch (seqs
        // 2..=2): valid then corrupted so decode fails AFTER the first batch.
        let good = vec![make_event(100, type_id.as_u16() as u8, 100)];
        let bad = vec![make_event(200, type_id.as_u16() as u8, 200)];
        let mut bytes = encode_batch(&good, 1, 100).unwrap();
        let good_len = bytes.len();
        bytes.extend(encode_batch(&bad, 2, 200).unwrap());
        // Corrupt the SECOND batch's checksum region (bytes 32..64 of it).
        for b in &mut bytes[good_len + 32..good_len + 64] {
            *b = b.wrapping_add(1);
        }

        let result = apply_payload(&bytes, ShardId::SINGLE, &ledger, &state, None);
        assert!(
            matches!(result, Err(WalError::Corruption { .. })),
            "mid-payload corruption must return Corruption, got {result:?}"
        );
        // NEITHER batch may be applied: the first (valid) batch must not have
        // been committed before the second batch failed to decode.
        assert!(
            !ledger
                .entries()
                .contains_key(&(EntityId::new(100), type_id)),
            "earlier valid batch must NOT be applied when a later batch is corrupt"
        );
        assert!(
            !ledger
                .entries()
                .contains_key(&(EntityId::new(200), type_id)),
            "corrupt batch must not be applied"
        );
        // The applied high-water-mark must NOT have advanced past 0.
        assert_eq!(
            state.applied_seqno(ShardId::SINGLE),
            Some(0),
            "half-applied payload must not advance the applied high-water-mark"
        );
    }

    /// WARNING (REPL): in `community_share_only` mode a segment whose events are
    /// all local filters to empty bytes that decode to ZERO batches. The
    /// receiver must still advance the lag gauge's leader high-water-mark from
    /// the segment's authoritative `leader_last_seq` (known before filtering),
    /// so the gauge does not under-report the leader's progress across the
    /// all-local segment.
    #[test]
    fn lag_gauge_tracks_leader_hwm_across_all_local_empty_segment() {
        use crate::replication::lag::ReplicationLagGauge;

        let schema = make_schema();
        let ledger = Arc::new(SignalLedger::new(schema, Box::new(NoopWalWriter)));
        let state = Arc::new(ReplicationState::new(&[ShardId::SINGLE]));
        let gauge = Arc::new(ReplicationLagGauge::new(
            ShardId::SINGLE,
            Arc::clone(&state),
        ));

        // An all-local segment filters to empty bytes: the leader's TRUE last
        // WAL seqno for it is 5 (e.g. a batch first_seq=4 with 2 events), but the
        // shipped bytes are empty.
        let empty_bytes: Vec<u8> = Vec::new();
        apply_payload_with_leader_seq(
            &empty_bytes,
            ShardId::SINGLE,
            &ledger,
            &state,
            Some(&gauge),
            Some(5),
        )
        .unwrap();

        assert_eq!(
            gauge.leader_seqno(),
            5,
            "leader HWM must advance from the authoritative leader_last_seq even \
             when the filtered segment decodes zero batches"
        );
        // No events were applied (empty segment), so the applied seqno stays 0.
        assert_eq!(state.applied_seqno(ShardId::SINGLE), Some(0));
        // The gauge therefore reports the leader being ahead of the follower.
        assert_eq!(
            gauge.lag_segments(),
            5,
            "leader at 5, applied 0 → 5 segments behind across the empty segment"
        );
    }

    /// A minimal transport that returns one payload then signals shutdown.
    struct OneShot {
        rx: crossbeam::channel::Receiver<crate::replication::WalSegmentPayload>,
    }

    impl Transport for OneShot {
        fn send_segment(
            &self,
            _to: ShardId,
            _payload: crate::replication::WalSegmentPayload,
        ) -> Result<(), crate::replication::TransportError> {
            Ok(())
        }
        fn recv_segment(&self) -> Option<crate::replication::WalSegmentPayload> {
            self.rx.recv().ok()
        }
        fn try_recv_segment(&self) -> Option<crate::replication::WalSegmentPayload> {
            self.rx.try_recv().ok()
        }
        fn local_shard(&self) -> ShardId {
            ShardId(1)
        }
    }

    /// REGRESSION (m11p1 coalescing, within-batch double-fold): a drained
    /// backlog may carry the SAME seqnos twice — a sender-retry duplicate of a
    /// whole run, and a heal-path single-event re-ship INSIDE an already-queued
    /// batched run. The idempotency gate reads `ReplicationState`, which only
    /// advances at commit time, so a naive coalesced apply would gate both
    /// copies before either committed and fold the shared events twice.
    /// `apply_drained` must keep each group's ranges disjoint and defer
    /// overlapping payloads to a later group (where the advanced state gates
    /// them out) — every entity's count must equal the leader's logical stream.
    #[test]
    fn coalesced_backlog_with_duplicates_and_subsets_folds_once() {
        let schema = make_schema();
        let ledger = Arc::new(SignalLedger::new(schema, Box::new(NoopWalWriter)));
        let state = Arc::new(ReplicationState::new(&[ShardId::SINGLE]));
        let tid = ledger.resolve_signal_type("view").unwrap().as_u16() as u8;

        // The leader's logical stream: seqnos 1..=6, one view per entity 1..=6,
        // shipped as two batched runs [1,3] and [4,6].
        let run = |first: u64, entities: &[u64]| {
            let events: Vec<EventRecord> = entities
                .iter()
                .enumerate()
                .map(|(i, &e)| make_event(e, tid, (first + i as u64) * 1_000_000))
                .collect();
            let last = first + entities.len() as u64 - 1;
            crate::replication::WalSegmentPayload {
                id: WalSegmentId::new(RegionId::SINGLE, ShardId::SINGLE, first),
                bytes: encode_batch(&events, first, 1).unwrap(),
                event_count: entities.len() as u64,
                leader_last_seq: last,
                stream_baseline: 0,
                term: 0,
                leader_region: 0,
            }
        };

        let backlog = vec![
            run(1, &[1, 2, 3]),
            run(1, &[1, 2, 3]), // exact retry duplicate of [1,3]
            run(4, &[4, 5, 6]),
            run(2, &[2]), // heal single-event re-ship INSIDE the [1,3] run
            run(5, &[5]), // heal single inside [4,6]
        ];
        apply_drained(backlog, &ledger, &state, None, None).unwrap();

        assert_eq!(
            state.applied_seqno(ShardId::SINGLE),
            Some(6),
            "the full stream must be applied"
        );
        for entity in 1..=6u64 {
            let count = ledger
                .read_windowed_count(EntityId::new(entity), "view", Window::AllTime)
                .unwrap();
            assert_eq!(
                count, 1,
                "entity {entity} must fold EXACTLY once despite the duplicate \
                 run and the overlapping heal singles"
            );
        }
    }

    #[test]
    fn receiver_thread_exits_on_transport_close() {
        let (tx, rx) = crossbeam::channel::bounded(4);
        let transport = Arc::new(OneShot { rx });

        let schema = make_schema();
        let ledger = Arc::new(SignalLedger::new(schema, Box::new(NoopWalWriter)));
        let state = Arc::new(ReplicationState::new(&[ShardId(0)]));

        let handle = spawn_receiver(
            Arc::clone(&transport),
            Arc::clone(&ledger),
            Arc::clone(&state),
            None,
            None,
        );

        // Send one segment.
        let type_id = ledger.resolve_signal_type("view").unwrap();
        let events = vec![make_event(99, type_id.as_u16() as u8, 100)];
        let payload_bytes = encode_batch(&events, 1, 1).unwrap();
        tx.send(crate::replication::WalSegmentPayload {
            id: WalSegmentId::new(RegionId::SINGLE, ShardId(0), 1),
            bytes: payload_bytes,
            event_count: 1,
            leader_last_seq: 1,
            stream_baseline: 0,
            term: 0,
            leader_region: 0,
        })
        .unwrap();

        // Give the receiver a moment to process.
        std::thread::sleep(Duration::from_millis(50));

        // Drop sender -- receiver's recv will return None.
        drop(tx);

        // The receiver should exit gracefully.
        handle.join().unwrap();

        // Verify the segment was applied.
        assert!(ledger.entries().contains_key(&(EntityId::new(99), type_id)));
        assert_eq!(state.applied_seqno(ShardId(0)), Some(1));
    }

    #[test]
    fn receiver_thread_exits_on_corrupt_payload() {
        let (tx, rx) = crossbeam::channel::bounded(4);
        let transport = Arc::new(OneShot { rx });

        let schema = make_schema();
        let ledger = Arc::new(SignalLedger::new(schema, Box::new(NoopWalWriter)));
        let state = Arc::new(ReplicationState::new(&[ShardId(0)]));

        let handle = spawn_receiver(
            Arc::clone(&transport),
            Arc::clone(&ledger),
            Arc::clone(&state),
            None,
            None,
        );

        // Build a valid batch then corrupt it.
        let type_id = ledger.resolve_signal_type("view").unwrap();
        let events = vec![make_event(55, type_id.as_u16() as u8, 100)];
        let mut corrupt_bytes = encode_batch(&events, 1, 1).unwrap();
        for b in &mut corrupt_bytes[32..64] {
            *b = b.wrapping_add(1);
        }

        tx.send(crate::replication::WalSegmentPayload {
            id: WalSegmentId::new(RegionId::SINGLE, ShardId(0), 1),
            bytes: corrupt_bytes,
            event_count: 1,
            leader_last_seq: 1,
            stream_baseline: 0,
            term: 0,
            leader_region: 0,
        })
        .unwrap();

        // Give thread time to process.
        std::thread::sleep(Duration::from_millis(50));
        drop(tx);

        let result = handle.join();
        assert!(
            matches!(result, Err(WalError::Corruption { .. })),
            "expected Corruption from join, got {result:?}"
        );
        // Entity 55 must NOT have been applied.
        assert!(!ledger.entries().contains_key(&(EntityId::new(55), type_id)));
    }

    /// C11: a receiver that halts on a corrupt segment must flip its `died`
    /// liveness latch WITHOUT the owner ever calling `join()` — so `health_check`
    /// can observe a silently-stalled follower while the node is still open,
    /// instead of the death being visible only at shutdown.
    #[test]
    fn receiver_died_latch_set_on_corrupt_without_join() {
        let (tx, rx) = crossbeam::channel::bounded(4);
        let transport = Arc::new(OneShot { rx });

        let schema = make_schema();
        let ledger = Arc::new(SignalLedger::new(schema, Box::new(NoopWalWriter)));
        let state = Arc::new(ReplicationState::new(&[ShardId(0)]));

        let handle = spawn_receiver(
            Arc::clone(&transport),
            Arc::clone(&ledger),
            Arc::clone(&state),
            None,
            None,
        );

        // Healthy until something breaks.
        assert!(!handle.died(), "fresh receiver must not report died");

        // Feed one corrupt segment.
        let type_id = ledger.resolve_signal_type("view").unwrap();
        let events = vec![make_event(55, type_id.as_u16() as u8, 100)];
        let mut corrupt_bytes = encode_batch(&events, 1, 1).unwrap();
        for b in &mut corrupt_bytes[32..64] {
            *b = b.wrapping_add(1);
        }
        tx.send(crate::replication::WalSegmentPayload {
            id: WalSegmentId::new(RegionId::SINGLE, ShardId(0), 1),
            bytes: corrupt_bytes,
            event_count: 1,
            leader_last_seq: 1,
            stream_baseline: 0,
            term: 0,
            leader_region: 0,
        })
        .unwrap();

        // Poll the latch (NOT join): the receiver must halt and set `died` while
        // the channel/transport stays open and the owner never calls join().
        let deadline = std::time::Instant::now() + Duration::from_secs(5);
        while !handle.died() && std::time::Instant::now() < deadline {
            std::thread::sleep(Duration::from_millis(5));
        }
        assert!(
            handle.died(),
            "a corrupt segment must flip the receiver's died latch without join()"
        );
        // The transport is still open (tx not dropped) — the death is observable
        // purely from the latch, exactly what health_check reads.
        drop(tx);
    }

    /// C11: a CLEAN shutdown (transport closed via `recv_segment()` == None) must
    /// NOT set the `died` latch — otherwise `health_check` would flag an
    /// intentional teardown as a fault.
    #[test]
    fn receiver_died_latch_unset_on_clean_shutdown() {
        let (tx, rx) = crossbeam::channel::bounded(4);
        let transport = Arc::new(OneShot { rx });

        let schema = make_schema();
        let ledger = Arc::new(SignalLedger::new(schema, Box::new(NoopWalWriter)));
        let state = Arc::new(ReplicationState::new(&[ShardId(0)]));

        let handle = spawn_receiver(
            Arc::clone(&transport),
            Arc::clone(&ledger),
            Arc::clone(&state),
            None,
            None,
        );

        // Clean shutdown: drop the sender so recv_segment() returns None.
        drop(tx);

        // The thread exits Ok; the latch must stay false.
        let deadline = std::time::Instant::now() + Duration::from_secs(5);
        while !handle.is_finished() && std::time::Instant::now() < deadline {
            std::thread::sleep(Duration::from_millis(5));
        }
        assert!(
            handle.is_finished(),
            "clean shutdown must let the thread exit"
        );
        assert!(
            !handle.died(),
            "a clean shutdown must NOT set the died latch (no false-positive fault)"
        );
        handle.join().unwrap();
    }

    /// obs-REPL-1: a follower that has received batches from the leader but not
    /// yet applied all of them reports a NON-ZERO replication lag; the lag
    /// returns to 0 once the follower catches up. This proves the receiver
    /// feeds the lag gauge's leader high-water-mark from the real apply path,
    /// not just from test-only `update_leader_seqno` calls.
    #[test]
    fn apply_payload_feeds_lag_gauge_until_caught_up() {
        use crate::replication::lag::ReplicationLagGauge;

        let schema = make_schema();
        let ledger = Arc::new(SignalLedger::new(schema, Box::new(NoopWalWriter)));
        let state = Arc::new(ReplicationState::new(&[ShardId::SINGLE]));
        let gauge = Arc::new(ReplicationLagGauge::new(
            ShardId::SINGLE,
            Arc::clone(&state),
        ));

        let type_id = ledger.resolve_signal_type("view").unwrap();

        // First segment carries WAL seqs 1..=2 (leader high-water = 2).
        let e1 = vec![make_event(1, type_id.as_u16() as u8, 100)];
        let e2 = vec![make_event(2, type_id.as_u16() as u8, 200)];
        let mut first = encode_batch(&e1, 1, 100).unwrap();
        first.extend(encode_batch(&e2, 2, 200).unwrap());

        // Apply the first segment: leader HWM 2, applied 2 → caught up.
        apply_payload(&first, ShardId::SINGLE, &ledger, &state, Some(&gauge)).unwrap();
        assert_eq!(gauge.leader_seqno(), 2);
        assert_eq!(gauge.applied_seqno(), 2);
        assert_eq!(gauge.lag_segments(), 0, "fully applied → no lag");

        // Simulate received-but-not-yet-applied backpressure: the leader has
        // shipped up through seq 5, but the follower has only applied 2.
        // Recording the leader HWM ahead of application is exactly what the
        // receiver does for each decoded batch before the apply step.
        gauge.update_leader_seqno(5);
        assert_eq!(
            gauge.lag_segments(),
            3,
            "leader at 5, applied 2 → 3 segments behind"
        );

        // Deliver the missing segment (WAL seqs 3..=5) and apply it: caught up.
        let e3 = vec![make_event(3, type_id.as_u16() as u8, 300)];
        let e4 = vec![make_event(4, type_id.as_u16() as u8, 400)];
        let e5 = vec![make_event(5, type_id.as_u16() as u8, 500)];
        let mut catchup = encode_batch(&e3, 3, 300).unwrap();
        catchup.extend(encode_batch(&e4, 4, 400).unwrap());
        catchup.extend(encode_batch(&e5, 5, 500).unwrap());

        apply_payload(&catchup, ShardId::SINGLE, &ledger, &state, Some(&gauge)).unwrap();
        assert_eq!(gauge.leader_seqno(), 5);
        assert_eq!(gauge.applied_seqno(), 5);
        assert_eq!(gauge.lag_segments(), 0, "caught up → lag back to 0");
    }

    /// m11p2: a payload mixing a kind-1 item blob, a kind-0 signal batch, and
    /// a kind-2 embedding blob applies atomically — blobs through the applier
    /// (in seqno order, BEFORE the signal fold), signals into the ledger — and
    /// advances the contiguous frontier across the whole range.
    #[test]
    fn mixed_kind_payload_applies_blobs_and_signals() {
        use std::sync::Mutex as StdMutex;

        use crate::replication::shard::RegionId;
        use crate::wal::format::batch::{
            EmbeddingRecord, ItemMetadataRecord, encode_embedding_batch, encode_item_metadata_batch,
        };

        #[derive(Default)]
        struct RecordingApplier {
            applied: StdMutex<Vec<String>>,
        }
        impl ReplicatedBlobApplier for RecordingApplier {
            fn apply_blobs(&self, records: Vec<BlobRecord>) -> crate::Result<()> {
                let lines: Vec<String> = records
                    .iter()
                    .map(|record| match record {
                        BlobRecord::ItemMetadata(r) => {
                            format!("item:{}:{}", r.entity_id, r.metadata_bytes.len())
                        }
                        BlobRecord::Embedding(r) => {
                            format!("emb:{}:{}", r.entity_id, r.values.len())
                        }
                        BlobRecord::TermMarker(r) => format!("term:{}", r.term),
                        BlobRecord::Membership(r) => {
                            format!("membership:v{}:{}", r.version, r.members.len())
                        }
                    })
                    .collect();
                self.applied.lock().unwrap().extend(lines);
                Ok(())
            }
        }

        let schema = make_schema();
        let ledger = Arc::new(SignalLedger::new(schema, Box::new(NoopWalWriter)));
        let state = ReplicationState::new(&[ShardId::SINGLE]);
        let tid = ledger.resolve_signal_type("view").unwrap().as_u16() as u8;

        // seq 1: item blob; seqs 2-3: two signal events; seq 4: embedding blob.
        let mut bytes = encode_item_metadata_batch(
            &ItemMetadataRecord {
                entity_id: 9,
                metadata_bytes: vec![0xAA; 16],
            },
            1,
            1,
            ShardId::SINGLE,
            RegionId::SINGLE,
        )
        .unwrap();
        bytes.extend(
            encode_batch(&[make_event(10, tid, 100), make_event(11, tid, 200)], 2, 1).unwrap(),
        );
        bytes.extend(
            encode_embedding_batch(
                &EmbeddingRecord {
                    entity_id: 9,
                    values: vec![0.5, 0.5],
                },
                4,
                1,
                ShardId::SINGLE,
                RegionId::SINGLE,
            )
            .unwrap(),
        );

        let applier = RecordingApplier::default();
        let payload = crate::replication::WalSegmentPayload {
            id: crate::replication::WalSegmentId::new(RegionId::SINGLE, ShardId::SINGLE, 1),
            bytes,
            event_count: 4,
            leader_last_seq: 4,
            stream_baseline: 0,
            term: 0,
            leader_region: 0,
        };
        apply_drained(vec![payload], &ledger, &state, None, Some(&applier)).unwrap();

        assert_eq!(
            state.applied_seqno(ShardId::SINGLE),
            Some(4),
            "the whole mixed range advances the frontier"
        );
        let applied = applier.applied.lock().unwrap().clone();
        assert_eq!(
            applied,
            vec!["item:9:16".to_string(), "emb:9:2".to_string()],
            "blobs apply in seqno order"
        );
        let count = |e: u64| {
            ledger
                .read_windowed_count(EntityId::new(e), "view", Window::AllTime)
                .unwrap()
        };
        assert_eq!(count(10), 1, "signal events folded");
        assert_eq!(count(11), 1);
    }

    /// m11p2: a blob batch arriving with NO applier wired halts the receiver
    /// loudly instead of silently dropping a replicated item.
    #[test]
    fn blob_payload_without_applier_halts() {
        use crate::replication::shard::RegionId;
        use crate::wal::format::batch::{ItemMetadataRecord, encode_item_metadata_batch};

        let schema = make_schema();
        let ledger = Arc::new(SignalLedger::new(schema, Box::new(NoopWalWriter)));
        let state = ReplicationState::new(&[ShardId::SINGLE]);
        let bytes = encode_item_metadata_batch(
            &ItemMetadataRecord {
                entity_id: 1,
                metadata_bytes: vec![1],
            },
            1,
            1,
            ShardId::SINGLE,
            RegionId::SINGLE,
        )
        .unwrap();
        let payload = crate::replication::WalSegmentPayload {
            id: crate::replication::WalSegmentId::new(RegionId::SINGLE, ShardId::SINGLE, 1),
            bytes,
            event_count: 1,
            leader_last_seq: 1,
            stream_baseline: 0,
            term: 0,
            leader_region: 0,
        };
        let err = apply_drained(vec![payload], &ledger, &state, None, None)
            .expect_err("a blob with no applier must halt");
        assert!(err.to_string().contains("no blob"), "got: {err}");
        assert_eq!(
            state.applied_seqno(ShardId::SINGLE),
            Some(0),
            "nothing recorded on the halt path"
        );
    }

    /// m11p2: a catch-up chunk announcing a stream baseline jumps the frontier
    /// past pre-stream history, so the chunk's own range applies contiguously
    /// instead of parking on a phantom gap.
    #[test]
    fn stream_baseline_jumps_frontier_past_prestream_history() {
        let schema = make_schema();
        let ledger = Arc::new(SignalLedger::new(schema, Box::new(NoopWalWriter)));
        let state = ReplicationState::new(&[ShardId::SINGLE]);
        let tid = ledger.resolve_signal_type("view").unwrap().as_u16() as u8;

        // A promoted leader's first chunk: range [11, 12] with baseline 10.
        let bytes =
            encode_batch(&[make_event(1, tid, 100), make_event(2, tid, 200)], 11, 1).unwrap();
        let payload = crate::replication::WalSegmentPayload {
            id: crate::replication::WalSegmentId::new(
                crate::replication::shard::RegionId::SINGLE,
                ShardId::SINGLE,
                11,
            ),
            bytes,
            event_count: 2,
            leader_last_seq: 12,
            stream_baseline: 10,
            term: 0,
            leader_region: 0,
        };
        apply_drained(vec![payload], &ledger, &state, None, None).unwrap();
        assert_eq!(
            state.applied_seqno(ShardId::SINGLE),
            Some(12),
            "baseline 10 + chunk [11,12] must land contiguously at 12"
        );
    }
}