hopr_transport_session/supervision/

supervisor.rs

//! Deterministic [`SessionPixSupervisor`] — the pure state machine for PIX
//! session lifecycle.
//!
//! All methods take explicit [`std::time::Instant`] timestamps and a
//! `served_total: u64` sample from the [`ServiceGate`](super::gate::ServiceGate).
//! No method sleeps, spawns, or performs I/O.

use std::time::{Duration, Instant};

use hopr_api::{HoprBalance, OffchainPublicKey, types::internal::prelude::HoprPseudonym};
use hopr_protocol_pix::{SsaId, SsaIndex, SsaRecoveryProgress};

use super::{
    PixParams, SessionPixAction, SessionPixCloseReason, SessionPixEvent, SupervisorConfig,
    fill::{FillPlanner, FillTarget},
    telemetry::{PixCycleOutcome, PixCyclePhase, PixCycleSummary, PixGateMode, PixSessionSnapshot, PixTurnEvents},
};

// ---------------------------------------------------------------------------
// SsaPhase
// ---------------------------------------------------------------------------

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum SsaPhase {
    /// Request has been sent; waiting for complete verifiable commitment.
    AwaitingCommitment,
    /// Commitment is verifiable; waiting for a sufficient deposit.
    AwaitingDeposit,
    /// Deposit confirmed; recovering shares.
    Recovering,
    /// SSA fully recovered; diagnostic tombstone until `tombstone_until`.
    /// Any paid FIFO remainder lives separately in [`PaidRecoveryTail`].
    Recovered { tombstone_until: Instant },
    /// Phase that will produce a close action on next deadline check.
    Closing,
}

/// Wei in one µHOPR, the unit `hopr_pix_deposits_*_uhopr_total` counts.
const WEI_PER_UHOPR: u128 = 1_000_000_000_000;

/// A deposit in µHOPR, saturating rather than wrapping.
///
/// Integer because the counters it feeds are, and µHOPR because wei overflows a `u64` at 18 HOPR
/// while whole HOPR would round every realistic deposit to nothing.
fn deposit_uhopr(amount: HoprBalance) -> u64 {
    let wei = amount.amount();
    let wei = if wei.bits() > 128 { u128::MAX } else { wei.low_u128() };
    (wei / WEI_PER_UHOPR).min(u64::MAX as u128) as u64
}

// ---------------------------------------------------------------------------
// PerSsaState
// ---------------------------------------------------------------------------

/// Internal state for one supervised SSA.
struct PerSsaState {
    ssa_id: SsaId<HoprPseudonym>,
    /// First SSA index allocated in this request batch.
    batch_id: u32,
    phase: SsaPhase,

    // Deadlines (None means not set for this phase).
    commitment_deadline: Option<Instant>,
    /// When to re-request the parts of this commitment that have not arrived, or `None` while there
    /// is nothing to ask for.
    ///
    /// Set only once *some* of the commitment has arrived, and pushed out by every further part, so
    /// it measures silence in the delivery rather than elapsed time. A cycle for which nothing was
    /// ever delivered leaves it unset and dies on `commitment_deadline`, which is all the Exit can
    /// do about one: it has no scope to name.
    recommit_deadline: Option<Instant>,
    /// Re-requests already made for this cycle, capped at
    /// [`MAX_COMMITMENT_RETRANSMISSIONS`](hopr_protocol_pix::MAX_COMMITMENT_RETRANSMISSIONS).
    ///
    /// The cap is a count and not a duration on purpose. `commitment_deadline` bounds the asks in
    /// *time*, which is what protects the Entry, but the packets they cost are this node's own: at a
    /// misconfigured millisecond interval inside a 20-second deadline the time bound alone permits
    /// twenty thousand of them. Matching the figure the Entry will answer means none of the asks is
    /// wasted either.
    recommit_attempts: u8,
    deposit_deadline: Option<Instant>,
    recovery_idle_deadline: Option<Instant>,
    recovery_hard_deadline: Option<Instant>,

    /// What the Entry deposited for this cycle, in µHOPR, or zero until the deposit confirms.
    deposit_uhopr: u64,
    /// When [`phase`](Self::phase) was entered, for `hopr_pix_cycle_phase_seconds`.
    phase_entered_at: Instant,

    // Progress tracking.
    largest_useful_shares: u64,
    /// Largest `shares_seen` observed for this cycle — the liveness counter, which moves for surplus
    /// shares too. Also the staleness discriminator for snapshots, since it moves for every share
    /// while `largest_useful_shares` does not.
    largest_shares_seen: u64,
    target_useful_shares: u64,
    recovered_polynomials: u16,

    // Overlap / deferred-request state.
    next_request_pending_deposit: bool,
    next_requested: bool,
    /// Set on the last cycle allocated by [`SessionPixSupervisor::emit_request_next_ssa`], and only
    /// on that one.
    ///
    /// The successor gate: a batch asks for its replacement once, when its *last* cycle is nearly
    /// recovered. Every other flag here is per-cycle, so without this one each member of a batch
    /// would answer its own `AlmostRecovered` with a whole batch of its own — `ssas_per_request`
    /// batches per batch, compounding, each cycle a separate on-chain deposit for the Entry.
    ///
    /// The last cycle rather than the first because the batch is served in index order: gating on
    /// the first would ask for the successor a whole batch too early, which is the exposure
    /// `ssas_per_request` deliberately bounds.
    ///
    /// If the two-generation admission bound defers that request, `next_requested` remains set and
    /// the supervisor-level `successor_request_deferred` flag owns the retry; the gate must not be
    /// handed to another cycle and spend the request twice.
    is_batch_last: bool,
    /// Set when `Recovered` arrives before deposit confirms (i.e. during
    /// `AwaitingCommitment` or `AwaitingDeposit`). Once deposit is confirmed
    /// and the SSA enters `Recovering`, this flag triggers an immediate
    /// tombstone transition — replaying the deferred `Recovered` event.
    recovered_pending: bool,

    // Service gating.
    served_total_at_last_progress: u64,
    /// The Session's served count when this cycle reached the front of the batch, or `None` while
    /// it is still queued behind one.
    ///
    /// Subtracted from the served count at finalization to give the cycle's own egress. Baselined
    /// at the front rather than at the request, because a queued cycle is not being served: the
    /// Entry's emission window is clamped to one cycle, so charging a batch member for everything
    /// its predecessors consumed would report the last cycle of a batch as having cost the whole
    /// batch. `None` on a cycle that never reached the front is therefore the honest answer, and it
    /// is reported as zero egress rather than as the Session's whole history.
    served_total_at_front: Option<u64>,
}

impl PerSsaState {
    fn new(ssa_id: SsaId<HoprPseudonym>, batch_id: u32, target_useful_shares: u64, now: Instant) -> Self {
        Self {
            ssa_id,
            batch_id,
            phase: SsaPhase::AwaitingCommitment,
            phase_entered_at: now,
            commitment_deadline: None,
            recommit_deadline: None,
            recommit_attempts: 0,
            deposit_uhopr: 0,
            deposit_deadline: None,
            recovery_idle_deadline: None,
            recovery_hard_deadline: None,
            largest_useful_shares: 0,
            largest_shares_seen: 0,
            target_useful_shares,
            recovered_polynomials: 0,
            next_request_pending_deposit: false,
            next_requested: false,
            is_batch_last: false,
            recovered_pending: false,
            served_total_at_last_progress: 0,
            served_total_at_front: None,
        }
    }

    /// Moves to `phase`, returning how long the outgoing one lasted in milliseconds.
    ///
    /// `None` for an outgoing `Recovered` or `Closing`, which have no [`PixCyclePhase`] label — both
    /// are retained state rather than live state, so neither is measured.
    fn enter_phase(&mut self, phase: SsaPhase, now: Instant) -> Option<(PixCyclePhase, u64)> {
        let left = match self.phase {
            SsaPhase::AwaitingCommitment => Some(PixCyclePhase::AwaitingCommitment),
            SsaPhase::AwaitingDeposit => Some(PixCyclePhase::AwaitingDeposit),
            SsaPhase::Recovering => Some(PixCyclePhase::Recovering),
            SsaPhase::Recovered { .. } | SsaPhase::Closing => None,
        };
        let elapsed = now.saturating_duration_since(self.phase_entered_at).as_millis() as u64;
        self.phase = phase;
        self.phase_entered_at = now;
        left.map(|phase| (phase, elapsed))
    }

    /// This cycle's coverage, for the histograms observed once when it finalizes.
    fn telemetry_summary(&self, outcome: PixCycleOutcome, served_total: u64) -> PixCycleSummary {
        PixCycleSummary {
            outcome,
            // Saturating rather than assuming monotonicity: the baseline is taken from a
            // `served_total` sample passed in by the worker, and a cycle retired on a deadline
            // sweep can be summarized from an older sample than the one that baselined it.
            egress_packets: self
                .served_total_at_front
                .map(|base| served_total.saturating_sub(base))
                .unwrap_or(0),
            accepted_shares: self.largest_shares_seen,
            useful_shares: self.largest_useful_shares,
            target_useful_shares: self.target_useful_shares,
            deposit_uhopr: self.deposit_uhopr,
        }
    }

    /// True if this SSA is past the recovery phase.
    fn is_terminal(&self) -> bool {
        matches!(self.phase, SsaPhase::Recovered { .. } | SsaPhase::Closing)
    }
}

/// Bounded handoff for SURBs minted against the cycle that has just recovered.
///
/// Cryptographic recovery and transport-buffer exhaustion are different instants. The former frees
/// the reconstructor's heavy state; this small receipt keeps the paid service front on the recovered
/// predecessor until either its negotiated tail is exhausted or any progress proves the successor
/// has reached the head of the FIFO. The latter need not be funded: observing an unfunded successor
/// is precisely when the predecessor's receipt must stop authorizing service.
struct PaidRecoveryTail {
    ssa_id: SsaId<HoprPseudonym>,
    /// When the tail took the paid front, for `hopr_pix_cycle_phase_seconds{phase="paid_tail"}`.
    entered_at: Instant,
    largest_shares_seen: u64,
    served_total_at_last_progress: u64,
    idle_deadline: Option<Instant>,
    hard_deadline: Option<Instant>,
}

// ---------------------------------------------------------------------------
// SessionPixSupervisor
// ---------------------------------------------------------------------------

/// Deterministic core of the PIX session supervisor.
pub struct SessionPixSupervisor {
    pub(crate) cfg: SupervisorConfig,
    pub(crate) dims: PixParams,
    pub(crate) pseudonym: HoprPseudonym,
    pub(crate) closed: bool,
    /// The Session is gone and only its funded front cycle is still being worked for.
    ///
    /// Set by [`on_session_closed`](Self::on_session_closed) and never cleared: a drain ends by
    /// closing, one way or another. While it is set the supervisor asks for no successor, plans fill
    /// at the ceiling, and stops service-gating the recovery idle deadline — see "Draining a closed
    /// Session" in this module's documentation for why each of those follows from the Session having
    /// left rather than being separate policy.
    pub(crate) draining: bool,
    next_ssa_index: u32,
    /// The cycle the two share-order counters below are measured against — the earliest unrecovered
    /// cycle, i.e. the one the Entry should currently be serving. `None` before the first batch exists.
    share_order_front: Option<SsaIndex>,
    /// Useful shares booked on [`Self::share_order_front`] since it took the front.
    front_useful: u64,
    /// Useful shares booked on any *other* live cycle over that same span.
    ///
    /// `off_front_useful / (front_useful + off_front_useful)` is the share-order ratio — see
    /// [`SupervisorConfig::max_off_front_share_fraction`] for what it detects and why it is a fraction
    /// rather than a count. Both counters reset when the front changes, which is what keeps the window
    /// bounded to a single cycle's service.
    off_front_useful: u64,
    /// Whether the service gate currently follows a funded front cycle.
    service_open: bool,
    /// Front cycle for which the current gate mode was emitted.
    ///
    /// Tracking the identity as well as the mode lets a funded-to-funded handoff rebaseline the
    /// served-without-progress ceiling for the newly promoted cycle.
    service_front: Option<SsaIndex>,
    /// A front cycle became paid and terminal in one event, so an unfunded successor still earns a
    /// freshly restored predeposit allowance even though the gate never entered funded mode.
    paid_front_handoff: bool,
    /// The sole recovered predecessor authorized to report buffered liveness.
    ///
    /// This never becomes a set: accepting arbitrary recovered tombstones would let the Entry replay
    /// old cycle tags to keep service open. Replacement on each in-order recovery is the supervisor's
    /// independent bound behind the reconstructor's per-polynomial surplus counters.
    paid_recovery_tail: Option<PaidRecoveryTail>,
    /// A successor batch was earned but could not yet fit the two-generation reservation.
    successor_request_deferred: bool,
    /// Ordered SSAs (oldest first, newest last), including short-lived recovered tombstones.
    ssas: Vec<PerSsaState>,
    /// Tracks the first failure reason when multiple SSAs fail, so the
    /// earliest cause is used for the final `Close` action rather than the last.
    first_failure_reason: Option<SessionPixCloseReason>,
    /// Cycles this Session has lost without recovering them, over its whole life.
    ///
    /// Cumulative, and that is the whole point: retiring a failed member leaves no trace on its
    /// siblings, so without a counter that survives the retirement an Entry can lose one cycle per
    /// batch indefinitely while a single funded sibling holds the Session open. Bounded by
    /// [`SupervisorConfig::max_failed_cycles`].
    failed_cycles: usize,
    /// Plans the rate at which this Exit originates its own fill keep-alives.
    ///
    /// Kept here rather than in the worker because the rate is a function of the lifecycle state
    /// this struct owns — which cycle is at the paid front and how far it has got — and because
    /// every close path has to be able to stop it in the same action vector that carries the close.
    fill: FillPlanner,
    /// Greatest SSA index that has been retired (closed and removed).
    ///
    /// Prevents a stale `SsaRequestSent` from resurrecting a closed SSA. A high-watermark rather
    /// than the set of every retired index: the set grew for the whole life of the Session and was
    /// scanned linearly on every request, one entry per cycle, without bound.
    ///
    /// A watermark alone would be wrong if it were consulted first — retirement is not monotone,
    /// since a later batch member can lose its deposit while an earlier one is still recovering, and
    /// "at or below the greatest retired index" would then condemn that live earlier cycle. The
    /// guard is therefore placed *after* the live-record lookup, so a cycle that still exists always
    /// wins and the watermark only ever answers for indices that have none.
    highest_retired_ssa_index: Option<SsaIndex>,
    /// Cycle lifecycle transitions since the worker last drained this.
    ///
    /// Latched rather than emitted because this type is a pure function of its inputs and a metric
    /// is not — the same arrangement as [`FillPlanner`]'s stall, which is latched here and drained
    /// in the worker. Always compiled, including without the `telemetry` feature, so the default
    /// test configuration exercises the counting rather than only the feature-enabled one.
    telemetry: PixTurnEvents,
}

impl SessionPixSupervisor {
    /// Create a new supervisor and emit the first `RequestSsa` action.
    pub fn new(
        cfg: SupervisorConfig,
        dims: PixParams,
        pseudonym: HoprPseudonym,
        now: Instant,
    ) -> (Self, Vec<SessionPixAction>) {
        let mut s = Self {
            fill: FillPlanner::new(&cfg, &dims, now),
            cfg,
            dims,
            pseudonym,
            next_ssa_index: 1,
            share_order_front: None,
            front_useful: 0,
            off_front_useful: 0,
            closed: false,
            draining: false,
            service_open: false,
            service_front: None,
            paid_front_handoff: false,
            paid_recovery_tail: None,
            successor_request_deferred: false,
            ssas: Vec::with_capacity(2),
            first_failure_reason: None,
            failed_cycles: 0,
            highest_retired_ssa_index: None,
            telemetry: PixTurnEvents::default(),
        };

        let actions = s.emit_request_next_ssa(now);
        s.refresh_share_order_front();
        (s, actions)
    }

    /// Handle a lifecycle event.
    pub fn handle_event(&mut self, ev: &SessionPixEvent, now: Instant, served_total: u64) -> Vec<SessionPixAction> {
        if self.closed {
            return Vec::new();
        }

        let actions = match ev {
            SessionPixEvent::SsaRequestSent(ssa_id) => self.on_ssa_request_sent(ssa_id, now),
            SessionPixEvent::CommitmentProgress(ssa_id) => self.on_commitment_progress(ssa_id, now),
            SessionPixEvent::CommitmentVerified(ssa_id) => self.on_commitment_verified(ssa_id, now),
            SessionPixEvent::DepositConfirmed { ssa_id, amount } => {
                self.on_deposit_confirmed(ssa_id, *amount, now, served_total)
            }
            SessionPixEvent::DepositObserverClosed(ssa_id) => {
                self.on_deposit_observer_closed(ssa_id, now, served_total)
            }
            SessionPixEvent::RecoveryProgress(progress) => self.on_recovery_progress(progress, now, served_total),
            SessionPixEvent::AlmostRecovered(ssa_id) => self.on_almost_recovered(ssa_id, now),
            SessionPixEvent::Recovered(ssa_id) => self.on_recovered(ssa_id, now, served_total),
            SessionPixEvent::UnverifiableShares {
                ssa_id,
                observed_total,
                peer,
            } => self.on_unverifiable_shares(ssa_id, *observed_total, peer, now),
            SessionPixEvent::SessionClosed { drainable_surbs } => self.on_session_closed(*drainable_surbs, now),
        };

        let mut lifecycle_actions = actions;
        lifecycle_actions.extend(self.retry_deferred_successor_request(now));
        self.consume_paid_tail_at_observed_boundary(now);
        self.arm_recovery_clocks_for_earliest(now, served_total);
        let mut actions = if lifecycle_actions
            .iter()
            .any(|action| matches!(action, SessionPixAction::Close(_)))
        {
            // A closing Session has no cycle left to fill for, and the fill stream outlives this
            // state machine by however long it takes the driver to tear the Session down. Silencing
            // it here rather than only there is what keeps a close from being followed by a burst of
            // keep-alives for a cycle that no longer exists.
            self.fill
                .stop()
                .map(SessionPixAction::SetFillRate)
                .into_iter()
                .collect()
        } else {
            self.sync_service_gate()
        };
        actions.extend(lifecycle_actions);
        self.refresh_share_order_front();
        actions
    }

    /// Raises the retired-index watermark.
    ///
    /// Takes the maximum rather than assuming the caller retires in order, because it does not: a
    /// later batch member losing its deposit is retired before an earlier one that is still
    /// recovering.
    fn note_retired_index(&mut self, index: SsaIndex) {
        self.highest_retired_ssa_index = Some(match self.highest_retired_ssa_index {
            Some(highest) if highest >= index => highest,
            _ => index,
        });
    }

    /// Index of the earliest cycle that has not finished recovering — the one the Entry should be
    /// serving, given that emission is clamped to a single cycle.
    fn earliest_live_idx(&self) -> Option<usize> {
        self.ssas
            .iter()
            .enumerate()
            .filter(|(_, s)| !s.is_terminal())
            .min_by_key(|(_, s)| s.ssa_id.ssa_index())
            .map(|(i, _)| i)
    }

    /// Consumes the predecessor receipt once the current live front has reported any share.
    ///
    /// This check is centralized rather than tied only to `RecoveryProgress`: a previously queued
    /// cycle can become the front because a sibling is retired or because its deposit arrives after
    /// its first snapshot. In every case the observation proves the transport crossed the cycle
    /// boundary. Keeping the old funded receipt beyond it would let an unfunded successor spend the
    /// predecessor's remaining allowance.
    /// Retires the paid tail, latching how long it held the front.
    fn release_paid_tail(&mut self, now: Instant) {
        if let Some(tail) = self.paid_recovery_tail.take() {
            let elapsed = now.saturating_duration_since(tail.entered_at).as_millis() as u64;
            self.telemetry.phase_durations.push((PixCyclePhase::PaidTail, elapsed));
        }
    }

    fn consume_paid_tail_at_observed_boundary(&mut self, now: Instant) {
        let Some(front_idx) = self.earliest_live_idx() else {
            return;
        };
        let front_id = self.ssas[front_idx].ssa_id;
        if self.ssas[front_idx].largest_shares_seen == 0 {
            return;
        }

        if let Some(predecessor) = self
            .paid_recovery_tail
            .as_ref()
            .map(|tail| tail.ssa_id)
            .filter(|predecessor| predecessor.ssa_index() < front_id.ssa_index())
        {
            tracing::debug!(
                %predecessor,
                successor = %front_id,
                "successor progress consumed the paid FIFO-tail handoff"
            );
            self.release_paid_tail(now);
        }
    }

    /// Keeps the service gate aligned with the funding state of the earliest live cycle.
    ///
    /// A funded-to-unfunded handoff restores the configured predeposit allowance. A
    /// funded-to-funded handoff emits `ReleaseService` too: the mode does not change, but the new
    /// front must receive its own served-without-progress ceiling rather than inheriting the tail of
    /// its predecessor's. An unfunded cycle that fails does not earn another allowance.
    fn sync_service_gate(&mut self) -> Vec<SessionPixAction> {
        if self.closed {
            return Vec::new();
        }

        let front = self.earliest_live_idx();
        let front_id = self
            .paid_recovery_tail
            .as_ref()
            .map(|tail| tail.ssa_id.ssa_index())
            .or_else(|| front.map(|idx| self.ssas[idx].ssa_id.ssa_index()));
        let front_changed = front_id != self.service_front;
        let front_funded =
            self.paid_recovery_tail.is_some() || front.is_some_and(|idx| self.ssas[idx].phase == SsaPhase::Recovering);
        let paid_front_handoff = std::mem::take(&mut self.paid_front_handoff);
        self.service_front = front_id;

        match (front_funded, self.service_open, front_changed) {
            (true, false, _) | (true, true, true) => {
                self.service_open = true;
                tracing::debug!(?front_id, "aligning service gate with funded front cycle");
                vec![SessionPixAction::ReleaseService]
            }
            (false, true, _) => {
                self.service_open = false;
                tracing::debug!(?front_id, "withholding service for unfunded front cycle");
                vec![SessionPixAction::WithholdService]
            }
            (false, false, _) if paid_front_handoff => {
                tracing::debug!(?front_id, "restoring predeposit service after paid front handoff");
                vec![SessionPixAction::WithholdService]
            }
            _ => Vec::new(),
        }
    }

    /// This Session's current contribution to the node-level live-set gauges.
    ///
    /// Recomputed from `ssas` and `paid_recovery_tail` rather than maintained incrementally, which
    /// is what makes the gauges drain on every terminal path without any of those paths having to
    /// know they exist. A Session holds at most `ssas_per_request × MAX_OVERLAPPING_BATCHES` cycles,
    /// so the scan is over a handful of entries once per turn.
    ///
    /// `predeposit_exposure` comes from the gate, which the caller reads — this type deliberately
    /// knows nothing about the gate's counters.
    pub fn telemetry_snapshot(&self, predeposit_exposure: u64) -> PixSessionSnapshot {
        let mut snapshot = PixSessionSnapshot {
            // The mode the gate is actually in, which is what an operator reading this gauge next to
            // `hopr_pix_predeposit_exposure_packets` is asking about. A paid recovered predecessor
            // still holding the front counts as funded, because service continues to be authorized.
            gate_mode: if self.service_open {
                PixGateMode::Funded
            } else {
                PixGateMode::Predeposit
            },
            paid_tail: u32::from(self.paid_recovery_tail.is_some()),
            predeposit_exposure_packets: predeposit_exposure,
            stalled_share: u32::from(self.fill.is_stalled()),
            ..Default::default()
        };

        for ssa in self.ssas.iter() {
            // Recovered tombstones and `Closing` records are retained diagnostics, not live cycles.
            // Counting them would leave `hopr_pix_cycles_active` unable to drain, since a tombstone
            // outlives the cycle by `tombstone_retention_window`.
            let bucket = match ssa.phase {
                SsaPhase::AwaitingCommitment => &mut snapshot.awaiting_commitment,
                SsaPhase::AwaitingDeposit => &mut snapshot.awaiting_deposit,
                SsaPhase::Recovering => &mut snapshot.recovering,
                SsaPhase::Recovered { .. } | SsaPhase::Closing => continue,
            };
            *bucket = bucket.saturating_add(1);
        }

        snapshot
    }

    /// Drains the cycle lifecycle transitions latched since the last call.
    pub fn take_telemetry_events(&mut self) -> PixTurnEvents {
        std::mem::take(&mut self.telemetry)
    }

    fn live_cycle_count(&self) -> usize {
        self.ssas.iter().filter(|ssa| !ssa.is_terminal()).count()
    }

    fn live_batch_count(&self) -> usize {
        let mut batches = Vec::with_capacity(crate::MAX_OVERLAPPING_BATCHES as usize);
        for ssa in self.ssas.iter().filter(|ssa| !ssa.is_terminal()) {
            if !batches.contains(&ssa.batch_id) {
                batches.push(ssa.batch_id);
            }
        }
        batches.len()
    }

    fn reserved_cycle_slots(&self) -> usize {
        self.cfg
            .ssas_per_request
            .clamp(1, crate::MAX_SSA_BATCH_SIZE)
            .saturating_mul(crate::MAX_OVERLAPPING_BATCHES as usize)
    }

    fn retry_deferred_successor_request(&mut self, now: Instant) -> Vec<SessionPixAction> {
        if self.closed || !self.successor_request_deferred {
            return Vec::new();
        }

        self.successor_request_deferred = false;
        self.emit_request_next_ssa(now)
    }

    /// Re-points the share-order accounting at the current front of the batch, clearing it if the front
    /// moved.
    ///
    /// Clearing on a change of front is what bounds the measurement window to one cycle's service, and
    /// it is why an Entry cannot launder a spree by eventually finishing a cycle: the reset is only
    /// reached by the front cycle *completing*, which is the thing the Exit wanted all along.
    ///
    /// Called after every event and every deadline sweep, so it covers both ways the front can move —
    /// the front cycle recovering, or being retired.
    fn refresh_share_order_front(&mut self) {
        let front = self.earliest_live_idx().map(|i| self.ssas[i].ssa_id.ssa_index());
        if front != self.share_order_front {
            self.share_order_front = front;
            self.front_useful = 0;
            self.off_front_useful = 0;
        }
    }

    /// Starts the recovery clocks of the earliest unrecovered cycle, if they are not running yet.
    ///
    /// A batch's cycles are served strictly in index order — the Entry's emission window is clamped to
    /// one cycle (see [`hopr_protocol_pix::SHARE_EMISSION_WINDOW`]) — so a cycle behind the front of
    /// the batch is *queued*, not stalled. Starting its clocks when its deposit confirmed, as this used
    /// to, measured the queue wait rather than the recovery: at deployed dimensions one cycle takes
    /// ~61 min of emission to exhaust, so every cycle after the first was retired by
    /// `max_recovery_time` — or, worse, by `max_recovery_idle` inside a minute, since the idle gate
    /// compares against *session-wide* service and a queued cycle sees plenty of that without any of it
    /// being its own.
    ///
    /// So the clock starts when the cycle reaches the front, which is when it can actually make
    /// progress, and [`SupervisorConfig::max_recovery_time`] then means what its name says: the ceiling
    /// on recovering a single cycle. `served_total_at_last_progress` is re-baselined at the same moment
    /// for the same reason — otherwise the idle gate would charge the queue wait against it.
    ///
    /// Called after every event and every deadline sweep, so it covers each way a cycle can reach the
    /// front: its predecessor recovering, or its predecessor being retired.
    fn arm_recovery_clocks_for_earliest(&mut self, now: Instant, served_total: u64) {
        if self.closed {
            return;
        }
        // The packets being served still carry the recovered predecessor's shares. Starting the
        // successor now would charge that pipeline delay against both its idle and hard recovery
        // budgets even though no successor share can possibly arrive yet.
        if self.paid_recovery_tail.is_some() {
            return;
        }
        let Some(idx) = self.earliest_live_idx() else {
            return;
        };

        let ssa = &mut self.ssas[idx];
        if ssa.phase != SsaPhase::Recovering || ssa.recovery_hard_deadline.is_some() {
            return;
        }

        ssa.recovery_idle_deadline = now.checked_add(self.cfg.max_recovery_idle);
        ssa.recovery_hard_deadline = now.checked_add(self.cfg.max_recovery_time);
        ssa.served_total_at_last_progress = served_total;
        // The same instant, for the same reason: this is when the cycle starts being served, so it
        // is when its own egress begins. `get_or_insert` rather than a plain assignment because a
        // cycle reaches the front once — the guard above returns early for one that already armed —
        // but stating it here means a future caller cannot rebaseline a cycle mid-flight and make
        // its recorded egress smaller than what it actually consumed.
        ssa.served_total_at_front.get_or_insert(served_total);
        tracing::debug!(ssa_id = %ssa.ssa_id, "recovery clocks started — cycle is at the front of the batch");
    }

    /// Check all deadlines and emit actions for any that have expired.
    pub fn handle_deadline(&mut self, now: Instant, served_total: u64) -> Vec<SessionPixAction> {
        if self.closed {
            return Vec::new();
        }

        let mut actions = Vec::new();
        let max_recovery_idle = self.cfg.max_recovery_idle;

        // While the recovered predecessor still owns the service front, its original recovery
        // clocks remain the resource bound. Progress refreshes only the idle clock; the hard clock
        // is never extended. Thus a large but paid surplus may drain, while an Entry that declares
        // one and then feeds neither tail nor successor cannot hold the Session forever.
        if let Some(tail) = self.paid_recovery_tail.as_mut() {
            let expired = if tail.hard_deadline.is_some_and(|deadline| now >= deadline) {
                Some(SessionPixCloseReason::RecoveryDeadline)
            } else if tail.idle_deadline.is_some_and(|deadline| now >= deadline) {
                Some(SessionPixCloseReason::RecoveryIdle)
            } else {
                None
            };

            if let Some(reason) = expired {
                if reason == SessionPixCloseReason::RecoveryIdle && served_total <= tail.served_total_at_last_progress {
                    tail.idle_deadline = now.checked_add(max_recovery_idle);
                } else {
                    tracing::warn!(
                        ssa_id = %tail.ssa_id,
                        ?reason,
                        largest_shares_seen = tail.largest_shares_seen,
                        served_total_at_last_progress = tail.served_total_at_last_progress,
                        "closing PIX Session while its paid recovery tail is stalled"
                    );
                    // The tail is ending here, not merely being abandoned with the Session, so it
                    // owes its duration like every other phase that reaches a terminal edge.
                    self.release_paid_tail(now);
                    self.closed = true;
                    return self.with_fill_stopped(vec![SessionPixAction::Close(reason)]);
                }
            }
        }

        let mut i = 0;
        while i < self.ssas.len() {
            if self.closed {
                break;
            }

            let expired = {
                let ssa = &self.ssas[i];
                if ssa.is_terminal() {
                    i += 1;
                    continue;
                }
                ssa.check_deadlines(now)
            };

            // A stalled commitment is re-requested rather than given up on, and the timer re-arms so
            // a repair that is itself lost is asked for again. This yields no close reason and is
            // bounded by nothing of its own: the absolute `commitment_deadline` still ends the cycle,
            // which is why it can repeat freely — and why it is skipped once that deadline has gone,
            // rather than putting an ask on the wire for a Session about to close.
            if expired.is_none()
                && self.ssas[i].phase == SsaPhase::AwaitingCommitment
                && self.ssas[i].recommit_deadline.is_some_and(|deadline| now >= deadline)
            {
                let ssa = &mut self.ssas[i];
                ssa.recommit_attempts += 1;
                // Disarmed rather than re-armed at the cap, so the worker stops waking for a timer
                // that would produce nothing. Further arriving parts cannot re-arm it either.
                ssa.recommit_deadline = (ssa.recommit_attempts < hopr_protocol_pix::MAX_COMMITMENT_RETRANSMISSIONS)
                    .then(|| now.checked_add(self.cfg.commitment_recommit_interval))
                    .flatten();
                tracing::debug!(
                    ssa_id = %ssa.ssa_id, attempt = ssa.recommit_attempts,
                    "commitment delivery stalled — re-requesting the parts that did not arrive"
                );
                actions.push(SessionPixAction::RequestCommitmentRetransmission {
                    ssa_id: ssa.ssa_id,
                    params: self.dims,
                });
            }

            if let Some(reason) = expired {
                // Service-gated idle: if no service consumed since last progress,
                // re-arm instead of closing.
                //
                // Not while draining, though. The re-arm asks "was any *gated* service consumed",
                // and a drained Session has no application left to consume any — `served_total` is
                // frozen for good, so the re-arm would fire for ever and the one bound on a drain
                // admitted against an over-estimated buffer would be gone with it. A *productive*
                // drain is unaffected: its own progress still refreshes `recovery_idle_deadline`
                // through `on_recovery_progress`.
                if reason == SessionPixCloseReason::RecoveryIdle
                    && !self.draining
                    && served_total <= self.ssas[i].served_total_at_last_progress
                {
                    self.ssas[i].recovery_idle_deadline = now.checked_add(max_recovery_idle);
                    i += 1;
                    continue;
                }

                actions.extend(self.close_ssa_and_collect(i, reason, served_total, now));
                continue;
            }
            i += 1;
        }

        // If the session is already closing, skip tombstone retirement.
        // Whole-session teardown retires everything via the retirement guard.
        if self.closed {
            return self.with_fill_stopped(actions);
        }

        // Remove tombstones that have expired and emit RetireSsa so the
        // reconstructor and observer state is released mid-session.
        let retired_ids: Vec<SsaId<_>> = self
            .ssas
            .iter()
            .filter(|ssa| matches!(ssa.phase, SsaPhase::Recovered { tombstone_until } if now >= tombstone_until))
            .map(|ssa| ssa.ssa_id)
            .collect();
        self.ssas
            .retain(|ssa| !matches!(ssa.phase, SsaPhase::Recovered { tombstone_until } if now >= tombstone_until));
        for id in retired_ids {
            self.note_retired_index(id.ssa_index());
            actions.push(SessionPixAction::RetireSsa(id));
        }

        // A terminal cycle may have released the older of two live batches. Keep retirement before
        // the replacement request in the action stream so the reconstructor drops the old guard
        // before allocating the successor.
        actions.extend(self.retry_deferred_successor_request(now));
        self.consume_paid_tail_at_observed_boundary(now);

        // If no SSAs remain, close.
        // Note: a successor request may still be in flight here (RequestSsa
        // emitted, SsaRequestSent not yet observed) — intentionally not
        // tracked. The tombstone retention window is the backstop; a request
        // delayed beyond it (>30 s transport stall on an otherwise healthy
        // session) is theoretical and does not warrant blocking normal
        // tombstone expiry.
        if self.ssas.is_empty() && !self.closed {
            actions.push(SessionPixAction::Close(
                self.first_failure_reason
                    .unwrap_or(SessionPixCloseReason::NoSsaRemaining),
            ));
            self.closed = true;
        }

        // Retiring or tombstoning a cycle can promote the next one to the front of the batch — and if
        // that one is already funded, promotion is the only moment left at which service can be
        // released, since its deposit event is long past.
        self.arm_recovery_clocks_for_earliest(now, served_total);
        let mut gate_actions = self.sync_service_gate();
        gate_actions.extend(actions);
        self.refresh_share_order_front();

        if self.closed {
            return self.with_fill_stopped(gate_actions);
        }
        gate_actions
    }

