frame-conv 0.2.0

//! Pure exchange-state and cursor tracking: request correlation states and
//! delivery-sequence gap detection. No I/O; unit-pinned in this file.

use std::collections::{BTreeSet, HashMap};

use crate::id::{ConversationSeq, CorrelationId};

/// Terminal state of one request exchange this handle issued.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum ExchangeState {
    /// Awaiting the first reply.
    Pending,
    /// Answered; further replies are duplicates.
    Replied,
    /// Deadline elapsed; a later reply is late.
    Elapsed,
}

/// Classification of one inbound reply against the exchange registry.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum ReplyClass {
    /// First reply to a pending exchange — the correlated answer.
    First,
    /// A second reply to an answered exchange.
    Duplicate,
    /// A reply to an exchange whose deadline already elapsed.
    Late,
    /// A correlation this handle never issued (another peer's exchange).
    Foreign,
}

/// Registry of exchanges this handle issued, keyed by correlation.
///
/// Completed exchanges are retained so late and duplicate replies stay
/// classifiable for the handle's lifetime — retention is bounded by the
/// number of requests the caller actually issued, and no cap is invented
/// here (a retention policy would be a deployment-owner decision).
#[derive(Debug, Default)]
pub(crate) struct ExchangeRegistry {
    states: HashMap<CorrelationId, ExchangeState>,
}

impl ExchangeRegistry {
    /// Opens a pending exchange.
    pub(crate) fn open(&mut self, correlation: CorrelationId) {
        self.states.insert(correlation, ExchangeState::Pending);
    }

    /// Classifies an inbound reply and advances the exchange state: the
    /// first reply to a pending exchange marks it replied.
    pub(crate) fn classify_reply(&mut self, correlation: CorrelationId) -> ReplyClass {
        match self.states.get(&correlation) {
            Some(ExchangeState::Pending) => {
                self.states.insert(correlation, ExchangeState::Replied);
                ReplyClass::First
            }
            Some(ExchangeState::Replied) => ReplyClass::Duplicate,
            Some(ExchangeState::Elapsed) => ReplyClass::Late,
            None => ReplyClass::Foreign,
        }
    }

    /// Marks a pending exchange elapsed (its deadline fired). A completed
    /// exchange is left untouched: exactly one terminal transition happens.
    pub(crate) fn close_elapsed(&mut self, correlation: CorrelationId) {
        if self.states.get(&correlation) == Some(&ExchangeState::Pending) {
            self.states.insert(correlation, ExchangeState::Elapsed);
        }
    }
}

/// A detected delivery-sequence discontinuity.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct SeqGap {
    /// The sequence position that was expected next.
    pub(crate) expected: ConversationSeq,
    /// The sequence position that was observed instead.
    pub(crate) observed: ConversationSeq,
}

/// The own-ahead ledger's named bound (tear-F1 fix). Entries exist only
/// while a FOREIGN delivery is genuinely in flight below an own committed
/// position — the transient held-fan-out window. A stall long enough to
/// accumulate this many own commits above the delivery frontier is
/// indistinguishable from delivery loss, and the honest surface for loss
/// is the typed gap (constraint 4), never unbounded memory.
pub(crate) const MAX_OWN_AHEAD: usize = 512;

/// Contiguity tracker over the conversation's single delivery sequence.
///
/// The substrate proved one ascending, contiguous per-conversation
/// sequence with the sender excluded from its own records (F-0c
/// assertion 5). TWO observation streams feed the tracker and they carry
/// NO cross-stream ordering guarantee (tear-F1, measured 2026-07-24):
/// fan-out deliveries (per-connection ordered, but the server's delivery
/// scheduler may HOLD a push across its fairness slices) and this
/// participant's own commit receipts (the request-reply path, never
/// held). An own receipt can therefore legitimately arrive while an
/// OLDER foreign delivery is still in flight; treating it as a delivery
/// frontier advance manufactures a false gap. The split: deliveries
/// advance the frontier ([`Self::observe`]); own receipts account their
/// position WITHOUT advancing past unobserved foreign positions
/// ([`Self::observe_own`]), parking above the frontier in a bounded
/// ledger the frontier drains through. The first observation of either
/// stream seeds the baseline (a joiner's sequence space begins at its
/// membership — the late-joiner finding); any forward skip of a
/// non-own position after that is a typed gap.
#[derive(Debug, Default)]
pub(crate) struct CursorTracker {
    next_expected: Option<u64>,
    /// Own committed positions strictly above the delivery frontier —
    /// excused from delivery (their receipt is their witness), drained
    /// as the frontier passes. Bounded by [`MAX_OWN_AHEAD`].
    own_ahead: BTreeSet<u64>,
}