    /// Runs the deadline sweep and, if one is due, the fill tick — in that order.
    ///
    /// The order is the point. The sweep is what arms a promoted cycle's recovery clocks, consumes a
    /// spent FIFO tail and re-aligns the service gate, so running it first means the tick plans
    /// against the lifecycle as it is *now* rather than as it was a second ago. A tick that ran first
    /// would spend one whole sampling interval filling for a cycle that had already recovered.
    ///
    /// This is the entry point the worker uses; [`handle_deadline`](Self::handle_deadline) remains
    /// separate because the deadline logic is meaningful — and tested — on its own.
    pub fn handle_timers(&mut self, now: Instant, served_total: u64) -> Vec<SessionPixAction> {
        let mut actions = self.handle_deadline(now, served_total);
        if self.closed {
            return actions;
        }
        if self.next_fill_tick().is_some_and(|tick| now >= tick) {
            actions.extend(self.handle_fill_tick(now, served_total));
        }
        actions
    }

    /// Re-plans the fill rate, emitting an action only when the rate actually changes.
    pub fn handle_fill_tick(&mut self, now: Instant, served_total: u64) -> Vec<SessionPixAction> {
        if self.closed {
            return Vec::new();
        }
        let target = self.fill_target();
        self.fill
            .plan(now, served_total, target)
            .map(|rate| vec![SessionPixAction::SetFillRate(rate)])
            .into_iter()
            .flatten()
            .collect()
    }

    /// Reports, once, that fill has fallen back because its target stopped progressing.
    ///
    /// Exposed for the worker, which is the layer that may have side effects; this state machine and
    /// the planner behind it are pure, so neither can record the metric itself.
    pub fn take_fill_stall(&mut self) -> bool {
        self.fill.take_stall_onset()
    }

    /// When the fill planner next wants to run, or `None` if there is nothing for it to do.
    ///
    /// `None` while no cycle is being filled for *and* the stream is already silent, which is what
    /// keeps an unfunded or non-PIX-idle Session from waking its worker once a second for nothing. A
    /// running stream always yields a tick even without a target, because it is owed the zero that
    /// stops it.
    pub fn next_fill_tick(&self) -> Option<Instant> {
        if self.closed || !self.fill.is_enabled() {
            return None;
        }
        if self.fill_target().is_none() && self.fill.is_idle() {
            return None;
        }
        Some(self.fill.next_tick())
    }

    /// The cycle the Exit should currently be filling for, if any.
    ///
    /// The recovered predecessor's FIFO tail wins whenever it exists, for two reasons that point the
    /// same way: its SURBs sit ahead of the successor's in the Exit's own buffer, so packets sent now
    /// carry *its* shares; and it holds the binding hard deadline, since the successor's clocks are
    /// deliberately not armed until the tail is exhausted (see
    /// [`arm_recovery_clocks_for_earliest`](Self::arm_recovery_clocks_for_earliest)).
    ///
    /// Otherwise it is the earliest live cycle, and only once it is both funded and clocked. An
    /// unfunded cycle has nothing at stake to be filled for, and a funded one whose clocks have not
    /// armed is queued behind a predecessor rather than being served — filling for it would spend
    /// SURBs on shares the Entry cannot yet be emitting.
    fn fill_target(&self) -> Option<FillTarget> {
        if let Some(tail) = self.paid_recovery_tail.as_ref() {
            return tail.hard_deadline.map(|hard_deadline| FillTarget {
                ssa_id: tail.ssa_id,
                largest_shares_seen: tail.largest_shares_seen,
                hard_deadline,
                drain: self.draining,
            });
        }

        let ssa = &self.ssas[self.earliest_live_idx()?];
        (ssa.phase == SsaPhase::Recovering)
            .then_some(ssa)
            .and_then(|ssa| ssa.recovery_hard_deadline.map(|hard_deadline| (ssa, hard_deadline)))
            .map(|(ssa, hard_deadline)| FillTarget {
                ssa_id: ssa.ssa_id,
                largest_shares_seen: ssa.largest_shares_seen,
                hard_deadline,
                drain: self.draining,
            })
    }

    /// Prepends the action that silences the fill stream, when there is one to silence.
    ///
    /// Returns `actions` untouched for a Session that never filled — a disabled one, or one that
    /// closed before its first cycle funded — so the action stream of every non-filling Session is
    /// exactly what it was before this existed.
    fn with_fill_stopped(&mut self, actions: Vec<SessionPixAction>) -> Vec<SessionPixAction> {
        match self.fill.stop() {
            Some(rate) => {
                let mut stopped = Vec::with_capacity(actions.len() + 1);
                stopped.push(SessionPixAction::SetFillRate(rate));
                stopped.extend(actions);
                stopped
            }
            None => actions,
        }
    }

    /// Returns the earliest deadline across all live SSAs, or `None`.
    pub fn next_deadline(&self) -> Option<Instant> {
        if self.closed || (self.ssas.is_empty() && self.paid_recovery_tail.is_none()) {
            return None;
        }

        let ssa_deadline = self
            .ssas
            .iter()
            .filter_map(|ssa| {
                if ssa.is_terminal() {
                    if let SsaPhase::Recovered { tombstone_until } = ssa.phase {
                        return Some(tombstone_until);
                    }
                    return None;
                }
                // Return the earliest set deadline. The re-request timer belongs here like the rest:
                // it is shorter than every other deadline of the phase it runs in, so leaving it out
                // would mean the worker slept past it and only re-requested when something else woke
                // it up.
                ssa.commitment_deadline
                    .into_iter()
                    .chain(ssa.recommit_deadline)
                    .chain(ssa.deposit_deadline)
                    .chain(ssa.recovery_idle_deadline)
                    .chain(ssa.recovery_hard_deadline)
                    .min()
            })
            .min();
        let tail_deadline = self
            .paid_recovery_tail
            .as_ref()
            .and_then(|tail| tail.idle_deadline.into_iter().chain(tail.hard_deadline).min());

        ssa_deadline.into_iter().chain(tail_deadline).min()
    }

    /// Feed back the result of executing an action.
    pub fn action_result(&mut self, action: &SessionPixAction, ok: bool, _now: Instant) -> Vec<SessionPixAction> {
        if self.closed {
            return Vec::new();
        }

        match action {
            SessionPixAction::RequestSsa { .. } if !ok => {
                self.closed = true;
                self.with_fill_stopped(vec![SessionPixAction::Close(
                    SessionPixCloseReason::SupervisorUnavailable,
                )])
            }
            SessionPixAction::Close(_) => {
                self.closed = true;
                Vec::new()
            }
            _ => Vec::new(),
        }
    }

    // ------------------------------------------------------------------
    // Internal event handlers
    // ------------------------------------------------------------------

    /// Arms the commitment deadline once the request this SSA was registered for has gone out.
    ///
    /// The record itself is created by [`emit_request_next_ssa`](Self::emit_request_next_ssa); this
    /// only starts the clock, because the Entry cannot be late answering a request it has not been
    /// sent yet.
    fn on_ssa_request_sent(&mut self, ssa_id: &SsaId<HoprPseudonym>, now: Instant) -> Vec<SessionPixAction> {
        // Validate pseudonym. Checked before the lookup: a foreign pseudonym never matches a record
        // of ours, so looking first would answer a cross-session confusion with silence.
        if ssa_id.pseudonym() != &self.pseudonym {
            return vec![SessionPixAction::Close(SessionPixCloseReason::InvalidTransition)];
        }

        // The live-record lookup comes first, and the staleness guard only answers for indices that
        // have no record. Retirement is not monotone — a later batch member can lose its deposit while
        // an earlier one is still recovering — so a watermark consulted first would silently ignore a
        // legitimate request for a cycle that is very much alive. Asking the records first removes the
        // assumption instead of relying on it.
        let Some(idx) = self.find_ssa_idx(ssa_id) else {
            // No live record. Either this index was retired and the event is a straggler, which is
            // benign and ignored, or we were never asked for it at all, which is a protocol fault.
            if self
                .highest_retired_ssa_index
                .is_some_and(|highest| ssa_id.ssa_index() <= highest)
            {
                return Vec::new();
            }
            return vec![SessionPixAction::Close(SessionPixCloseReason::InvalidTransition)];
        };

        // Idempotent: a repeated confirmation must not extend the deadline, and one arriving after
        // the SSA has moved on must not resurrect a phase it has left.
        if self.ssas[idx].phase == SsaPhase::AwaitingCommitment && self.ssas[idx].commitment_deadline.is_none() {
            self.ssas[idx].commitment_deadline = now.checked_add(crate::supervision::scaled_deadline(
                self.cfg.max_ssa_delivery_time,
                self.cfg.ssas_per_request,
            ));
        }

        Vec::new()
    }

    /// Arms — or re-arms — the re-request timer on a commitment that is under way but not complete.
    ///
    /// This is where the issue's rule lives: a cycle that received *nothing* never reaches here, so
    /// it never asks, which is correct because there is no scope for it to name. And because every
    /// arriving part pushes the timer out, the first ask cannot land in the middle of the burst —
    /// what the timer measures is a stall in delivery.
    ///
    /// Unscaled by the batch size, unlike the two absolute deadlines: this is a per-cycle stall
    /// detector, and a batch's commitment sets do not queue behind one another in a way that would
    /// make one cycle's silence expected.
    fn on_commitment_progress(&mut self, ssa_id: &SsaId<HoprPseudonym>, now: Instant) -> Vec<SessionPixAction> {
        if let Some(idx) = self.find_ssa_idx(ssa_id)
            && self.ssas[idx].phase == SsaPhase::AwaitingCommitment
            && self.ssas[idx].recommit_attempts < hopr_protocol_pix::MAX_COMMITMENT_RETRANSMISSIONS
        {
            self.ssas[idx].recommit_deadline = now.checked_add(self.cfg.commitment_recommit_interval);
        }

        Vec::new()
    }

    fn on_commitment_verified(&mut self, ssa_id: &SsaId<HoprPseudonym>, now: Instant) -> Vec<SessionPixAction> {
        let idx = match self.find_ssa_idx(ssa_id) {
            Some(i) => i,
            None => return Vec::new(),
        };
        let ssa = &mut self.ssas[idx];
        if ssa.phase != SsaPhase::AwaitingCommitment {
            return Vec::new();
        }

        let deposit_deadline = now.checked_add(crate::supervision::scaled_deadline(
            self.cfg.max_deposit_wait,
            self.cfg.ssas_per_request,
        ));
        let phase_sample = ssa.enter_phase(SsaPhase::AwaitingDeposit, now);
        ssa.deposit_deadline = deposit_deadline;
        ssa.commitment_deadline = None;
        // Nothing is missing any more, so nothing is to be asked for.
        ssa.recommit_deadline = None;
        self.telemetry.phase_durations.extend(phase_sample);

        // Behind the phase guard above, so a repeated `CommitmentVerified` for a cycle that has
        // already moved on counts nothing.
        self.telemetry.committed = self.telemetry.committed.saturating_add(1);

        Vec::new()
    }

    /// Funds the cycle on the deposit pool's verdict. The supervisor does not second-guess the amount.
    ///
    /// Sufficiency is priced and judged outside this workspace. The Exit's pool is handed the
    /// [`DepositUpdated`](hopr_api::node::DepositUpdated) sender together with the byte quota when the
    /// deposit address arrives, and it sends on that channel once the deposit clears the price it
    /// charges for that quota — so by the time this runs, the question has been answered by the only
    /// component that can answer it. Nothing in the Session layer prices a byte.
    ///
    /// This used to compare the running total against `max(expected_deposit, min_deposit)`. Both are
    /// gone, and not because they were merely unused: a configured floor here is not a backstop behind
    /// the pool but a second authority alongside it, and one set above the pool's price fails a cycle
    /// that has already been paid for — silently, on `max_deposit_wait`, with neither side able to
    /// break the tie. `expected_deposit` was the same idea arriving over the wire, and the commitment
    /// never carried an amount to fill it.
    ///
    /// What remains is the one check that needs no price: a zero balance asserts that nothing was
    /// deposited, so it is not a verdict and cannot release service. The deadline keeps running and a
    /// later confirmation can still fund the cycle.
    fn on_deposit_confirmed(
        &mut self,
        ssa_id: &SsaId<HoprPseudonym>,
        amount: HoprBalance,
        now: Instant,
        served_total: u64,
    ) -> Vec<SessionPixAction> {
        let idx = match self.find_ssa_idx(ssa_id) {
            Some(i) => i,
            None => return Vec::new(),
        };
        let was_front = self.earliest_live_idx() == Some(idx);

        let ssa = &mut self.ssas[idx];
        if ssa.phase != SsaPhase::AwaitingDeposit {
            return Vec::new();
        }

        if amount == HoprBalance::zero() {
            return Vec::new();
        }

        // Transition to Recovering. The two recovery clocks are *not* started here — see
        // `arm_recovery_clocks_for_earliest`, which starts them when this cycle's turn comes.
        let phase_sample = ssa.enter_phase(SsaPhase::Recovering, now);
        ssa.deposit_deadline = None;
        ssa.served_total_at_last_progress = served_total;
        self.telemetry.phase_durations.extend(phase_sample);

        // Behind both the phase guard and the zero-amount guard above, so a duplicate confirmation
        // and a zero balance — which is not a verdict — both count nothing.
        ssa.deposit_uhopr = deposit_uhopr(amount);
        self.telemetry.funded = self.telemetry.funded.saturating_add(1);
        self.telemetry.deposit_confirmed_uhopr =
            self.telemetry.deposit_confirmed_uhopr.saturating_add(ssa.deposit_uhopr);

        // If recovery completed before the deposit arrived, immediately
        // tombstone the SSA — the Recovered event was deferred.
        let recovered_pending = ssa.recovered_pending;

        let pending = ssa.next_request_pending_deposit;
        if pending {
            ssa.next_request_pending_deposit = false;
        }
        // End the mutable borrow on ssas[idx].
        let _ = ssa;

        let mut actions = Vec::new();

        if recovered_pending {
            if was_front {
                self.paid_front_handoff = true;
            }
            actions.extend(self.perform_recovered_transition(idx, now, served_total));
        }

        if pending {
            // When recovered_pending is also set, perform_recovered_transition
            // won't emit RequestSsa (next_requested is already set by
            // on_recovered), so emit it here.
            actions.extend(self.emit_request_next_ssa(now));
        }

        actions
    }

    fn on_deposit_observer_closed(
        &mut self,
        ssa_id: &SsaId<HoprPseudonym>,
        now: Instant,
        served_total: u64,
    ) -> Vec<SessionPixAction> {
        let idx = match self.find_ssa_idx(ssa_id) {
            Some(i) => i,
            None => return Vec::new(),
        };

        if self.ssas[idx].phase != SsaPhase::AwaitingDeposit {
            return Vec::new();
        }

        self.close_ssa_and_collect(idx, SessionPixCloseReason::DepositObserverClosed, served_total, now)
    }

    fn on_recovery_progress(
        &mut self,
        progress: &SsaRecoveryProgress<HoprPseudonym>,
        now: Instant,
        served_total: u64,
    ) -> Vec<SessionPixAction> {
        if self
            .paid_recovery_tail
            .as_ref()
            .is_some_and(|tail| tail.ssa_id == progress.ssa_id)
        {
            return self.on_recovered_tail_progress(progress, now, served_total);
        }

        let idx = match self.find_ssa_idx(&progress.ssa_id) {
            Some(i) => i,
            None => return Vec::new(),
        };

        // Absorb late progress on tombstones — the SSA is already fully
        // recovered and should not reset the session-wide gate watermark.
        if self.ssas[idx].is_terminal() {
            return Vec::new();
        }

        // Validate every dimension-derived bound before mutating. Progress is produced locally, but
        // all of its source fields ultimately came from peer-tagged shares. Keeping the supervisor
        // as an independent backstop means a reconstructor regression cannot turn an oversized
        // counter into unbounded service.
        let target = self.dims.target_useful_shares();
        let max_shares_seen = self.dims.polys_per_ssa() as u64 * self.dims.emitted_shares_per_poly() as u64;
        if progress.target_useful_shares != target
            || progress.useful_shares > target
            || progress.shares_seen < progress.useful_shares
            || progress.shares_seen > max_shares_seen
            || progress.recovered_polynomials > self.dims.polys_per_ssa()
        {
            return vec![SessionPixAction::Close(SessionPixCloseReason::CounterRegression)];
        }

        let ssa = &mut self.ssas[idx];
        let new_useful = progress.useful_shares;
        let new_seen = progress.shares_seen;

        // Counter regression check, on the liveness counter because it is the one that moves for every
        // share — a snapshot can be newer than the last one while carrying the same `useful_shares`.
        //
        // The relay-as-Exit pipeline processes acknowledgement batches with
        // for_each_concurrent, so absolute progress snapshots from different
        // batches can arrive out of order. Treat a stale snapshot as benign
        // noise rather than a protocol violation.
        if new_seen <= ssa.largest_shares_seen {
            return Vec::new();
        }

        // The accepted delta, taken here because this is the one point at which a snapshot is known
        // to be both validated and newer than everything seen before. The absolute/reordered
        // semantics the issue asks for fall out of the two high-water marks rather than needing
        // their own machinery: a repeated or stale snapshot returned at the guard above and adds
        // zero, and a snapshot arriving after a dropped one carries the whole missing delta,
        // because a mark is a maximum and never a running sum.
        let accepted_delta = new_seen - ssa.largest_shares_seen;
        // Clamped to the accepted delta, which is not defensive clutter. The two marks are
        // independent maxima over possibly different snapshots, so a reconstructor regression could
        // present `(seen 10, useful 2)` followed by `(seen 11, useful 11)` — a useful delta of nine
        // against an accepted delta of one — and mint a negative surplus. This file is an
        // independent backstop against exactly that class of fault everywhere else; it is one here
        // too.
        let useful_delta = new_useful.saturating_sub(ssa.largest_useful_shares).min(accepted_delta);
        self.telemetry.useful_shares = self.telemetry.useful_shares.saturating_add(useful_delta);
        self.telemetry.surplus_shares = self
            .telemetry
            .surplus_shares
            .saturating_add(accepted_delta - useful_delta);

        ssa.largest_shares_seen = new_seen;

        // Liveness tier. A share arrived, useful or not, so the Entry is serving this cycle — which is
        // the question the egress gate and the recovery-idle deadline are asking. Keying either of them
        // on `useful_shares` instead makes a conforming Entry's surplus run look like silence, and for
        // the gate that is self-inflicted: withholding service removes the only thing that could
        // produce the next useful share. See `SsaRecoveryProgress::shares_seen`.
        ssa.served_total_at_last_progress = served_total;
        // Refreshing is not arming: only `arm_recovery_clocks_for_earliest` starts a cycle's clocks,
        // and `recovery_hard_deadline` is how a cycle records that it has reached the front. Without
        // that second condition a stray reordered share *starts* the idle clock on a cycle still
        // queued behind the front — which the Entry cannot then feed, because emission is clamped to
        // one cycle — and 60 s later the idle gate sees session-wide service climbing and retires it.
        // That is the queue-wait failure `arm_recovery_clocks_for_earliest` documents, reached
        // through the refresh path instead of the deposit one. The clocks arm together or not at all.
        if ssa.phase == SsaPhase::Recovering && ssa.recovery_hard_deadline.is_some() {
            ssa.recovery_idle_deadline = now.checked_add(self.cfg.max_recovery_idle);
        }

        // Payment tier. Only shares that advanced reconstruction are worth anything, so only they move
        // the recovery counters or count towards the share-order ratio.
        if new_useful > ssa.largest_useful_shares {
            let delta = new_useful - ssa.largest_useful_shares;
            ssa.largest_useful_shares = new_useful;
            ssa.recovered_polynomials = progress.recovered_polynomials;

            if let Some(reason) = self.book_share_order_progress(&progress.ssa_id, delta) {
                return vec![SessionPixAction::Close(reason)];
            }
        }

        let is_front = self.earliest_live_idx() == Some(idx);

        // Only funded progress on the cycle service is currently charged against may reopen the
        // Session-wide ceiling. Unfunded progress is the H11 bypass; off-front progress belongs to a
        // queued cycle and must not buy service for the front either. The outer event transition
        // consumes any predecessor tail before synchronizing the gate, even for unfunded progress.
        if self.ssas[idx].phase == SsaPhase::Recovering && is_front {
            vec![SessionPixAction::ProgressNotification]
        } else {
            Vec::new()
        }
    }

    /// Applies a progress snapshot from the sole recovered predecessor authorized to drain.
    ///
    /// The reconstructor enforces the strong per-polynomial budget. These aggregate checks are a
    /// second security boundary: even if that component regresses, the Entry cannot make a recovered
    /// cycle report more than `polys × (threshold + surplus)` packets, change its terminal useful
    /// count, or revive any older tombstone.
    fn on_recovered_tail_progress(
        &mut self,
        progress: &SsaRecoveryProgress<HoprPseudonym>,
        now: Instant,
        served_total: u64,
    ) -> Vec<SessionPixAction> {
        let target = self.dims.target_useful_shares();
        let expected_polynomials = self.dims.polys_per_ssa();
        let max_shares_seen = expected_polynomials as u64 * self.dims.emitted_shares_per_poly() as u64;

        if progress.target_useful_shares != target
            || progress.useful_shares > target
            || progress.recovered_polynomials > expected_polynomials
            || progress.shares_seen < progress.useful_shares
            || progress.shares_seen > max_shares_seen
        {
            tracing::warn!(
                ssa_id = %progress.ssa_id,
                useful_shares = progress.useful_shares,
                shares_seen = progress.shares_seen,
                recovered_polynomials = progress.recovered_polynomials,
                "rejected impossible progress from a recovered PIX tail"
            );
            return vec![SessionPixAction::Close(SessionPixCloseReason::CounterRegression)];
        }

        // Absolute snapshots from concurrent acknowledgement batches may arrive after the
        // `Recovered` event that overtook them. A non-terminal snapshot cannot buy liveness here,
        // but it is not evidence of peer misconduct either. Ignore it exactly as the live path
        // ignores an older absolute counter. Only a terminal snapshot can touch the paid receipt.
        if progress.useful_shares != target || progress.recovered_polynomials != expected_polynomials {
            return Vec::new();
        }

        let (exhausted, accepted_delta) = {
            let tail = self
                .paid_recovery_tail
                .as_mut()
                .expect("caller matched the paid recovery tail");
            if progress.shares_seen <= tail.largest_shares_seen {
                return Vec::new();
            }

            let accepted_delta = progress.shares_seen - tail.largest_shares_seen;
            tail.largest_shares_seen = progress.shares_seen;
            tail.served_total_at_last_progress = served_total;
            tail.idle_deadline = now.checked_add(self.cfg.max_recovery_idle);
            (progress.shares_seen == max_shares_seen, accepted_delta)
        };

        // All surplus, by definition: this cycle has already reached its useful-share target, which
        // is what the guard above requires before a snapshot may touch the receipt at all. These
        // are the negotiated tail a conforming Entry keeps emitting while the FIFO drains — real
        // accepted traffic, and the evidence that keeps the paid front open, but no longer worth
        // anything towards recovery.
        self.telemetry.surplus_shares = self.telemetry.surplus_shares.saturating_add(accepted_delta);

        // Keep the tombstone's diagnostics monotone while it is retained, but do not require the
        // record: tombstone cleanup is intentionally shorter than a Session's possible FIFO drain.
        if let Some(idx) = self.find_ssa_idx(&progress.ssa_id) {
            self.ssas[idx].largest_useful_shares = target;
            self.ssas[idx].largest_shares_seen = progress.shares_seen;
            self.ssas[idx].recovered_polynomials = expected_polynomials;
            self.ssas[idx].served_total_at_last_progress = served_total;
        }

        if exhausted {
            self.release_paid_tail(now);
        }
        vec![SessionPixAction::ProgressNotification]
    }

    /// Attributes `delta` useful shares to the front of the batch or behind it, and judges the ratio.
    ///
    /// The Entry serves a batch strictly in index order, so progress on anything but the front cycle is
    /// service the Exit cannot yet be paid for. See
    /// [`SupervisorConfig::max_off_front_share_fraction`] for the threat and for why this is measured
    /// as a fraction; the judgement waits for
    /// [`SupervisorConfig::min_share_order_sample`] shares of evidence.
    fn book_share_order_progress(
        &mut self,
        ssa_id: &SsaId<HoprPseudonym>,
        delta: u64,
    ) -> Option<SessionPixCloseReason> {
        let front = self.share_order_front?;
        if ssa_id.ssa_index() == front {
            self.front_useful = self.front_useful.saturating_add(delta);
        } else {
            self.off_front_useful = self.off_front_useful.saturating_add(delta);
        }

        // Judged on every event, not only off-front ones: the sample can cross the floor on a share
        // that lands *on* the front cycle, and the ratio is already whatever it is by then. Waiting for
        // the next off-front share would only delay the same verdict.
        let total = self.front_useful.saturating_add(self.off_front_useful);
        if total < self.cfg.min_share_order_sample || total == 0 {
            return None;
        }

        let fraction = self.off_front_useful as f64 / total as f64;
        if fraction > self.cfg.max_off_front_share_fraction {
            tracing::error!(
                %ssa_id, %front, fraction,
                limit = self.cfg.max_off_front_share_fraction,
                front_useful = self.front_useful,
                off_front_useful = self.off_front_useful,
                "entry is serving the batch out of order"
            );
            return Some(SessionPixCloseReason::BatchServedOutOfOrder);
        }

        tracing::trace!(%ssa_id, %front, fraction, "progress behind the front of the batch");
        None
    }

    fn on_almost_recovered(&mut self, ssa_id: &SsaId<HoprPseudonym>, now: Instant) -> Vec<SessionPixAction> {
        let idx = match self.find_ssa_idx(ssa_id) {
            Some(i) => i,
            None => return Vec::new(),
        };

        // If already recovered (e.g. a concurrent batch delivered Recovered
        // before this AlmostRecovered), the next SSA is already requested —
        // no-op.
        if self.ssas[idx].is_terminal() {
            return Vec::new();
        }

        // The successor gate. A cycle in the middle of a batch reaching its early-recovery threshold
        // says nothing about the batch as a whole: the ones behind it still have their full quota to
        // serve. Only the last cycle standing in for the batch may ask for a replacement.
        if !self.ssas[idx].is_batch_last {
            return Vec::new();
        }

        let next_requested = self.ssas[idx].next_requested;
        let phase = self.ssas[idx].phase;

        if next_requested {
            return Vec::new();
        }

        match phase {
            SsaPhase::Recovering => {
                self.ssas[idx].next_requested = true;
                self.emit_request_next_ssa(now)
            }
            SsaPhase::AwaitingDeposit => {
                self.ssas[idx].next_requested = true;
                self.ssas[idx].next_request_pending_deposit = true;
                Vec::new()
            }
            // AwaitingCommitment or unknown phase: the next SSA request will be
            // triggered when the recovered transition fires (normal Recovering
            // path or deferred replay via recovered_pending). Set the deferred
            // flag so that commitment_verified can propagate it forward.
            _ => {
                self.ssas[idx].next_request_pending_deposit = true;
                self.ssas[idx].next_requested = true;
                Vec::new()
            }
        }
    }

    fn on_recovered(
        &mut self,
        ssa_id: &SsaId<HoprPseudonym>,
        now: Instant,
        served_total: u64,
    ) -> Vec<SessionPixAction> {
        let idx = match self.find_ssa_idx(ssa_id) {
            Some(i) => i,
            None => return Vec::new(),
        };

        // Guard against duplicate recovery events (possible with concurrent
        // batch processing). If already recovered, this is a no-op.
        if self.ssas[idx].is_terminal() {
            return Vec::new();
        }

        match self.ssas[idx].phase {
            SsaPhase::Recovering => {
                // Normal path — transition to tombstone directly.
                self.perform_recovered_transition(idx, now, served_total)
            }
            SsaPhase::AwaitingDeposit => {
                // Recovery completed before deposit confirmed. Record the
                // pending flag so that when the deposit arrives we replay
                // the transition immediately after entering Recovering.
                self.ssas[idx].recovered_pending = true;
                // Also ensure the next SSA request is deferred — the
                // eventual tombstone will emit it. Subject to the same successor gate as
                // `on_almost_recovered`: a non-last cycle of a batch never asks, deferred or not.
                if !self.ssas[idx].next_requested && self.ssas[idx].is_batch_last {
                    self.ssas[idx].next_request_pending_deposit = true;
                    self.ssas[idx].next_requested = true;
                }
                Vec::new()
            }
            SsaPhase::AwaitingCommitment => {
                // Recovery outpaced commitment verification. Same semantics
                // as AwaitingDeposit: set pending flag so the transition
                // replays once commitment and deposit have both arrived.
                self.ssas[idx].recovered_pending = true;
                Vec::new()
            }
            // Closing / already Recovered: handled by the terminal guard above.
            _ => Vec::new(),
        }
    }

    /// Books what a recovered cycle owes the counters and retires it into its tombstone.
    ///
    /// Split out because a drain returns before the rest of the transition, and would otherwise
    /// never book the deposit it just unlocked — making the one outcome the drain exists for read as
    /// stranded value in `hopr_pix_deposits_recovered_uhopr_total`.
    fn book_recovered_cycle(&mut self, idx: usize, now: Instant, served_total: u64) {
        // `Recovered` is a local cryptographic fact, so it supplies the lower bound even if its
        // final absolute Progress snapshot is delivered just afterwards by a concurrent worker.
        // Every later tail snapshot must keep these terminal fields exact.
        let target = self.dims.target_useful_shares();
        self.ssas[idx].largest_useful_shares = target;
        self.ssas[idx].largest_shares_seen = self.ssas[idx].largest_shares_seen.max(target);
        self.ssas[idx].recovered_polynomials = self.dims.polys_per_ssa();

        // Counted here rather than on the `Recovered` event, because that event is not the
        // transition: `on_recovered` defers it when the cycle is still awaiting its commitment or
        // deposit, and replays it through this function once both arrive. This is the one place a
        // cycle becomes terminal-by-recovery, and its callers all guard against reaching it twice.
        self.telemetry.recovered = self.telemetry.recovered.saturating_add(1);
        // Summarized before the tombstone transition below, and before `paid_recovery_tail` starts
        // collecting the drain, which belongs to the FIFO rather than to this cycle's recovery. The
        // terminal fields were just pinned to the target above, so a recovered cycle's useful
        // fraction is exactly one by construction.
        let summary = self.ssas[idx].telemetry_summary(PixCycleOutcome::Recovered, served_total);
        self.telemetry.finalized.push(summary);

        let phase_sample = self.ssas[idx].enter_phase(
            SsaPhase::Recovered {
                tombstone_until: now
                    .checked_add(self.cfg.tombstone_retention_window)
                    .unwrap_or_else(|| now + Duration::from_secs(86400 * 365)),
            },
            now,
        );
        self.telemetry.phase_durations.extend(phase_sample);
    }

    /// Perform the terminal tombstone transition for a fully-recovered SSA.
    /// Called from `on_recovered` (normal Recovering path) and replayed from
    /// `on_deposit_confirmed`/`on_commitment_verified` when `recovered_pending`
    /// was set earlier.
    fn perform_recovered_transition(&mut self, idx: usize, now: Instant, served_total: u64) -> Vec<SessionPixAction> {
        let next_requested = self.ssas[idx].next_requested;
        // Read before the cycle is retired below, which would make it terminal and change the answer.
        let was_front = self.earliest_live_idx() == Some(idx);
        let max_shares_seen = self.dims.polys_per_ssa() as u64 * self.dims.emitted_shares_per_poly() as u64;

        self.book_recovered_cycle(idx, now, served_total);

        // The end of a drain, and the outcome it exists for: the deposit this cycle was paid is
        // released rather than stranded. None of the bookkeeping below is worth doing — the tail
        // receipt, the gate notification and the successor request all serve a Session that is about
        // to be torn down, and `closed` gates every entry point from here on. Returning before the
        // tail is created also guarantees a drain can never leave one behind.
        if self.draining {
            tracing::info!(
                ssa_id = %self.ssas[idx].ssa_id,
                "the funded cycle of a closed session recovered — the SURB drain is complete"
            );
            self.closed = true;
            return self.with_fill_stopped(vec![SessionPixAction::Close(SessionPixCloseReason::Drained)]);
        }

        if was_front {
            // Replacing rather than accumulating is intentional. Only the immediate recovered
            // predecessor can occupy a FIFO immediately ahead of the live front; accepting an
            // arbitrary tombstone here would turn old-cycle replay into gate liveness.
            self.paid_recovery_tail =
                (self.ssas[idx].largest_shares_seen < max_shares_seen).then(|| PaidRecoveryTail {
                    ssa_id: self.ssas[idx].ssa_id,
                    entered_at: now,
                    largest_shares_seen: self.ssas[idx].largest_shares_seen,
                    served_total_at_last_progress: served_total,
                    // Recovery itself is fresh, cryptographic progress. Reusing the prior idle
                    // instant would let delayed/dropped progress snapshots make an honest tail
                    // expire immediately after recovery. Refresh this clock once; unlike the hard
                    // deadline below, it is the clock progress is meant to extend.
                    idle_deadline: now.checked_add(self.cfg.max_recovery_idle),
                    hard_deadline: self.ssas[idx]
                        .recovery_hard_deadline
                        .or_else(|| now.checked_add(self.cfg.max_recovery_time)),
                });
        }

        self.ssas[idx].commitment_deadline = None;
        self.ssas[idx].deposit_deadline = None;
        self.ssas[idx].recovery_idle_deadline = None;
        self.ssas[idx].recovery_hard_deadline = None;
        self.ssas[idx].recovered_pending = false;

        let mut actions = Vec::new();
        // A cycle that recovered with nothing buffered behind it leaves the Exit with nothing to fill
        // for until its successor reaches the front and arms its own clocks — which is at least a
        // deposit away. Stopping now rather than waiting for the successor keeps those SURBs for the
        // cycle that will actually be paid for; the next tick starts the ramp again from the
        // successor's true remainder.
        if was_front && self.paid_recovery_tail.is_none() {
            actions.extend(self.fill.stop().map(SessionPixAction::SetFillRate));
        }
        if was_front {
            // `Recovered` is stronger evidence than an ordinary progress snapshot. Emit this even
            // though the reconstructor normally sends the final snapshot first: progress delivery
            // is deliberately lossy under command-channel pressure, and without this reset a gate
            // already at its served-without-progress ceiling cannot spend the first buffered SURB
            // needed to produce tail liveness. If the successor is unfunded, `sync_service_gate`
            // prepends `WithholdService`; notifying afterwards changes only the dormant funded
            // watermark and cannot restore service.
            actions.push(SessionPixAction::ProgressNotification);
        }
        // Full recovery is the fallback trigger for the early signal, so it carries the same
        // successor gate: a non-last cycle of a batch is tombstoned without asking for anything.
        if !next_requested && self.ssas[idx].is_batch_last {
            self.ssas[idx].next_requested = true;
            actions.extend(self.emit_request_next_ssa(now));
        }

        actions
    }

    /// Closes the Session on the first unverifiable-share report. There is no tolerance to configure.
    ///
    /// Shares are not checked on arrival, so a report is not "a bad share" — it means a whole
    /// polynomial's share set failed to open its commitment, which surfaces only once `threshold` of
    /// them have been interpolated. `SsaPart::add_share` then marks that part failed and releases its
    /// shares, and nothing ever clears the flag; since the SSA is the sum of *every* polynomial's
    /// constant term, the cycle can no longer be reconstructed by any means and will never pay.
    ///
    /// So serving on is not tolerance, it is donation: every packet past this point is unpaid with no
    /// recovery to preserve at the end of it. That holds however the failure arose, which is why this
    /// is not a knob — a false positive from a verification bug leaves the part just as permanently
    /// failed, so a tolerance would buy unpaid service rather than the cycle it was raised to save.
    /// Closing here caps the exposure at the `threshold` packets already served rather than a multiple
    /// of it.
    ///
    /// `observed_total` is the reconstructor's cross-peer aggregate, and with no limit to compare it
    /// against it is logged rather than accumulated: it says how many polynomials the batch that
    /// triggered this took down, which is the difference between one bad relayer and a peer sending
    /// garbage wholesale.
    ///
    /// Note this closes the Session directly rather than through
    /// [`close_ssa_and_collect`](Self::close_ssa_and_collect), so a batch's surviving siblings do not
    /// keep it alive: unlike a lost deposit, this fault is evidence about the peer rather than about
    /// one cycle.
    fn on_unverifiable_shares(
        &mut self,
        ssa_id: &SsaId<HoprPseudonym>,
        observed_total: u64,
        peer: &OffchainPublicKey,
        _now: Instant,
    ) -> Vec<SessionPixAction> {
        let idx = match self.find_ssa_idx(ssa_id) {
            Some(i) => i,
            None => return Vec::new(),
        };

        // Absorb late reports on a tombstoned or already-closing SSA: there is no service left to
        // stop, and the cycle they condemn has already left. Concurrent ack batches make these
        // ordinary rather than exceptional.
        if self.ssas[idx].is_terminal() {
            return Vec::new();
        }

        tracing::warn!(
            %ssa_id,
            %peer,
            observed_total,
            phase = ?self.ssas[idx].phase,
            "closing PIX session: a polynomial's share set failed to open its commitment"
        );

        self.with_fill_stopped(vec![SessionPixAction::Close(SessionPixCloseReason::UnverifiableShares)])
    }

    /// Decides whether a Session closed at the Exit is worth draining its buffered SURBs into.
    ///
    /// `available` is the manager's estimate of the SURBs still spendable, already net of the fill
    /// reserve. The cycle it is measured against is the earliest live one, and only when it is both
    /// funded and clocked — i.e. exactly the cycle fill would have been working for. Two shapes have
    /// nothing a drain could buy and close at once: an unfunded front, where no deposit is at stake;
    /// and a recovered predecessor's paid FIFO tail, whose key is already recovered and behind which
    /// the successor is queued without clocks of its own.
    ///
    /// The comparison is against the *whole* remaining emission, with no loss margin. That is
    /// conservative in both directions on purpose: the estimate is an upper bound (see
    /// [`SessionPixEvent::SessionClosed`]), while in-flight packets are already subtracted from it
    /// and full recovery in practice needs fewer than `E` shares. A drain that runs out partway has
    /// spent the Session's SURBs down to the reserve and recovered nothing, which is strictly worse
    /// than never starting.
    ///
    /// Siblings are retired directly rather than through
    /// [`close_ssa_and_collect`](Self::close_ssa_and_collect), which would be wrong twice over: it
    /// charges [`failed_cycles`](Self::failed_cycles), and these cycles were retired by an operator
    /// closing a Session rather than lost by an Entry; and its `ssas.len() == 1` branch would close
    /// the Session outright. Recovered tombstones are left alone — they hold no reconstructor state
    /// and expire on their own timer with an idempotent `RetireSsa`.
    ///
    /// `_now` is taken for symmetry with every other handler; nothing here is timed. A drain inherits
    /// the target's existing clocks rather than starting any of its own, which is the point — the
    /// resource bound on the cycle must not be extended by the Session ending.
    fn on_session_closed(&mut self, available: u64, _now: Instant) -> Vec<SessionPixAction> {
        let target = self
            .paid_recovery_tail
            .is_none()
            .then(|| self.earliest_live_idx())
            .flatten()
            .filter(|&idx| {
                self.ssas[idx].phase == SsaPhase::Recovering && self.ssas[idx].recovery_hard_deadline.is_some()
            });

        let cycle_shares = self.dims.polys_per_ssa() as u64 * self.dims.emitted_shares_per_poly() as u64;
        let remaining = target.map(|idx| cycle_shares.saturating_sub(self.ssas[idx].largest_shares_seen));

        let Some((idx, remaining)) = target.zip(remaining).filter(|&(_, remaining)| available >= remaining) else {
            tracing::info!(
                available,
                ?remaining,
                target = ?target.map(|idx| self.ssas[idx].ssa_id),
                "closing PIX session: the exit-side close left nothing worth draining for"
            );
            self.closed = true;
            return self.with_fill_stopped(vec![SessionPixAction::Close(SessionPixCloseReason::SessionClosed)]);
        };

        let target_id = self.ssas[idx].ssa_id;
        tracing::info!(
            %target_id,
            available,
            remaining,
            "draining buffered SURBs into the funded cycle of a session closed at the exit"
        );

        self.draining = true;
        // A deferred request can never be retried now, and leaving the flag set would have
        // `retry_deferred_successor_request` call into the (now silent) request path every event.
        self.successor_request_deferred = false;

        // Every sibling is behind the target in index order — the target *is* `earliest_live_idx` —
        // so removing them cannot move the front, and the tail of `handle_event` re-derives the same
        // gate state and share-order front it already had.
        let mut retired = Vec::new();
        self.ssas.retain(|ssa| {
            if ssa.ssa_id == target_id || ssa.is_terminal() {
                true
            } else {
                retired.push(ssa.ssa_id);
                false
            }
        });

        retired
            .into_iter()
            .map(|ssa_id| {
                self.note_retired_index(ssa_id.ssa_index());
                SessionPixAction::RetireSsa(ssa_id)
            })
            .collect()
    }

    // ------------------------------------------------------------------
    // Deadline helpers
    // ------------------------------------------------------------------

    /// Close the SSA at `idx` and return close actions.
    fn close_ssa_and_collect(
        &mut self,
        idx: usize,
        reason: SessionPixCloseReason,
        served_total: u64,
        now: Instant,
    ) -> Vec<SessionPixAction> {
        if matches!(self.ssas[idx].phase, SsaPhase::Closing) {
            return Vec::new();
        }

        // Track the first close reason so it isn't lost when multiple SSAs close.
        self.first_failure_reason.get_or_insert(reason);
        let close_reason = self.first_failure_reason.unwrap_or(reason);

        // Charged before the branches below, because the whole point is a count that outlives the
        // record being retired. Every path into this function is a cycle lost without recovering —
        // a recovered cycle leaves through tombstone expiry in `handle_deadline`, not through here.
        self.failed_cycles += 1;
        // Same transition, so the same guard covers both: the `Closing` check at the top of this
        // function is what makes a repeated close of one cycle count once. The summary rides along
        // for the same reason — this is the one place a cycle is retired without recovering, and
        // the counters are still intact because nothing below has cleared them yet.
        self.telemetry.failed = self.telemetry.failed.saturating_add(1);
        let summary = self.ssas[idx].telemetry_summary(PixCycleOutcome::Failed, served_total);
        self.telemetry.finalized.push(summary);

        // Warn-level diagnostic with full SSA state before closing.
        let ssa = &self.ssas[idx];
        tracing::warn!(
            ssa_id = %ssa.ssa_id,
            ?reason,
            phase = ?ssa.phase,
            largest_useful_shares = ssa.largest_useful_shares,
            target_useful_shares = ssa.target_useful_shares,
            recovered_polynomials = ssa.recovered_polynomials,
            served_total_at_last_progress = ssa.served_total_at_last_progress,
            ?ssa.commitment_deadline,
            ?ssa.deposit_deadline,
            ?ssa.recovery_idle_deadline,
            ?ssa.recovery_hard_deadline,
            "closing PIX SSA"
        );

        let phase_sample = self.ssas[idx].enter_phase(SsaPhase::Closing, now);
        self.telemetry.phase_durations.extend(phase_sample);

        if self.ssas.len() == 1 {
            self.closed = true;
            return self.with_fill_stopped(vec![SessionPixAction::Close(close_reason)]);
        }

        // A batch has siblings to fall back on, which is what makes retiring one member survivable —
        // but only up to a point. Past the limit the Session goes, with the *first* failure's reason
        // rather than this one's, so the report names the cause instead of the last symptom.
        //
        // Strictly greater: the field is how many failures are *tolerated*, so the shipping value of
        // one keeps the existing "retire the member and carry on" behaviour for a batch's first loss
        // — including handing on the successor gate below — and closes on the second.
        if self.failed_cycles > self.cfg.max_failed_cycles.max(1) {
            tracing::warn!(
                failed_cycles = self.failed_cycles,
                max_failed_cycles = self.cfg.max_failed_cycles,
                ?close_reason,
                "closing PIX session: too many cycles lost without recovering"
            );
            self.closed = true;
            return self.with_fill_stopped(vec![SessionPixAction::Close(close_reason)]);
        }

        // Clear deadlines on this SSA.
        self.ssas[idx].commitment_deadline = None;
        self.ssas[idx].deposit_deadline = None;
        self.ssas[idx].recovery_idle_deadline = None;
        self.ssas[idx].recovery_hard_deadline = None;

        // If all SSAs are terminal, close the session.
        if self
            .ssas
            .iter()
            .all(|s| matches!(s.phase, SsaPhase::Closing | SsaPhase::Recovered { .. }))
        {
            self.closed = true;
            return self.with_fill_stopped(vec![SessionPixAction::Close(close_reason)]);
        }

        // Remove this closing SSA and emit RetireSsa so the reconstructor
        // releases its builder/verifier/counter state mid-session.
        // Record the retired index to prevent stale SsaRequestSent events
        // from resurrecting it.
        let retired = self.ssas[idx].ssa_id;
        self.note_retired_index(retired.ssa_index());
        let carried_successor_gate = self.ssas[idx].is_batch_last && !self.ssas[idx].next_requested;
        self.ssas.remove(idx);

        // Retiring the cycle that carried the successor gate would strand the batch: its surviving
        // siblings are barred from asking, so nothing would ever request a replacement and the
        // Session would die on a recovery timeout that names the timer rather than the cause. Hand
        // the gate to the newest cycle still standing, which is now the last of the batch that will
        // actually serve its quota — so the request still cannot come early. `next_requested` guards
        // it: a gate already spent must not be handed on and fire a second time.
        if carried_successor_gate
            && let Some(newest) = self
                .ssas
                .iter_mut()
                .filter(|s| !s.is_terminal())
                .max_by_key(|s| s.ssa_id.ssa_index())
        {
            tracing::debug!(
                %retired, promoted = %newest.ssa_id,
                "successor gate handed on from a retired cycle"
            );
            newest.is_batch_last = true;
        }
        vec![SessionPixAction::RetireSsa(retired)]
    }

    /// Allocates the next batch of SSA indices and asks for all of them in one action.
    ///
    /// The batch size is [`SupervisorConfig::ssas_per_request`], clamped rather than trusted since a
    /// supervisor can be built from a config that never went through `validate_pix_supervision`.
    ///
    /// Exactly one cycle of the batch — the last one allocated — carries
    /// [`PerSsaState::is_batch_last`], and only that cycle may ask for the successor batch. Without
    /// it every member would ask for one of its own, since the request flags are per-cycle; see that
    /// field for why the last rather than the first.
    ///
    /// Before allocating, this enforces both halves of the admission reservation: no more than
    /// [`crate::MAX_OVERLAPPING_BATCHES`] live generations and no more than that many full batches'
    /// live cycles. A request earned while the reservation is full is retried after a generation
    /// releases its reconstructor state.
    ///
    /// What batching does change is the exposure *within* a batch: every cycle in it is unfunded at
    /// once, so the ceiling is `ssas_per_request` SSA quotas rather than one. That is the trade the
    /// knob exists to make, and it is why both deadlines are scaled by the same factor.
    fn emit_request_next_ssa(&mut self, now: Instant) -> Vec<SessionPixAction> {
        // A drain has no Session left to serve a successor's quota, so asking for one would commit
        // the Entry to a deposit against a cycle that could never be served. Guarded here rather than
        // at each of the four call sites — the early signal, the recovered transition, the deferred
        // retry and the deposit replay — so no route into a request can be missed.
        if self.draining {
            return Vec::new();
        }

        let batch = self.cfg.ssas_per_request.clamp(1, crate::MAX_SSA_BATCH_SIZE);
        let live_cycles = self.live_cycle_count();
        let live_batches = self.live_batch_count();
        if live_batches >= crate::MAX_OVERLAPPING_BATCHES as usize
            || live_cycles.saturating_add(batch) > self.reserved_cycle_slots()
        {
            self.successor_request_deferred = true;
            tracing::debug!(
                live_cycles,
                live_batches,
                requested = batch,
                reserved = self.reserved_cycle_slots(),
                "deferring successor SSA batch until an older batch is released"
            );
            return Vec::new();
        }

        self.successor_request_deferred = false;
        let batch_id = self.next_ssa_index;
        let mut ssa_ids = Vec::with_capacity(batch);

        for _ in 0..batch {
            let index = self.next_ssa_index;

            let Ok(ssa_index) = SsaIndex::try_from(index) else {
                break;
            };
            let Some(next) = index.checked_add(1) else {
                // Index space exhausted mid-batch. Stop here rather than wrapping: a reused index
                // would collide with a live cycle. Whatever was allocated still goes out.
                break;
            };
            self.next_ssa_index = next;

            let ssa_id = SsaId::new(self.pseudonym, ssa_index);

            // Register each SSA now rather than on `SsaRequestSent`. Carrying out the action registers
            // an Exit commitment, which is what makes shares for that SSA processable — so
            // observations about it can reach us before the confirmation that we asked for it does.
            // Every handler ignores an SSA it has no record of, and for `UnverifiableShares` that
            // would mean failing open on exactly the signal that must fail closed.
            //
            // No deadline yet: the request has not gone out, so there is nothing to be late for. That
            // is what `SsaRequestSent` adds, per index.
            self.ssas.push(PerSsaState::new(
                ssa_id,
                batch_id,
                self.dims.target_useful_shares(),
                now,
            ));
            // Counted at allocation rather than at `SsaRequestSent`, for the same reason the record
            // is created here: a cycle is registered before its request goes out, and observations
            // about it can arrive before the confirmation that it was asked for. A batch truncated
            // by index exhaustion counts only what it allocated.
            self.telemetry.requested = self.telemetry.requested.saturating_add(1);
            ssa_ids.push(ssa_id);
        }

        // Nothing allocated means the index space is spent, and no further cycle can ever be funded.
        if ssa_ids.is_empty() {
            self.closed = true;
            return vec![SessionPixAction::Close(SessionPixCloseReason::InvalidTransition)];
        }

        // The last cycle *actually* allocated carries the successor gate, which is what makes a
        // batch truncated by index exhaustion still able to ask for the next one.
        self.ssas
            .last_mut()
            .expect("a non-empty batch has just been pushed")
            .is_batch_last = true;

        vec![SessionPixAction::RequestSsa {
            ssa_ids,
            params: self.dims,
        }]
    }

    fn find_ssa_idx(&self, ssa_id: &SsaId<HoprPseudonym>) -> Option<usize> {
        self.ssas.iter().position(|s| s.ssa_id == *ssa_id)
    }
}

// ---------------------------------------------------------------------------
// PerSsaState — deadline check (borrows immutably from self)
// ---------------------------------------------------------------------------

impl PerSsaState {
    /// Check which deadline expired, if any.
    fn check_deadlines(&self, now: Instant) -> Option<SessionPixCloseReason> {
        match self.phase {
            SsaPhase::AwaitingCommitment => self
                .commitment_deadline
                .filter(|d| now >= *d)
                .map(|_| SessionPixCloseReason::CommitmentTimeout),
            SsaPhase::AwaitingDeposit => self
                .deposit_deadline
                .filter(|d| now >= *d)
                .map(|_| SessionPixCloseReason::DepositTimeout),
            SsaPhase::Recovering => {
                // Hard deadline is immutable.
                if let Some(d) = self.recovery_hard_deadline
                    && now >= d
                {
                    return Some(SessionPixCloseReason::RecoveryDeadline);
                }
                // Idle deadline — service gating happens in handle_deadline.
                if let Some(d) = self.recovery_idle_deadline
                    && now >= d
                {
                    return Some(SessionPixCloseReason::RecoveryIdle);
                }
                None
            }
            SsaPhase::Recovered { .. } | SsaPhase::Closing => None,
        }
    }
}

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

#[cfg(test)]
mod tests {
    use std::time::{Duration, Instant};

    use hopr_api::{
        HoprBalance,
        types::{crypto_random::Randomizable, internal::prelude::HoprPseudonym},
    };
    use hopr_protocol_pix::{SsaId, SsaIndex, SsaRecoveryProgress};

    use super::{
        super::{FillRate, ORGANIC_WINDOW, PixFillConfig, SAMPLING_INTERVAL},
        *,
    };

    /// A compact baseline for the state-machine tests — *not* [`SupervisorConfig::default`].
    ///
    /// Deadlines and budgets are shrunk to keep the tests short; the tolerances that remain are at
    /// their shipped values. Tests that care about a specific one say so by name.
    fn default_cfg() -> SupervisorConfig {
        SupervisorConfig {
            ssas_per_request: 1,
            allow_dynamic_ssa_batches: true,
            max_failed_cycles: 1,
            max_ssa_delivery_time: Duration::from_secs(20),
            commitment_recommit_interval: Duration::from_secs(3),
            max_deposit_wait: Duration::from_secs(60),
            max_recovery_idle: Duration::from_secs(60),
            max_recovery_time: Duration::from_secs(3600),
            max_off_front_share_fraction: 0.25,
            min_share_order_sample: 16384,
            max_predeposit_packets: 1024,
            max_served_without_progress: 256,
            tombstone_retention_window: Duration::from_secs(30),
            fill: PixFillConfig::default(),
        }
    }

    /// Test dimensions, with a deliberately non-zero surplus.
    ///
    /// Surplus never enters the payment target, but it does bound the paid FIFO tail after recovery.
    /// A non-zero value makes both properties visible: `target_useful_shares` stays threshold-only,
    /// while accepted liveness cannot pass the emitted cycle volume.
    fn dims(polys: u16, threshold: u8) -> PixParams {
        PixParams::try_new(polys, threshold, 7, crate::types::LOCAL_PIX_SUITE).expect("test dimensions must be valid")
    }

    fn pseudonym() -> HoprPseudonym {
        HoprPseudonym::random()
    }

    fn ssa_id(p: HoprPseudonym, idx: u32) -> SsaId<HoprPseudonym> {
        SsaId::new(p, SsaIndex::new(idx).unwrap())
    }

    /// A snapshot in which every share seen was useful — the shape of a cycle's first
    /// `threshold` shares per polynomial, and what every test predating the liveness split means.
    fn make_progress(
        ssa_id: SsaId<HoprPseudonym>,
        useful: u64,
        target: u64,
        recovered_polys: u16,
    ) -> SsaRecoveryProgress<HoprPseudonym> {
        make_progress_seen(ssa_id, useful, useful, target, recovered_polys)
    }

    /// A snapshot where `seen` and `useful` can differ, i.e. the Entry is serving surplus.
    fn make_progress_seen(
        ssa_id: SsaId<HoprPseudonym>,
        useful: u64,
        seen: u64,
        target: u64,
        recovered_polys: u16,
    ) -> SsaRecoveryProgress<HoprPseudonym> {
        SsaRecoveryProgress {
            ssa_id,
            useful_shares: useful,
            shares_seen: seen,
            target_useful_shares: target,
            recovered_polynomials: recovered_polys,
        }
    }

    fn make_recovered_tail_progress(
        sup: &SessionPixSupervisor,
        ssa_id: SsaId<HoprPseudonym>,
        shares_seen: u64,
    ) -> SsaRecoveryProgress<HoprPseudonym> {
        SsaRecoveryProgress {
            ssa_id,
            useful_shares: sup.dims.target_useful_shares(),
            shares_seen,
            target_useful_shares: sup.dims.target_useful_shares(),
            recovered_polynomials: sup.dims.polys_per_ssa(),
        }
    }

    fn max_cycle_shares(sup: &SessionPixSupervisor) -> u64 {
        sup.dims.polys_per_ssa() as u64 * sup.dims.emitted_shares_per_poly() as u64
    }

    fn sufficient_balance() -> HoprBalance {
        HoprBalance::new_base(1000)
    }

    /// Drives every cycle of a batch to `Recovering`, so only the batch's own gates hold anything back.
    fn fund_batch(sup: &mut SessionPixSupervisor, p: HoprPseudonym, batch: u32, now: Instant) {
        for idx in 1..=batch {
            let id = ssa_id(p, idx);
            sup.handle_event(&SessionPixEvent::SsaRequestSent(id), now, 0);
            sup.handle_event(&SessionPixEvent::CommitmentVerified(id), now, 0);
            sup.handle_event(
                &SessionPixEvent::DepositConfirmed {
                    ssa_id: id,
                    amount: sufficient_balance(),
                },
                now,
                0,
            );
        }
    }

    /// Drives a range of a batch's cycles to `AwaitingDeposit` and leaves them there.
    ///
    /// The complement of [`fund_batch`]: this is the state a cycle is *lost* from, and a funded one
    /// cannot be — `DepositObserverClosed` on a recovering cycle is a no-op.
    fn commit_unfunded(
        sup: &mut SessionPixSupervisor,
        p: HoprPseudonym,
        indices: std::ops::RangeInclusive<u32>,
        now: Instant,
    ) {
        for idx in indices {
            let id = ssa_id(p, idx);
            sup.handle_event(&SessionPixEvent::SsaRequestSent(id), now, 0);
            sup.handle_event(&SessionPixEvent::CommitmentVerified(id), now, 0);
        }
    }

    fn _small_balance() -> HoprBalance {
        HoprBalance::new_base(1)
    }

    /// Each phase is timed from when it was entered to when it was left, and only on leaving.
    #[test]
    fn a_cycle_is_timed_in_each_phase_it_leaves() {
        let p = pseudonym();
        let t0 = Instant::now();
        let (mut sup, _) = SessionPixSupervisor::new(default_cfg(), dims(10, 5), p, t0);
        let id = ssa_id(p, 1);

        sup.handle_event(&SessionPixEvent::SsaRequestSent(id), t0, 0);
        assert!(
            sup.take_telemetry_events().phase_durations.is_empty(),
            "a cycle that has not left a phase has nothing to time"
        );

        // Three seconds awaiting the commitment, then seven awaiting the deposit.
        sup.handle_event(&SessionPixEvent::CommitmentVerified(id), t0 + Duration::from_secs(3), 0);
        assert_eq!(
            vec![(PixCyclePhase::AwaitingCommitment, 3_000)],
            sup.take_telemetry_events().phase_durations
        );

        sup.handle_event(
            &SessionPixEvent::DepositConfirmed {
                ssa_id: id,
                amount: sufficient_balance(),
            },
            t0 + Duration::from_secs(10),
            0,
        );
        assert_eq!(
            vec![(PixCyclePhase::AwaitingDeposit, 7_000)],
            sup.take_telemetry_events().phase_durations,
            "the clock restarts at each transition rather than measuring from the request"
        );

        // Recovery runs for a minute. The tombstone it leaves is not a phase, so nothing further is
        // timed for this cycle.
        sup.handle_event(&SessionPixEvent::Recovered(id), t0 + Duration::from_secs(70), 0);
        assert_eq!(
            vec![(PixCyclePhase::Recovering, 60_000)],
            sup.take_telemetry_events().phase_durations
        );
    }

    // ---------------------------------------------------------------
    // new / initial state
    // ---------------------------------------------------------------

    #[test]
    fn new_emits_initial_request_for_index_one() {
        let p = pseudonym();
        let (sup, actions) = SessionPixSupervisor::new(default_cfg(), dims(10, 5), p, Instant::now());

        assert_eq!(actions.len(), 1);
        match &actions[0] {
            SessionPixAction::RequestSsa { ssa_ids, params } => {
                assert_eq!(*params, dims(10, 5));
                assert_eq!(ssa_ids.len(), 1, "the default batch is a single SSA");
                assert_eq!(ssa_ids[0].ssa_index(), SsaIndex::new(1).unwrap());
            }
            other => panic!("expected RequestSsa, got {other:?}"),
        }

        assert_eq!(sup.next_ssa_index, 2);
        assert!(!sup.closed);

        // The SSA is tracked from the moment it is requested, so that an observation about it
        // cannot arrive before there is anywhere to record it.
        assert_eq!(sup.ssas.len(), 1);
        assert_eq!(sup.ssas[0].phase, SsaPhase::AwaitingCommitment);
        assert!(
            sup.ssas[0].commitment_deadline.is_none(),
            "the delivery clock starts when the request goes out, not when it is queued"
        );
    }

    /// A batch is allocated and asked for as a unit: one action, `ssas_per_request` contiguous
    /// indices, one per-SSA record each, and the index counter advanced past all of them.
    ///
    /// One action rather than N is what puts the whole batch in a single `SsaRequest`, which is the
    /// point of the knob — and what makes the Entry's per-message cap the thing that has to accept it.
    #[test]
    fn a_batch_is_allocated_and_requested_as_one_action() {
        const BATCH: usize = 4;

        let p = pseudonym();
        let cfg = SupervisorConfig {
            ssas_per_request: BATCH,
            ..default_cfg()
        };
        let (sup, actions) = SessionPixSupervisor::new(cfg, dims(10, 5), p, Instant::now());

        assert_eq!(actions.len(), 1, "the batch must travel as a single action");
        match &actions[0] {
            SessionPixAction::RequestSsa { ssa_ids, .. } => {
                assert_eq!(
                    ssa_ids.iter().map(|i| i.ssa_index().get()).collect::<Vec<_>>(),
                    (1..=BATCH as u32).collect::<Vec<_>>(),
                    "the batch must cover contiguous indices from 1"
                );
            }
            other => panic!("expected RequestSsa, got {other:?}"),
        }

        assert_eq!(
            sup.next_ssa_index,
            BATCH as u32 + 1,
            "the index counter must advance past the whole batch"
        );
        assert_eq!(
            sup.ssas.len(),
            BATCH,
            "every SSA in the batch needs its own record, or observations about it fail open"
        );
        assert!(sup.ssas.iter().all(|s| s.phase == SsaPhase::AwaitingCommitment));
        assert!(!sup.closed);
    }

    /// Both per-cycle deadlines are multiplied by the batch size.
    ///
    /// The commitment clock must scale because a batch's clocks all start together while the Entry has
    /// that many commitment sets to produce; the deposit clock must scale because an Entry funding a
    /// batch in order finishes the last one that many deposits after the first. Without either, a peer
    /// answering correctly but in sequence is closed for being slow.
    #[test]
    fn batching_scales_both_per_cycle_deadlines() {
        const BATCH: usize = 3;

        let p = pseudonym();
        let unit_commit = default_cfg().max_ssa_delivery_time;
        let unit_deposit = default_cfg().max_deposit_wait;
        let cfg = SupervisorConfig {
            ssas_per_request: BATCH,
            ..default_cfg()
        };
        let now = Instant::now();
        let (mut sup, actions) = SessionPixSupervisor::new(cfg, dims(10, 5), p, now);
        let SessionPixAction::RequestSsa { ssa_ids, .. } = &actions[0] else {
            panic!("expected RequestSsa");
        };
        let ssa_ids = ssa_ids.clone();

        // Commitment clock: armed per index when the request goes out, at the scaled duration.
        for id in &ssa_ids {
            sup.handle_event(&SessionPixEvent::SsaRequestSent(*id), now, 0);
        }
        for (i, id) in ssa_ids.iter().enumerate() {
            let idx = sup.find_ssa_idx(id).expect("record must exist");
            assert_eq!(
                sup.ssas[idx].commitment_deadline,
                Some(now + BATCH as u32 * unit_commit),
                "cycle {i} must get the batch-scaled commitment deadline"
            );
        }

        // Deposit clock: armed when that cycle's commitment verifies, also at the scaled duration.
        let later = now + Duration::from_secs(1);
        sup.handle_event(&SessionPixEvent::CommitmentVerified(ssa_ids[0]), later, 0);
        let idx = sup.find_ssa_idx(&ssa_ids[0]).expect("record must exist");
        assert_eq!(sup.ssas[idx].phase, SsaPhase::AwaitingDeposit);
        assert_eq!(
            sup.ssas[idx].deposit_deadline,
            Some(later + BATCH as u32 * unit_deposit),
            "the deposit deadline must be batch-scaled too"
        );
    }

    /// An unvalidated `ssas_per_request` must not reach the deadline arithmetic or the allocator.
    ///
    /// A supervisor can be built from a config that never went through `validate_pix_supervision`, so
    /// zero must still request one SSA rather than none, and a value above the ceiling must be clamped
    /// rather than allocating an arbitrary batch or scaling a deadline past the duration cap.
    #[test]
    fn an_unvalidated_batch_size_is_clamped() {
        for (configured, expected) in [
            (0usize, 1usize),
            (crate::MAX_SSA_BATCH_SIZE + 5, crate::MAX_SSA_BATCH_SIZE),
        ] {
            let p = pseudonym();
            let cfg = SupervisorConfig {
                ssas_per_request: configured,
                ..default_cfg()
            };
            let (sup, actions) = SessionPixSupervisor::new(cfg, dims(10, 5), p, Instant::now());
            match &actions[0] {
                SessionPixAction::RequestSsa { ssa_ids, .. } => assert_eq!(
                    ssa_ids.len(),
                    expected,
                    "a configured batch of {configured} must be clamped to {expected}"
                ),
                other => panic!("expected RequestSsa, got {other:?}"),
            }
            assert_eq!(sup.ssas.len(), expected);
        }
    }

    // ---------------------------------------------------------------
    // SsaRequestSent
    // ---------------------------------------------------------------

    #[test]
    fn request_sent_starts_commitment_deadline() {
        let p = pseudonym();
        let (mut sup, _) = SessionPixSupervisor::new(default_cfg(), dims(10, 5), p, Instant::now());
        let now = Instant::now();
        let id = ssa_id(p, 1);

        let actions = sup.handle_event(&SessionPixEvent::SsaRequestSent(id), now, 0);
        assert!(actions.is_empty());

        let ssa = sup.ssas.iter().find(|s| s.ssa_id == id).unwrap();
        assert_eq!(ssa.phase, SsaPhase::AwaitingCommitment);
        assert!(ssa.commitment_deadline.is_some());
    }

    #[test]
    fn request_sent_is_idempotent() {
        let p = pseudonym();
        let (mut sup, _) = SessionPixSupervisor::new(default_cfg(), dims(10, 5), p, Instant::now());
        let now = Instant::now();
        let id = ssa_id(p, 1);

        sup.handle_event(&SessionPixEvent::SsaRequestSent(id), now, 0);
        let actions = sup.handle_event(&SessionPixEvent::SsaRequestSent(id), now, 0);
        assert!(actions.is_empty());
        assert_eq!(sup.ssas.len(), 1);
    }

    #[test]
    fn request_sent_wrong_pseudonym_closes() {
        let p = pseudonym();
        let (mut sup, _) = SessionPixSupervisor::new(default_cfg(), dims(10, 5), p, Instant::now());
        let wrong_p = pseudonym();
        let now = Instant::now();

        let actions = sup.handle_event(&SessionPixEvent::SsaRequestSent(ssa_id(wrong_p, 1)), now, 0);
        assert_eq!(actions.len(), 1);
        assert!(matches!(actions[0], SessionPixAction::Close(_)));
    }

    // ---------------------------------------------------------------
    // CommitmentVerified
    // ---------------------------------------------------------------

    #[test]
    fn commitment_verified_starts_the_deposit_deadline() {
        let p = pseudonym();
        let (mut sup, _) = SessionPixSupervisor::new(default_cfg(), dims(10, 5), p, Instant::now());
        let now = Instant::now();
        let id = ssa_id(p, 1);

        sup.handle_event(&SessionPixEvent::SsaRequestSent(id), now, 0);

        let actions = sup.handle_event(&SessionPixEvent::CommitmentVerified(id), now, 0);
        assert!(actions.is_empty());

        let ssa = sup.ssas.iter().find(|s| s.ssa_id == id).unwrap();
        assert_eq!(ssa.phase, SsaPhase::AwaitingDeposit);
        assert!(ssa.deposit_deadline.is_some());
        assert!(ssa.commitment_deadline.is_none());
    }

    #[test]
    fn commitment_verified_is_idempotent() {
        let p = pseudonym();
        let (mut sup, _) = SessionPixSupervisor::new(default_cfg(), dims(10, 5), p, Instant::now());
        let now = Instant::now();
        let id = ssa_id(p, 1);

        sup.handle_event(&SessionPixEvent::SsaRequestSent(id), now, 0);
        sup.handle_event(&SessionPixEvent::CommitmentVerified(id), now, 0);
        let actions = sup.handle_event(&SessionPixEvent::CommitmentVerified(id), now, 0);
        assert!(actions.is_empty());
        assert_eq!(
            sup.ssas.iter().find(|s| s.ssa_id == id).unwrap().phase,
            SsaPhase::AwaitingDeposit
        );
    }

    // ---------------------------------------------------------------
    // DepositConfirmed
    // ---------------------------------------------------------------