impl CursorTracker {
    /// Observes one DELIVERED sequence position. Returns a gap when a
    /// non-own position was skipped; regressions and duplicates are
    /// ignored (the substrate replays the unacked window on resume, so
    /// re-observing an old position is not an anomaly).
    pub(crate) fn observe(&mut self, seq: ConversationSeq) -> Option<SeqGap> {
        let value = seq.value();
        match self.next_expected {
            None => {
                self.advance_past(value);
                None
            }
            Some(expected) if value == expected => {
                self.advance_past(value);
                None
            }
            Some(expected) if value > expected => {
                // Positions in (expected..value) that are our own are
                // accounted by their receipts, not owed as deliveries; a
                // skip of exclusively-own positions is not a hole.
                let hole = (expected..value).find(|position| !self.own_ahead.contains(position));
                self.advance_past(value);
                hole.map(|position| SeqGap {
                    expected: ConversationSeq::new(position),
                    observed: seq,
                })
            }
            Some(_) => None,
        }
    }

    /// Observes one of this participant's OWN committed positions (its
    /// receipt is the only witness — own records are never delivered).
    /// At the frontier it advances like a delivery; above the frontier
    /// it parks in the bounded own-ahead ledger without advancing past
    /// unobserved foreign positions. Returns a gap only when the ledger
    /// overflows [`MAX_OWN_AHEAD`] — a delivery stall indistinguishable
    /// from loss, surfaced loudly instead of remembered unboundedly.
    pub(crate) fn observe_own(&mut self, seq: ConversationSeq) -> Option<SeqGap> {
        let value = seq.value();
        match self.next_expected {
            None => {
                self.advance_past(value);
                None
            }
            Some(expected) if value == expected => {
                self.advance_past(value);
                None
            }
            Some(expected) if value > expected => {
                self.own_ahead.insert(value);
                if self.own_ahead.len() > MAX_OWN_AHEAD {
                    // The stalled foreign span converts to the typed gap
                    // surface; the frontier re-seeds past this position.
                    self.advance_past(value);
                    return Some(SeqGap {
                        expected: ConversationSeq::new(expected),
                        observed: seq,
                    });
                }
                None
            }
            Some(_) => None,
        }
    }

    /// Advances the frontier past `value`, draining own-ahead positions
    /// the new frontier reaches (and discarding any at or below it).
    fn advance_past(&mut self, value: u64) {
        let mut next = value.saturating_add(1);
        self.own_ahead = self.own_ahead.split_off(&next);
        while self.own_ahead.remove(&next) {
            next = next.saturating_add(1);
        }
        self.next_expected = Some(next);
    }

    /// Test-only view of the own-ahead ledger size (the bounded-by-
    /// construction witness).
    #[cfg(test)]
    pub(crate) fn own_ahead_len(&self) -> usize {
        self.own_ahead.len()
    }
}

#[cfg(test)]
mod tests {
    #![allow(clippy::expect_used)] // test code — the same latitude the tests/ tree declares

    use super::*;

    #[test]
    fn first_reply_is_first_then_duplicate() {
        let mut registry = ExchangeRegistry::default();
        let correlation = CorrelationId::mint();
        registry.open(correlation);
        assert_eq!(registry.classify_reply(correlation), ReplyClass::First);
        assert_eq!(registry.classify_reply(correlation), ReplyClass::Duplicate);
    }

    #[test]
    fn reply_after_elapse_is_late_and_stays_late() {
        let mut registry = ExchangeRegistry::default();
        let correlation = CorrelationId::mint();
        registry.open(correlation);
        registry.close_elapsed(correlation);
        assert_eq!(registry.classify_reply(correlation), ReplyClass::Late);
        assert_eq!(registry.classify_reply(correlation), ReplyClass::Late);
    }

    #[test]
    fn elapse_after_reply_does_not_reopen_the_exchange() {
        let mut registry = ExchangeRegistry::default();
        let correlation = CorrelationId::mint();
        registry.open(correlation);
        assert_eq!(registry.classify_reply(correlation), ReplyClass::First);
        registry.close_elapsed(correlation);
        assert_eq!(registry.classify_reply(correlation), ReplyClass::Duplicate);
    }