    #[test]
    fn sufficient_deposit_enters_recovering_starts_idle_and_hard() {
        let p = pseudonym();
        let (mut sup, _) = SessionPixSupervisor::new(default_cfg(), dims(10, 5), p, Instant::now());
        let now = Instant::now();
        let id = ssa_id(p, 1);

        sup.handle_event(&SessionPixEvent::SsaRequestSent(id), now, 0);
        sup.handle_event(&SessionPixEvent::CommitmentVerified(id), now, 0);

        let actions = sup.handle_event(
            &SessionPixEvent::DepositConfirmed {
                ssa_id: id,
                amount: sufficient_balance(),
            },
            now,
            0,
        );

        assert_eq!(actions.len(), 1);
        assert!(matches!(actions[0], SessionPixAction::ReleaseService));

        let ssa = sup.ssas.iter().find(|s| s.ssa_id == id).unwrap();
        assert_eq!(ssa.phase, SsaPhase::Recovering);
        assert!(ssa.recovery_idle_deadline.is_some());
        assert!(ssa.recovery_hard_deadline.is_some());
        assert!(ssa.deposit_deadline.is_none());
    }

    #[test]
    fn first_funding_emits_release_service_once() {
        let p = pseudonym();
        let (mut sup, _) = SessionPixSupervisor::new(default_cfg(), dims(10, 5), p, Instant::now());
        let now = Instant::now();
        let id1 = ssa_id(p, 1);

        sup.handle_event(&SessionPixEvent::SsaRequestSent(id1), now, 0);
        sup.handle_event(&SessionPixEvent::CommitmentVerified(id1), now, 0);

        let actions = sup.handle_event(
            &SessionPixEvent::DepositConfirmed {
                ssa_id: id1,
                amount: sufficient_balance(),
            },
            now,
            0,
        );
        assert_eq!(actions.len(), 1);
        assert!(matches!(actions[0], SessionPixAction::ReleaseService));

        // Second deposit on a different SSA should not emit ReleaseService again.
        sup.handle_event(&SessionPixEvent::SsaRequestSent(ssa_id(p, 2)), now, 0);
        sup.handle_event(&SessionPixEvent::CommitmentVerified(ssa_id(p, 2)), now, 0);
        let actions = sup.handle_event(
            &SessionPixEvent::DepositConfirmed {
                ssa_id: ssa_id(p, 2),
                amount: sufficient_balance(),
            },
            now,
            0,
        );
        assert!(actions.iter().all(|a| !matches!(a, SessionPixAction::ReleaseService)));
    }

    #[test]
    fn a_funded_successor_gets_a_fresh_service_ceiling_when_it_reaches_the_front() {
        let mut cfg = default_cfg();
        cfg.ssas_per_request = 2;
        cfg.max_served_without_progress = 2;
        let p = pseudonym();
        let now = Instant::now();
        let (mut sup, _) = SessionPixSupervisor::new(cfg, dims(10, 5), p, now);

        fund_batch(&mut sup, p, 2, now);
        let actions = sup.handle_event(&SessionPixEvent::Recovered(ssa_id(p, 1)), now, 0);
        assert!(
            matches!(actions.as_slice(), [SessionPixAction::ProgressNotification]),
            "recovery resets the existing gate but is not the FIFO boundary: {actions:?}"
        );

        let successor = sup.ssas.iter().find(|ssa| ssa.ssa_id == ssa_id(p, 2)).unwrap();
        assert_eq!(
            (successor.recovery_idle_deadline, successor.recovery_hard_deadline),
            (None, None),
            "successor clocks must not run while every served SURB still belongs to its predecessor"
        );

        let target = sup.dims.target_useful_shares();
        let max_seen = max_cycle_shares(&sup);
        for (offset, seen) in ((target + 1)..=max_seen).enumerate() {
            let at = now + Duration::from_secs(offset as u64 + 1);
            let actions = sup.handle_event(
                &SessionPixEvent::RecoveryProgress(make_recovered_tail_progress(&sup, ssa_id(p, 1), seen)),
                at,
                offset as u64 + 1,
            );
            assert!(
                actions
                    .iter()
                    .any(|action| matches!(action, SessionPixAction::ProgressNotification)),
                "buffered paid share {seen} must refresh liveness: {actions:?}"
            );
            assert!(
                actions
                    .iter()
                    .all(|action| !matches!(action, SessionPixAction::Close(_))),
                "a tail longer than the gate ceiling must not close an honest Session: {actions:?}"
            );
            if seen == max_seen {
                assert!(
                    actions
                        .iter()
                        .any(|action| matches!(action, SessionPixAction::ReleaseService)),
                    "the funded successor must receive a fresh service ceiling at the exact FIFO boundary: {actions:?}"
                );
            }
        }

        assert!(
            sup.paid_recovery_tail.is_none(),
            "the exact negotiated tail bound must finish the handoff"
        );
        let successor = sup.ssas.iter().find(|ssa| ssa.ssa_id == ssa_id(p, 2)).unwrap();
        assert!(
            successor.recovery_idle_deadline.is_some() && successor.recovery_hard_deadline.is_some(),
            "successor clocks start only after the final predecessor-bound SURB"
        );
    }

    /// Any non-zero amount funds the cycle, because the amount is not the supervisor's to judge.
    ///
    /// The pool prices the quota and sends only once the deposit clears that price, so a confirmation
    /// is a verdict rather than evidence. A dust amount arriving here means the pool decided dust was
    /// enough — a pool bug, or an operator's deliberate pricing, and either way not something a floor
    /// in this crate could correct without overruling the component that knows the price.
    #[test]
    fn any_nonzero_deposit_funds_the_cycle() {
        for amount in [HoprBalance::new_base(1), sufficient_balance()] {
            let p = pseudonym();
            let (mut sup, _) = SessionPixSupervisor::new(default_cfg(), dims(10, 5), p, Instant::now());
            let now = Instant::now();
            let id = ssa_id(p, 1);

            sup.handle_event(&SessionPixEvent::SsaRequestSent(id), now, 0);
            sup.handle_event(&SessionPixEvent::CommitmentVerified(id), now, 0);

            let actions = sup.handle_event(&SessionPixEvent::DepositConfirmed { ssa_id: id, amount }, now, 0);

            assert_eq!(
                sup.ssas.iter().find(|s| s.ssa_id == id).unwrap().phase,
                SsaPhase::Recovering,
                "a confirmation of {amount} is the pool's verdict and must fund the cycle"
            );
            assert!(!actions.is_empty());
        }
    }

    /// A zero balance is the one confirmation that asserts nothing, and must not release service.
    ///
    /// It also must not consume the cycle's chance to be funded: the deadline keeps running, and a
    /// later confirmation still counts. That is what makes it safe for the observer to forward every
    /// message the pool sends rather than latching on the first.
    #[test]
    fn a_zero_deposit_is_not_a_verdict_and_leaves_the_cycle_fundable() {
        let p = pseudonym();
        let (mut sup, _) = SessionPixSupervisor::new(default_cfg(), dims(10, 5), p, Instant::now());
        let now = Instant::now();
        let id = ssa_id(p, 1);

        sup.handle_event(&SessionPixEvent::SsaRequestSent(id), now, 0);
        sup.handle_event(&SessionPixEvent::CommitmentVerified(id), now, 0);

        let deadline_before = sup.ssas.iter().find(|s| s.ssa_id == id).unwrap().deposit_deadline;

        let actions = sup.handle_event(
            &SessionPixEvent::DepositConfirmed {
                ssa_id: id,
                amount: HoprBalance::zero(),
            },
            now,
            0,
        );
        assert!(actions.is_empty());

        let ssa = sup.ssas.iter().find(|s| s.ssa_id == id).unwrap();
        assert_eq!(ssa.phase, SsaPhase::AwaitingDeposit);
        assert_eq!(
            ssa.deposit_deadline, deadline_before,
            "a zero confirmation must not extend or clear the deadline"
        );

        // The next confirmation still funds it.
        let actions = sup.handle_event(
            &SessionPixEvent::DepositConfirmed {
                ssa_id: id,
                amount: sufficient_balance(),
            },
            now,
            0,
        );
        assert_eq!(
            sup.ssas.iter().find(|s| s.ssa_id == id).unwrap().phase,
            SsaPhase::Recovering
        );
        assert!(!actions.is_empty());
    }

    /// Funding a queued cycle must not release service while an unfunded predecessor is live, but
    /// that funding must take effect if the predecessor is later retired. Otherwise the one-shot
    /// Session gate remains in predeposit mode even though the newly promoted front is funded, and
    /// no further deposit event exists to wake it.
    #[test]
    fn funded_successor_releases_service_when_unfunded_front_is_retired() {
        let mut cfg = default_cfg();
        cfg.ssas_per_request = 2;
        let p = pseudonym();
        let now = Instant::now();
        let (mut sup, _) = SessionPixSupervisor::new(cfg, dims(10, 5), p, now);
        let front = ssa_id(p, 1);
        let successor = ssa_id(p, 2);

        for id in [front, successor] {
            sup.handle_event(&SessionPixEvent::SsaRequestSent(id), now, 0);
            sup.handle_event(&SessionPixEvent::CommitmentVerified(id), now, 0);
        }

        let actions = sup.handle_event(
            &SessionPixEvent::DepositConfirmed {
                ssa_id: successor,
                amount: sufficient_balance(),
            },
            now,
            0,
        );
        assert!(
            actions.iter().all(|a| !matches!(a, SessionPixAction::ReleaseService)),
            "funding a cycle behind an unfunded front must not release service"
        );

        let actions = sup.handle_event(&SessionPixEvent::DepositObserverClosed(front), now, 0);
        assert!(
            actions.iter().any(|a| matches!(a, SessionPixAction::ReleaseService)),
            "retiring the unfunded front must release service for its already-funded successor; got {actions:?}"
        );
    }

    /// A cycle that funds and goes terminal in the same call must release only its own paid tail.
    ///
    /// `Recovered` can arrive while the cycle is still `AwaitingDeposit` — the Exit finishes
    /// reconstructing before the chain observer reports the deposit — and is deferred until the
    /// deposit lands. That one call then both funds the cycle and retires its heavy state, but the
    /// buffered SURBs were included in its paid quota. The gate stays funded only through that
    /// bounded tail, then restores the unfunded successor's predeposit allowance.
    #[test]
    fn a_cycle_that_funds_and_recovers_at_once_rearms_for_its_unfunded_successor() {
        let mut cfg = default_cfg();
        cfg.ssas_per_request = 2;
        let p = pseudonym();
        let now = Instant::now();
        let (mut sup, _) = SessionPixSupervisor::new(cfg, dims(10, 5), p, now);
        let front = ssa_id(p, 1);
        let successor = ssa_id(p, 2);

        for id in [front, successor] {
            sup.handle_event(&SessionPixEvent::SsaRequestSent(id), now, 0);
            sup.handle_event(&SessionPixEvent::CommitmentVerified(id), now, 0);
        }

        // Recovery completes before the deposit is observed, so the event is held.
        let actions = sup.handle_event(&SessionPixEvent::Recovered(front), now, 0);
        assert!(
            actions.iter().all(|a| !matches!(a, SessionPixAction::ReleaseService)),
            "recovering an unfunded cycle must not release service"
        );

        // The deposit lands: the cycle funds and drops its heavy state in the same call. The compact
        // predecessor receipt may now release only the SURBs already paid by that deposit.
        let actions = sup.handle_event(
            &SessionPixEvent::DepositConfirmed {
                ssa_id: front,
                amount: sufficient_balance(),
            },
            now,
            0,
        );
        assert!(
            matches!(
                actions.as_slice(),
                [SessionPixAction::ReleaseService, SessionPixAction::ProgressNotification]
            ),
            "the paid handoff must open only the bounded predecessor tail, got {actions:?}"
        );

        let max_seen = max_cycle_shares(&sup);
        let actions = sup.handle_event(
            &SessionPixEvent::RecoveryProgress(make_recovered_tail_progress(&sup, front, max_seen)),
            now,
            max_seen,
        );
        assert!(
            actions
                .iter()
                .any(|action| matches!(action, SessionPixAction::WithholdService)),
            "exhausting the paid tail must restore the unfunded successor's allowance: {actions:?}"
        );

        let actions = sup.handle_event(
            &SessionPixEvent::DepositConfirmed {
                ssa_id: successor,
                amount: sufficient_balance(),
            },
            now,
            0,
        );
        assert!(
            actions.iter().any(|a| matches!(a, SessionPixAction::ReleaseService)),
            "the successor's own deposit must open service, got {actions:?}"
        );
    }

    #[test]
    fn progress_on_an_unfunded_front_does_not_notify_the_service_gate() {
        let mut cfg = default_cfg();
        cfg.ssas_per_request = 2;
        cfg.max_off_front_share_fraction = 1.0;
        let p = pseudonym();
        let now = Instant::now();
        let (mut sup, _) = SessionPixSupervisor::new(cfg, dims(10, 5), p, now);
        let funded_front = ssa_id(p, 1);
        let unfunded_successor = ssa_id(p, 2);

        for id in [funded_front, unfunded_successor] {
            sup.handle_event(&SessionPixEvent::SsaRequestSent(id), now, 0);
            sup.handle_event(&SessionPixEvent::CommitmentVerified(id), now, 0);
        }
        sup.handle_event(
            &SessionPixEvent::DepositConfirmed {
                ssa_id: funded_front,
                amount: sufficient_balance(),
            },
            now,
            0,
        );
        sup.handle_event(&SessionPixEvent::Recovered(funded_front), now, 0);

        let actions = sup.handle_event(
            &SessionPixEvent::RecoveryProgress(make_progress(
                unfunded_successor,
                1,
                sup.dims.target_useful_shares(),
                0,
            )),
            now,
            1,
        );

        assert!(
            actions
                .iter()
                .all(|action| !matches!(action, SessionPixAction::ProgressNotification)),
            "an unfunded front must not reopen the funded service ceiling, got {actions:?}"
        );
        assert!(
            actions
                .iter()
                .any(|action| matches!(action, SessionPixAction::WithholdService)),
            "the first successor-bound packet must close the predecessor's funded gate: {actions:?}"
        );
        assert!(
            sup.paid_recovery_tail.is_none(),
            "an unfunded successor must not trick the Exit into spending its predecessor's remaining allowance"
        );
    }

    #[test]
    fn progress_behind_the_front_does_not_notify_the_service_gate() {
        let mut cfg = default_cfg();
        cfg.ssas_per_request = 2;
        cfg.max_off_front_share_fraction = 1.0;
        let p = pseudonym();
        let now = Instant::now();
        let (mut sup, _) = SessionPixSupervisor::new(cfg, dims(10, 5), p, now);

        fund_batch(&mut sup, p, 2, now);
        let actions = sup.handle_event(
            &SessionPixEvent::RecoveryProgress(make_progress(ssa_id(p, 2), 1, sup.dims.target_useful_shares(), 0)),
            now,
            1,
        );

        assert!(
            actions.is_empty(),
            "off-front progress must not reopen the gate, got {actions:?}"
        );
    }

    /// A live cycle below the retired watermark must still accept its own request event.
    ///
    /// Retirement is not monotone: a later batch member can lose its deposit — timeout, observer
    /// closure — while an earlier one is still recovering, which puts the watermark *above* a live
    /// index. A staleness guard consulted before the record lookup would then silently drop the
    /// earlier cycle's `SsaRequestSent`, leaving its commitment deadline unarmed and the Entry free
    /// to never answer. Looking the record up first is what removes the assumption.
    #[test]
    fn a_live_cycle_below_the_retired_watermark_still_arms_its_deadline() {
        let mut cfg = default_cfg();
        cfg.ssas_per_request = 2;
        let p = pseudonym();
        let now = Instant::now();
        let (mut sup, _) = SessionPixSupervisor::new(cfg, dims(10, 5), p, now);
        let earlier = ssa_id(p, 1);
        let later = ssa_id(p, 2);

        // The later member is retired first, which puts the watermark above the earlier one.
        sup.handle_event(&SessionPixEvent::SsaRequestSent(later), now, 0);
        sup.handle_event(&SessionPixEvent::CommitmentVerified(later), now, 0);
        sup.handle_event(&SessionPixEvent::DepositObserverClosed(later), now, 0);
        assert_eq!(
            Some(later.ssa_index()),
            sup.highest_retired_ssa_index,
            "the later member must have set the watermark"
        );

        // The earlier member is still live, and its request event must still arm the clock.
        sup.handle_event(&SessionPixEvent::SsaRequestSent(earlier), now, 0);
        let armed = sup
            .ssas
            .iter()
            .find(|s| s.ssa_id == earlier)
            .expect("the earlier cycle must still be live")
            .commitment_deadline;
        assert!(
            armed.is_some(),
            "a live cycle below the watermark must still have its commitment deadline armed"
        );
    }

    /// And a straggler for an index that really is gone stays benign rather than closing the Session.
    ///
    /// A batch of two, so retiring one member leaves the Session alive — retiring the last live cycle
    /// closes it outright, and a closed supervisor ignores everything, which would make the second
    /// half of this test vacuous.
    #[test]
    fn a_request_event_for_a_retired_index_is_ignored() {
        let mut cfg = default_cfg();
        cfg.ssas_per_request = 2;
        let p = pseudonym();
        let now = Instant::now();
        let (mut sup, _) = SessionPixSupervisor::new(cfg, dims(10, 5), p, now);
        let retired = ssa_id(p, 1);

        sup.handle_event(&SessionPixEvent::SsaRequestSent(retired), now, 0);
        sup.handle_event(&SessionPixEvent::CommitmentVerified(retired), now, 0);
        sup.handle_event(&SessionPixEvent::DepositObserverClosed(retired), now, 0);
        assert!(!sup.closed, "the second batch member must keep the Session alive");

        let actions = sup.handle_event(&SessionPixEvent::SsaRequestSent(retired), now, 0);
        assert!(
            actions.iter().all(|a| !matches!(a, SessionPixAction::Close(_))),
            "a straggling request for a retired index must be ignored, not fatal; got {actions:?}"
        );

        // An index we were never asked for is a different matter — above the watermark and with no
        // record, it can only be an SSA this supervisor never registered.
        let actions = sup.handle_event(&SessionPixEvent::SsaRequestSent(ssa_id(p, 9)), now, 0);
        assert!(
            actions
                .iter()
                .any(|a| matches!(a, SessionPixAction::Close(SessionPixCloseReason::InvalidTransition))),
            "an index above the watermark with no record is a protocol fault; got {actions:?}"
        );
    }

    #[test]
    fn duplicate_deposit_confirmation_is_idempotent() {
        let p = pseudonym();
        let (mut sup, _) = SessionPixSupervisor::new(default_cfg(), dims(10, 5), p, Instant::now());
        let now = Instant::now();
        let id = ssa_id(p, 1);

        sup.handle_event(&SessionPixEvent::SsaRequestSent(id), now, 0);
        sup.handle_event(&SessionPixEvent::CommitmentVerified(id), now, 0);

        sup.handle_event(
            &SessionPixEvent::DepositConfirmed {
                ssa_id: id,
                amount: sufficient_balance(),
            },
            now,
            0,
        );

        let actions = sup.handle_event(
            &SessionPixEvent::DepositConfirmed {
                ssa_id: id,
                amount: sufficient_balance(),
            },
            now,
            0,
        );
        assert!(actions.is_empty());
    }

    #[test]
    fn wrong_ssa_deposit_does_not_transition_any_ssa() {
        let p = pseudonym();
        let (mut sup, _) = SessionPixSupervisor::new(default_cfg(), dims(10, 5), p, Instant::now());
        let now = Instant::now();

        sup.handle_event(&SessionPixEvent::SsaRequestSent(ssa_id(p, 1)), now, 0);

        let actions = sup.handle_event(
            &SessionPixEvent::DepositConfirmed {
                ssa_id: ssa_id(p, 99),
                amount: sufficient_balance(),
            },
            now,
            0,
        );
        assert!(actions.is_empty());
    }

    // ---------------------------------------------------------------
    // Deadlines
    // ---------------------------------------------------------------

    #[test]
    fn commitment_deadline_expiry_closes_awaiting_commitment() {
        let p = pseudonym();
        let (mut sup, _) = SessionPixSupervisor::new(default_cfg(), dims(10, 5), p, Instant::now());
        let start = Instant::now();
        let id = ssa_id(p, 1);

        sup.handle_event(&SessionPixEvent::SsaRequestSent(id), start, 0);

        let actions = sup.handle_deadline(start + Duration::from_secs(10), 0);
        assert!(actions.is_empty());

        let actions = sup.handle_deadline(start + Duration::from_secs(21), 0);
        assert_eq!(actions.len(), 1);
        assert!(matches!(
            actions[0],
            SessionPixAction::Close(SessionPixCloseReason::CommitmentTimeout)
        ));
    }

    /// A commitment that is under way but has gone quiet is re-requested, repeatedly, without ever
    /// being closed for it — and the absolute deadline is still what ends the cycle.
    #[test]
    fn a_stalled_commitment_is_re_requested_until_the_delivery_deadline() {
        let p = pseudonym();
        let cfg = default_cfg();
        let (mut sup, _) = SessionPixSupervisor::new(cfg.clone(), dims(10, 5), p, Instant::now());
        let start = Instant::now();
        let id = ssa_id(p, 1);

        sup.handle_event(&SessionPixEvent::SsaRequestSent(id), start, 0);

        // Nothing has arrived, so there is no scope to ask for and no timer to expire.
        assert!(
            sup.handle_deadline(start + cfg.commitment_recommit_interval * 3, 0)
                .is_empty(),
            "a commitment nothing was delivered for must not be re-requested"
        );

        // The first part arrives; the timer is armed from *that* instant.
        let first_part = start + Duration::from_secs(1);
        sup.handle_event(&SessionPixEvent::CommitmentProgress(id), first_part, 0);
        assert!(
            sup.handle_deadline(first_part + cfg.commitment_recommit_interval / 2, 0)
                .is_empty(),
            "the interval must be measured from the last part that arrived"
        );

        // More of the burst lands, pushing the timer out again: the first ask must not land in the
        // middle of delivery.
        let last_part = first_part + cfg.commitment_recommit_interval;
        sup.handle_event(&SessionPixEvent::CommitmentProgress(id), last_part, 0);
        assert!(
            sup.handle_deadline(last_part + cfg.commitment_recommit_interval / 2, 0)
                .is_empty(),
            "an arriving part must re-arm the timer"
        );

        // Delivery stalls. Every interval from here produces one ask, and no close, for as long as
        // the delivery deadline has room — at the shipped 3 s inside 20 s that is five of them, so
        // the count cap is not what binds at the defaults.
        let delivery_deadline =
            start + crate::supervision::scaled_deadline(cfg.max_ssa_delivery_time, cfg.ssas_per_request);
        let mut now = last_part;
        let mut asks = 0;
        while now + cfg.commitment_recommit_interval < delivery_deadline {
            now += cfg.commitment_recommit_interval;
            let actions = sup.handle_deadline(now, 0);
            assert!(
                matches!(
                    actions.as_slice(),
                    [SessionPixAction::RequestCommitmentRetransmission { ssa_id, params }]
                        if *ssa_id == id && *params == dims(10, 5)
                ),
                "ask {} must name the missing parts, got {actions:?}",
                asks + 1
            );
            asks += 1;
        }
        assert!(asks > 1, "the interval must fit several asks inside the deadline");

        // The delivery deadline is untouched by any of it, and remains the only thing that ends the
        // cycle.
        let actions = sup.handle_deadline(delivery_deadline, 0);
        assert!(matches!(
            actions.as_slice(),
            [SessionPixAction::Close(SessionPixCloseReason::CommitmentTimeout)]
        ));
    }

    /// Asks stop at [`MAX_COMMITMENT_RETRANSMISSIONS`], which is what the Entry will answer.
    ///
    /// The delivery deadline bounds the asks in *time*, and that is what protects the Entry — but the
    /// packets they cost are the Exit's own, and a short interval inside a long deadline leaves time
    /// for far more of them than any Entry answers. Hence a count.
    #[test]
    fn asks_for_a_stalled_commitment_are_capped_by_count() {
        let p = pseudonym();
        // Short enough that the whole budget is spent well inside the delivery deadline, so it is the
        // count and not the clock that stops it.
        let cfg = SupervisorConfig {
            commitment_recommit_interval: Duration::from_secs(1),
            ..default_cfg()
        };
        let (mut sup, _) = SessionPixSupervisor::new(cfg.clone(), dims(10, 5), p, Instant::now());
        let start = Instant::now();
        let id = ssa_id(p, 1);

        sup.handle_event(&SessionPixEvent::SsaRequestSent(id), start, 0);
        sup.handle_event(&SessionPixEvent::CommitmentProgress(id), start, 0);

        let mut now = start;
        for attempt in 1..=hopr_protocol_pix::MAX_COMMITMENT_RETRANSMISSIONS {
            now += cfg.commitment_recommit_interval;
            let actions = sup.handle_deadline(now, 0);
            assert!(
                matches!(
                    actions.as_slice(),
                    [SessionPixAction::RequestCommitmentRetransmission { ssa_id, .. }] if *ssa_id == id
                ),
                "attempt {attempt} must ask for the missing parts, got {actions:?}"
            );
        }

        now += cfg.commitment_recommit_interval;
        assert!(
            sup.handle_deadline(now, 0).is_empty(),
            "the ask budget must be spent, with the delivery deadline still far off"
        );
        assert_eq!(
            Some(start + crate::supervision::scaled_deadline(cfg.max_ssa_delivery_time, cfg.ssas_per_request)),
            sup.next_deadline(),
            "a spent budget must leave no timer for the worker to wake on"
        );

        // Further parts of the burst cannot buy more asks either.
        sup.handle_event(&SessionPixEvent::CommitmentProgress(id), now, 0);
        assert!(
            sup.handle_deadline(now + cfg.commitment_recommit_interval * 2, 0)
                .is_empty(),
            "an arriving part must not re-open a spent budget"
        );
    }

    /// The re-request timer is the earliest deadline of the phase it runs in, so the worker has to be
    /// woken by it — and it must stop existing the moment the commitment completes.
    #[test]
    fn the_re_request_timer_leads_the_deadlines_and_ends_with_the_commitment() {
        let p = pseudonym();
        let cfg = default_cfg();
        let (mut sup, _) = SessionPixSupervisor::new(cfg.clone(), dims(10, 5), p, Instant::now());
        let start = Instant::now();
        let id = ssa_id(p, 1);

        sup.handle_event(&SessionPixEvent::SsaRequestSent(id), start, 0);
        sup.handle_event(&SessionPixEvent::CommitmentProgress(id), start, 0);
        assert_eq!(
            Some(start + cfg.commitment_recommit_interval),
            sup.next_deadline(),
            "the worker must wake for the re-request before the delivery deadline"
        );

        sup.handle_event(&SessionPixEvent::CommitmentVerified(id), start, 0);
        assert_eq!(
            Some(start + crate::supervision::scaled_deadline(cfg.max_deposit_wait, cfg.ssas_per_request)),
            sup.next_deadline(),
            "a completed commitment leaves only the deposit clock"
        );
        assert!(
            sup.handle_deadline(start + cfg.commitment_recommit_interval, 0)
                .is_empty(),
            "nothing may be re-requested once the commitment verified"
        );

        // A straggling part of the burst arrives after completion — which retransmission makes
        // routine — and must not resurrect the timer.
        sup.handle_event(
            &SessionPixEvent::CommitmentProgress(id),
            start + cfg.commitment_recommit_interval,
            0,
        );
        assert!(
            sup.handle_deadline(start + cfg.commitment_recommit_interval * 3, 0)
                .is_empty(),
            "a late part must not re-arm a timer the phase has left"
        );
    }

    #[test]
    fn deposit_deadline_expiry_closes_awaiting_deposit() {
        let p = pseudonym();
        let (mut sup, _) = SessionPixSupervisor::new(default_cfg(), dims(10, 5), p, Instant::now());
        let start = Instant::now();
        let id = ssa_id(p, 1);

        sup.handle_event(&SessionPixEvent::SsaRequestSent(id), start, 0);
        sup.handle_event(&SessionPixEvent::CommitmentVerified(id), start, 0);

        let actions = sup.handle_deadline(start + Duration::from_secs(61), 0);
        assert_eq!(actions.len(), 1);
        assert!(matches!(
            actions[0],
            SessionPixAction::Close(SessionPixCloseReason::DepositTimeout)
        ));
    }

    // ---------------------------------------------------------------
    // DepositObserverClosed
    // ---------------------------------------------------------------

    #[test]
    fn deposit_observer_closed_closes_with_distinct_reason() {
        let p = pseudonym();
        let (mut sup, _) = SessionPixSupervisor::new(default_cfg(), dims(10, 5), p, Instant::now());
        let now = Instant::now();
        let id = ssa_id(p, 1);

        sup.handle_event(&SessionPixEvent::SsaRequestSent(id), now, 0);
        sup.handle_event(&SessionPixEvent::CommitmentVerified(id), now, 0);

        let actions = sup.handle_event(&SessionPixEvent::DepositObserverClosed(id), now, 0);
        assert_eq!(actions.len(), 1);
        assert!(matches!(
            actions[0],
            SessionPixAction::Close(SessionPixCloseReason::DepositObserverClosed)
        ));
    }

    // ---------------------------------------------------------------
    // Stale timer safety
    // ---------------------------------------------------------------

    #[test]
    fn stale_timer_wake_after_transition_does_not_close() {
        let p = pseudonym();
        let (mut sup, _) = SessionPixSupervisor::new(default_cfg(), dims(10, 5), p, Instant::now());
        let start = Instant::now();
        let id = ssa_id(p, 1);

        sup.handle_event(&SessionPixEvent::SsaRequestSent(id), start, 0);
        sup.handle_event(&SessionPixEvent::CommitmentVerified(id), start, 0);

        // Deposit arrives before deadline.
        sup.handle_event(
            &SessionPixEvent::DepositConfirmed {
                ssa_id: id,
                amount: sufficient_balance(),
            },
            start,
            0,
        );

        let actions = sup.handle_deadline(start + Duration::from_secs(120), 0);
        assert!(
            actions
                .iter()
                .all(|a| { !matches!(a, SessionPixAction::Close(SessionPixCloseReason::DepositTimeout)) })
        );
    }

    // ---------------------------------------------------------------
    // Progress
    // ---------------------------------------------------------------

    #[test]
    fn useful_progress_extends_idle_only_hard_never_moves() {
        let p = pseudonym();
        let mut cfg = default_cfg();
        cfg.max_recovery_idle = Duration::from_secs(60);
        let (mut sup, _) = SessionPixSupervisor::new(cfg, dims(10, 5), p, Instant::now());
        let start = Instant::now();
        let id = ssa_id(p, 1);

        sup.handle_event(&SessionPixEvent::SsaRequestSent(id), start, 0);
        sup.handle_event(&SessionPixEvent::CommitmentVerified(id), start, 0);
        sup.handle_event(
            &SessionPixEvent::DepositConfirmed {
                ssa_id: id,
                amount: sufficient_balance(),
            },
            start,
            0,
        );

        let hard_before = sup.ssas.iter().find(|s| s.ssa_id == id).unwrap().recovery_hard_deadline;
        let idle_before = sup.ssas.iter().find(|s| s.ssa_id == id).unwrap().recovery_idle_deadline;

        sup.handle_event(
            &SessionPixEvent::RecoveryProgress(make_progress(id, 10, 50, 1)),
            start + Duration::from_secs(55),
            5,
        );

        let ssa = sup.ssas.iter().find(|s| s.ssa_id == id).unwrap();
        assert_eq!(ssa.recovery_hard_deadline, hard_before);
        assert!(ssa.recovery_idle_deadline > idle_before);
    }

    #[test]
    fn hard_deadline_immutable_under_trickle_progress() {
        let p = pseudonym();
        let mut cfg = default_cfg();
        cfg.max_recovery_idle = Duration::from_secs(10);
        cfg.max_recovery_time = Duration::from_secs(30);
        let (mut sup, _) = SessionPixSupervisor::new(cfg, dims(10, 5), p, Instant::now());
        let start = Instant::now();
        let id = ssa_id(p, 1);

        sup.handle_event(&SessionPixEvent::SsaRequestSent(id), start, 0);
        sup.handle_event(&SessionPixEvent::CommitmentVerified(id), start, 0);
        sup.handle_event(
            &SessionPixEvent::DepositConfirmed {
                ssa_id: id,
                amount: sufficient_balance(),
            },
            start,
            0,
        );

        for secs in [9u64, 19, 29] {
            sup.handle_event(
                &SessionPixEvent::RecoveryProgress(make_progress(id, secs + 1, 50, 1)),
                start + Duration::from_secs(secs),
                secs,
            );
        }

        let actions = sup.handle_deadline(start + Duration::from_secs(31), 30);
        assert_eq!(actions.len(), 1);
        match &actions[0] {
            SessionPixAction::Close(SessionPixCloseReason::RecoveryDeadline) => {}
            other => panic!("expected RecoveryDeadline, got {other:?}"),
        }
    }

    #[test]
    fn idle_expiry_without_service_since_progress_rearms() {
        let p = pseudonym();
        let mut cfg = default_cfg();
        cfg.max_recovery_idle = Duration::from_secs(10);
        let (mut sup, _) = SessionPixSupervisor::new(cfg, dims(10, 5), p, Instant::now());
        let start = Instant::now();
        let id = ssa_id(p, 1);

        sup.handle_event(&SessionPixEvent::SsaRequestSent(id), start, 0);
        sup.handle_event(&SessionPixEvent::CommitmentVerified(id), start, 0);
        sup.handle_event(
            &SessionPixEvent::DepositConfirmed {
                ssa_id: id,
                amount: sufficient_balance(),
            },
            start,
            0,
        );

        let actions = sup.handle_deadline(start + Duration::from_secs(11), 0);
        assert!(
            !actions
                .iter()
                .any(|a| matches!(a, SessionPixAction::Close(SessionPixCloseReason::RecoveryIdle)))
        );

        // SSA should still be in Recovering.
        assert_eq!(
            sup.ssas.iter().find(|s| s.ssa_id == id).unwrap().phase,
            SsaPhase::Recovering
        );
    }