    #[test]
    fn unissued_correlation_is_foreign() {
        let mut registry = ExchangeRegistry::default();
        assert_eq!(
            registry.classify_reply(CorrelationId::mint()),
            ReplyClass::Foreign
        );
    }

    /// TEAR-F1 (red at 328ba5b, green here): the publisher's own commit
    /// receipt racing a HELD foreign delivery must not mint a gap. The
    /// exact broadcast red at the tear seat: the publisher is delivered
    /// a-attach (seq 2, frontier -> 3); the server's delivery scheduler
    /// holds b-attach (seq 3) past the publish answer (held pushes /
    /// fairness slices — no cross-stream order between commit answers
    /// and fan-out deliveries); the publisher's OWN commit receipt
    /// (seq 4) reaches the tracker first, now through the split
    /// `observe_own` seam. Nothing was lost — seq 3 arrives moments
    /// later — an own position is ACCOUNTED BY ITS RECEIPT (it is never
    /// delivered) and never advances the delivery frontier past an
    /// unobserved foreign position.
    #[test]
    fn own_receipt_racing_held_foreign_delivery_is_not_a_gap() {
        let mut tracker = CursorTracker::default();
        // The publisher's delivery stream: a-attach at seq 2.
        assert_eq!(tracker.observe(ConversationSeq::new(2)), None);
        // Its own commit receipt at seq 4, while foreign delivery 3 is
        // held by the server's delivery scheduler.
        assert_eq!(
            tracker.observe_own(ConversationSeq::new(4)),
            None,
            "an own receipt ahead of a held foreign delivery is not a hole"
        );
        // The held delivery arrives: contiguous, no gap; the frontier
        // drains THROUGH the own position (4 is never delivered — its
        // receipt already accounted for it) to expect 5 next.
        assert_eq!(tracker.observe(ConversationSeq::new(3)), None);
        assert_eq!(tracker.own_ahead_len(), 0);
        assert_eq!(tracker.observe(ConversationSeq::new(5)), None);
    }

    /// A sole publisher's contiguous own commits advance the frontier
    /// directly — the own-ahead ledger stays EMPTY (entries exist only
    /// while a foreign delivery is genuinely in flight below an own
    /// position).
    #[test]
    fn contiguous_own_commits_advance_without_ledger_growth() {
        let mut tracker = CursorTracker::default();
        assert_eq!(tracker.observe_own(ConversationSeq::new(2)), None);
        assert_eq!(tracker.observe_own(ConversationSeq::new(3)), None);
        assert_eq!(tracker.observe_own(ConversationSeq::new(4)), None);
        assert_eq!(tracker.own_ahead_len(), 0);
        // A real foreign skip after own commits still gaps.
        assert_eq!(
            tracker.observe(ConversationSeq::new(7)),
            Some(SeqGap {
                expected: ConversationSeq::new(5),
                observed: ConversationSeq::new(7),
            })
        );
    }

    /// A genuine foreign hole below an own position still gaps: the
    /// ledger excuses OWN positions only, never a missing foreign one —
    /// and the gap names the missing FOREIGN position, skipping over
    /// receipt-accounted own ones.
    #[test]
    fn true_foreign_hole_below_an_own_position_still_gaps() {
        let mut tracker = CursorTracker::default();
        assert_eq!(tracker.observe(ConversationSeq::new(2)), None);
        assert_eq!(tracker.observe_own(ConversationSeq::new(5)), None);
        // Delivery arrives at 6: position 3 and 4 are neither delivered
        // nor own — a true hole, named at its first missing position.
        assert_eq!(
            tracker.observe(ConversationSeq::new(6)),
            Some(SeqGap {
                expected: ConversationSeq::new(3),
                observed: ConversationSeq::new(6),
            })
        );
        // Frontier re-seeded past the far edge; the ledger drained.
        assert_eq!(tracker.own_ahead_len(), 0);
        assert_eq!(tracker.observe(ConversationSeq::new(7)), None);
    }

    /// A skip of EXCLUSIVELY own positions is not a hole: delivery
    /// resumes past receipt-accounted positions without a gap.
    #[test]
    fn skip_of_exclusively_own_positions_is_not_a_hole() {
        let mut tracker = CursorTracker::default();
        assert_eq!(tracker.observe(ConversationSeq::new(2)), None);
        assert_eq!(tracker.observe_own(ConversationSeq::new(3)), None);
        assert_eq!(tracker.observe_own(ConversationSeq::new(4)), None);
        // Both own positions were at the frontier — drained immediately;
        // the next delivery is contiguous.
        assert_eq!(tracker.observe(ConversationSeq::new(5)), None);

        // And when own positions park ABOVE a held foreign one, the
        // arriving foreign delivery drains through them.
        let mut parked = CursorTracker::default();
        assert_eq!(parked.observe(ConversationSeq::new(10)), None);
        assert_eq!(parked.observe_own(ConversationSeq::new(12)), None);
        assert_eq!(parked.observe_own(ConversationSeq::new(13)), None);
        assert_eq!(parked.own_ahead_len(), 2);
        assert_eq!(parked.observe(ConversationSeq::new(11)), None);
        assert_eq!(parked.own_ahead_len(), 0);
        assert_eq!(parked.observe(ConversationSeq::new(14)), None);
    }

    /// The own-ahead ledger is bounded by the named cap: a delivery
    /// stall long enough to accumulate `MAX_OWN_AHEAD` own positions
    /// above the frontier converts to the LOUD typed-gap surface
    /// (constraint 4's law) instead of unbounded memory.
    #[test]
    fn own_ahead_overflow_is_a_loud_gap_never_unbounded_memory() {
        let mut tracker = CursorTracker::default();
        assert_eq!(tracker.observe(ConversationSeq::new(1)), None);
        // Own commits leapfrog a stalled foreign position at seq 2.
        let mut fired = None;
        for offset in 0..=u64::try_from(MAX_OWN_AHEAD).expect("cap fits u64") {
            if let Some(gap) = tracker.observe_own(ConversationSeq::new(3 + offset)) {
                fired = Some((offset, gap));
                break;
            }
        }
        let (at, gap) = fired.expect("the cap must fire before unbounded growth");
        assert_eq!(
            gap.expected,
            ConversationSeq::new(2),
            "the overflow gap names the stalled foreign position"
        );
        assert_eq!(
            usize::try_from(at).expect("fits usize"),
            MAX_OWN_AHEAD,
            "the ledger held exactly the cap before converting to the typed gap"
        );
        assert!(tracker.own_ahead_len() <= MAX_OWN_AHEAD);
        // Post-overflow the frontier is re-seeded past the fired own
        // position and the stream continues.
        let next = 3 + at + 1;
        assert_eq!(tracker.observe(ConversationSeq::new(next)), None);
    }

    #[test]
    fn contiguous_sequences_raise_no_gap() {
        let mut tracker = CursorTracker::default();
        assert_eq!(tracker.observe(ConversationSeq::new(3)), None);
        assert_eq!(tracker.observe(ConversationSeq::new(4)), None);
        assert_eq!(tracker.observe(ConversationSeq::new(5)), None);
    }

    #[test]
    fn membership_forward_baseline_is_first_observation() {
        let mut tracker = CursorTracker::default();
        // A late joiner's first delivery may sit anywhere in the sequence;
        // the baseline seeds there, and pre-membership history is not a gap.
        assert_eq!(tracker.observe(ConversationSeq::new(41)), None);
        assert_eq!(tracker.observe(ConversationSeq::new(42)), None);
    }

    #[test]
    fn forward_skip_is_a_typed_gap() {
        let mut tracker = CursorTracker::default();
        assert_eq!(tracker.observe(ConversationSeq::new(1)), None);
        assert_eq!(
            tracker.observe(ConversationSeq::new(4)),
            Some(SeqGap {
                expected: ConversationSeq::new(2),
                observed: ConversationSeq::new(4),
            })
        );
        // The tracker re-seeds past the gap; the stream continues.
        assert_eq!(tracker.observe(ConversationSeq::new(5)), None);
    }

    #[test]
    fn replayed_positions_are_not_gaps() {
        let mut tracker = CursorTracker::default();
        assert_eq!(tracker.observe(ConversationSeq::new(6)), None);
        assert_eq!(tracker.observe(ConversationSeq::new(7)), None);
        // Resume replays the unacked window; an old position re-observed is
        // benign.
        assert_eq!(tracker.observe(ConversationSeq::new(7)), None);
    }
}