    #[test]
    fn idle_expiry_with_service_and_no_progress_closes_recovery_idle() {
        let p = pseudonym();
        let mut cfg = default_cfg();
        cfg.max_recovery_idle = Duration::from_secs(10);
        let (mut sup, _) = SessionPixSupervisor::new(cfg, dims(10, 5), p, Instant::now());
        let start = Instant::now();
        let id = ssa_id(p, 1);

        sup.handle_event(&SessionPixEvent::SsaRequestSent(id), start, 0);
        sup.handle_event(&SessionPixEvent::CommitmentVerified(id), start, 0);
        sup.handle_event(
            &SessionPixEvent::DepositConfirmed {
                ssa_id: id,
                amount: sufficient_balance(),
            },
            start,
            5,
        );

        // Idle fires with served_total=5 (same as watermark) → re-arm.
        let actions = sup.handle_deadline(start + Duration::from_secs(11), 5);
        assert!(
            !actions
                .iter()
                .any(|a| matches!(a, SessionPixAction::Close(SessionPixCloseReason::RecoveryIdle)))
        );

        // Now served_total increased to 10 (5 consumed since progress) → close.
        let actions = sup.handle_deadline(start + Duration::from_secs(22), 10);
        assert!(
            actions
                .iter()
                .any(|a| matches!(a, SessionPixAction::Close(SessionPixCloseReason::RecoveryIdle)))
        );
    }

    #[test]
    fn progress_resamples_served_total_watermark() {
        let p = pseudonym();
        let mut cfg = default_cfg();
        cfg.max_recovery_idle = Duration::from_secs(10);
        let (mut sup, _) = SessionPixSupervisor::new(cfg, dims(10, 5), p, Instant::now());
        let start = Instant::now();
        let id = ssa_id(p, 1);

        sup.handle_event(&SessionPixEvent::SsaRequestSent(id), start, 0);
        sup.handle_event(&SessionPixEvent::CommitmentVerified(id), start, 0);
        sup.handle_event(
            &SessionPixEvent::DepositConfirmed {
                ssa_id: id,
                amount: sufficient_balance(),
            },
            start,
            10,
        );

        sup.handle_event(
            &SessionPixEvent::RecoveryProgress(make_progress(id, 5, 50, 1)),
            start + Duration::from_secs(5),
            15,
        );

        let actions = sup.handle_deadline(start + Duration::from_secs(16), 15);
        assert!(
            !actions
                .iter()
                .any(|a| matches!(a, SessionPixAction::Close(SessionPixCloseReason::RecoveryIdle)))
        );
    }

    /// A snapshot carrying only surplus must keep the cycle alive without paying it anything.
    ///
    /// This is the whole of C1 at the supervisor. A conforming Entry drains an emission window's
    /// surplus in one contiguous run — `surplus × min(polys, 256)` packets, 4096 at the shipped
    /// dimensions — during which `useful_shares` does not move at all. Every one of those packets is
    /// service the Exit is consuming, so a supervisor that keyed liveness on `useful_shares` would
    /// let the run exhaust `max_served_without_progress`, park the writer, and then close an honest
    /// Session with `RecoveryIdle` — the re-arm branch cannot save it, because service *was*
    /// consumed. Liveness must therefore move on `shares_seen`, and payment must not.
    #[test]
    fn a_surplus_only_snapshot_is_liveness_without_payment() {
        let p = pseudonym();
        let mut cfg = default_cfg();
        cfg.max_recovery_idle = Duration::from_secs(10);
        let (mut sup, _) = SessionPixSupervisor::new(cfg, dims(10, 5), p, Instant::now());
        let start = Instant::now();
        let id = ssa_id(p, 1);

        sup.handle_event(&SessionPixEvent::SsaRequestSent(id), start, 0);
        sup.handle_event(&SessionPixEvent::CommitmentVerified(id), start, 0);
        sup.handle_event(
            &SessionPixEvent::DepositConfirmed {
                ssa_id: id,
                amount: sufficient_balance(),
            },
            start,
            0,
        );

        // One useful share, to establish a baseline both counters agree on.
        sup.handle_event(
            &SessionPixEvent::RecoveryProgress(make_progress(id, 1, 50, 0)),
            start + Duration::from_secs(1),
            10,
        );

        // Then a run of pure surplus: `shares_seen` climbs, `useful_shares` is pinned, and service is
        // being consumed throughout.
        for (i, seen) in (2..=6u64).enumerate() {
            let actions = sup.handle_event(
                &SessionPixEvent::RecoveryProgress(make_progress_seen(id, 1, seen, 50, 0)),
                start + Duration::from_secs(2 + i as u64),
                20 + 10 * i as u64,
            );
            assert!(
                actions
                    .iter()
                    .any(|a| matches!(a, SessionPixAction::ProgressNotification)),
                "a surplus share must reset the gate's served-without-progress ceiling"
            );
            assert!(
                !actions.iter().any(|a| matches!(a, SessionPixAction::Close(_))),
                "a surplus share must never close the Session"
            );
        }

        // The idle deadline was refreshed by the surplus run, so it has not expired — even though
        // `served_total` has grown well past the watermark of the last *useful* share.
        let actions = sup.handle_deadline(start + Duration::from_secs(12), 70);
        assert!(
            !actions
                .iter()
                .any(|a| matches!(a, SessionPixAction::Close(SessionPixCloseReason::RecoveryIdle))),
            "an Entry serving its negotiated surplus is not idle"
        );

        // Payment did not move: the surplus bought the Entry time, not credit.
        let ssa = sup.ssas.iter().find(|s| s.ssa_id == id).expect("cycle must be live");
        assert_eq!(1, ssa.largest_useful_shares, "surplus must not count as useful shares");
        assert_eq!(6, ssa.largest_shares_seen, "every share must count as liveness");
    }

    /// Once the surplus run stops, the idle rule must still fire.
    ///
    /// The liveness split widens what counts as "the Entry is alive"; it must not make the anti-drip
    /// bound unreachable. With no snapshot of any kind and service still being consumed, the Session
    /// closes exactly as it did before.
    #[test]
    fn silence_after_a_surplus_run_still_closes_the_session() {
        let p = pseudonym();
        let mut cfg = default_cfg();
        cfg.max_recovery_idle = Duration::from_secs(10);
        let (mut sup, _) = SessionPixSupervisor::new(cfg, dims(10, 5), p, Instant::now());
        let start = Instant::now();
        let id = ssa_id(p, 1);

        sup.handle_event(&SessionPixEvent::SsaRequestSent(id), start, 0);
        sup.handle_event(&SessionPixEvent::CommitmentVerified(id), start, 0);
        sup.handle_event(
            &SessionPixEvent::DepositConfirmed {
                ssa_id: id,
                amount: sufficient_balance(),
            },
            start,
            0,
        );
        sup.handle_event(
            &SessionPixEvent::RecoveryProgress(make_progress_seen(id, 1, 4, 50, 0)),
            start + Duration::from_secs(1),
            10,
        );

        // Nothing further arrives, and service keeps being consumed.
        let actions = sup.handle_deadline(start + Duration::from_secs(12), 500);
        assert!(
            actions
                .iter()
                .any(|a| matches!(a, SessionPixAction::Close(SessionPixCloseReason::RecoveryIdle))),
            "a genuinely silent Entry must still be caught"
        );
    }

    #[test]
    fn equal_snapshot_is_noop() {
        let p = pseudonym();
        let (mut sup, _) = SessionPixSupervisor::new(default_cfg(), dims(10, 5), p, Instant::now());
        let now = Instant::now();
        let id = ssa_id(p, 1);

        sup.handle_event(&SessionPixEvent::SsaRequestSent(id), now, 0);
        sup.handle_event(&SessionPixEvent::CommitmentVerified(id), now, 0);
        sup.handle_event(
            &SessionPixEvent::DepositConfirmed {
                ssa_id: id,
                amount: sufficient_balance(),
            },
            now,
            0,
        );

        let progress = make_progress(id, 10, 50, 1);
        sup.handle_event(&SessionPixEvent::RecoveryProgress(progress), now, 5);

        let idle_before = sup.ssas.iter().find(|s| s.ssa_id == id).unwrap().recovery_idle_deadline;

        let actions = sup.handle_event(&SessionPixEvent::RecoveryProgress(progress), now, 5);
        assert!(actions.is_empty());

        assert_eq!(
            sup.ssas.iter().find(|s| s.ssa_id == id).unwrap().recovery_idle_deadline,
            idle_before
        );
    }

    #[test]
    fn lower_snapshot_is_ignored_as_stale() {
        let p = pseudonym();
        let (mut sup, _) = SessionPixSupervisor::new(default_cfg(), dims(10, 5), p, Instant::now());
        let now = Instant::now();
        let id = ssa_id(p, 1);

        sup.handle_event(&SessionPixEvent::SsaRequestSent(id), now, 0);
        sup.handle_event(&SessionPixEvent::RecoveryProgress(make_progress(id, 10, 50, 1)), now, 0);

        // Stale snapshot from concurrent processing is silently ignored.
        // Close-on-regression was rejected because ack batches are processed
        // with for_each_concurrent, so out-of-order arrival is possible.
        let actions = sup.handle_event(&SessionPixEvent::RecoveryProgress(make_progress(id, 5, 50, 1)), now, 0);
        assert!(actions.is_empty(), "stale snapshot should be ignored, got: {actions:?}");
    }

    #[test]
    fn inconsistent_target_closes_as_counter_regression() {
        let p = pseudonym();
        let (mut sup, _) = SessionPixSupervisor::new(default_cfg(), dims(10, 5), p, Instant::now());
        let now = Instant::now();
        let id = ssa_id(p, 1);

        // Must register the SSA first.
        sup.handle_event(&SessionPixEvent::SsaRequestSent(id), now, 0);
        sup.handle_event(&SessionPixEvent::CommitmentVerified(id), now, 0);
        sup.handle_event(
            &SessionPixEvent::DepositConfirmed {
                ssa_id: id,
                amount: sufficient_balance(),
            },
            now,
            0,
        );

        // RecoveryProgress with target != dims.target_useful_shares() = 50.
        let actions = sup.handle_event(&SessionPixEvent::RecoveryProgress(make_progress(id, 1, 99, 0)), now, 0);
        assert_eq!(actions.len(), 1);
        assert!(matches!(
            actions[0],
            SessionPixAction::Close(SessionPixCloseReason::CounterRegression)
        ));
    }

    // ---------------------------------------------------------------
    // Fault tests
    // ---------------------------------------------------------------

    /// A fault reported before the request confirmation still counts.
    ///
    /// Carrying out `RequestSsa` registers the Exit commitment, which is what makes shares for that
    /// SSA processable — so a share can fail verification and be reported before `SsaRequestSent`
    /// gets back. Every handler ignores an SSA it has no record of, which on this event would mean
    /// failing open on the one signal that has to fail closed.
    #[test]
    fn an_unverifiable_share_counts_even_before_the_request_is_confirmed() {
        let p = pseudonym();
        let (mut sup, actions) = SessionPixSupervisor::new(SupervisorConfig::default(), dims(10, 5), p, Instant::now());
        let now = Instant::now();

        let id = match actions.as_slice() {
            [SessionPixAction::RequestSsa { ssa_ids, .. }] => ssa_ids[0],
            other => panic!("expected one RequestSsa, got {other:?}"),
        };

        // No `SsaRequestSent` — the confirmation is still in flight.
        let actions = sup.handle_event(
            &SessionPixEvent::UnverifiableShares {
                ssa_id: id,
                observed_total: 1,
                peer: crate::supervision::test_peer(),
            },
            now,
            0,
        );

        assert!(
            matches!(
                actions.as_slice(),
                [SessionPixAction::Close(SessionPixCloseReason::UnverifiableShares)]
            ),
            "expected a close, got {actions:?}"
        );
    }

    /// The report closes whatever it says, because the count is not what makes it fatal.
    ///
    /// `observed_total` is a cross-peer aggregate assembled from concurrently processed ack batches,
    /// so it can arrive as any value and out of order. It used to be charged as a delta against a
    /// running maximum and compared to a tolerance; with no tolerance left it is a log field, and the
    /// close must not become conditional on it again — a total of one is a doomed cycle exactly as
    /// much as a total of a thousand.
    #[test]
    fn any_observed_total_closes_on_the_first_report() {
        for observed_total in [1, 2, u64::MAX] {
            let p = pseudonym();
            let (mut sup, _) = SessionPixSupervisor::new(default_cfg(), dims(10, 5), p, Instant::now());
            let now = Instant::now();
            let id = ssa_id(p, 1);

            sup.handle_event(&SessionPixEvent::SsaRequestSent(id), now, 0);

            let actions = sup.handle_event(
                &SessionPixEvent::UnverifiableShares {
                    ssa_id: id,
                    observed_total,
                    peer: crate::supervision::test_peer(),
                },
                now,
                0,
            );

            assert!(
                matches!(
                    actions.as_slice(),
                    [SessionPixAction::Close(SessionPixCloseReason::UnverifiableShares)]
                ),
                "observed_total {observed_total} must close on the first report, got {actions:?}"
            );
        }
    }

    /// A batch's healthy siblings must not keep the Session open.
    ///
    /// Every other per-cycle fault routes through `close_ssa_and_collect`, which retires the member
    /// and lets `max_failed_cycles` siblings carry on. This one does not, and deliberately: a
    /// polynomial that fails to open its commitment is evidence about the *peer*, and the same peer
    /// holds every other cycle in the batch.
    #[test]
    fn an_unverifiable_share_closes_the_session_even_with_live_siblings() {
        let p = pseudonym();
        let mut cfg = default_cfg();
        cfg.ssas_per_request = 3;
        cfg.max_failed_cycles = 10;
        let (mut sup, actions) = SessionPixSupervisor::new(cfg, dims(10, 5), p, Instant::now());
        let now = Instant::now();

        let ids = match actions.as_slice() {
            [SessionPixAction::RequestSsa { ssa_ids, .. }] => ssa_ids.clone(),
            other => panic!("expected one RequestSsa, got {other:?}"),
        };
        assert_eq!(3, ids.len());
        for id in &ids {
            sup.handle_event(&SessionPixEvent::SsaRequestSent(*id), now, 0);
        }

        // The fault lands on the *last* member, so two healthy cycles remain in front of it.
        let actions = sup.handle_event(
            &SessionPixEvent::UnverifiableShares {
                ssa_id: ids[2],
                observed_total: 1,
                peer: crate::supervision::test_peer(),
            },
            now,
            0,
        );

        assert!(
            matches!(
                actions.as_slice(),
                [SessionPixAction::Close(SessionPixCloseReason::UnverifiableShares)]
            ),
            "a live batch must not absorb an unverifiable-share report, got {actions:?}"
        );
    }

    // ---------------------------------------------------------------
    // AlmostRecovered / Recovered
    // ---------------------------------------------------------------

    #[test]
    fn almost_recovered_while_recovering_requests_next_once() {
        let p = pseudonym();
        let (mut sup, _) = SessionPixSupervisor::new(default_cfg(), dims(10, 5), p, Instant::now());
        let now = Instant::now();
        let id1 = ssa_id(p, 1);

        sup.handle_event(&SessionPixEvent::SsaRequestSent(id1), now, 0);
        sup.handle_event(&SessionPixEvent::CommitmentVerified(id1), now, 0);
        sup.handle_event(
            &SessionPixEvent::DepositConfirmed {
                ssa_id: id1,
                amount: sufficient_balance(),
            },
            now,
            0,
        );

        let actions = sup.handle_event(&SessionPixEvent::AlmostRecovered(id1), now, 0);
        assert_eq!(actions.len(), 1);
        match &actions[0] {
            SessionPixAction::RequestSsa { ssa_ids, .. } => {
                assert_eq!(ssa_ids[0].ssa_index(), SsaIndex::new(2).unwrap());
            }
            other => panic!("expected RequestSsa, got {other:?}"),
        }

        let actions = sup.handle_event(&SessionPixEvent::AlmostRecovered(id1), now, 0);
        assert!(actions.is_empty());
    }

    /// Config for the share-order tests: a batch of three, generous dimensions so a cycle can carry a
    /// meaningful sample, and a small floor so the tests need few events.
    fn share_order_cfg(sample: u64, fraction: f64) -> SupervisorConfig {
        SupervisorConfig {
            ssas_per_request: 3,
            min_share_order_sample: sample,
            max_off_front_share_fraction: fraction,
            ..default_cfg()
        }
    }

    /// Serving the front cycle, with only a trace of progress behind it, must never trip the ratio.
    ///
    /// The trace stands in for what the mixnet produces on its own: SURBs are permuted outbound and
    /// their acknowledgements again inbound, so a conforming Entry still shows *some* off-front
    /// progress at a cycle boundary. The point of measuring a fraction rather than a count is that a
    /// bounded permutation stays a vanishing share of a growing window however the mixer is tuned.
    #[test]
    fn conforming_service_never_trips_the_share_order_ratio() {
        let p = pseudonym();
        let (mut sup, _) = SessionPixSupervisor::new(share_order_cfg(1000, 0.25), dims(100, 10), p, Instant::now());
        let now = Instant::now();
        fund_batch(&mut sup, p, 3, now);
        let target = sup.dims.target_useful_shares();

        for useful in [200, 400, 600, 800] {
            let actions = sup.handle_event(
                &SessionPixEvent::RecoveryProgress(make_progress(ssa_id(p, 1), useful, target, 10)),
                now,
                useful,
            );
            assert!(
                matches!(actions.as_slice(), [SessionPixAction::ProgressNotification]),
                "front-cycle service is what is supposed to happen, got {actions:?}"
            );
        }

        // A boundary's worth of reordering: 50 shares against 800, far inside the ceiling.
        let actions = sup.handle_event(
            &SessionPixEvent::RecoveryProgress(make_progress(ssa_id(p, 2), 50, target, 1)),
            now,
            850,
        );
        assert!(
            actions.is_empty(),
            "off-front progress must neither convict nor reopen the gate"
        );
        assert!(!sup.closed);
    }

    /// An Entry spreading a batch across all its cycles must be caught once the evidence is in.
    ///
    /// Three cycles served in parallel put two thirds of all progress behind the front — the signature
    /// the ratio exists for, and one that neither the service gate nor `max_recovery_idle` can see,
    /// since the front cycle keeps progressing and keeps refreshing its own idle timer.
    #[test]
    fn spreading_a_batch_across_its_cycles_trips_the_share_order_ratio() {
        const SAMPLE: u64 = 1000;

        let p = pseudonym();
        let (mut sup, _) = SessionPixSupervisor::new(share_order_cfg(SAMPLE, 0.25), dims(100, 10), p, Instant::now());
        let now = Instant::now();
        fund_batch(&mut sup, p, 3, now);
        let target = sup.dims.target_useful_shares();

        let mut closed_at = None;
        'outer: for round in 1..=5u64 {
            for cycle in 1..=3u32 {
                let actions = sup.handle_event(
                    &SessionPixEvent::RecoveryProgress(make_progress(ssa_id(p, cycle), round * 100, target, 10)),
                    now,
                    round * 300,
                );
                let sample = sup.front_useful + sup.off_front_useful;
                if let [SessionPixAction::Close(reason)] = actions.as_slice() {
                    assert_eq!(*reason, SessionPixCloseReason::BatchServedOutOfOrder);
                    closed_at = Some(sample);
                    break 'outer;
                }
                assert!(
                    sample < SAMPLE,
                    "at {sample} shares the evidence was in and the 2:1 split should have closed the Session"
                );
            }
        }

        let closed_at = closed_at.expect("spreading across three cycles must close the Session");
        assert!(
            closed_at >= SAMPLE,
            "conviction at {closed_at} shares is below the evidence floor of {SAMPLE}"
        );
    }

    /// The accounting is scoped to one cycle's service: completing the front cycle clears it.
    ///
    /// That bound is what stops an Entry laundering a spree of out-of-order service — the only way to
    /// reach the reset is to finish the front cycle, which is to say to earn the traffic it took.
    #[test]
    fn share_order_accounting_resets_when_the_front_cycle_completes() {
        let p = pseudonym();
        let (mut sup, _) = SessionPixSupervisor::new(share_order_cfg(1000, 0.25), dims(100, 10), p, Instant::now());
        let now = Instant::now();
        fund_batch(&mut sup, p, 3, now);
        let target = sup.dims.target_useful_shares();

        sup.handle_event(
            &SessionPixEvent::RecoveryProgress(make_progress(ssa_id(p, 1), 500, target, 50)),
            now,
            500,
        );
        sup.handle_event(
            &SessionPixEvent::RecoveryProgress(make_progress(ssa_id(p, 2), 100, target, 10)),
            now,
            600,
        );
        assert_eq!(sup.share_order_front, Some(SsaIndex::new(1).unwrap()));
        assert_eq!((sup.front_useful, sup.off_front_useful), (500, 100));

        sup.handle_event(&SessionPixEvent::Recovered(ssa_id(p, 1)), now, 600);
        assert_eq!(
            sup.share_order_front,
            Some(SsaIndex::new(2).unwrap()),
            "cycle 2 is the front now"
        );
        assert_eq!(
            (sup.front_useful, sup.off_front_useful),
            (0, 0),
            "the window is one cycle's service, so it starts empty on the new front"
        );
    }

    /// Below the evidence floor nothing is judged, however lopsided the split.
    ///
    /// This is what keeps a cycle made unrecoverable by loss from convicting the Entry: it stops
    /// progressing while later cycles continue, so the fraction goes to 1.0 — and the floor outlasts
    /// the service-gated `max_recovery_idle` window that retires it, after which the front moves and
    /// the accounting resets. The ceiling here is 0.0, the strictest possible, so only the floor can be
    /// what holds the verdict back.
    #[test]
    fn no_verdict_is_reached_below_the_evidence_floor() {
        const SAMPLE: u64 = 1000;

        let p = pseudonym();
        let (mut sup, _) = SessionPixSupervisor::new(share_order_cfg(SAMPLE, 0.0), dims(100, 10), p, Instant::now());
        let now = Instant::now();
        fund_batch(&mut sup, p, 3, now);
        let target = sup.dims.target_useful_shares();

        // Every share lands behind the front, and still nothing is decided while the sample is thin.
        for useful in [300, 600, 900] {
            let actions = sup.handle_event(
                &SessionPixEvent::RecoveryProgress(make_progress(ssa_id(p, 3), useful, target, 10)),
                now,
                useful,
            );
            assert!(actions.is_empty(), "{useful} shares is below the floor and off-front");
        }

        let actions = sup.handle_event(
            &SessionPixEvent::RecoveryProgress(make_progress(ssa_id(p, 3), SAMPLE, target, 10)),
            now,
            SAMPLE,
        );
        assert!(
            matches!(
                actions.as_slice(),
                [SessionPixAction::Close(SessionPixCloseReason::BatchServedOutOfOrder)]
            ),
            "at the floor the verdict must be reached, got {actions:?}"
        );
    }

    /// A stray reordered share must not put a *queued* cycle on the idle clock.
    ///
    /// The sibling test below fixes the arming path: clocks start when a cycle reaches the front. But
    /// refreshing is not arming, and `on_recovery_progress` used to set `recovery_idle_deadline` for
    /// any cycle merely in `Recovering`. One share crossing a cycle boundary out of order — the exact
    /// thing [`SupervisorConfig::min_share_order_sample`] exists to tolerate — therefore *started* the
    /// idle clock on a cycle the Entry cannot serve, since emission is clamped to one cycle. A
    /// `max_recovery_idle` later the gate saw session-wide service climbing on the front cycle's
    /// behalf and retired the queued one, charging it for a queue wait it had no way to end.
    ///
    /// The two clocks arm together or not at all, so the hard deadline is the witness for "has been at
    /// the front", and a queued cycle must come out of this with neither.
    #[test]
    fn a_stray_share_on_a_queued_cycle_must_not_start_its_idle_clock() {
        const BATCH: u32 = 3;

        let p = pseudonym();
        let cfg = SupervisorConfig {
            ssas_per_request: BATCH as usize,
            max_recovery_idle: Duration::from_secs(10),
            max_recovery_time: Duration::from_secs(3600),
            ..default_cfg()
        };
        let (mut sup, _) = SessionPixSupervisor::new(cfg, dims(10, 5), p, Instant::now());
        let start = Instant::now();
        fund_batch(&mut sup, p, BATCH, start);
        let target = sup.dims.target_useful_shares();

        // One share for queued cycle 2 arrives while cycle 1 holds the front.
        sup.handle_event(
            &SessionPixEvent::RecoveryProgress(make_progress(ssa_id(p, 2), 1, target, 0)),
            start + Duration::from_secs(1),
            1_000,
        );

        let queued = sup.ssas.iter().find(|s| s.ssa_id == ssa_id(p, 2)).expect("cycle 2");
        assert_eq!(
            (queued.recovery_hard_deadline, queued.recovery_idle_deadline),
            (None, None),
            "an off-front share must leave a queued cycle off both clocks, not just the hard one"
        );

        // Cycle 1 is served normally, so session-wide service climbs past cycle 2's baseline — which is
        // what would convict it if the idle clock were running.
        for step in 1..=3u64 {
            sup.handle_event(
                &SessionPixEvent::RecoveryProgress(make_progress(ssa_id(p, 1), step, target, 0)),
                start + Duration::from_secs(1 + step * 3),
                1_000 + step * 1_000,
            );
        }

        let actions = sup.handle_deadline(start + Duration::from_secs(12), 5_000);
        assert!(
            actions.is_empty(),
            "no cycle may be retired or closed while the front is being served, got {actions:?}"
        );
        assert_eq!(sup.ssas.len(), BATCH as usize, "the whole batch must still be live");
        assert_eq!(sup.failed_cycles, 0, "nothing may be charged as a failed cycle");
    }

    /// A cycle queued behind the front of its batch must not be charged for the wait.
    ///
    /// The Entry serves a batch strictly in index order, so at deployed dimensions cycle 2 of a batch
    /// sits idle for the ~61 min it takes cycle 1's shares to be emitted. Starting its recovery clocks
    /// when its deposit confirmed measured that queue wait: `max_recovery_idle` would retire it inside
    /// a minute — the idle gate compares against *session-wide* service, which is plentiful while cycle
    /// 1 is served — and `max_recovery_time` would finish the job. Batching would have paid for
    /// `ssas_per_request` deposits and been able to use one.
    ///
    /// So the clocks start when the cycle reaches the front, and `max_recovery_time` bounds the
    /// recovery of a single cycle rather than its position in a queue.
    #[test]
    fn a_queued_cycle_starts_its_recovery_clocks_only_at_the_front_of_the_batch() {
        const BATCH: u32 = 3;

        let p = pseudonym();
        let cfg = SupervisorConfig {
            ssas_per_request: BATCH as usize,
            max_recovery_idle: Duration::from_secs(10),
            max_recovery_time: Duration::from_secs(3600),
            ..default_cfg()
        };
        let (mut sup, _) = SessionPixSupervisor::new(cfg, dims(10, 5), p, Instant::now());
        let start = Instant::now();
        fund_batch(&mut sup, p, BATCH, start);
        let target = sup.dims.target_useful_shares();

        let armed = |sup: &SessionPixSupervisor, idx: u32| {
            let ssa = sup
                .ssas
                .iter()
                .find(|s| s.ssa_id == ssa_id(p, idx))
                .expect("cycle must exist");
            (
                ssa.recovery_hard_deadline.is_some(),
                ssa.recovery_idle_deadline.is_some(),
            )
        };

        assert_eq!(
            armed(&sup, 1),
            (true, true),
            "the front cycle's clocks run from funding"
        );
        for queued in 2..=BATCH {
            assert_eq!(
                armed(&sup, queued),
                (false, false),
                "cycle {queued} is queued behind the front and must not be on the clock yet"
            );
        }

        // Cycle 1 is served for five times `max_recovery_idle`, with session-wide service climbing the
        // whole way. Nothing queued may be retired for that.
        for step in 1..=10u64 {
            let at = start + Duration::from_secs(step * 5);
            let served = step * 1000;
            sup.handle_event(
                &SessionPixEvent::RecoveryProgress(make_progress(ssa_id(p, 1), step, target, 1)),
                at,
                served,
            );
            let actions = sup.handle_deadline(at, served);
            assert!(
                !actions
                    .iter()
                    .any(|a| matches!(a, SessionPixAction::Close(_) | SessionPixAction::RetireSsa(_))),
                "no cycle may be closed or retired at +{}s, got {actions:?}",
                step * 5
            );
        }
        assert_eq!(sup.ssas.len(), BATCH as usize, "the whole batch must still be live");

        // Cycle 1 recovers cryptographically, but its already-minted SURBs still own the FIFO and
        // cycle 2 must remain off the clock until that paid tail drains.
        let handover = start + Duration::from_secs(60);
        sup.handle_event(&SessionPixEvent::Recovered(ssa_id(p, 1)), handover, 10_000);
        assert_eq!(
            armed(&sup, 2),
            (false, false),
            "cycle 2 is still behind the paid FIFO tail"
        );
        assert_eq!(
            armed(&sup, 3),
            (false, false),
            "cycle 3 is still queued and must stay off the clock"
        );

        let fifo_boundary = handover + Duration::from_secs(5);
        let max_seen = max_cycle_shares(&sup);
        sup.handle_event(
            &SessionPixEvent::RecoveryProgress(make_recovered_tail_progress(&sup, ssa_id(p, 1), max_seen)),
            fifo_boundary,
            10_100,
        );
        assert_eq!(armed(&sup, 2), (true, true), "cycle 2 starts only at the FIFO boundary");

        let ssa2 = sup.ssas.iter().find(|s| s.ssa_id == ssa_id(p, 2)).expect("cycle 2");
        assert_eq!(
            ssa2.recovery_hard_deadline,
            fifo_boundary.checked_add(Duration::from_secs(3600)),
            "cycle 2's ceiling must be measured from its own turn, not from its deposit"
        );
        assert_eq!(
            ssa2.served_total_at_last_progress, 10_100,
            "the idle gate's baseline must be re-based at the handover, or the queue wait is charged to it"
        );
    }

    /// At `ssas_per_request > 1`, only the batch's **last** SSA may ask for the successor batch.
    ///
    /// `next_requested` is per-`PerSsaState`, so without a batch-wide gate every member of a batch
    /// answers its own `AlmostRecovered` with a fresh batch of `ssas_per_request`: one batch of 3
    /// becomes three, then nine, each SSA a separate on-chain deposit demanded of the Entry. At
    /// `ssas_per_request == 1`, "one requesting cycle" and "one requesting batch" coincide, which is
    /// why every unbatched test misses the distinction.
    ///
    /// All three cycles are funded and `Recovering` before the first event, i.e. the case where the
    /// deferral in [`almost_recovered_while_awaiting_deposit_defers_request`] does *not* apply and
    /// only the batch gate can hold the request back.
    #[test]
    fn only_the_last_ssa_of_a_batch_asks_for_the_next_one() {
        const BATCH: usize = 3;

        let p = pseudonym();
        let cfg = SupervisorConfig {
            ssas_per_request: BATCH,
            ..default_cfg()
        };
        let (mut sup, _) = SessionPixSupervisor::new(cfg, dims(10, 5), p, Instant::now());
        let now = Instant::now();

        // Fund every cycle of the batch, so each one is in `Recovering`.
        for idx in 1..=BATCH as u32 {
            let id = ssa_id(p, idx);
            sup.handle_event(&SessionPixEvent::SsaRequestSent(id), now, 0);
            sup.handle_event(&SessionPixEvent::CommitmentVerified(id), now, 0);
            sup.handle_event(
                &SessionPixEvent::DepositConfirmed {
                    ssa_id: id,
                    amount: sufficient_balance(),
                },
                now,
                0,
            );
        }
        assert_eq!(sup.next_ssa_index, BATCH as u32 + 1, "one batch allocated so far");

        // Every member but the last is silent, in the order the Exit actually reconstructs them.
        for idx in 1..BATCH as u32 {
            let actions = sup.handle_event(&SessionPixEvent::AlmostRecovered(ssa_id(p, idx)), now, 0);
            assert!(
                actions.is_empty(),
                "SSA {idx} is not the last of the batch and must not ask for a successor, got {actions:?}"
            );
            assert_eq!(
                sup.next_ssa_index,
                BATCH as u32 + 1,
                "SSA {idx} must not allocate any index"
            );
        }

        // The last member is the one that pipelines the refill.
        let last = ssa_id(p, BATCH as u32);
        let actions = sup.handle_event(&SessionPixEvent::AlmostRecovered(last), now, 0);
        assert_eq!(actions.len(), 1, "the last SSA of the batch must ask exactly once");
        match &actions[0] {
            SessionPixAction::RequestSsa { ssa_ids, .. } => assert_eq!(
                ssa_ids.iter().map(|i| i.ssa_index().get()).collect::<Vec<_>>(),
                (BATCH as u32 + 1..=2 * BATCH as u32).collect::<Vec<_>>(),
                "the successor batch must be the next contiguous run"
            ),
            other => panic!("expected RequestSsa, got {other:?}"),
        }

        // Once-only is preserved for the last member too.
        assert!(
            sup.handle_event(&SessionPixEvent::AlmostRecovered(last), now, 0)
                .is_empty(),
            "a repeated AlmostRecovered on the last member must not ask again"
        );
        assert_eq!(sup.next_ssa_index, 2 * BATCH as u32 + 1);
        assert_eq!(
            sup.live_cycle_count(),
            sup.reserved_cycle_slots(),
            "two full batches sit exactly on the admission reservation"
        );
    }

    #[test]
    fn a_third_live_batch_is_deferred_even_when_share_order_policy_allows_it() {
        const BATCH: usize = 2;

        let p = pseudonym();
        let cfg = SupervisorConfig {
            ssas_per_request: BATCH,
            max_off_front_share_fraction: 1.0,
            ..default_cfg()
        };
        let now = Instant::now();
        let (mut sup, _) = SessionPixSupervisor::new(cfg, dims(10, 5), p, now);

        fund_batch(&mut sup, p, BATCH as u32, now);
        let actions = sup.handle_event(&SessionPixEvent::AlmostRecovered(ssa_id(p, 2)), now, 0);
        let second_batch = match actions.as_slice() {
            [SessionPixAction::RequestSsa { ssa_ids, .. }] => ssa_ids.clone(),
            other => panic!("expected the second batch, got {other:?}"),
        };
        for id in second_batch {
            sup.handle_event(&SessionPixEvent::SsaRequestSent(id), now, 0);
            sup.handle_event(&SessionPixEvent::CommitmentVerified(id), now, 0);
            sup.handle_event(
                &SessionPixEvent::DepositConfirmed {
                    ssa_id: id,
                    amount: sufficient_balance(),
                },
                now,
                0,
            );
        }

        let actions = sup.handle_event(&SessionPixEvent::AlmostRecovered(ssa_id(p, 4)), now, 0);
        assert!(
            actions
                .iter()
                .all(|action| !matches!(action, SessionPixAction::RequestSsa { .. })),
            "two live batches already consume the reservation, got {actions:?}"
        );
        assert_eq!(sup.ssas.len(), 2 * BATCH, "a third batch must not be allocated");
    }

    #[test]
    fn a_deferred_batch_retries_only_after_the_older_generation_is_released() {
        let p = pseudonym();
        let cfg = SupervisorConfig {
            ssas_per_request: 2,
            max_failed_cycles: 2,
            max_off_front_share_fraction: 1.0,
            ..default_cfg()
        };
        let now = Instant::now();
        let (mut sup, _) = SessionPixSupervisor::new(cfg, dims(10, 5), p, now);

        // Leave cycle 1 unfunded and fund cycle 2, whose early signal earns batch 2.
        commit_unfunded(&mut sup, p, 1..=1, now);
        let second = ssa_id(p, 2);
        sup.handle_event(&SessionPixEvent::SsaRequestSent(second), now, 0);
        sup.handle_event(&SessionPixEvent::CommitmentVerified(second), now, 0);
        sup.handle_event(
            &SessionPixEvent::DepositConfirmed {
                ssa_id: second,
                amount: sufficient_balance(),
            },
            now,
            0,
        );
        assert!(
            sup.handle_event(&SessionPixEvent::AlmostRecovered(second), now, 0)
                .iter()
                .any(|action| matches!(action, SessionPixAction::RequestSsa { .. }))
        );

        for idx in 3..=4 {
            let id = ssa_id(p, idx);
            sup.handle_event(&SessionPixEvent::SsaRequestSent(id), now, 0);
            sup.handle_event(&SessionPixEvent::CommitmentVerified(id), now, 0);
            sup.handle_event(
                &SessionPixEvent::DepositConfirmed {
                    ssa_id: id,
                    amount: sufficient_balance(),
                },
                now,
                0,
            );
        }
        assert!(
            sup.handle_event(&SessionPixEvent::AlmostRecovered(ssa_id(p, 4)), now, 0)
                .iter()
                .all(|action| !matches!(action, SessionPixAction::RequestSsa { .. }))
        );

        // Releasing only one member of the old generation is insufficient.
        let actions = sup.handle_event(&SessionPixEvent::DepositObserverClosed(ssa_id(p, 1)), now, 0);
        assert!(
            actions
                .iter()
                .all(|action| !matches!(action, SessionPixAction::RequestSsa { .. }))
        );

        // The remaining old member times out. Its RetireSsa must precede the now-admissible request,
        // otherwise the action driver would allocate the replacement before dropping the old guard.
        let actions = sup.handle_deadline(now + default_cfg().max_recovery_idle, 1);
        let retire_position = actions
            .iter()
            .position(|action| matches!(action, SessionPixAction::RetireSsa(id) if *id == second));
        let request_position = actions.iter().position(|action| {
            matches!(action, SessionPixAction::RequestSsa { ssa_ids, .. }
                if ssa_ids.iter().map(|id| id.ssa_index().get()).eq(5..=6))
        });
        assert!(
            matches!((retire_position, request_position), (Some(retire), Some(request)) if retire < request),
            "the old generation must retire before batch 3 is requested, got {actions:?}"
        );
        assert_eq!(sup.live_batch_count(), 2);
        assert_eq!(sup.live_cycle_count(), 4);
    }

    /// Retiring the cycle that holds the successor gate must hand it to the newest survivor.
    ///
    /// Otherwise the batch is stranded: its siblings are barred from asking, so no replacement is
    /// ever requested and the Session dies on a recovery timeout instead of on its actual cause.
    /// The promoted cycle is the last of the batch that will really serve its quota, so the request
    /// still cannot come early.
    #[test]
    fn retiring_the_last_ssa_of_a_batch_hands_the_successor_gate_on() {
        const BATCH: usize = 3;

        let p = pseudonym();
        let cfg = SupervisorConfig {
            ssas_per_request: BATCH,
            ..default_cfg()
        };
        let (mut sup, _) = SessionPixSupervisor::new(cfg, dims(10, 5), p, Instant::now());
        let now = Instant::now();

        // Cycles 1 and 2 get funded; 3 — the gate holder — is left awaiting its deposit.
        for idx in 1..BATCH as u32 {
            let id = ssa_id(p, idx);
            sup.handle_event(&SessionPixEvent::SsaRequestSent(id), now, 0);
            sup.handle_event(&SessionPixEvent::CommitmentVerified(id), now, 0);
            sup.handle_event(
                &SessionPixEvent::DepositConfirmed {
                    ssa_id: id,
                    amount: sufficient_balance(),
                },
                now,
                0,
            );
        }
        let last = ssa_id(p, BATCH as u32);
        sup.handle_event(&SessionPixEvent::SsaRequestSent(last), now, 0);
        sup.handle_event(&SessionPixEvent::CommitmentVerified(last), now, 0);

        // Its deposit observer gives up, so the gate holder is retired mid-batch.
        let actions = sup.handle_event(&SessionPixEvent::DepositObserverClosed(last), now, 0);
        assert!(
            actions
                .iter()
                .any(|a| matches!(a, SessionPixAction::RetireSsa(id) if *id == last)),
            "the unfunded last cycle must be retired, not close the Session, got {actions:?}"
        );
        assert!(!sup.closed, "surviving siblings must keep the Session alive");

        // The gate is now cycle 2's, and cycle 1 is still barred.
        assert!(
            sup.handle_event(&SessionPixEvent::AlmostRecovered(ssa_id(p, 1)), now, 0)
                .is_empty(),
            "cycle 1 is still not the last survivor and must stay silent"
        );
        let actions = sup.handle_event(&SessionPixEvent::AlmostRecovered(ssa_id(p, 2)), now, 0);
        assert_eq!(
            actions.len(),
            1,
            "the newest survivor must have inherited the successor gate, got {actions:?}"
        );
        match &actions[0] {
            SessionPixAction::RequestSsa { ssa_ids, .. } => assert_eq!(
                ssa_ids.iter().map(|i| i.ssa_index().get()).collect::<Vec<_>>(),
                (BATCH as u32 + 1..=2 * BATCH as u32).collect::<Vec<_>>(),
                "the successor batch still starts past the whole retired batch"
            ),
            other => panic!("expected RequestSsa, got {other:?}"),
        }
    }

    /// A second lost cycle closes a batched Session, however many funded siblings are still standing.
    ///
    /// Retiring a member and carrying on is what keeps one bad cycle from costing a whole Session,
    /// but it leaves no trace on the survivors — so without a count that outlives the retirement an
    /// Entry can lose one cycle per batch forever, paying for a fraction of what it is served. The
    /// counter is cumulative for exactly that reason: here the two losses are in *different* batches,
    /// and nothing in a per-batch view would connect them.
    ///
    /// The close carries the *first* failure's reason, so the report names the cause rather than the
    /// last symptom.
    #[test]
    fn the_second_lost_cycle_closes_a_batched_session() {
        const BATCH: usize = 4;

        let p = pseudonym();
        let cfg = SupervisorConfig {
            ssas_per_request: BATCH,
            max_failed_cycles: 1,
            ..default_cfg()
        };
        let (mut sup, _) = SessionPixSupervisor::new(cfg, dims(10, 5), p, Instant::now());
        let now = Instant::now();

        // Cycles 1 and 2 are funded and recovering; 3 and 4 are stuck awaiting their deposits, which
        // is the state an unpaid cycle is lost from.
        fund_batch(&mut sup, p, 2, now);
        commit_unfunded(&mut sup, p, 3..=BATCH as u32, now);

        // First loss: survivable, and the batch carries on.
        let actions = sup.handle_event(&SessionPixEvent::DepositObserverClosed(ssa_id(p, 3)), now, 0);
        assert!(
            actions
                .iter()
                .any(|a| matches!(a, SessionPixAction::RetireSsa(id) if *id == ssa_id(p, 3))),
            "the first lost cycle must be retired rather than close the Session, got {actions:?}"
        );
        assert!(!sup.closed, "one loss is inside the limit");

        // Second loss, with two funded siblings still recovering: over the limit, so the Session
        // goes anyway — which is the point, since those siblings are what would otherwise keep an
        // Entry losing one cycle per batch alive indefinitely.
        let actions = sup.handle_event(&SessionPixEvent::DepositObserverClosed(ssa_id(p, 4)), now, 0);
        assert!(
            actions
                .iter()
                .any(|a| matches!(a, SessionPixAction::Close(SessionPixCloseReason::DepositObserverClosed))),
            "the second lost cycle must close the Session, got {actions:?}"
        );
        assert!(sup.closed);
        assert_eq!(2, sup.failed_cycles);
    }

    /// Raising the limit buys exactly the extra losses it names, and no more.
    ///
    /// The counter has to survive the retirement of the cycle that incremented it, so this drives
    /// three separate losses through a batch large enough that a survivor always remains — a
    /// per-batch or per-cycle count would never reach the limit.
    #[test]
    fn a_raised_failure_limit_tolerates_exactly_that_many_losses() {
        const BATCH: usize = 6;

        let p = pseudonym();
        let cfg = SupervisorConfig {
            ssas_per_request: BATCH,
            max_failed_cycles: 3,
            ..default_cfg()
        };
        let (mut sup, _) = SessionPixSupervisor::new(cfg, dims(10, 5), p, Instant::now());
        let now = Instant::now();

        // One funded cycle, so the Session always has a survivor and only the limit can close it.
        fund_batch(&mut sup, p, 1, now);
        commit_unfunded(&mut sup, p, 2..=BATCH as u32, now);

        for idx in 2..=4 {
            sup.handle_event(&SessionPixEvent::DepositObserverClosed(ssa_id(p, idx)), now, 0);
            assert!(!sup.closed, "loss {} of 3 is inside the limit", idx - 1);
        }

        sup.handle_event(&SessionPixEvent::DepositObserverClosed(ssa_id(p, 5)), now, 0);
        assert!(sup.closed, "the fourth loss is over a limit of three");
    }

    #[test]
    fn almost_recovered_while_awaiting_deposit_defers_request() {
        let p = pseudonym();
        let (mut sup, _) = SessionPixSupervisor::new(default_cfg(), dims(10, 5), p, Instant::now());
        let now = Instant::now();
        let id1 = ssa_id(p, 1);

        sup.handle_event(&SessionPixEvent::SsaRequestSent(id1), now, 0);
        sup.handle_event(&SessionPixEvent::CommitmentVerified(id1), now, 0);

        let actions = sup.handle_event(&SessionPixEvent::AlmostRecovered(id1), now, 0);
        assert!(actions.is_empty());

        let actions = sup.handle_event(
            &SessionPixEvent::DepositConfirmed {
                ssa_id: id1,
                amount: sufficient_balance(),
            },
            now,
            0,
        );
        assert_eq!(actions.len(), 2);
        assert!(matches!(actions[0], SessionPixAction::ReleaseService));
        match &actions[1] {
            SessionPixAction::RequestSsa { ssa_ids, .. } => {
                assert_eq!(ssa_ids[0].ssa_index(), SsaIndex::new(2).unwrap());
            }
            other => panic!("expected RequestSsa, got {other:?}"),
        }
    }

    #[test]
    fn recovered_without_prior_early_event_falls_back_to_request() {
        let p = pseudonym();
        let (mut sup, _) = SessionPixSupervisor::new(default_cfg(), dims(10, 5), p, Instant::now());
        let now = Instant::now();
        let id1 = ssa_id(p, 1);

        sup.handle_event(&SessionPixEvent::SsaRequestSent(id1), now, 0);
        sup.handle_event(&SessionPixEvent::CommitmentVerified(id1), now, 0);
        sup.handle_event(
            &SessionPixEvent::DepositConfirmed {
                ssa_id: id1,
                amount: sufficient_balance(),
            },
            now,
            0,
        );

        let actions = sup.handle_event(&SessionPixEvent::Recovered(id1), now, 0);
        assert_eq!(actions.len(), 2, "expected gate reset followed by the fallback request");
        assert!(matches!(actions[0], SessionPixAction::ProgressNotification));
        match &actions[1] {
            SessionPixAction::RequestSsa { ssa_ids, .. } => {
                assert_eq!(ssa_ids[0].ssa_index(), SsaIndex::new(2).unwrap());
            }
            other => panic!("expected RequestSsa, got {other:?}"),
        }
    }

    #[test]
    fn recovered_with_prior_early_event_does_not_fallback() {
        let p = pseudonym();
        let (mut sup, _) = SessionPixSupervisor::new(default_cfg(), dims(10, 5), p, Instant::now());
        let now = Instant::now();
        let id1 = ssa_id(p, 1);

        sup.handle_event(&SessionPixEvent::SsaRequestSent(id1), now, 0);
        sup.handle_event(&SessionPixEvent::CommitmentVerified(id1), now, 0);
        sup.handle_event(
            &SessionPixEvent::DepositConfirmed {
                ssa_id: id1,
                amount: sufficient_balance(),
            },
            now,
            0,
        );

        sup.handle_event(&SessionPixEvent::AlmostRecovered(id1), now, 0);
        let actions = sup.handle_event(&SessionPixEvent::Recovered(id1), now, 0);
        assert!(
            matches!(actions.as_slice(), [SessionPixAction::ProgressNotification]),
            "the existing gate is reset and now belongs to the predecessor's bounded tail: {actions:?}"
        );
    }

    // ---------------------------------------------------------------
    // Close behavior
    // ---------------------------------------------------------------

    #[test]
    fn close_action_emitted_at_most_once() {
        let p = pseudonym();
        let (mut sup, _) = SessionPixSupervisor::new(default_cfg(), dims(10, 5), p, Instant::now());
        let now = Instant::now();
        let id = ssa_id(p, 1);

        sup.handle_event(&SessionPixEvent::SsaRequestSent(id), now, 0);
        sup.handle_deadline(now + Duration::from_secs(100), 0);
        assert!(sup.closed);

        let empty = sup.handle_event(&SessionPixEvent::SsaRequestSent(ssa_id(p, 999)), now, 0);
        assert!(empty.is_empty());
        let empty = sup.handle_deadline(now + Duration::from_secs(200), 0);
        assert!(empty.is_empty());
    }

    #[test]
    fn all_events_after_close_are_ignored() {
        let p = pseudonym();
        let (mut sup, _) = SessionPixSupervisor::new(default_cfg(), dims(10, 5), p, Instant::now());
        let now = Instant::now();
        let id = ssa_id(p, 1);

        sup.handle_event(&SessionPixEvent::SsaRequestSent(id), now, 0);
        sup.handle_deadline(now + Duration::from_secs(100), 0);
        assert!(sup.closed);

        for ev in &[
            SessionPixEvent::SsaRequestSent(ssa_id(p, 2)),
            SessionPixEvent::CommitmentVerified(ssa_id(p, 2)),
            SessionPixEvent::DepositConfirmed {
                ssa_id: ssa_id(p, 2),
                amount: sufficient_balance(),
            },
            SessionPixEvent::DepositObserverClosed(ssa_id(p, 2)),
            SessionPixEvent::RecoveryProgress(make_progress(ssa_id(p, 2), 1, 50, 0)),
            SessionPixEvent::AlmostRecovered(ssa_id(p, 2)),
            SessionPixEvent::Recovered(ssa_id(p, 2)),
            SessionPixEvent::UnverifiableShares {
                ssa_id: ssa_id(p, 2),
                observed_total: 1,
                peer: crate::supervision::test_peer(),
            },
        ] {
            assert!(
                sup.handle_event(ev, now, 0).is_empty(),
                "event should be ignored after close"
            );
        }
    }

    // ---------------------------------------------------------------
    // next_deadline
    // ---------------------------------------------------------------

    #[test]
    fn next_deadline_none_when_no_ssas() {
        let (sup, _) = SessionPixSupervisor::new(default_cfg(), dims(10, 5), pseudonym(), Instant::now());
        assert!(sup.next_deadline().is_none());
    }

    #[test]
    fn next_deadline_returns_earliest() {
        let p = pseudonym();
        let (mut sup, _) = SessionPixSupervisor::new(default_cfg(), dims(10, 5), p, Instant::now());
        let now = Instant::now();

        // Two SSAs in flight with deadlines 40 s apart: the later one is armed second, so returning
        // the earliest is a real choice rather than the only candidate.
        let id1 = ssa_id(p, 1);
        sup.handle_event(&SessionPixEvent::SsaRequestSent(id1), now, 0);
        let id2 = ssa_id(p, 2);
        sup.handle_event(&SessionPixEvent::SsaRequestSent(id2), now + Duration::from_secs(40), 0);

        let dl = sup.next_deadline().unwrap();

        let expected = now + Duration::from_secs(20);
        assert!((dl - expected).as_millis() < 10, "expected {expected:?}, got {dl:?}");
    }

    /// Pipelining a second SSA must not disturb the first one's deadlines.
    ///
    /// This used to be a `SessionManager` test asserting that two per-index `PixKillSwitch` abort
    /// handles coexisted. The deadlines are the supervisor's now, so the invariant is stated here
    /// against the state it actually lives in — and it is stronger, because a shared timer would
    /// satisfy "both handles present" but not "both instants unchanged".
    #[test]
    fn pipelining_a_second_ssa_leaves_the_first_ones_deadlines_alone() {
        let p = pseudonym();
        let (mut sup, _) = SessionPixSupervisor::new(default_cfg(), dims(10, 5), p, Instant::now());
        let now = Instant::now();

        let id1 = ssa_id(p, 1);
        sup.handle_event(&SessionPixEvent::SsaRequestSent(id1), now, 0);
        sup.handle_event(&SessionPixEvent::CommitmentVerified(id1), now, 0);
        sup.handle_event(
            &SessionPixEvent::DepositConfirmed {
                ssa_id: id1,
                amount: sufficient_balance(),
            },
            now,
            0,
        );

        let before = sup.ssas.iter().find(|s| s.ssa_id == id1).unwrap();
        assert_eq!(before.phase, SsaPhase::Recovering);
        let (hard_before, idle_before) = (before.recovery_hard_deadline, before.recovery_idle_deadline);

        // Early recovery on a funded SSA is what pipelines the next one.
        let later = now + Duration::from_secs(5);
        let actions = sup.handle_event(&SessionPixEvent::AlmostRecovered(id1), later, 0);
        let id2 = match actions.as_slice() {
            [SessionPixAction::RequestSsa { ssa_ids, .. }] => ssa_ids[0],
            other => panic!("expected exactly one RequestSsa, got {other:?}"),
        };
        assert_eq!(id2.ssa_index().get(), 2, "the pipelined SSA must take the next index");

        sup.handle_event(&SessionPixEvent::SsaRequestSent(id2), later, 0);
        sup.handle_event(&SessionPixEvent::CommitmentVerified(id2), later, 0);

        let after = sup.ssas.iter().find(|s| s.ssa_id == id1).unwrap();
        assert_eq!(
            after.phase,
            SsaPhase::Recovering,
            "pipelining moved the first SSA's phase"
        );
        assert_eq!(
            after.recovery_hard_deadline, hard_before,
            "pipelining moved the first SSA's hard recovery deadline"
        );
        assert_eq!(
            after.recovery_idle_deadline, idle_before,
            "pipelining moved the first SSA's idle recovery deadline"
        );

        // ...and the second one is independently armed rather than sharing the first's timers.
        let second = sup.ssas.iter().find(|s| s.ssa_id == id2).unwrap();
        assert_eq!(second.phase, SsaPhase::AwaitingDeposit);
        assert!(second.deposit_deadline.is_some());
    }

    // ---------------------------------------------------------------
    // SsaIndex overflow
    // ---------------------------------------------------------------

    #[test]
    fn ssa_index_overflow_fails_closed() {
        let p = pseudonym();
        let (mut sup, _) = SessionPixSupervisor::new(default_cfg(), dims(10, 5), p, Instant::now());
        let now = Instant::now();

        let id1 = ssa_id(p, 1);
        sup.handle_event(&SessionPixEvent::SsaRequestSent(id1), now, 0);
        sup.handle_event(&SessionPixEvent::CommitmentVerified(id1), now, 0);
        sup.handle_event(
            &SessionPixEvent::DepositConfirmed {
                ssa_id: id1,
                amount: sufficient_balance(),
            },
            now,
            0,
        );

        // Wind the allocator to the last usable index, so pipelining the next one overflows.
        sup.next_ssa_index = u32::MAX;

        let actions = sup.handle_event(&SessionPixEvent::AlmostRecovered(id1), now, 0);
        assert!(
            matches!(
                actions.as_slice(),
                [SessionPixAction::Close(SessionPixCloseReason::InvalidTransition)]
            ),
            "expected a single Close, got {actions:?}"
        );
        assert!(sup.closed);
        assert_eq!(
            sup.ssas.len(),
            1,
            "an index that could not be allocated must not leave a record behind"
        );
    }

    // ---------------------------------------------------------------
    // action_result
    // ---------------------------------------------------------------

    #[test]
    fn request_failure_result_closes() {
        let p = pseudonym();
        let (mut sup, _) = SessionPixSupervisor::new(default_cfg(), dims(10, 5), p, Instant::now());
        let now = Instant::now();

        let actions = sup.action_result(
            &SessionPixAction::RequestSsa {
                ssa_ids: vec![ssa_id(p, 1)],
                params: dims(10, 5),
            },
            false,
            now,
        );
        assert_eq!(actions.len(), 1);
        assert!(matches!(
            actions[0],
            SessionPixAction::Close(SessionPixCloseReason::SupervisorUnavailable)
        ));
    }

    // ---------------------------------------------------------------
    // Tombstone and multi-SSA lifecycle
    // ---------------------------------------------------------------

    #[test]
    fn tombstone_expiry_clears_recovered_ssa() {
        let mut cfg = default_cfg();
        cfg.tombstone_retention_window = Duration::from_secs(10);
        let p = pseudonym();
        let (mut sup, _) = SessionPixSupervisor::new(cfg, dims(10, 5), p, Instant::now());
        let start = Instant::now();
        let id = ssa_id(p, 1);

        sup.handle_event(&SessionPixEvent::SsaRequestSent(id), start, 0);
        sup.handle_event(&SessionPixEvent::CommitmentVerified(id), start, 0);
        sup.handle_event(
            &SessionPixEvent::DepositConfirmed {
                ssa_id: id,
                amount: sufficient_balance(),
            },
            start,
            0,
        );
        // Recovery pipelines the next SSA, which is tracked from the moment it is requested.
        sup.handle_event(&SessionPixEvent::Recovered(id), start, 0);
        assert_eq!(sup.ssas.len(), 2, "the recovered SSA's tombstone plus its successor");

        // Before tombstone expires — still present.
        let actions = sup.handle_deadline(start + Duration::from_secs(5), 0);
        assert!(actions.is_empty());
        assert_eq!(sup.ssas.len(), 2);

        // After tombstone expires — RetireSsa, and nothing else: the successor is still pending, so
        // the Session has not run out of SSAs. Closing here would kill a Session over a request
        // that is merely in flight.
        let actions = sup.handle_deadline(start + Duration::from_secs(11), 5);
        assert_eq!(actions.len(), 1, "expected RetireSsa alone, got {actions:?}");
        assert!(
            matches!(&actions[0], SessionPixAction::RetireSsa(rid) if *rid == id),
            "action should be RetireSsa({id}), got {:?}",
            actions[0]
        );
        assert_eq!(
            sup.ssas.len(),
            1,
            "the successor must survive its predecessor's retirement"
        );

        let seen = sup.dims.target_useful_shares() + 1;
        let actions = sup.handle_event(
            &SessionPixEvent::RecoveryProgress(make_recovered_tail_progress(&sup, id, seen)),
            start + Duration::from_secs(12),
            6,
        );
        assert!(
            actions
                .iter()
                .any(|action| matches!(action, SessionPixAction::ProgressNotification)),
            "short tombstone cleanup must not truncate a longer paid FIFO drain: {actions:?}"
        );
    }

    #[test]
    fn all_ssas_terminal_closes_session_after_multi_ssa_close() {
        let p = pseudonym();
        let (mut sup, _) = SessionPixSupervisor::new(default_cfg(), dims(10, 5), p, Instant::now());
        let start = Instant::now();
        let id1 = ssa_id(p, 1);

        // Set up first SSA through recovery.
        sup.handle_event(&SessionPixEvent::SsaRequestSent(id1), start, 0);
        sup.handle_event(&SessionPixEvent::CommitmentVerified(id1), start, 0);
        sup.handle_event(
            &SessionPixEvent::DepositConfirmed {
                ssa_id: id1,
                amount: sufficient_balance(),
            },
            start,
            0,
        );
        // AlmostRecovered triggers next SSA request.
        let actions = sup.handle_event(&SessionPixEvent::AlmostRecovered(id1), start, 0);
        assert_eq!(actions.len(), 1);
        assert!(matches!(actions[0], SessionPixAction::RequestSsa { .. }));
        sup.handle_event(&SessionPixEvent::SsaRequestSent(ssa_id(p, 2)), start, 0);

        // Both deadlines have expired by now (SSA 1 hard deadline at start + 3600s,
        // SSA 2 commitment deadline at start + 20s). The loop processes SSA 1 first
        // (RecoveryDeadline → removed), then SSA 2 (CommitmentTimeout → session close).
        let actions = sup.handle_deadline(start + Duration::from_secs(7200), 5);
        assert_eq!(actions.len(), 2, "expected RetireSsa(id1) + Close(RecoveryDeadline)");
        assert!(
            matches!(&actions[0], SessionPixAction::RetireSsa(rid) if *rid == id1),
            "first action should be RetireSsa({id1}), got {:?}",
            actions[0]
        );
        // SSA 1 fails with RecoveryDeadline first, so that reason is surfaced.
        assert!(matches!(
            actions[1],
            SessionPixAction::Close(SessionPixCloseReason::RecoveryDeadline)
        ));
        assert!(sup.closed);
    }

    #[test]
    fn next_deadline_none_when_after_tombstone_expiry() {
        let mut cfg = default_cfg();
        cfg.tombstone_retention_window = Duration::from_secs(10);
        let p = pseudonym();
        let (mut sup, _) = SessionPixSupervisor::new(cfg, dims(10, 5), p, Instant::now());
        let start = Instant::now();
        let id = ssa_id(p, 1);

        sup.handle_event(&SessionPixEvent::SsaRequestSent(id), start, 0);
        sup.handle_event(&SessionPixEvent::CommitmentVerified(id), start, 0);
        sup.handle_event(
            &SessionPixEvent::DepositConfirmed {
                ssa_id: id,
                amount: sufficient_balance(),
            },
            start,
            0,
        );
        sup.handle_event(&SessionPixEvent::Recovered(id), start, 0);

        // Finish the paid FIFO tail; this test is about tombstone scheduling, not drain liveness.
        let max_seen = max_cycle_shares(&sup);
        sup.handle_event(
            &SessionPixEvent::RecoveryProgress(make_recovered_tail_progress(&sup, id, max_seen)),
            start,
            max_seen,
        );

        // While tombstone is alive, next_deadline returns the tombstone_until.
        assert!(sup.next_deadline().is_some());

        // After tombstone expires, handle_deadline removes it and closes.
        sup.handle_deadline(start + Duration::from_secs(11), 0);

        assert!(sup.next_deadline().is_none());
    }

    #[test]
    fn action_result_close_sets_closed() {
        let p = pseudonym();
        let (mut sup, _actions) = SessionPixSupervisor::new(default_cfg(), dims(10, 5), p, Instant::now());
        let now = Instant::now();
        assert!(!sup.closed);

        // First fake-close via action_result.
        let _ = sup.action_result(&SessionPixAction::Close(SessionPixCloseReason::RecoveryIdle), true, now);
        assert!(sup.closed);

        // After close, all subsequent calls are no-ops.
        let actions = sup.handle_deadline(now, 0);
        assert!(actions.is_empty());
        let actions = sup.handle_event(&SessionPixEvent::SsaRequestSent(ssa_id(p, 1)), now, 0);
        assert!(actions.is_empty());
    }

    // -------------------------------------------------------------------
    // M-02: Event ordering guards
    // -------------------------------------------------------------------

    #[test]
    fn recovered_before_commitment_is_ignored() {
        let p = pseudonym();
        let (mut sup, _) = SessionPixSupervisor::new(default_cfg(), dims(10, 5), p, Instant::now());
        let now = Instant::now();
        let id = ssa_id(p, 1);

        sup.handle_event(&SessionPixEvent::SsaRequestSent(id), now, 0);

        // Recovered arrives before CommitmentVerified — should be ignored.
        let actions = sup.handle_event(&SessionPixEvent::Recovered(id), now, 0);
        assert!(actions.is_empty(), "recovered before commitment should be ignored");
        assert!(!sup.closed);
    }

    #[test]
    fn recovered_before_deposit_is_ignored() {
        let p = pseudonym();
        let (mut sup, _) = SessionPixSupervisor::new(default_cfg(), dims(10, 5), p, Instant::now());
        let now = Instant::now();
        let id = ssa_id(p, 1);

        sup.handle_event(&SessionPixEvent::SsaRequestSent(id), now, 0);
        sup.handle_event(&SessionPixEvent::CommitmentVerified(id), now, 0);

        // Recovered arrives before DepositConfirmed — should be ignored.
        let actions = sup.handle_event(&SessionPixEvent::Recovered(id), now, 0);
        assert!(actions.is_empty(), "recovered before deposit should be ignored");
    }

    /// A fully recovered and subsequently funded SSA must not lead to the
    /// session being closed with `RecoveryIdle`. The `Recovered` event
    /// arriving during `AwaitingDeposit` is deferred via `recovered_pending`
    /// — once the deposit confirms and the SSA enters `Recovering`, the
    /// deferred tombstone transition fires immediately. Its recovery clocks then bound only the
    /// compact paid tail: with no service consumed, idle expiry re-arms rather than closing.
    #[test]
    fn recovered_before_deposit_then_funded_session_survives() {
        let p = pseudonym();
        let start = Instant::now();
        let (mut sup, _) = SessionPixSupervisor::new(default_cfg(), dims(2, 2), p, start);
        let id = ssa_id(p, 1);

        sup.handle_event(&SessionPixEvent::SsaRequestSent(id), start, 0);
        sup.handle_event(&SessionPixEvent::CommitmentVerified(id), start, 0);

        // Entry delivered all shares before the on-chain deposit confirmed.
        // Recovered arrives while still AwaitingDeposit — deferred.
        sup.handle_event(&SessionPixEvent::Recovered(id), start, 0);

        // The deposit confirms — triggers deferred tombstone + RequestSsa.
        let actions = sup.handle_event(
            &SessionPixEvent::DepositConfirmed {
                ssa_id: id,
                amount: sufficient_balance(),
            },
            start,
            10,
        );
        assert!(!sup.closed, "session must be alive after funding");

        // The first SSA should have been tombstoned (not stuck in Recovering).
        let ssa1_phase = sup.ssas.iter().find(|s| s.ssa_id == id).map(|s| &s.phase);
        assert!(
            matches!(ssa1_phase, Some(SsaPhase::Recovered { .. })),
            "SSA 1 should be tombstoned, got {ssa1_phase:?}"
        );

        // A RequestSsa for the next SSA must have been emitted (so the
        // driver can start the successor).
        let next_id = ssa_id(p, 2);
        assert!(
            actions
                .iter()
                .any(|a| matches!(a, SessionPixAction::RequestSsa { ssa_ids, .. } if ssa_ids.contains(&next_id))),
            "expected RequestSsa for SSA 2, got actions: {actions:?}"
        );

        // No packet was served after the handoff baseline, so the tail idle clock re-arms. Service
        // consumption without a corresponding old-tail or successor signal would close instead;
        // that is the bound preventing a declared surplus from becoming indefinite free service.
        let deadline_actions = sup.handle_deadline(start + Duration::from_secs(61), 10);
        assert!(
            !deadline_actions
                .iter()
                .any(|a| matches!(a, SessionPixAction::Close(SessionPixCloseReason::RecoveryIdle))),
            "must not close with RecoveryIdle, got: {deadline_actions:?}"
        );
    }

    #[test]
    fn the_immediate_recovered_tail_reports_bounded_liveness() {
        let p = pseudonym();
        let (mut sup, _) = SessionPixSupervisor::new(default_cfg(), dims(10, 5), p, Instant::now());
        let start = Instant::now();
        let id = ssa_id(p, 1);

        sup.handle_event(&SessionPixEvent::SsaRequestSent(id), start, 0);
        sup.handle_event(&SessionPixEvent::CommitmentVerified(id), start, 0);
        sup.handle_event(
            &SessionPixEvent::DepositConfirmed {
                ssa_id: id,
                amount: sufficient_balance(),
            },
            start,
            0,
        );
        sup.handle_event(&SessionPixEvent::Recovered(id), start, 0);

        let seen = sup.dims.target_useful_shares() + 1;
        let actions = sup.handle_event(
            &SessionPixEvent::RecoveryProgress(make_recovered_tail_progress(&sup, id, seen)),
            start,
            100,
        );
        assert!(
            actions
                .iter()
                .any(|action| matches!(action, SessionPixAction::ProgressNotification)),
            "the authorized predecessor must keep the gate alive while its FIFO tail drains: {actions:?}"
        );
    }

    /// A concurrent acknowledgement batch can publish an older absolute snapshot after the
    /// terminal event. It must not be accepted as tail liveness, but neither may it close an honest
    /// Session merely because `Recovered` overtook it.
    #[test]
    fn a_pre_recovery_snapshot_overtaken_by_recovered_is_absorbed() {
        let mut cfg = default_cfg();
        cfg.ssas_per_request = 2;
        let p = pseudonym();
        let start = Instant::now();
        let (mut sup, _) = SessionPixSupervisor::new(cfg, dims(10, 5), p, start);
        let recovered = ssa_id(p, 1);
        fund_batch(&mut sup, p, 2, start);
        sup.handle_event(&SessionPixEvent::Recovered(recovered), start, 0);

        let target = sup.dims.target_useful_shares();
        let stale = SsaRecoveryProgress {
            ssa_id: recovered,
            useful_shares: target - 1,
            shares_seen: target + 1,
            target_useful_shares: target,
            recovered_polynomials: sup.dims.polys_per_ssa() - 1,
        };
        let actions = sup.handle_event(&SessionPixEvent::RecoveryProgress(stale), start, 1);
        assert!(
            actions.is_empty(),
            "an overtaken non-terminal snapshot must be stale noise, not liveness or a close: {actions:?}"
        );
        assert_eq!(
            sup.paid_recovery_tail.as_ref().map(|tail| tail.largest_shares_seen),
            Some(target),
            "the stale snapshot must not consume any paid tail"
        );

        let actions = sup.handle_event(
            &SessionPixEvent::RecoveryProgress(make_recovered_tail_progress(&sup, recovered, target + 2)),
            start,
            2,
        );
        assert!(
            actions
                .iter()
                .any(|action| matches!(action, SessionPixAction::ProgressNotification)),
            "a later terminal snapshot must still consume the legitimate tail: {actions:?}"
        );
    }

    /// `Recovered` is itself proof of fresh progress, even if its immediately preceding progress
    /// snapshot was dropped. The tail gets a new idle interval, while retaining the original hard
    /// deadline.
    #[test]
    fn recovery_refreshes_the_tail_idle_clock_once() {
        let mut cfg = default_cfg();
        cfg.ssas_per_request = 2;
        cfg.max_recovery_idle = Duration::from_secs(10);
        let p = pseudonym();
        let start = Instant::now();
        let (mut sup, _) = SessionPixSupervisor::new(cfg, dims(10, 5), p, start);
        let recovered = ssa_id(p, 1);
        fund_batch(&mut sup, p, 2, start);

        let recovered_at = start + Duration::from_secs(9);
        sup.handle_event(&SessionPixEvent::Recovered(recovered), recovered_at, 5);
        let actions = sup.handle_deadline(start + Duration::from_secs(11), 6);
        assert!(
            actions
                .iter()
                .all(|action| !matches!(action, SessionPixAction::Close(SessionPixCloseReason::RecoveryIdle))),
            "the pre-recovery idle instant must not expire a freshly recovered tail: {actions:?}"
        );

        let actions = sup.handle_deadline(start + Duration::from_secs(20), 6);
        assert!(
            matches!(
                actions.as_slice(),
                [SessionPixAction::Close(SessionPixCloseReason::RecoveryIdle)]
            ),
            "the one-time refresh must not weaken the ordinary stalled-tail bound: {actions:?}"
        );
    }

    /// Defense in depth: even a forged local snapshot cannot exceed the negotiated cycle volume.
    #[test]
    fn a_recovered_tail_cannot_report_more_service_than_the_entry_paid_for() {
        let mut cfg = default_cfg();
        cfg.ssas_per_request = 2;
        let p = pseudonym();
        let start = Instant::now();
        let (mut sup, _) = SessionPixSupervisor::new(cfg, dims(10, 5), p, start);
        let recovered = ssa_id(p, 1);
        fund_batch(&mut sup, p, 2, start);
        sup.handle_event(&SessionPixEvent::Recovered(recovered), start, 0);

        let impossible = max_cycle_shares(&sup) + 1;
        let actions = sup.handle_event(
            &SessionPixEvent::RecoveryProgress(make_recovered_tail_progress(&sup, recovered, impossible)),
            start,
            1,
        );
        assert!(
            matches!(
                actions.as_slice(),
                [SessionPixAction::Close(SessionPixCloseReason::CounterRegression)]
            ),
            "over-budget recovered progress must fail closed, never become gate liveness: {actions:?}"
        );
        assert_eq!(
            sup.paid_recovery_tail.as_ref().map(|tail| tail.largest_shares_seen),
            Some(sup.dims.target_useful_shares()),
            "the rejected snapshot must not mutate the authorized absolute watermark"
        );
    }

    /// Declaring a paid remainder is not itself liveness: silence while consuming service closes.
    #[test]
    fn a_stalled_recovered_tail_cannot_hold_the_session_open() {
        let mut cfg = default_cfg();
        cfg.ssas_per_request = 2;
        cfg.max_recovery_idle = Duration::from_secs(10);
        let p = pseudonym();
        let start = Instant::now();
        let (mut sup, _) = SessionPixSupervisor::new(cfg, dims(10, 5), p, start);
        let recovered = ssa_id(p, 1);
        fund_batch(&mut sup, p, 2, start);
        sup.handle_event(&SessionPixEvent::Recovered(recovered), start, 0);

        let actions = sup.handle_deadline(start + Duration::from_secs(11), 1);
        assert!(
            matches!(
                actions.as_slice(),
                [SessionPixAction::Close(SessionPixCloseReason::RecoveryIdle)]
            ),
            "the Entry cannot turn an unspent declared surplus into indefinite funded service: {actions:?}"
        );
    }

    /// Valid tail traffic may refresh the idle timer, but it cannot stretch the paid predecessor's
    /// original absolute resource deadline.
    #[test]
    fn a_trickling_recovered_tail_cannot_extend_the_hard_deadline() {
        let mut cfg = default_cfg();
        cfg.ssas_per_request = 2;
        cfg.max_recovery_idle = Duration::from_secs(10);
        cfg.max_recovery_time = Duration::from_secs(20);
        let p = pseudonym();
        let start = Instant::now();
        let (mut sup, _) = SessionPixSupervisor::new(cfg, dims(10, 5), p, start);
        let recovered = ssa_id(p, 1);
        fund_batch(&mut sup, p, 2, start);
        sup.handle_event(
            &SessionPixEvent::Recovered(recovered),
            start + Duration::from_secs(1),
            0,
        );

        let target = sup.dims.target_useful_shares();
        for (at, seen, served) in [(9, target + 1, 1), (18, target + 2, 2)] {
            let actions = sup.handle_event(
                &SessionPixEvent::RecoveryProgress(make_recovered_tail_progress(&sup, recovered, seen)),
                start + Duration::from_secs(at),
                served,
            );
            assert!(
                actions
                    .iter()
                    .all(|action| !matches!(action, SessionPixAction::Close(_))),
                "legitimate tail progress before the hard limit must remain live: {actions:?}"
            );
        }

        let actions = sup.handle_deadline(start + Duration::from_secs(21), 3);
        assert!(
            matches!(
                actions.as_slice(),
                [SessionPixAction::Close(SessionPixCloseReason::RecoveryDeadline)]
            ),
            "tail trickle must not move the predecessor's absolute deadline: {actions:?}"
        );
        // The tail took the front at +1 s and ends here, so it owes its 20 s like any other phase
        // that reaches a terminal edge — being ended by a deadline is still ending.
        assert_eq!(
            Some(&(PixCyclePhase::PaidTail, 20_000)),
            sup.take_telemetry_events()
                .phase_durations
                .iter()
                .find(|(phase, _)| *phase == PixCyclePhase::PaidTail)
        );
    }

    /// Recovered does not mean replayable: only the immediate predecessor owns the one tail slot.
    #[test]
    fn an_older_recovered_cycle_cannot_reopen_service() {
        let mut cfg = default_cfg();
        cfg.ssas_per_request = 2;
        cfg.max_off_front_share_fraction = 1.0;
        let p = pseudonym();
        let start = Instant::now();
        let (mut sup, _) = SessionPixSupervisor::new(cfg, dims(10, 5), p, start);
        let first = ssa_id(p, 1);
        let second = ssa_id(p, 2);
        fund_batch(&mut sup, p, 2, start);

        sup.handle_event(&SessionPixEvent::Recovered(first), start, 0);
        sup.handle_event(
            &SessionPixEvent::RecoveryProgress(make_progress(second, 1, sup.dims.target_useful_shares(), 0)),
            start,
            1,
        );
        sup.handle_event(&SessionPixEvent::Recovered(second), start, 1);

        let actions = sup.handle_event(
            &SessionPixEvent::RecoveryProgress(make_recovered_tail_progress(
                &sup,
                first,
                sup.dims.target_useful_shares() + 1,
            )),
            start,
            2,
        );
        assert!(
            actions.is_empty(),
            "an obsolete recovered-cycle tag must not refresh or reopen the Exit's service gate: {actions:?}"
        );
    }

    #[test]
    fn late_tombstone_unverifiable_shares_are_absorbed() {
        let p = pseudonym();
        let (mut sup, _) = SessionPixSupervisor::new(default_cfg(), dims(10, 5), p, Instant::now());
        let start = Instant::now();
        let id = ssa_id(p, 1);

        sup.handle_event(&SessionPixEvent::SsaRequestSent(id), start, 0);
        sup.handle_event(&SessionPixEvent::CommitmentVerified(id), start, 0);
        sup.handle_event(
            &SessionPixEvent::DepositConfirmed {
                ssa_id: id,
                amount: sufficient_balance(),
            },
            start,
            0,
        );
        sup.handle_event(&SessionPixEvent::Recovered(id), start, 0);

        // Late unverifiable shares on tombstone — should be absorbed.
        let actions = sup.handle_event(
            &SessionPixEvent::UnverifiableShares {
                ssa_id: id,
                observed_total: 5,
                peer: crate::supervision::test_peer(),
            },
            start,
            100,
        );
        assert!(
            actions.is_empty(),
            "late tombstone unverifiable shares should be absorbed"
        );
    }

    // ---------------------------------------------------------------
    // PIX fill
    // ---------------------------------------------------------------

    /// Dimensions wide enough that the required fill rate is a whole number of packets per second.
    ///
    /// [`dims`]`(10, 5)` is 120 packets over an hour, which the planner correctly expresses as one
    /// packet every twenty-odd seconds — true, but not a rate any assertion about a *ramp* can be
    /// written against. These are the profiled dimensions of the worked example: 145 408 packets,
    /// which needs about 57 packets/s to clear the aim point.
    fn fill_dims() -> PixParams {
        dims(2048, 64)
    }

    /// The rate carried by the last `SetFillRate` in `actions`, if any.
    fn planned_rate(actions: &[SessionPixAction]) -> Option<FillRate> {
        actions
            .iter()
            .filter_map(|action| match action {
                SessionPixAction::SetFillRate(rate) => Some(*rate),
                _ => None,
            })
            .next_back()
    }

    /// A funded, clocked cycle at the front, with the fill planner untouched.
    ///
    /// Returns the supervisor and the cycle's id. `now` is both the construction instant and the
    /// moment the deposit confirms, so the cycle's hard deadline is exactly `max_recovery_time` out.
    fn funded_at_front(cfg: SupervisorConfig, dims: PixParams, now: Instant) -> (SessionPixSupervisor, HoprPseudonym) {
        let p = pseudonym();
        let (mut sup, _) = SessionPixSupervisor::new(cfg, dims, p, now);
        fund_batch(&mut sup, p, 1, now);
        (sup, p)
    }

    /// Feeds `seen` shares of liveness to `id`, as the reconstructor would.
    fn feed_seen(sup: &mut SessionPixSupervisor, id: SsaId<HoprPseudonym>, seen: u64, now: Instant, served: u64) {
        let target = sup.dims.target_useful_shares();
        let progress = make_progress_seen(id, seen.min(target), seen, target, 0);
        sup.handle_event(&SessionPixEvent::RecoveryProgress(progress), now, served);
    }

    /// Nothing is filled for a Session whose first cycle has not been paid for.
    ///
    /// Fill exists to protect a *deposit*, so an unfunded cycle has nothing at stake and must cost
    /// nothing. The stronger half of the property is the tick itself: `next_fill_tick` returning
    /// `None` is what keeps every unfunded PIX Session — including every one that will never be
    /// funded — from waking its worker once a second for the whole of its commitment and deposit
    /// deadlines.
    #[test]
    fn no_fill_is_planned_or_ticked_before_a_cycle_is_funded() {
        let t0 = Instant::now();
        let p = pseudonym();
        let (mut sup, _) = SessionPixSupervisor::new(default_cfg(), fill_dims(), p, t0);

        assert_eq!(None, sup.next_fill_tick(), "a fresh Session has nothing to fill for");

        commit_unfunded(&mut sup, p, 1..=1, t0);
        assert_eq!(
            None,
            sup.next_fill_tick(),
            "a committed but unfunded cycle still has nothing at stake"
        );

        // Ticked by hand anyway, in case a caller drives the timers without consulting the schedule.
        let actions = sup.handle_timers(t0 + Duration::from_secs(2), 0);
        assert_eq!(None, planned_rate(&actions), "an unfunded cycle must not be filled");
    }

    /// A cycle that recovers with nothing buffered behind it stops fill until its successor is paid.
    ///
    /// Recovery frees the reconstructor's state, but it does not put the successor at the front: the
    /// successor's clocks arm only once the predecessor's FIFO tail is exhausted, and until then there
    /// is no deadline to plan against. Filling on regardless would spend SURBs on a cycle nobody has
    /// deposited for yet — and `shares_seen` at the full emission is exactly the case where no tail is
    /// created, so this is the handoff with no cover at all.
    #[test]
    fn a_recovered_cycle_without_a_tail_stops_fill_and_plans_nothing_for_its_successor() {
        let t0 = Instant::now();
        let (mut sup, p) = funded_at_front(default_cfg(), fill_dims(), t0);
        let id = ssa_id(p, 1);

        let running = planned_rate(&sup.handle_timers(t0 + SAMPLING_INTERVAL, 0))
            .expect("a funded cycle at the front must be filled");
        assert!(!running.is_zero());

        // Fully emitted, so `perform_recovered_transition` leaves no paid tail behind.
        let whole_cycle = max_cycle_shares(&sup);
        feed_seen(&mut sup, id, whole_cycle, t0 + Duration::from_secs(2), 0);
        let actions = sup.handle_event(&SessionPixEvent::Recovered(id), t0 + Duration::from_secs(2), 0);
        assert_eq!(
            Some(FillRate::ZERO),
            planned_rate(&actions),
            "recovery without a tail must silence fill in the same action vector"
        );

        assert_eq!(
            None,
            sup.next_fill_tick(),
            "an unfunded successor is not something to wake up for"
        );
        let actions = sup.handle_timers(t0 + Duration::from_secs(4), 0);
        assert_eq!(None, planned_rate(&actions), "and nothing is planned for it either");
    }

    /// The whole point: an idle funded cycle is carried to completion inside its own deadline.
    ///
    /// The Session sends not one application packet — `served_total` never moves — so every share
    /// that comes back was earned by a fill keep-alive. The simulation closes the loop the deployment
    /// does: whatever fill emits in a second arrives as liveness in the next one, and the planner
    /// re-derives the rate from the remainder that leaves.
    ///
    /// Two properties, and the second is why `max_rate` is a configuration item rather than a
    /// constant: the cycle must finish *before the aim point*, and it must do so without the planner
    /// ever asking for more than the ceiling it was given.
    #[test]
    fn an_idle_funded_cycle_is_filled_to_completion_before_its_aim_point() {
        let cfg = default_cfg();
        let t0 = Instant::now();
        let (mut sup, p) = funded_at_front(cfg.clone(), fill_dims(), t0);
        let id = ssa_id(p, 1);

        let cycle = max_cycle_shares(&sup);
        let aim_point = cfg.max_recovery_time.mul_f64(cfg.fill.finish_fraction);

        let mut rate = FillRate::ZERO;
        let mut sent = 0.0_f64;
        let mut completed_at = None;
        for second in 1..=aim_point.as_secs() {
            let now = t0 + Duration::from_secs(second);
            if let Some(planned) = planned_rate(&sup.handle_timers(now, 0)) {
                assert!(
                    planned.as_packets_per_sec() <= cfg.fill.max_rate as f64,
                    "fill asked for {} packets/s against a ceiling of {}",
                    planned.as_packets_per_sec(),
                    cfg.fill.max_rate
                );
                rate = planned;
            }

            // A lossless return path, so the packets sent in this second are the shares seen in it.
            sent += rate.as_packets_per_sec();
            if sent as u64 >= cycle {
                completed_at = Some(second);
                break;
            }
            feed_seen(&mut sup, id, sent as u64, now, 0);
        }

        let completed_at = completed_at.unwrap_or_else(|| {
            panic!("an idle cycle of {cycle} packets was not filled to completion within {aim_point:?}")
        });
        assert!(
            Duration::from_secs(completed_at) < aim_point,
            "the cycle completed at {completed_at}s, which is not inside the {aim_point:?} aim point"
        );
    }

    /// An application already sending faster than the requirement is not helped, only accompanied.
    ///
    /// Fill is `required − organic`, so a Session whose own traffic covers the cycle needs no help and
    /// must not be charged for any: what is left is the heartbeat, which exists for the Entry's idle
    /// timer and its SURB balancer rather than for the deadline. Getting this wrong is not a small
    /// waste — it doubles the return traffic of every busy PIX Session on the node.
    #[test]
    fn organic_traffic_above_the_requirement_holds_fill_at_the_heartbeat() {
        let cfg = default_cfg();
        let t0 = Instant::now();
        let (mut sup, p) = funded_at_front(cfg.clone(), fill_dims(), t0);
        let id = ssa_id(p, 1);

        // Comfortably above the ~57 packets/s the cycle needs, and sustained past the averaging
        // window so the estimate is the application's rate rather than its first second.
        const ORGANIC: u64 = 200;
        let mut rate = None;
        for second in 1..=(4 * ORGANIC_WINDOW.as_secs()) {
            let now = t0 + Duration::from_secs(second);
            let served = ORGANIC * second;
            if let Some(planned) = planned_rate(&sup.handle_timers(now, served)) {
                rate = Some(planned);
            }
            feed_seen(&mut sup, id, served, now, served);
        }

        assert_eq!(
            Some(FillRate::once_per(cfg.fill.heartbeat)),
            rate,
            "an application outrunning the requirement must leave fill at its heartbeat"
        );
    }

    /// When the application stops, fill picks the cycle up from the remainder that is actually left.
    ///
    /// This is the case the deployment is full of — a VPN Session that transfers and then idles — and
    /// the trap in it is arithmetic rather than plumbing: the planner must re-derive the rate from
    /// `E − shares seen`, not from the rate it would have asked for at the start. A planner that
    /// replayed its opening figure would over-send by however much the application already
    /// contributed.
    #[test]
    fn fill_resumes_from_the_true_remainder_when_organic_traffic_stops() {
        let cfg = default_cfg();
        let t0 = Instant::now();
        let (mut sup, p) = funded_at_front(cfg.clone(), fill_dims(), t0);
        let id = ssa_id(p, 1);

        const ORGANIC: u64 = 200;
        const BUSY_FOR: u64 = 300;
        let mut seen = 0;
        for second in 1..=BUSY_FOR {
            let now = t0 + Duration::from_secs(second);
            seen = ORGANIC * second;
            sup.handle_timers(now, seen);
            feed_seen(&mut sup, id, seen, now, seen);
        }

        // The application stops dead. `served_total` is frozen from here on.
        let mut rate = None;
        for second in (BUSY_FOR + 1)..=(BUSY_FOR + 4 * ORGANIC_WINDOW.as_secs()) {
            let now = t0 + Duration::from_secs(second);
            if let Some(planned) = planned_rate(&sup.handle_timers(now, seen)) {
                rate = Some(planned);
            }
        }

        let elapsed = Duration::from_secs(BUSY_FOR + 4 * ORGANIC_WINDOW.as_secs());
        let horizon = cfg.max_recovery_time.mul_f64(cfg.fill.finish_fraction) - elapsed;
        let expected = (max_cycle_shares(&sup) - seen) as f64 * (1.0 + cfg.fill.loss_margin) / horizon.as_secs_f64();
        let actual = rate
            .expect("fill must resume once the application stops")
            .as_packets_per_sec();
        assert!(
            (actual - expected).abs() <= 0.15 * expected,
            "fill resumed at {actual} packets/s against a remainder that needs {expected}"
        );
    }

    /// The recovered predecessor's tail is filled against *its* clock, and the successor waits.
    ///
    /// The SURBs the Exit holds when a cycle recovers were minted for that cycle and sit ahead of the
    /// successor's in the same FIFO, so packets sent now carry the predecessor's shares and count
    /// against the predecessor's deadline — which is why the successor's clocks are deliberately not
    /// armed until the tail is spent. Planning against the successor here would aim at a deadline
    /// two hours further out and under-send the tail that is actually draining.
    #[test]
    fn the_paid_recovery_tail_is_filled_against_its_predecessors_deadline() {
        let cfg = default_cfg();
        let t0 = Instant::now();
        let (mut sup, p) = funded_at_front(cfg.clone(), fill_dims(), t0);
        let first = ssa_id(p, 1);
        let hard = sup.ssas[0]
            .recovery_hard_deadline
            .expect("a funded front cycle is clocked");

        // Recovered with emission still outstanding, which is what leaves a paid tail behind.
        let partial = max_cycle_shares(&sup) / 2;
        feed_seen(&mut sup, first, partial, t0 + Duration::from_secs(1), 0);
        sup.handle_event(&SessionPixEvent::Recovered(first), t0 + Duration::from_secs(1), 0);

        let target = sup.fill_target().expect("the paid tail is the thing being filled for");
        assert_eq!(first, target.ssa_id, "the tail, not the successor, is the target");
        assert_eq!(
            hard, target.hard_deadline,
            "the tail must be planned against the clock it inherited, not a fresh one"
        );

        // The successor is funded and reaches the front only once the tail is consumed, which any
        // share on the successor proves.
        let second = ssa_id(p, 2);
        sup.handle_event(&SessionPixEvent::SsaRequestSent(second), t0 + Duration::from_secs(2), 0);
        sup.handle_event(
            &SessionPixEvent::CommitmentVerified(second),
            t0 + Duration::from_secs(2),
            0,
        );
        sup.handle_event(
            &SessionPixEvent::DepositConfirmed {
                ssa_id: second,
                amount: sufficient_balance(),
            },
            t0 + Duration::from_secs(2),
            0,
        );
        assert_eq!(
            Some(first),
            sup.fill_target().map(|target| target.ssa_id),
            "a funded successor does not take the front while the tail is still draining"
        );

        feed_seen(&mut sup, second, 1, t0 + Duration::from_secs(3), 0);
        let target = sup.fill_target().expect("the successor is now the front");
        assert_eq!(second, target.ssa_id);
        assert!(
            target.hard_deadline > hard,
            "the successor's clock starts at the FIFO boundary, so it must be later than its predecessor's"
        );
    }

    /// Within a batch, only the cycle at the front is filled for.
    ///
    /// A batch is served in index order because the Entry's emission window is clamped to one cycle,
    /// so its later members are queued rather than stalled — they have no clocks and can receive no
    /// shares. Filling for one of them would spend SURBs the Entry cannot answer.
    #[test]
    fn only_the_front_cycle_of_a_batch_is_filled_for() {
        let cfg = SupervisorConfig {
            ssas_per_request: 2,
            ..default_cfg()
        };
        let t0 = Instant::now();
        let p = pseudonym();
        let (mut sup, _) = SessionPixSupervisor::new(cfg, fill_dims(), p, t0);
        fund_batch(&mut sup, p, 2, t0);

        assert_eq!(
            Some(ssa_id(p, 1)),
            sup.fill_target().map(|target| target.ssa_id),
            "the second cycle of a batch is queued, not being served"
        );
    }

    /// A cycle that has stopped moving is filled at the heartbeat and no faster.
    ///
    /// The idle rule cannot reach this state: it re-arms whenever no *gated* service was consumed
    /// since the last progress, and fill is ungated, so a Session whose only traffic is fill looks
    /// perfectly quiet to it. Without the stall rule an Exit would spend `max_rate × max_recovery_time`
    /// — half a million packets at the shipped values — on a cycle whose polynomial has already failed
    /// or whose Entry is returning SURBs that carry nothing.
    ///
    /// The second half of the property matters just as much: a cycle that resumes must be filled
    /// again. A stall latch that never released would turn one slow minute into a lost deposit.
    #[test]
    fn a_stalled_cycle_falls_back_to_the_heartbeat_and_recovers_when_it_moves() {
        let cfg = default_cfg();
        let t0 = Instant::now();
        let (mut sup, p) = funded_at_front(cfg.clone(), fill_dims(), t0);
        let id = ssa_id(p, 1);

        let running = planned_rate(&sup.handle_timers(t0 + SAMPLING_INTERVAL, 0)).expect("fill must start");
        assert!(
            running.as_packets_per_sec() > FillRate::once_per(cfg.fill.heartbeat).as_packets_per_sec(),
            "the fixture needs a rate above the heartbeat for the fallback to be observable"
        );

        // Not one share arrives for the whole stall horizon.
        let mut rate = None;
        for second in 2..=(cfg.max_recovery_idle.as_secs() + 2) {
            if let Some(planned) = planned_rate(&sup.handle_timers(t0 + Duration::from_secs(second), 0)) {
                rate = Some(planned);
            }
        }
        assert_eq!(
            Some(FillRate::once_per(cfg.fill.heartbeat)),
            rate,
            "a motionless cycle must be held at the heartbeat"
        );

        // And a single share is enough to put it back to work.
        let resumed_at = t0 + Duration::from_secs(cfg.max_recovery_idle.as_secs() + 3);
        feed_seen(&mut sup, id, 1, resumed_at, 0);
        let rate = planned_rate(&sup.handle_timers(resumed_at + SAMPLING_INTERVAL, 0))
            .expect("progress must lift the fallback");
        assert!(
            rate.as_packets_per_sec() > FillRate::once_per(cfg.fill.heartbeat).as_packets_per_sec(),
            "a cycle that resumed must be filled again, got {rate:?}"
        );
    }

    /// A rate that has barely moved is not re-emitted.
    ///
    /// The planner runs every second for hours and the required rate drifts continuously, so without
    /// hysteresis this would put a per-second action on a channel sized for lifecycle transitions —
    /// and ask the driver to reconfigure a rate controller for a change no packet schedule can
    /// express.
    #[test]
    fn a_rate_that_barely_moved_is_not_re_emitted() {
        let t0 = Instant::now();
        let (mut sup, _) = funded_at_front(default_cfg(), fill_dims(), t0);

        assert!(
            planned_rate(&sup.handle_timers(t0 + SAMPLING_INTERVAL, 0)).is_some(),
            "the first plan is always a change"
        );
        for second in 2..=10 {
            assert_eq!(
                None,
                planned_rate(&sup.handle_timers(t0 + Duration::from_secs(second), 0)),
                "second {second} re-emitted a rate that had not meaningfully changed"
            );
        }
    }

    /// Closing the Session silences fill in the same action vector that carries the close.
    ///
    /// The fill stream outlives this state machine by however long the action driver takes to tear
    /// the Session down, and it is the one egress on the Exit with no application behind it to stop
    /// on its own. A close that did not carry the zero would be followed by keep-alives for a cycle
    /// that no longer exists.
    #[test]
    fn a_close_silences_fill_before_it_is_delivered() {
        let t0 = Instant::now();
        let (mut sup, p) = funded_at_front(default_cfg(), fill_dims(), t0);
        assert!(planned_rate(&sup.handle_timers(t0 + SAMPLING_INTERVAL, 0)).is_some());

        let actions = sup.handle_event(
            &SessionPixEvent::UnverifiableShares {
                ssa_id: ssa_id(p, 1),
                observed_total: 1,
                peer: crate::supervision::test_peer(),
            },
            t0 + Duration::from_secs(2),
            0,
        );
        assert!(
            matches!(actions.first(), Some(SessionPixAction::SetFillRate(rate)) if rate.is_zero()),
            "the zero must lead the close, got {actions:?}"
        );
        assert!(
            actions.iter().any(|action| matches!(
                action,
                SessionPixAction::Close(SessionPixCloseReason::UnverifiableShares)
            )),
            "the close itself must still be there, got {actions:?}"
        );
    }

    /// The same, on the path where a retired cycle exhausts the failure budget.
    ///
    /// A different close path with a different owner — `close_ssa_and_collect` rather than a fault
    /// handler — and the one an Entry can actually reach without misbehaving, by simply losing two
    /// deposits.
    #[test]
    fn retiring_past_the_failure_budget_silences_fill() {
        const BATCH: u32 = 4;

        let cfg = SupervisorConfig {
            ssas_per_request: BATCH as usize,
            max_failed_cycles: 1,
            ..default_cfg()
        };
        let t0 = Instant::now();
        let p = pseudonym();
        let (mut sup, _) = SessionPixSupervisor::new(cfg, fill_dims(), p, t0);

        // Cycles 1 and 2 funded and recovering, so the front is being filled for; 3 and 4 stuck
        // awaiting their deposits, which is the state an unpaid cycle is lost from.
        fund_batch(&mut sup, p, 2, t0);
        commit_unfunded(&mut sup, p, 3..=BATCH, t0);
        assert!(
            planned_rate(&sup.handle_timers(t0 + SAMPLING_INTERVAL, 0)).is_some(),
            "the funded front must be filling before the losses"
        );

        // The first loss is survivable; the second is over the budget and takes the Session with it,
        // funded siblings and all.
        sup.handle_event(&SessionPixEvent::DepositObserverClosed(ssa_id(p, 3)), t0, 0);
        assert!(!sup.closed, "one loss is inside the limit");
        let actions = sup.handle_event(&SessionPixEvent::DepositObserverClosed(ssa_id(p, BATCH)), t0, 0);

        assert!(
            actions
                .iter()
                .any(|action| matches!(action, SessionPixAction::Close(_))),
            "the second lost cycle must close the Session, got {actions:?}"
        );
        assert!(
            matches!(actions.first(), Some(SessionPixAction::SetFillRate(rate)) if rate.is_zero()),
            "the zero must lead the close, got {actions:?}"
        );
    }

    // ---------------------------------------------------------------
    // Draining a closed Session
    // ---------------------------------------------------------------

    /// A closed Session whose buffer covers the rest of the cycle keeps working for it.
    ///
    /// This is the whole feature in one test: the deposit both sides have already paid for is
    /// recoverable, the SURBs to recover it are already held, and the only thing that had been
    /// spending them was a Session that has now gone. Nothing about the close makes the cycle
    /// un-completable, so the supervisor neither closes nor asks for a successor it could never
    /// serve — it plans the remainder at the ceiling and works it off.
    #[test]
    fn a_closed_session_with_enough_surbs_drains_its_funded_cycle() {
        let cfg = default_cfg();
        let t0 = Instant::now();
        let (mut sup, _) = funded_at_front(cfg.clone(), fill_dims(), t0);

        let actions = sup.handle_event(
            &SessionPixEvent::SessionClosed {
                drainable_surbs: max_cycle_shares(&sup),
            },
            t0 + Duration::from_secs(1),
            0,
        );

        assert!(!sup.closed, "a drain leaves the state machine running");
        assert!(sup.draining);
        assert!(
            !actions
                .iter()
                .any(|action| matches!(action, SessionPixAction::Close(_) | SessionPixAction::RequestSsa { .. })),
            "a drain neither closes the Session nor asks for a successor, got {actions:?}"
        );

        // The next tick asks for the whole remainder, which the ceiling then paces.
        let rate =
            planned_rate(&sup.handle_timers(t0 + Duration::from_secs(2), 0)).expect("a draining cycle must be filled");
        assert_eq!(FillRate::per_second(cfg.fill.max_rate), rate);
    }

    /// A buffer one SURB short of the remainder is not worth draining.
    ///
    /// The estimate is an upper bound and the comparison is against the *whole* remaining emission,
    /// because a drain that runs out partway has spent the Session's SURBs down to the reserve and
    /// recovered nothing — strictly worse than closing at once, which is what it must do instead.
    #[test]
    fn a_closed_session_without_enough_surbs_closes_instead_of_draining() {
        let t0 = Instant::now();
        let (mut sup, p) = funded_at_front(default_cfg(), fill_dims(), t0);
        let id = ssa_id(p, 1);

        // Filling before the close, so the refusal has a stream to silence.
        assert!(planned_rate(&sup.handle_timers(t0 + SAMPLING_INTERVAL, 0)).is_some());

        let seen = max_cycle_shares(&sup) / 2;
        feed_seen(&mut sup, id, seen, t0 + Duration::from_secs(2), 0);
        let one_short = max_cycle_shares(&sup) - seen - 1;

        let actions = sup.handle_event(
            &SessionPixEvent::SessionClosed {
                drainable_surbs: one_short,
            },
            t0 + Duration::from_secs(3),
            0,
        );

        assert!(sup.closed);
        assert!(!sup.draining);
        assert!(
            matches!(actions.first(), Some(SessionPixAction::SetFillRate(rate)) if rate.is_zero()),
            "the zero must lead the close, got {actions:?}"
        );
        assert!(
            actions
                .iter()
                .any(|action| matches!(action, SessionPixAction::Close(SessionPixCloseReason::SessionClosed))),
            "got {actions:?}"
        );
    }

    /// With no funded cycle to work for, a close is just a close.
    ///
    /// Two ways to have none. An unfunded front has nothing at stake — no deposit has been paid, so
    /// there is nothing to strand. A paid recovery tail means the front's key is *already* recovered,
    /// so draining it would buy nothing either; and its successor is queued behind the tail with no
    /// clocks of its own, which is precisely the state fill refuses to plan against.
    #[test]
    fn a_closed_session_with_no_funded_cycle_closes_instead_of_draining() {
        let plenty = u64::MAX;

        let t0 = Instant::now();
        let p = pseudonym();
        let (mut sup, _) = SessionPixSupervisor::new(default_cfg(), fill_dims(), p, t0);
        commit_unfunded(&mut sup, p, 1..=1, t0);

        let actions = sup.handle_event(
            &SessionPixEvent::SessionClosed {
                drainable_surbs: plenty,
            },
            t0 + Duration::from_secs(1),
            0,
        );
        assert!(
            sup.closed && !sup.draining,
            "an unfunded front has nothing to drain for"
        );
        assert!(
            actions
                .iter()
                .any(|action| matches!(action, SessionPixAction::Close(SessionPixCloseReason::SessionClosed))),
            "got {actions:?}"
        );

        // And the same with a recovered predecessor's paid tail in front.
        let t0 = Instant::now();
        let (mut sup, p) = funded_at_front(default_cfg(), fill_dims(), t0);
        let first = ssa_id(p, 1);
        let half = max_cycle_shares(&sup) / 2;
        feed_seen(&mut sup, first, half, t0 + Duration::from_secs(1), 0);
        sup.handle_event(&SessionPixEvent::Recovered(first), t0 + Duration::from_secs(1), 0);
        assert!(
            sup.paid_recovery_tail.is_some(),
            "the fixture must leave a paid tail behind for this half to mean anything"
        );

        let actions = sup.handle_event(
            &SessionPixEvent::SessionClosed {
                drainable_surbs: plenty,
            },
            t0 + Duration::from_secs(2),
            0,
        );
        assert!(sup.closed && !sup.draining, "a recovered key is not worth draining for");
        assert!(
            actions
                .iter()
                .any(|action| matches!(action, SessionPixAction::Close(SessionPixCloseReason::SessionClosed))),
            "got {actions:?}"
        );
    }

    /// Everything but the cycle being drained is retired at once, and none of it counts as a failure.
    ///
    /// A drain works for exactly one cycle: the batch's later members are queued behind it and can
    /// receive no share while it drains, and there is no successor coming, so holding their
    /// reconstructor state open would cost memory for the length of the drain and buy nothing. The
    /// failure counter must not move either — these cycles were retired by an operator closing a
    /// Session, not lost by an Entry, and charging them would make a second drain on the same node
    /// look like an Entry over its budget.
    #[test]
    fn a_drain_retires_every_cycle_but_the_one_it_is_recovering() {
        const BATCH: u32 = 4;

        let cfg = SupervisorConfig {
            ssas_per_request: BATCH as usize,
            ..default_cfg()
        };
        let t0 = Instant::now();
        let p = pseudonym();
        let (mut sup, _) = SessionPixSupervisor::new(cfg, fill_dims(), p, t0);
        fund_batch(&mut sup, p, 2, t0);
        commit_unfunded(&mut sup, p, 3..=BATCH, t0);

        let actions = sup.handle_event(
            &SessionPixEvent::SessionClosed {
                drainable_surbs: max_cycle_shares(&sup),
            },
            t0 + Duration::from_secs(1),
            0,
        );

        let retired = actions
            .iter()
            .filter_map(|action| match action {
                SessionPixAction::RetireSsa(id) => Some(id.ssa_index().get()),
                _ => None,
            })
            .collect::<Vec<_>>();
        assert_eq!(
            vec![2, 3, 4],
            retired,
            "every cycle but the one being drained must be retired, got {actions:?}"
        );
        assert!(!sup.closed && sup.draining);
        assert_eq!(0, sup.failed_cycles, "a drain retires siblings, it does not lose them");
        assert_eq!(1, sup.ssas.len());
        assert_eq!(ssa_id(p, 1), sup.ssas[0].ssa_id);
    }

    /// A drain that finishes closes on its own reason and never asks for a successor.
    ///
    /// Both halves matter. `Drained` is what tells an operator the close was the good outcome — the
    /// deposit was recovered — rather than one more Session lost on a timer. And the successor
    /// request has to be suppressed on *every* route into it, the early-recovery signal included:
    /// there is no Session left to serve a successor's quota, so asking for one would commit the
    /// Entry to a deposit against a cycle that can never be served.
    #[test]
    fn a_completed_drain_closes_the_session_without_asking_for_a_successor() {
        let t0 = Instant::now();
        let (mut sup, p) = funded_at_front(default_cfg(), fill_dims(), t0);
        let id = ssa_id(p, 1);
        assert!(planned_rate(&sup.handle_timers(t0 + SAMPLING_INTERVAL, 0)).is_some());

        sup.handle_event(
            &SessionPixEvent::SessionClosed {
                drainable_surbs: max_cycle_shares(&sup),
            },
            t0 + Duration::from_secs(2),
            0,
        );
        assert!(sup.draining);

        let early = sup.handle_event(&SessionPixEvent::AlmostRecovered(id), t0 + Duration::from_secs(3), 0);
        assert!(
            early.is_empty(),
            "the early-recovery signal must not order a successor for a Session that is gone, got {early:?}"
        );

        let actions = sup.handle_event(&SessionPixEvent::Recovered(id), t0 + Duration::from_secs(4), 0);
        assert!(sup.closed);
        assert!(
            matches!(actions.first(), Some(SessionPixAction::SetFillRate(rate)) if rate.is_zero()),
            "the zero must lead the close, got {actions:?}"
        );
        assert!(
            actions
                .iter()
                .any(|action| matches!(action, SessionPixAction::Close(SessionPixCloseReason::Drained))),
            "got {actions:?}"
        );
        assert!(
            !actions
                .iter()
                .any(|action| matches!(action, SessionPixAction::RequestSsa { .. })),
            "got {actions:?}"
        );
    }

    /// A completed drain books its deposit as recovered, not as stranded.
    ///
    /// The drain exists to un-strand a deposit, so it is the one close path that must never leave
    /// its value out of `hopr_pix_deposits_recovered_uhopr_total`. It returns early from the
    /// recovered transition, which is what previously skipped the summary the counter reads.
    #[test]
    fn a_completed_drain_books_the_deposit_it_unlocked_as_recovered() {
        let t0 = Instant::now();
        let (mut sup, p) = funded_at_front(default_cfg(), fill_dims(), t0);
        let id = ssa_id(p, 1);
        // `funded_at_front` confirms `sufficient_balance()`, which is 1000 HOPR.
        assert_eq!(1_000_000_000, sup.take_telemetry_events().deposit_confirmed_uhopr);

        sup.handle_event(
            &SessionPixEvent::SessionClosed {
                drainable_surbs: max_cycle_shares(&sup),
            },
            t0 + Duration::from_secs(2),
            0,
        );
        sup.handle_event(&SessionPixEvent::Recovered(id), t0 + Duration::from_secs(4), 0);

        let events = sup.take_telemetry_events();
        assert_eq!(
            1, events.recovered,
            "a drained cycle recovered, so it counts as recovered"
        );
        let recovered: Vec<_> = events
            .finalized
            .iter()
            .filter(|summary| summary.outcome == PixCycleOutcome::Recovered)
            .collect();
        assert_eq!(
            1,
            recovered.len(),
            "the drain must summarize its cycle, or the value it unlocked reads as stranded"
        );
        assert_eq!(1_000_000_000, recovered[0].deposit_uhopr);
    }

    /// A drain that stops making progress is closed by the idle deadline rather than re-armed.
    ///
    /// The idle rule normally re-arms whenever no *gated* service was consumed since the last
    /// progress, so that a merely quiet Entry is never disconnected. A drained Session has no
    /// application left to consume any, so `served_total` is frozen for good and the re-arm would
    /// never stop firing — the one bound on a drain admitted against an over-estimated buffer would
    /// be gone with it.
    #[test]
    fn a_stalled_drain_closes_on_the_recovery_idle_deadline() {
        let cfg = default_cfg();
        let t0 = Instant::now();
        let (mut sup, _) = funded_at_front(cfg.clone(), fill_dims(), t0);

        sup.handle_event(
            &SessionPixEvent::SessionClosed {
                drainable_surbs: max_cycle_shares(&sup),
            },
            t0,
            0,
        );
        assert!(sup.draining);

        // Not one share, and not one packet of gated service either — which is exactly the shape the
        // re-arm branch was written for, and the shape it must no longer answer to.
        let actions = sup.handle_deadline(t0 + cfg.max_recovery_idle, 0);
        assert!(sup.closed);
        assert!(
            actions
                .iter()
                .any(|action| matches!(action, SessionPixAction::Close(SessionPixCloseReason::RecoveryIdle))),
            "got {actions:?}"
        );
    }

    /// And the hard deadline still ends a drain that is making progress but will not finish.
    ///
    /// The immutable per-cycle backstop is not relaxed for a drain: the reconstructor state it holds
    /// open is the resource the deadline exists to bound, and a closed Session is no reason to hold
    /// it longer. `check_deadlines` tests the hard clock before the idle one, which is why this
    /// reports the deadline rather than the silence that came with it.
    #[test]
    fn a_drain_still_ends_at_its_hard_recovery_deadline() {
        let cfg = default_cfg();
        let t0 = Instant::now();
        let (mut sup, _) = funded_at_front(cfg.clone(), fill_dims(), t0);

        sup.handle_event(
            &SessionPixEvent::SessionClosed {
                drainable_surbs: max_cycle_shares(&sup),
            },
            t0,
            0,
        );
        assert!(sup.draining);

        let actions = sup.handle_deadline(t0 + cfg.max_recovery_time, 0);
        assert!(sup.closed);
        assert!(
            actions
                .iter()
                .any(|action| matches!(action, SessionPixAction::Close(SessionPixCloseReason::RecoveryDeadline))),
            "got {actions:?}"
        );
    }

    /// With fill disabled the supervisor behaves exactly as it did before fill existed.
    ///
    /// Not merely "sends no keep-alives": it must also never schedule a tick, or every PIX Session on
    /// a node that turned the feature off would still wake its worker once a second for the whole of
    /// its life.
    #[test]
    fn a_disabled_filler_plans_nothing_and_schedules_nothing() {
        let cfg = SupervisorConfig {
            fill: PixFillConfig {
                enabled: false,
                ..PixFillConfig::default()
            },
            ..default_cfg()
        };
        let t0 = Instant::now();
        let (mut sup, _) = funded_at_front(cfg, fill_dims(), t0);

        assert_eq!(None, sup.next_fill_tick(), "a disabled filler has no schedule");
        for second in 1..=5 {
            let actions = sup.handle_timers(t0 + Duration::from_secs(second), 0);
            assert_eq!(None, planned_rate(&actions), "a disabled filler must plan nothing");
        }
    }

    // -----------------------------------------------------------------------
    // Bounded aggregate telemetry — the census and the event latch
    // -----------------------------------------------------------------------

    /// The census as the worker reads it, with no predeposit exposure to complicate the comparison.
    fn census(sup: &SessionPixSupervisor) -> PixSessionSnapshot {
        sup.telemetry_snapshot(0)
    }

    /// A census naming the gate mode and the three live phase counts, for comparing against
    /// [`census`].
    ///
    /// The mode is a parameter rather than defaulted because it is half of what these tests are
    /// checking: an `hopr_pix_sessions_active` that stayed in the predeposit bucket after a deposit
    /// confirmed would tell an operator the opposite of the truth.
    fn phases(
        gate_mode: PixGateMode,
        awaiting_commitment: u32,
        awaiting_deposit: u32,
        recovering: u32,
    ) -> PixSessionSnapshot {
        PixSessionSnapshot {
            gate_mode,
            awaiting_commitment,
            awaiting_deposit,
            recovering,
            ..Default::default()
        }
    }

    /// One Session walked end to end, asserting the census and the latch at every step.
    ///
    /// This is the test the whole census design exists for: a gauge that fails to come back down is
    /// invisible to an operator until they compare it against reality, and there is no reality to
    /// compare a metric against in production.
    #[test]
    fn a_session_walks_the_census_from_request_to_rollover() {
        let t0 = Instant::now();
        let p = pseudonym();
        let (mut sup, _) = SessionPixSupervisor::new(default_cfg(), dims(64, 8), p, t0);
        let first = ssa_id(p, 1);

        // `new` allocates and requests the first cycle, so it is already awaiting its commitment.
        assert_eq!(phases(PixGateMode::Predeposit, 1, 0, 0), census(&sup));
        assert_eq!(
            PixTurnEvents {
                requested: 1,
                ..Default::default()
            },
            sup.take_telemetry_events(),
            "allocation is what counts as requested, not the confirmation that it was sent"
        );
        assert!(
            sup.take_telemetry_events().is_empty(),
            "the latch must drain rather than accumulate"
        );

        sup.handle_event(&SessionPixEvent::SsaRequestSent(first), t0, 0);
        assert_eq!(phases(PixGateMode::Predeposit, 1, 0, 0), census(&sup));
        assert!(
            sup.take_telemetry_events().is_empty(),
            "confirming the send is not a lifecycle transition"
        );

        sup.handle_event(&SessionPixEvent::CommitmentVerified(first), t0, 0);
        assert_eq!(phases(PixGateMode::Predeposit, 0, 1, 0), census(&sup));
        assert_eq!(
            PixTurnEvents {
                committed: 1,
                // Zero because this fixture drives every transition at `t0`; that the sample is
                // latched at all is the assertion. `a_cycle_is_timed_in_each_phase_it_leaves`
                // covers the durations themselves.
                phase_durations: vec![(PixCyclePhase::AwaitingCommitment, 0)],
                ..Default::default()
            },
            sup.take_telemetry_events()
        );
        assert_eq!(
            PixGateMode::Predeposit,
            census(&sup).gate_mode,
            "a committed but unfunded front is still spending its predeposit allowance"
        );

        sup.handle_event(
            &SessionPixEvent::DepositConfirmed {
                ssa_id: first,
                amount: sufficient_balance(),
            },
            t0,
            0,
        );
        assert_eq!(phases(PixGateMode::Funded, 0, 0, 1), census(&sup));
        assert_eq!(
            PixTurnEvents {
                funded: 1,
                // `sufficient_balance()` is 1000 HOPR; confirming it books the value alongside the
                // count, in the same turn.
                deposit_confirmed_uhopr: 1_000_000_000,
                phase_durations: vec![(PixCyclePhase::AwaitingDeposit, 0)],
                ..Default::default()
            },
            sup.take_telemetry_events()
        );
        assert_eq!(PixGateMode::Funded, census(&sup).gate_mode);

        // Recovery retires the cycle into a tombstone and asks for its successor, so the census
        // swaps one recovering cycle for one awaiting a commitment — never showing both.
        sup.handle_event(&SessionPixEvent::Recovered(first), t0, 0);
        let after = census(&sup);
        assert_eq!(
            1,
            after.awaiting_commitment + after.awaiting_deposit + after.recovering,
            "a tombstone is retained state, not live state: {after:?}"
        );
        assert_eq!(0, after.recovering, "the recovered cycle must leave the live census");

        let events = sup.take_telemetry_events();
        assert_eq!(
            (1, 1),
            (events.recovered, events.requested),
            "the successor is requested in the same turn the predecessor recovers"
        );
        assert_eq!(
            vec![PixCycleSummary {
                outcome: PixCycleOutcome::Recovered,
                // This cycle was never served: the fixture drives the lifecycle directly, with a
                // served count of zero throughout.
                egress_packets: 0,
                accepted_shares: sup.dims.target_useful_shares(),
                useful_shares: sup.dims.target_useful_shares(),
                target_useful_shares: sup.dims.target_useful_shares(),
                // `sufficient_balance()` is 1000 HOPR, so the deposit the fixture confirmed reaches
                // the summary that decides how much value this recovery unlocked.
                deposit_uhopr: 1_000_000_000,
            }],
            events.finalized,
            "a recovered cycle is summarized exactly once, at full useful coverage"
        );
        assert_eq!(
            Some(1.0),
            events.finalized[0].useful_fraction(),
            "a recovered cycle's useful fraction is one by construction"
        );
    }

    /// A cycle that recovers with its negotiated tail outstanding is reported as `paid_tail`.
    ///
    /// The distinction matters to an operator: the Session is still being served, and still funded,
    /// but by a receipt from a cycle that has already reconstructed rather than by a live one.
    #[test]
    fn a_paid_recovered_predecessor_is_censused_as_a_paid_tail() {
        let t0 = Instant::now();
        let p = pseudonym();
        let (mut sup, _) = SessionPixSupervisor::new(default_cfg(), dims(64, 8), p, t0);
        let first = ssa_id(p, 1);

        fund_batch(&mut sup, p, 1, t0);
        // Some surplus is outstanding, so recovery leaves a receipt behind rather than nothing.
        sup.handle_event(
            &SessionPixEvent::RecoveryProgress(make_progress_seen(
                first,
                sup.dims.target_useful_shares(),
                sup.dims.target_useful_shares(),
                sup.dims.target_useful_shares(),
                sup.dims.polys_per_ssa(),
            )),
            t0,
            0,
        );
        sup.handle_event(&SessionPixEvent::Recovered(first), t0, 0);

        assert_eq!(1, census(&sup).paid_tail, "the receipt must be visible");
        assert_eq!(
            PixGateMode::Funded,
            census(&sup).gate_mode,
            "a paid tail still authorizes service"
        );

        // Draining the receipt to its emitted ceiling consumes it, and the census follows.
        sup.handle_event(
            &SessionPixEvent::RecoveryProgress(make_recovered_tail_progress(&sup, first, max_cycle_shares(&sup))),
            t0,
            0,
        );
        assert_eq!(0, census(&sup).paid_tail, "an exhausted receipt must leave the census");
    }

    /// Every terminal path drains the live census, and each lost cycle is counted exactly once.
    ///
    /// Parameterised over the failure modes rather than written out per reason, because the property
    /// is the same for all of them and the risk is that one path forgets.
    #[test]
    fn every_terminal_path_drains_the_census_and_counts_its_loss_once() {
        let t0 = Instant::now();

        // Each case drives a single-cycle Session into a terminal state a different way, and says
        // which of the two routes its cycle takes out. Stating it is what keeps a case honest: the
        // sum invariant below holds even for a case that has silently stopped reaching its
        // terminal path at all, because a cycle that stays live is absorbed by `still_live`.
        //
        // `Retired` — the supervisor retired the cycle itself, charging it as it went.
        // `LiveAtClose` — the Session closed outright with the cycle still live, so the charge is
        // `PixSessionTelemetry::release`'s to make from the census.
        #[derive(Clone, Copy, PartialEq, Eq, Debug)]
        enum Route {
            Retired,
            LiveAtClose,
        }
        type Kill = fn(&mut SessionPixSupervisor, HoprPseudonym, Instant);
        let cases: [(&str, Route, Kill); 5] = [
            ("commitment timeout", Route::Retired, |sup, p, t0| {
                // `SsaRequestSent` is what arms `commitment_deadline` — a cycle registered by
                // `emit_request_next_ssa` has none, because its request has not gone out and there
                // is nothing for it to be late for. Without this the sweep below expires nothing
                // and the case silently degenerates into "an unarmed cycle stays live".
                sup.handle_event(&SessionPixEvent::SsaRequestSent(ssa_id(p, 1)), t0, 0);
                sup.handle_timers(t0 + Duration::from_secs(3600), 0);
            }),
            ("deposit observer closed", Route::Retired, |sup, p, t0| {
                let id = ssa_id(p, 1);
                sup.handle_event(&SessionPixEvent::SsaRequestSent(id), t0, 0);
                sup.handle_event(&SessionPixEvent::CommitmentVerified(id), t0, 0);
                sup.handle_event(&SessionPixEvent::DepositObserverClosed(id), t0, 0);
            }),
            ("deposit timeout", Route::Retired, |sup, p, t0| {
                let id = ssa_id(p, 1);
                sup.handle_event(&SessionPixEvent::SsaRequestSent(id), t0, 0);
                sup.handle_event(&SessionPixEvent::CommitmentVerified(id), t0, 0);
                sup.handle_timers(t0 + Duration::from_secs(7200), 0);
            }),
            ("recovery hard deadline", Route::Retired, |sup, p, t0| {
                fund_batch(sup, p, 1, t0);
                sup.handle_timers(t0 + Duration::from_secs(100_000), 0);
            }),
            ("unverifiable shares", Route::LiveAtClose, |sup, p, t0| {
                fund_batch(sup, p, 1, t0);
                sup.handle_event(
                    &SessionPixEvent::UnverifiableShares {
                        ssa_id: ssa_id(p, 1),
                        observed_total: 1,
                        peer: crate::supervision::test_peer(),
                    },
                    t0,
                    0,
                );
            }),
        ];

        for (name, route, kill) in cases {
            let p = pseudonym();
            let (mut sup, _) = SessionPixSupervisor::new(default_cfg(), dims(64, 8), p, t0);
            let requested = sup.take_telemetry_events().requested;
            assert_eq!(1, requested, "{name}: one cycle was allocated");

            kill(&mut sup, p, t0);

            let events = sup.take_telemetry_events();
            assert_eq!(0, events.recovered, "{name}: nothing recovered");

            let charged_here = events.failed;
            let still_live = census(&sup).live_cycles();

            // Asserted before the sum, because the sum alone cannot tell a case that reached its
            // terminal path from one that never got there.
            let (expected_charged, expected_live) = match route {
                Route::Retired => (1, 0),
                Route::LiveAtClose => (0, 1),
            };
            assert_eq!(
                (expected_charged, expected_live),
                (charged_here, still_live),
                "{name}: expected the {route:?} route"
            );

            // And the invariant the aggregate actually reports: whichever route the cycle took, it
            // is accounted for exactly once. `UnverifiableShares` closes the Session without
            // retiring its cycles — the fault is evidence about the peer rather than about one
            // cycle — so its charge is `release`'s to make from the census.
            assert_eq!(
                requested,
                charged_here + still_live,
                "{name}: every requested cycle must be accounted for exactly once, got {charged_here} retired and \
                 {still_live} live"
            );
        }
    }

    /// Tombstones and closing records never appear in the live census.
    ///
    /// Stated on its own because it is the failure the census design was chosen to avoid: a
    /// tombstone outlives its cycle by `tombstone_retention_window`, so counting one would leave
    /// `hopr_pix_cycles_active` reporting a cycle the Exit has already finished with.
    #[test]
    fn retained_but_finished_cycles_are_not_live() {
        let t0 = Instant::now();
        let p = pseudonym();
        let cfg = SupervisorConfig {
            ssas_per_request: 2,
            ..default_cfg()
        };
        let (mut sup, _) = SessionPixSupervisor::new(cfg, dims(64, 8), p, t0);

        fund_batch(&mut sup, p, 2, t0);
        assert_eq!(2, census(&sup).recovering);

        sup.handle_event(&SessionPixEvent::Recovered(ssa_id(p, 1)), t0, 0);
        assert!(
            sup.ssas.iter().any(|s| matches!(s.phase, SsaPhase::Recovered { .. })),
            "the tombstone must still be retained, or this test proves nothing"
        );
        assert_eq!(
            1,
            census(&sup).recovering,
            "only the surviving sibling is live; the tombstone is retained state"
        );
    }

    /// A repeated lifecycle event counts its transition once.
    ///
    /// Concurrent acknowledgement batches make duplicates ordinary rather than exceptional, and the
    /// counters must not treat replay as progress.
    #[test]
    fn replayed_lifecycle_events_do_not_double_count() {
        let t0 = Instant::now();
        let p = pseudonym();
        let (mut sup, _) = SessionPixSupervisor::new(default_cfg(), dims(64, 8), p, t0);
        let first = ssa_id(p, 1);
        sup.take_telemetry_events();

        for _ in 0..3 {
            sup.handle_event(&SessionPixEvent::SsaRequestSent(first), t0, 0);
            sup.handle_event(&SessionPixEvent::CommitmentVerified(first), t0, 0);
        }
        assert_eq!(
            1,
            sup.take_telemetry_events().committed,
            "a repeated CommitmentVerified must count one transition"
        );

        for _ in 0..3 {
            sup.handle_event(
                &SessionPixEvent::DepositConfirmed {
                    ssa_id: first,
                    amount: sufficient_balance(),
                },
                t0,
                0,
            );
        }
        assert_eq!(
            1,
            sup.take_telemetry_events().funded,
            "a duplicate deposit confirmation must count one transition"
        );

        for _ in 0..3 {
            sup.handle_event(&SessionPixEvent::Recovered(first), t0, 0);
        }
        assert_eq!(
            1,
            sup.take_telemetry_events().recovered,
            "a duplicate Recovered must count one transition"
        );
    }

    /// A zero deposit is not a verdict, so it must not be counted as funding.
    #[test]
    fn a_zero_deposit_counts_no_funding() {
        let t0 = Instant::now();
        let p = pseudonym();
        let (mut sup, _) = SessionPixSupervisor::new(default_cfg(), dims(64, 8), p, t0);
        let first = ssa_id(p, 1);

        sup.handle_event(&SessionPixEvent::SsaRequestSent(first), t0, 0);
        sup.handle_event(&SessionPixEvent::CommitmentVerified(first), t0, 0);
        sup.take_telemetry_events();

        sup.handle_event(
            &SessionPixEvent::DepositConfirmed {
                ssa_id: first,
                amount: HoprBalance::zero(),
            },
            t0,
            0,
        );

        assert_eq!(0, sup.take_telemetry_events().funded);
        assert_eq!(
            phases(PixGateMode::Predeposit, 0, 1, 0),
            census(&sup),
            "the cycle stays fundable, awaiting a real deposit"
        );
    }

    // -----------------------------------------------------------------------
    // Share coverage — absolute snapshots into positive deltas
    // -----------------------------------------------------------------------

    /// A funded, recovering single-cycle Session, ready to be fed progress snapshots.
    fn recovering_session(p: HoprPseudonym, t0: Instant) -> SessionPixSupervisor {
        let (mut sup, _) = SessionPixSupervisor::new(default_cfg(), dims(64, 8), p, t0);
        fund_batch(&mut sup, p, 1, t0);
        sup.take_telemetry_events();
        sup
    }

    /// Feeds one snapshot and returns the `(useful, surplus)` it booked.
    fn feed(sup: &mut SessionPixSupervisor, id: SsaId<HoprPseudonym>, useful: u64, seen: u64) -> (u64, u64) {
        let target = sup.dims.target_useful_shares();
        let polys = sup.dims.polys_per_ssa();
        // `recovered_polynomials` is scaled with `useful` so the snapshot stays internally
        // consistent; the validator rejects one that is not, and this test is about deltas.
        let recovered = ((useful * polys as u64) / target.max(1)) as u16;
        sup.handle_event(
            &SessionPixEvent::RecoveryProgress(make_progress_seen(id, useful, seen, target, recovered)),
            Instant::now(),
            0,
        );
        let events = sup.take_telemetry_events();
        (events.useful_shares, events.surplus_shares)
    }

    /// Absolute snapshots become positive deltas, and every degenerate ordering adds zero.
    ///
    /// The property the issue names: `shares_seen` is absolute and acknowledgement batches arrive
    /// out of order or coalesced, so adding it to a counter would over-count wildly. What is added
    /// is the difference against a high-water mark.
    #[test]
    fn absolute_snapshots_become_positive_deltas() {
        let t0 = Instant::now();
        let p = pseudonym();
        let mut sup = recovering_session(p, t0);
        let id = ssa_id(p, 1);

        assert_eq!((10, 0), feed(&mut sup, id, 10, 10), "the first snapshot is all delta");
        assert_eq!((5, 0), feed(&mut sup, id, 15, 15), "and the next is the difference");

        // Repeated: the same snapshot again.
        assert_eq!((0, 0), feed(&mut sup, id, 15, 15), "a repeated snapshot adds nothing");
        // Stale: reordered behind one already seen.
        assert_eq!((0, 0), feed(&mut sup, id, 12, 12), "a stale snapshot adds nothing");
        // Still stale on the liveness counter even though useful is unchanged.
        assert_eq!(
            (0, 0),
            feed(&mut sup, id, 15, 14),
            "an older liveness count adds nothing"
        );

        // Coalesced: a jump standing in for snapshots that were dropped. The whole missing delta
        // arrives with it, because a high-water mark is a maximum and not a running sum.
        assert_eq!(
            (85, 0),
            feed(&mut sup, id, 100, 100),
            "a jump carries the missing delta"
        );
    }

    /// Shares that arrive but advance nothing are surplus, and are counted as such.
    ///
    /// This is the tail of every conforming cycle — `surplus` shares per polynomial that arrive
    /// after its threshold is met — so an Exit whose `useful` and `surplus` rates are both healthy
    /// is being served correctly, and one with surplus but no useful progress is not.
    #[test]
    fn shares_that_advance_nothing_are_counted_as_surplus() {
        let t0 = Instant::now();
        let p = pseudonym();
        let mut sup = recovering_session(p, t0);
        let id = ssa_id(p, 1);

        assert_eq!((20, 0), feed(&mut sup, id, 20, 20));
        // Liveness advances, payment does not: the conforming surplus run.
        assert_eq!(
            (0, 5),
            feed(&mut sup, id, 20, 25),
            "a surplus-only snapshot books surplus"
        );
        // A mixed snapshot splits.
        assert_eq!((10, 5), feed(&mut sup, id, 30, 40), "a mixed snapshot splits both ways");
    }

    /// Useful and surplus always partition the accepted delta, over an arbitrary sequence.
    ///
    /// Stated as an invariant over a generated sequence rather than as another worked example,
    /// because the failure it guards against is an accounting slip in one of the branches rather
    /// than a wrong answer to one case.
    #[test]
    fn useful_and_surplus_always_partition_the_accepted_delta() {
        let t0 = Instant::now();
        let p = pseudonym();
        let mut sup = recovering_session(p, t0);
        let id = ssa_id(p, 1);
        let target = sup.dims.target_useful_shares();
        // The greatest liveness count these dimensions can produce: anything above it is a protocol
        // violation the validator closes on, so a walk must stay inside it to be booking deltas at
        // all rather than testing the rejection path.
        let max_seen = max_cycle_shares(&sup);

        let mut booked_useful = 0;
        let mut booked_surplus = 0;
        let mut highest_seen = 0;

        // A deterministic walk that revisits, regresses and jumps, which is what real delivery does.
        for step in [
            (5u64, 5u64),
            (5, 9),
            (3, 4),
            (9, 9),
            (9, 20),
            (9, 20),
            (30, 41),
            (12, 12),
            (target, max_seen),
            (target, max_seen),
        ] {
            let (useful, seen) = step;
            let (u, s) = feed(&mut sup, id, useful, seen);
            assert_eq!(
                seen.saturating_sub(highest_seen),
                u + s,
                "snapshot {step:?} must book exactly its accepted delta"
            );
            highest_seen = highest_seen.max(seen);
            booked_useful += u;
            booked_surplus += s;
        }

        assert_eq!(
            highest_seen,
            booked_useful + booked_surplus,
            "over the whole sequence the two must sum to the greatest liveness count seen"
        );
        assert_eq!(
            target, booked_useful,
            "and the useful half must sum to the greatest useful count seen"
        );
    }

    /// A recovered cycle's drain is all surplus.
    ///
    /// Its useful counter is pinned at the target by recovery, so every further share the FIFO
    /// delivers is accepted traffic that can no longer advance anything.
    #[test]
    fn a_recovered_tail_books_only_surplus() {
        let t0 = Instant::now();
        let p = pseudonym();
        let mut sup = recovering_session(p, t0);
        let id = ssa_id(p, 1);
        let target = sup.dims.target_useful_shares();

        feed(&mut sup, id, target, target);
        sup.handle_event(&SessionPixEvent::Recovered(id), t0, 0);
        sup.take_telemetry_events();
        assert!(sup.paid_recovery_tail.is_some(), "the fixture must leave a receipt");

        sup.handle_event(
            &SessionPixEvent::RecoveryProgress(make_recovered_tail_progress(&sup, id, target + 40)),
            t0,
            0,
        );
        let events = sup.take_telemetry_events();
        assert_eq!(0, events.useful_shares, "a recovered cycle can advance nothing further");
        assert_eq!(40, events.surplus_shares);
    }

    // -----------------------------------------------------------------------
    // Cycle finalization
    // -----------------------------------------------------------------------

    /// A cycle retired without recovering is summarized once, with what it actually got.
    #[test]
    fn a_failed_cycle_is_summarized_with_its_partial_coverage() {
        let t0 = Instant::now();
        let p = pseudonym();
        let mut sup = recovering_session(p, t0);
        let id = ssa_id(p, 1);
        let target = sup.dims.target_useful_shares();

        // Reached the front and got a quarter of the way before stalling. `recovering_session`
        // funded it at a served count of zero, so that is where its egress is measured from.
        feed(&mut sup, id, target / 4, target / 2);

        // Retire it on the hard recovery deadline, with the served count having moved on.
        let actions = sup.handle_timers(t0 + Duration::from_secs(100_000), 5_000);
        assert!(
            actions
                .iter()
                .any(|action| matches!(action, SessionPixAction::Close(_))),
            "the fixture must actually retire the cycle, got {actions:?}"
        );

        let events = sup.take_telemetry_events();
        assert_eq!(1, events.failed);
        assert_eq!(1, events.finalized.len(), "summarized exactly once");
        let summary = events.finalized[0];
        assert_eq!(PixCycleOutcome::Failed, summary.outcome);
        assert_eq!(target / 4, summary.useful_shares);
        assert_eq!(target / 2, summary.accepted_shares);
        assert_eq!(Some(0.25), summary.useful_fraction());
        assert_eq!(Some(0.5), summary.accepted_fraction());
        assert_eq!(
            5_000, summary.egress_packets,
            "egress is the Session's served count measured from when this cycle reached the front"
        );
    }

    /// Egress is measured from the front, not from the Session's start.
    ///
    /// The companion to the test above: there, the cycle reached the front at a served count of
    /// zero, so the two readings coincide and the property is invisible. Here a predecessor has
    /// already consumed 1 000 packets before this cycle is armed, and only the 4 000 after that
    /// belong to it.
    #[test]
    fn cycle_egress_is_measured_from_the_front_not_the_session_start() {
        let t0 = Instant::now();
        let p = pseudonym();
        let (mut sup, _) = SessionPixSupervisor::new(default_cfg(), dims(64, 8), p, t0);
        let id = ssa_id(p, 1);

        // Funded at a served count of 1 000: the Session has already been passing data.
        sup.handle_event(&SessionPixEvent::SsaRequestSent(id), t0, 1_000);
        sup.handle_event(&SessionPixEvent::CommitmentVerified(id), t0, 1_000);
        sup.handle_event(
            &SessionPixEvent::DepositConfirmed {
                ssa_id: id,
                amount: sufficient_balance(),
            },
            t0,
            1_000,
        );
        sup.take_telemetry_events();

        sup.handle_timers(t0 + Duration::from_secs(100_000), 5_000);

        let events = sup.take_telemetry_events();
        assert_eq!(1, events.finalized.len());
        assert_eq!(
            4_000, events.finalized[0].egress_packets,
            "the 1 000 packets served before this cycle reached the front are not its cost"
        );
    }

    /// A cycle that never reached the front is charged no egress.
    ///
    /// The Entry's emission window is clamped to one cycle, so a queued batch member is not being
    /// served at all. Charging it for its predecessors' traffic would report the last cycle of a
    /// batch as having cost the whole batch.
    #[test]
    fn a_queued_cycle_is_charged_no_egress() {
        let t0 = Instant::now();
        let p = pseudonym();
        let cfg = SupervisorConfig {
            ssas_per_request: 2,
            max_failed_cycles: 4,
            ..default_cfg()
        };
        let (mut sup, _) = SessionPixSupervisor::new(cfg, dims(64, 8), p, t0);

        // Only the first cycle is funded, so the second never reaches the front.
        commit_unfunded(&mut sup, p, 1..=2, t0);
        sup.take_telemetry_events();
        sup.handle_event(&SessionPixEvent::DepositObserverClosed(ssa_id(p, 2)), t0, 9_999);

        let events = sup.take_telemetry_events();
        assert_eq!(1, events.finalized.len());
        assert_eq!(
            0, events.finalized[0].egress_packets,
            "a cycle that never reached the front consumed none of the Session's egress"
        );
    }

    /// An undefined coverage ratio is reported as absent rather than as zero.
    #[test]
    fn a_cycle_with_no_target_has_no_coverage_fraction() {
        let summary = PixCycleSummary {
            outcome: PixCycleOutcome::Failed,
            egress_packets: 10,
            accepted_shares: 0,
            useful_shares: 0,
            target_useful_shares: 0,
            deposit_uhopr: 0,
        };
        assert_eq!(None, summary.useful_fraction());
        assert_eq!(None, summary.accepted_fraction());
    }

    /// Accepted coverage above one is the conforming case, not an error.
    #[test]
    fn accepted_coverage_may_exceed_one() {
        let summary = PixCycleSummary {
            outcome: PixCycleOutcome::Recovered,
            egress_packets: 4096,
            accepted_shares: 768,
            useful_shares: 512,
            target_useful_shares: 512,
            deposit_uhopr: 0,
        };
        assert_eq!(Some(1.0), summary.useful_fraction());
        assert_eq!(Some(1.5), summary.accepted_fraction());
    }

    /// The snapshot reports the gate's exposure as given, without interpreting it.
    #[test]
    fn predeposit_exposure_is_carried_through_from_the_gate() {
        let t0 = Instant::now();
        let p = pseudonym();
        let (sup, _) = SessionPixSupervisor::new(default_cfg(), dims(64, 8), p, t0);

        assert_eq!(0, sup.telemetry_snapshot(0).predeposit_exposure_packets);
        assert_eq!(917, sup.telemetry_snapshot(917).predeposit_exposure_packets);
    }
}