simulator_api/

lib.rs

use std::{
    collections::{BTreeMap, BTreeSet, HashMap, HashSet},
    fmt,
    str::FromStr,
    sync::Arc,
};

use base64::{
    DecodeError as Base64DecodeError, Engine as _, engine::general_purpose::STANDARD as BASE64,
};
use parking_lot::RwLock;
use serde::{Deserialize, Serialize};
use solana_account::AccountSharedData;
use solana_address::{Address, error::ParseAddressError};

mod session_id;
mod session_type;
mod slot_range;

pub use session_id::{ControlSessionId, SessionId};
pub use session_type::SessionType;
pub use slot_range::{InvertedSlotRange, SlotRange};

pub mod client_identity;
pub mod geyser_route;
pub mod subscribe_config;
pub mod usage;
pub mod ws_compression;

/// Backtest RPC methods exposed to the client.
#[derive(Debug, Serialize, Deserialize)]
#[serde(tag = "method", content = "params", rename_all = "camelCase")]
pub enum BacktestRequest {
    CreateBacktestSession(CreateBacktestSessionRequest),
    Continue(ContinueParams),
    ContinueTo(ContinueToParams),
    ContinueSessionV1(ContinueSessionRequestV1),
    ContinueToSessionV1(ContinueToSessionRequestV1),
    CloseBacktestSession,
    CloseSessionV1(CloseSessionRequestV1),
    AttachBacktestSession {
        session_id: SessionId,
        /// Last sequence number the client received. Responses after this sequence
        /// will be replayed from the session's buffer. None = replay entire buffer.
        last_sequence: Option<u64>,
    },
    /// Sent after reattaching and rebuilding any dependent subscriptions.
    /// Allows the manager to resume a session that was paused for handoff.
    ResumeAttachedSession,
    AttachParallelControlSessionV2 {
        control_session_id: ControlSessionId,
        /// Last per-session sequence number received by the client. Responses after
        /// these sequence numbers will be replayed from the manager's per-session
        /// replay store. Missing sessions replay their entire retained history.
        #[serde(default)]
        last_sequences: BTreeMap<SessionId, u64>,
    },
}

/// Versioned payload for `CreateBacktestSession`.
///
/// - `V0` keeps backwards-compatible shape by using `CreateSessionParams` directly.
/// - `V1` keeps the same shape and adds `parallel`.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(untagged)]
pub enum CreateBacktestSessionRequest {
    V1(CreateBacktestSessionRequestV1),
    V0(CreateSessionParams),
}

impl CreateBacktestSessionRequest {
    pub fn into_request_options(self) -> CreateBacktestSessionRequestOptions {
        match self {
            Self::V0(request) => CreateBacktestSessionRequestOptions {
                request,
                parallel: false,
            },
            Self::V1(CreateBacktestSessionRequestV1 { request, parallel }) => {
                CreateBacktestSessionRequestOptions { request, parallel }
            }
        }
    }

    pub fn into_request_and_parallel(self) -> (CreateSessionParams, bool) {
        let options = self.into_request_options();
        (options.request, options.parallel)
    }
}

impl From<CreateSessionParams> for CreateBacktestSessionRequest {
    fn from(value: CreateSessionParams) -> Self {
        Self::V0(value)
    }
}

impl From<CreateBacktestSessionRequestV1> for CreateBacktestSessionRequest {
    fn from(value: CreateBacktestSessionRequestV1) -> Self {
        Self::V1(value)
    }
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct CreateBacktestSessionRequestV1 {
    #[serde(flatten)]
    pub request: CreateSessionParams,
    pub parallel: bool,
}

#[derive(Debug, Clone)]
pub struct CreateBacktestSessionRequestOptions {
    pub request: CreateSessionParams,
    pub parallel: bool,
}

#[derive(Debug, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ContinueSessionRequestV1 {
    pub session_id: SessionId,
    pub request: ContinueParams,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ContinueToSessionRequestV1 {
    pub session_id: SessionId,
    pub request: ContinueToParams,
}

#[derive(Debug, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct CloseSessionRequestV1 {
    pub session_id: SessionId,
}

/// A filter registered at session creation describing which upcoming batches
/// the session should announce ahead of execution via
/// [`BacktestResponse::DiscoveryBatch`] (and its session-event twins). Each
/// filter describes an event of interest (e.g. a specific program executing);
/// the session "discovers" the batch in which that event will occur and
/// emits a `DiscoveryBatch` so the client can pause immediately before it.
#[serde_with::serde_as]
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(tag = "kind", content = "value", rename_all = "camelCase")]
pub enum DiscoveryFilter {
    /// Discover batches containing a transaction that invokes this program.
    ProgramExecuted(#[serde_as(as = "serde_with::DisplayFromStr")] Address),
}

/// Per-transaction facts a [`DiscoveryFilter`] can inspect when deciding
/// whether to match. Callers build this once per transaction and feed it to
/// every registered filter; new variants add fields here rather than growing
/// the [`DiscoveryFilter::matches`] signature.
pub struct TxMatchContext<'a> {
    /// Programs invoked by the transaction (top-level + CPI observed in logs).
    pub invoked_programs: &'a HashSet<Address>,
}

impl DiscoveryFilter {
    /// Return `true` when this filter is satisfied by the transaction
    /// described by `ctx`.
    pub fn matches(&self, ctx: &TxMatchContext<'_>) -> bool {
        match self {
            Self::ProgramExecuted(target) => ctx.invoked_programs.contains(target),
        }
    }
}

/// The two mints a swap trades, without a direction: a maker quotes a pair both ways, so
/// `(a, b)` and `(b, a)` are the same pair. Normalized on construction and on deserialization,
/// which is what makes the set lookup direction-agnostic.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
#[serde(try_from = "[String; 2]", into = "[String; 2]")]
pub struct MintPair([Address; 2]);

impl MintPair {
    pub fn new(one: Address, other: Address) -> Self {
        Self(if one <= other {
            [one, other]
        } else {
            [other, one]
        })
    }

    pub fn mints(&self) -> (Address, Address) {
        (self.0[0], self.0[1])
    }
}

#[derive(Debug, thiserror::Error)]
#[error("invalid mint {mint} in pair")]
pub struct ParseMintPairError {
    mint: String,
    source: ParseAddressError,
}

impl TryFrom<[String; 2]> for MintPair {
    type Error = ParseMintPairError;

    fn try_from([one, other]: [String; 2]) -> Result<Self, Self::Error> {
        fn parse(mint: String) -> Result<Address, ParseMintPairError> {
            Address::from_str(&mint).map_err(|source| ParseMintPairError { mint, source })
        }
        Ok(Self::new(parse(one)?, parse(other)?))
    }
}

impl From<MintPair> for [String; 2] {
    fn from(pair: MintPair) -> Self {
        let (one, other) = pair.mints();
        [one.to_string(), other.to_string()]
    }
}

/// Restricts which detected swaps are re-quoted. A transaction is rerouted when at least one
/// of its swaps trades a listed pair; the rest replay unchanged, counted as `swaps_filtered`.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct RerouteFilter {
    /// A swap matches when the mints it trades are one of these pairs, either direction.
    #[serde(default)]
    pub pairs: BTreeSet<MintPair>,
}

impl RerouteFilter {
    /// An empty filter admits nothing, which creation-time validation rejects: omit the filter
    /// to admit everything.
    pub fn is_empty(&self) -> bool {
        self.pairs.is_empty()
    }

    pub fn matches_mints(&self, input: &Address, output: &Address) -> bool {
        self.pairs.contains(&MintPair::new(*input, *output))
    }

    /// Whether any listed pair trades `mint`. An arbitrage cycle trades one mint with itself, so
    /// it is admitted by the book naming that mint rather than by a pair it can never match.
    pub fn mentions_mint(&self, mint: &Address) -> bool {
        self.pairs.iter().any(|pair| {
            let (one, other) = pair.mints();
            one == *mint || other == *mint
        })
    }
}

/// How an action's transactions run. Only meaningful for an action that carries
/// transactions: one that carries none is a schedule entry for `account_overrides`.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub enum ActionKind {
    /// Runs against session state without committing. Each transaction's writes are
    /// visible to the next one in the same action and to nothing else, so the replay is
    /// unaffected.
    #[default]
    Simulate,
    /// Executes and commits to the session ledger. The writes are visible to every later
    /// slot, including the historical transactions that follow — a deliberate change to
    /// the replayed world, not a simulation of one. Rejected under `replay_account_state`,
    /// where the recording restamps account state and the commit would be lost.
    Send,
}

/// Where a [`ScheduledAction`] fires.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "at", rename_all = "camelCase")]
pub enum ActionAnchor {
    /// At the start of each slot, before its transactions execute and before
    /// reroute quoting. Fires the entry for the current slot, if any.
    BeforeSlot {
        #[serde(default, with = "compressed")]
        slots: Vec<u64>,
    },
    /// At the end of each slot, after its transactions execute but before it
    /// finalizes. Fires the entry for the current slot, if any.
    AfterSlot {
        #[serde(default, with = "compressed")]
        slots: Vec<u64>,
    },
    /// Before each batch matching `filter` executes. The whole sequence fires at
    /// each matching batch.
    BeforeMatch { filter: DiscoveryFilter },
    /// After each batch matching `filter` commits. The whole sequence fires at
    /// each matching batch.
    AfterMatch { filter: DiscoveryFilter },
}

/// Upper bound on a decompressed transactions blob, guarding the server against
/// a decompression bomb in client input.
const MAX_ACTION_TRANSACTIONS_BYTES: u64 = 64 * 1024 * 1024;

/// serde adapter that stores any value as a base64-encoded, zstd-compressed JSON
/// blob on the wire, so transaction bytes never travel raw. Used via
/// `#[serde(with = "compressed")]` on [`ScheduledAction`]'s `transactions` field
/// and [`ActionAnchor`]'s `slots` field.
mod compressed {
    use std::io::Read;

    use base64::Engine as _;
    use serde::{Deserialize, Deserializer, Serialize, Serializer, de::DeserializeOwned};

    use super::{BASE64, MAX_ACTION_TRANSACTIONS_BYTES};

    /// zstd compression level; `0` selects zstd's default (level 3).
    const COMPRESSION_LEVEL: i32 = 0;

    pub fn serialize<T, S>(value: &T, serializer: S) -> Result<S::Ok, S::Error>
    where
        T: Serialize,
        S: Serializer,
    {
        let json = serde_json::to_vec(value).map_err(serde::ser::Error::custom)?;
        let compressed = zstd::stream::encode_all(json.as_slice(), COMPRESSION_LEVEL)
            .map_err(serde::ser::Error::custom)?;
        serializer.serialize_str(&BASE64.encode(compressed))
    }

    pub fn deserialize<'de, T, D>(deserializer: D) -> Result<T, D::Error>
    where
        T: DeserializeOwned,
        D: Deserializer<'de>,
    {
        let blob = String::deserialize(deserializer)?;
        let compressed = BASE64
            .decode(blob.as_bytes())
            .map_err(serde::de::Error::custom)?;
        let mut json = Vec::new();
        zstd::stream::read::Decoder::new(compressed.as_slice())
            .and_then(|d| Read::take(d, MAX_ACTION_TRANSACTIONS_BYTES).read_to_end(&mut json))
            .map_err(serde::de::Error::custom)?;
        serde_json::from_slice(&json).map_err(serde::de::Error::custom)
    }
}

/// A sequence of transactions the server runs automatically during a backtest,
/// with results streamed over the `actionSubscribe` subscription. Under
/// `feeds_reroute` the state it ends with also reaches the session's override
/// schedule, which every simulation surface reads — reroute candidates and later
/// scheduled actions alike. An action carrying only `account_overrides` is purely
/// declarative: it contributes to that schedule with or without the flag, and
/// emits no result.
#[serde_with::serde_as]
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ScheduledAction {
    pub anchor: ActionAnchor,
    /// Ignored when `transactions` is empty — that action is a schedule entry, not a run.
    #[serde(default)]
    pub kind: ActionKind,
    /// Base64-encoded transactions, carried over the wire as a base64-encoded,
    /// zstd-compressed blob. For a slot anchor (`beforeSlot`/`afterSlot`),
    /// entry `i` fires at `slots[i]` — the two lists must be the same length,
    /// and repeating a slot runs its entries there in order as one sequence.
    /// For match anchors the whole list fires, in order, at each matching
    /// batch. For `simulate`, each transaction's writes are visible to the next
    /// without committing to the session ledger. Empty under `feeds_reroute`
    /// makes the action declarative: `account_overrides` is then the whole payload.
    #[serde(default, with = "compressed")]
    pub transactions: Vec<String>,
    /// Account state overrides seeding the first transaction's simulation, per
    /// `simulateTransaction` `modifyAccountStates` semantics. Invalid for
    /// `kind: send`. Without transactions they are themselves the contribution to
    /// the session's override schedule, `feeds_reroute` or not.
    #[serde(default)]
    pub account_overrides: AccountModifications,
    /// Feed the account state this action ends with to the session's override
    /// schedule, where it applies from the anchored slot to every simulation —
    /// reroute candidates and later scheduled actions — and carries forward per
    /// account until a later action names the same one. Requires a `beforeSlot`
    /// anchor.
    ///
    /// With no transactions that state is `account_overrides` itself, known at session
    /// creation. With transactions it is their simulated post-state, so the slot's
    /// overrides are not known until the action runs at slot start.
    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
    pub feeds_reroute: bool,
    /// Accounts whose post-execution state to return, following
    /// `simulateTransaction` semantics; reflects cumulative state after the
    /// final transaction.
    #[serde_as(as = "Vec<serde_with::DisplayFromStr>")]
    #[serde(default)]
    pub return_accounts: Vec<Address>,
    /// Optional client tag echoed back on each `actionSubscribe` result to correlate it to this action.
    #[serde(default)]
    pub label: Option<String>,
}

impl ScheduledAction {
    /// Whether the action's resulting state has to be simulated before it can be
    /// published — a slot's overrides are then unknown until the action runs.
    pub fn is_computed_override(&self) -> bool {
        self.feeds_reroute && !self.transactions.is_empty()
    }
}

/// One ordered contribution to a slot's override state.
#[derive(Debug, Clone)]
enum OverrideContribution {
    Static(HashMap<Address, AccountSharedData>),
    Pending,
    Resolved(HashMap<Address, AccountSharedData>),
    Failed,
}

#[derive(Debug, Clone, Default)]
struct OverrideScheduleInner {
    entries: BTreeMap<u64, Vec<(u32, OverrideContribution)>>,
    /// Every account some contribution names. The geyser rewrite path asks `active_for` about each
    /// account event it sees, and almost none are overridden; without this the miss walks every
    /// entry up to the slot.
    keys: HashSet<Address>,
}

impl OverrideScheduleInner {
    fn new(entries: BTreeMap<u64, Vec<(u32, OverrideContribution)>>) -> Self {
        let keys = entries
            .values()
            .flatten()
            .filter_map(|(_, contribution)| match contribution {
                OverrideContribution::Static(accounts)
                | OverrideContribution::Resolved(accounts) => Some(accounts.keys().copied()),
                OverrideContribution::Pending | OverrideContribution::Failed => None,
            })
            .flatten()
            .collect();
        Self { entries, keys }
    }
}

/// Slot-keyed account overrides, carried forward per account. Clones share the same runtime
/// resolutions so the session, reroute engine, and geyser pump observe one schedule.
#[derive(Debug, Clone, Default)]
pub struct OverrideSchedule {
    inner: Arc<RwLock<OverrideScheduleInner>>,
}

impl From<BTreeMap<u64, HashMap<Address, AccountSharedData>>> for OverrideSchedule {
    fn from(entries: BTreeMap<u64, HashMap<Address, AccountSharedData>>) -> Self {
        Self {
            inner: Arc::new(RwLock::new(OverrideScheduleInner::new(
                entries
                    .into_iter()
                    .map(|(slot, accounts)| {
                        (slot, vec![(0, OverrideContribution::Static(accounts))])
                    })
                    .collect(),
            ))),
        }
    }
}

impl OverrideSchedule {
    pub fn is_empty(&self) -> bool {
        self.inner.read().entries.is_empty()
    }

    /// Slots at which some entry activates, ascending.
    pub fn slots(&self) -> impl Iterator<Item = u64> + '_ {
        self.inner
            .read()
            .entries
            .keys()
            .copied()
            .collect::<Vec<_>>()
            .into_iter()
    }

    /// Everything active at `slot`, resolved per account.
    pub fn active_at(&self, slot: u64) -> HashMap<Address, AccountSharedData> {
        self.inner.read().entries.range(..=slot).fold(
            HashMap::new(),
            |mut active, (_, contributions)| {
                active.extend(fold_contributions(contributions));
                active
            },
        )
    }

    pub fn active_for(&self, key: &Address, slot: u64) -> Option<AccountSharedData> {
        let inner = self.inner.read();
        if !inner.keys.contains(key) {
            return None;
        }
        inner
            .entries
            .range(..=slot)
            .rev()
            .find_map(|(_, contributions)| {
                contributions
                    .iter()
                    .rev()
                    .find_map(|(_, contribution)| match contribution {
                        OverrideContribution::Static(accounts)
                        | OverrideContribution::Resolved(accounts) => accounts.get(key).cloned(),
                        OverrideContribution::Pending | OverrideContribution::Failed => None,
                    })
            })
    }

    /// Final values activated in `(after, up_to]`, deduplicated per account.
    pub fn newly_active(
        &self,
        after: Option<u64>,
        up_to: u64,
    ) -> Vec<(Address, AccountSharedData)> {
        let lower = after.map_or(std::ops::Bound::Unbounded, std::ops::Bound::Excluded);
        self.inner
            .read()
            .entries
            .range((lower, std::ops::Bound::Included(up_to)))
            .flat_map(|(_, contributions)| fold_contributions(contributions))
            .collect::<HashMap<_, _>>()
            .into_iter()
            .collect()
    }

    /// Resolve one computed contribution; `None` marks the action failed, contributing no accounts.
    /// Counting failures is the caller's job — it also sees the errors returned here.
    pub fn resolve(
        &self,
        slot: u64,
        action_index: u32,
        accounts: Option<HashMap<Address, AccountSharedData>>,
    ) -> Result<(), String> {
        let mut inner = self.inner.write();
        let OverrideScheduleInner { entries, keys } = &mut *inner;
        let contribution = entries
            .get_mut(&slot)
            .and_then(|contributions| {
                contributions
                    .iter_mut()
                    .find(|(index, _)| *index == action_index)
                    .map(|(_, contribution)| contribution)
            })
            .ok_or_else(|| {
                format!("override action {action_index} has no contribution at slot {slot}")
            })?;
        if !matches!(contribution, OverrideContribution::Pending) {
            return Err(format!(
                "override action {action_index} at slot {slot} is not pending"
            ));
        }
        *contribution = match accounts {
            Some(accounts) => {
                keys.extend(accounts.keys().copied());
                OverrideContribution::Resolved(accounts)
            }
            None => OverrideContribution::Failed,
        };
        Ok(())
    }
}

fn fold_contributions(
    contributions: &[(u32, OverrideContribution)],
) -> HashMap<Address, AccountSharedData> {
    let mut accounts = HashMap::new();
    for (_, contribution) in contributions {
        match contribution {
            OverrideContribution::Static(next) | OverrideContribution::Resolved(next) => {
                accounts.extend(next.iter().map(|(key, account)| (*key, account.clone())));
            }
            OverrideContribution::Pending | OverrideContribution::Failed => {}
        }
    }
    accounts
}

/// Validate and compile the publishing actions into the slot-keyed override schedule every
/// simulation surface consumes — reroute candidates and scheduled-action runs alike. Static
/// contributions are immediately available; computed contributions retain their action order as
/// pending entries until the action scheduler resolves them.
///
/// `feeds_reroute` actions contribute their end-state; a declarative override action (account
/// overrides, no transactions) contributes without the flag, so overrides stay usable in a
/// session that never reroutes.
fn decode_override_actions(actions: &[ScheduledAction]) -> Result<OverrideSchedule, String> {
    let mut entries: BTreeMap<u64, Vec<(u32, OverrideContribution)>> = BTreeMap::new();
    for (action_index, action) in actions.iter().enumerate().filter(|(_, action)| {
        action.feeds_reroute
            || (action.transactions.is_empty() && !action.account_overrides.0.is_empty())
    }) {
        // Without this an action that meant to send would feed the schedule instead, silently.
        if action.kind == ActionKind::Send {
            return Err("account overrides are not valid for kind: send".to_string());
        }
        if action.transactions.is_empty() && action.account_overrides.0.is_empty() {
            return Err(
                "feedsReroute needs something to feed: transactions to simulate, or \
                 accountOverrides"
                    .to_string(),
            );
        }
        let ActionAnchor::BeforeSlot { slots } = &action.anchor else {
            return Err("scheduled account overrides require a beforeSlot anchor".to_string());
        };
        let accounts = decode_override_accounts(&action.account_overrides)?;
        let contribution = if action.transactions.is_empty() {
            OverrideContribution::Static(accounts)
        } else {
            OverrideContribution::Pending
        };
        // Deduplicated: a slot repeated in one action would push two contributions sharing an
        // action index, and `resolve` only ever finds the first — leaving the second pending.
        for slot in slots.iter().copied().collect::<BTreeSet<_>>() {
            entries
                .entry(slot)
                .or_default()
                .push((action_index as u32, contribution.clone()));
        }
    }
    Ok(OverrideSchedule {
        inner: Arc::new(RwLock::new(OverrideScheduleInner::new(entries))),
    })
}

/// Decode account modifications into engine override form, rejecting
/// executable/program accounts (cached bytecode makes program-code overrides
/// stale).
#[allow(
    deprecated,
    reason = "`PROGRAM_OWNERS` remains the canonical loader-owner list"
)]
pub fn decode_account_modifications(
    modifications: &AccountModifications,
) -> Result<HashMap<Address, AccountSharedData>, String> {
    modifications
        .0
        .iter()
        .map(|(address, data)| {
            let account = data
                .to_account()
                .map_err(|error| format!("invalid account data for {address}: {error}"))?;
            if account.executable || solana_account::PROGRAM_OWNERS.contains(&account.owner) {
                return Err(format!(
                    "overriding executable/program accounts is not supported (account {address})"
                ));
            }
            Ok((*address, AccountSharedData::from(account)))
        })
        .collect()
}

/// [`decode_account_modifications`] plus rejection of token accounts, which the
/// reroute-only read path cannot honor consistently.
fn decode_override_accounts(
    modifications: &AccountModifications,
) -> Result<HashMap<Address, AccountSharedData>, String> {
    let accounts = decode_account_modifications(modifications)?;
    if let Some(address) = modifications
        .0
        .iter()
        .find_map(|(address, data)| TOKEN_PROGRAM_IDS.contains(&data.owner).then_some(address))
    {
        return Err(format!(
            "override for {address}: token accounts cannot be overridden \
             (prepare-stage reads would desync from simulation)"
        ));
    }
    Ok(accounts)
}

/// Server-side ceiling on [`CreateSessionParams::disconnect_timeout_secs`]:
/// 900 seconds (15 minutes). A larger request is silently clamped to this, so
/// clients must keep their reconnect budget under it to survive a detach.
pub const MAX_DISCONNECT_TIMEOUT_SECS: u16 = 900;

/// A router whose historical swaps a session can re-quote. The wire spelling; the session maps
/// each onto its own venue type.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
#[serde(rename_all = "lowercase", try_from = "String")]
pub enum SwapVenue {
    Jupiter,
    Okx,
    Titan,
    Dflow,
}

impl SwapVenue {
    /// Matches the serialized spelling and the venue labels in the per-requote rows.
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::Jupiter => "jupiter",
            Self::Okx => "okx",
            Self::Titan => "titan",
            Self::Dflow => "dflow",
        }
    }
}

impl std::fmt::Display for SwapVenue {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(self.as_str())
    }
}

#[derive(Debug, thiserror::Error)]
pub enum ParseVenueError {
    #[error("unknown swap venue {0:?}; expected one of jupiter, okx, titan, dflow")]
    Unknown(String),
    /// A run that admits no venue drives the whole reroute apparatus and re-quotes nothing.
    #[error("name at least one venue (jupiter, okx, titan, dflow)")]
    NoVenues,
}

impl TryFrom<String> for SwapVenue {
    type Error = ParseVenueError;

    fn try_from(value: String) -> Result<Self, Self::Error> {
        value.parse()
    }
}

impl std::str::FromStr for SwapVenue {
    type Err = ParseVenueError;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        match s.trim().to_ascii_lowercase().as_str() {
            "jupiter" => Ok(Self::Jupiter),
            "okx" => Ok(Self::Okx),
            "titan" => Ok(Self::Titan),
            "dflow" => Ok(Self::Dflow),
            _ => Err(ParseVenueError::Unknown(s.to_string())),
        }
    }
}

/// The venues a session re-quotes. Jupiter alone by default: the router builds Jupiter
/// replacements, so other venues' order flow is a cross-router counterfactual a run asks for
/// deliberately. Parsed from one comma-separated string so a flag or an env var carries the set.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(transparent)]
pub struct RerouteVenues(BTreeSet<SwapVenue>);

impl Default for RerouteVenues {
    fn default() -> Self {
        Self([SwapVenue::Jupiter].into_iter().collect())
    }
}

impl RerouteVenues {
    pub fn new(venues: impl IntoIterator<Item = SwapVenue>) -> Self {
        Self(venues.into_iter().collect())
    }

    pub fn contains(&self, venue: SwapVenue) -> bool {
        self.0.contains(&venue)
    }

    pub fn is_empty(&self) -> bool {
        self.0.is_empty()
    }
}

impl std::fmt::Display for RerouteVenues {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let mut venues = self.0.iter();
        if let Some(first) = venues.next() {
            write!(f, "{first}")?;
        }
        venues.try_for_each(|venue| write!(f, ",{venue}"))
    }
}

impl std::str::FromStr for RerouteVenues {
    type Err = ParseVenueError;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        let venues: BTreeSet<SwapVenue> = s
            .split(',')
            .filter(|venue| !venue.trim().is_empty())
            .map(str::parse)
            .collect::<Result<_, _>>()?;
        if venues.is_empty() {
            return Err(ParseVenueError::NoVenues);
        }
        Ok(Self(venues))
    }
}

/// Re-quoting historical swaps through the metis router: each admitted swap is re-quoted and the
/// routed transaction is *simulated* just before the original executes — never committed, so the
/// replay is unaffected. Each simulation reads the state of the batch its original belonged to;
/// reroutes never observe one another.
///
/// Presence is the switch, which makes "a knob but no rerouting" unrepresentable rather than a
/// validation error.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct RerouteParams {
    #[serde(default)]
    pub venues: RerouteVenues,
    /// Restrict re-quoting to the mint pairs of interest — a maker's own book. Absent: every
    /// detected swap on an admitted venue is re-quoted.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub filter: Option<RerouteFilter>,
    /// Also re-quote swaps whose L1 transaction failed, when the allowlist admits the failure as
    /// one re-quoting can answer (a slippage rejection, or a failure after the swap already ran).
    ///
    /// Off by default: L1 realised nothing for these, so every comparison against the original is
    /// undefined and they are tracked as their own population — `swaps_detected` and
    /// `swaps_rerouted` exclude them, and every row carries `original_failed`.
    #[serde(default)]
    pub detect_failed_l1_swaps: bool,
    /// Also re-quote arbitrage cycles — swaps whose input and output mint are the same. Off by
    /// default: the router prices a cycle and then refuses to build the route it found.
    #[serde(default)]
    pub circular_arbs: bool,
}

/// Parameters required to start a new backtest session.
#[serde_with::serde_as]
#[derive(Debug, Clone, Serialize, Deserialize, bon::Builder)]
#[serde(rename_all = "camelCase")]
pub struct CreateSessionParams {
    /// First slot (inclusive) to replay.
    pub start_slot: u64,
    /// Last slot (inclusive) to replay.
    pub end_slot: u64,
    #[serde_as(as = "BTreeSet<serde_with::DisplayFromStr>")]
    #[serde(default)]
    #[builder(default)]
    /// Skip transactions signed by these addresses.
    pub signer_filter: BTreeSet<Address>,
    /// When true, include a session summary with transaction statistics in client-facing
    /// `Completed` responses. Summary generation remains enabled internally for metrics.
    #[serde(default)]
    #[builder(default)]
    pub send_summary: bool,
    /// Re-quote historical swaps through the metis router. Absent: plain replay.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub reroute: Option<RerouteParams>,
    /// Maximum seconds to wait for ECS capacity-related startup retries before
    /// failing session creation. If not set (or 0), capacity errors fail immediately.
    #[serde(default)]
    pub capacity_wait_timeout_secs: Option<u16>,
    /// Maximum seconds to keep the session alive after the control websocket disconnects.
    /// If not set (or 0), the session tears down immediately on disconnect.
    /// Clamped to [`MAX_DISCONNECT_TIMEOUT_SECS`] (900 seconds / 15 minutes).
    #[serde(default)]
    pub disconnect_timeout_secs: Option<u16>,
    /// Extra compute units to add to each transaction's `SetComputeUnitLimit` budget.
    /// Useful when replaying with an account override whose program uses more CU than
    /// the original, causing otherwise-healthy transactions to run out of budget.
    /// Only applied when a `SetComputeUnitLimit` instruction is already present.
    #[serde(default)]
    pub extra_compute_units: Option<u32>,
    /// Events of interest the session should watch for. When an upcoming
    /// batch matches any filter, the server emits
    /// [`BacktestResponse::DiscoveryBatch`] (and its session-event twins)
    /// ahead of execution so the client can follow up with
    /// [`BacktestRequest::ContinueTo`] to pause before the batch. Empty
    /// means no batch discoveries are performed (existing behaviour).
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    #[builder(default)]
    pub discoveries: Vec<DiscoveryFilter>,
    /// Transactions the server runs automatically during the backtest, with
    /// results streamed over `actionSubscribe`.
    ///
    /// Compressed as a whole rather than per action: a per-slot schedule repeats
    /// near-identical account bytes thousands of times, and that redundancy is
    /// only visible across actions. Compressing each action alone leaves a
    /// 10k-slot schedule near the websocket message limit; compressing the
    /// vector takes the same schedule from ~28 MiB to ~1 MiB.
    #[serde(default, with = "compressed", skip_serializing_if = "Vec::is_empty")]
    #[builder(default)]
    pub actions: Vec<ScheduledAction>,
    /// Replay accounts states for this range instead of executing transactions
    /// — a state-only acceleration.
    #[serde(default)]
    #[builder(default)]
    pub replay_account_state: bool,
}

/// What account state replay cannot serve, worded once. Client, manager and
/// session all reject through here, so a user meets the same explanation
/// wherever the request is stopped.
pub mod replay_support {
    use super::{ActionKind, CreateSessionParams};

    /// `getTransaction`/`getSignatureStatuses`: nothing executed, so nothing is found.
    pub const NO_TRANSACTIONS: &str =
        "Create the session without replayAccountState to execute and persist transactions.";

    pub const SEND_TRANSACTION: &str = "Use simulateTransaction to run it against the replayed state, or create the session \
         without replayAccountState.";

    pub const SEND_ACTION: &str = "Use kind: simulate, which runs against the replayed state without committing, post state \
         directly with an action carrying accountOverrides and no transactions, or create the \
         session without replayAccountState.";

    /// States the operation, why replay cannot serve it, then the way forward.
    pub fn unsupported(operation: &str, remedy: &str) -> String {
        format!(
            "{operation} is not supported while account state replay is enabled: replay \
             reconstructs account state from the recorded delta bundle instead of executing \
             transactions, so this session never executes or persists any. {remedy}"
        )
    }

    /// The one rule decidable from the request alone, so it can be caught before
    /// a session — or a runtime to host it — is ever created.
    pub fn send_action_conflict(params: &CreateSessionParams) -> Option<String> {
        (params.replay_account_state
            && params
                .actions
                .iter()
                .any(|action| action.kind == ActionKind::Send))
        .then(|| unsupported("A scheduled action with kind: send", SEND_ACTION))
    }
}

impl CreateSessionParams {
    /// Whether a session needs the metis sidecar (reroute quotes through it).
    pub fn requires_sidecar(&self) -> bool {
        self.reroute_order_flow()
    }

    pub fn reroute_order_flow(&self) -> bool {
        self.reroute.is_some()
    }

    /// The Override actions in [`Self::actions`], compiled into the slot-keyed schedule the
    /// reroute layer consumes. Same-slot actions merge account-wise, later wins.
    pub fn override_schedule(&self) -> Result<OverrideSchedule, String> {
        decode_override_actions(&self.actions)
    }

    /// The mint pairs re-quoting is restricted to, if the run named any.
    pub fn reroute_filter(&self) -> Option<&RerouteFilter> {
        self.reroute.as_ref()?.filter.as_ref()
    }
}

/// What counts as a "non-benign" divergence for the backtest session fail-fast
/// behaviour (see `BacktestSessionOptions::fail_fast`).
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
#[serde(rename_all = "kebab-case")]
pub enum FailFastDivergenceKind {
    /// Any divergence other than a benign log diff (the full `is_divergent()` set:
    /// log mismatch, error mismatch, balance-diff mismatch).
    #[default]
    AnyNonBenign,
    /// Only divergences on transactions that touch an account currently watched by an
    /// account-diff subscription (subscribed directly, or owned by a subscribed program).
    Tracked,
}

impl FailFastDivergenceKind {
    /// Stable string form used for CLI/env-var serialization. Matches the serde
    /// `kebab-case` representation.
    pub fn as_str(self) -> &'static str {
        match self {
            Self::AnyNonBenign => "any-non-benign",
            Self::Tracked => "tracked",
        }
    }

    /// Inverse of [`Self::as_str`]; returns `None` for an unrecognized value.
    pub fn from_str_opt(value: &str) -> Option<Self> {
        match value {
            "any-non-benign" => Some(Self::AnyNonBenign),
            "tracked" => Some(Self::Tracked),
            _ => None,
        }
    }
}

/// Account state modifications to apply.
#[serde_with::serde_as]
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct AccountModifications(
    #[serde_as(as = "BTreeMap<serde_with::DisplayFromStr, _>")]
    #[serde(default)]
    pub BTreeMap<Address, AccountData>,
);

const TOKEN_PROGRAM_IDS: [Address; 2] = [spl_token_interface::ID, spl_token_2022_interface::ID];

impl From<BTreeMap<Address, AccountData>> for AccountModifications {
    fn from(value: BTreeMap<Address, AccountData>) -> Self {
        Self(value)
    }
}

/// Arguments used to continue an existing session.
#[serde_with::serde_as]
#[derive(Debug, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ContinueParams {
    #[serde(default = "ContinueParams::default_advance_count")]
    /// Number of blocks to advance before waiting.
    pub advance_count: u64,
    #[serde(default)]
    /// Base64-encoded transactions to execute before advancing.
    pub transactions: Vec<String>,
    #[serde(default)]
    /// Account state overrides to apply ahead of execution.
    pub modify_account_states: AccountModifications,
}

impl Default for ContinueParams {
    fn default() -> Self {
        Self {
            advance_count: Self::default_advance_count(),
            transactions: Vec::new(),
            modify_account_states: AccountModifications(BTreeMap::new()),
        }
    }
}

impl ContinueParams {
    pub fn default_advance_count() -> u64 {
        1
    }
}

/// Payload emitted when a session halts at a caller-specified point.
/// `batch_index` is `None` for block-boundary pauses and `Some(n)` when the
/// session stopped *before* batch `n` within `slot` (no transaction from
/// batch `n` has been applied). While paused, RPC reads against the session
/// observe partial state up through batch `n - 1`.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PausedEvent {
    pub slot: u64,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub batch_index: Option<u32>,
}

/// Payload emitted when the session has *discovered* an upcoming batch that
/// matches one or more registered [`DiscoveryFilter`]s from session creation
/// (for example, a batch containing a transaction that invokes a program of
/// interest). The `(slot, batch_index)` pair can be fed directly to
/// [`BacktestRequest::ContinueTo`] to pause immediately before the batch
/// executes. After each `Continue` / `ContinueTo`, the session emits the
/// next `DiscoveryBatchEvent` ahead of the next matching batch, enabling a
/// reactive "pause on every discovery" driver loop.
#[serde_with::serde_as]
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct DiscoveryBatchEvent {
    pub slot: u64,
    pub batch_index: u32,
    /// Filters that matched this batch (always non-empty).
    pub matched: Vec<DiscoveryFilter>,
    /// Encoded transactions in this batch that triggered the match. Each
    /// entry carries the serialized `VersionedTransaction` bytes paired with
    /// the encoding used.
    pub transactions: Vec<EncodedBinary>,
}

/// Arguments used to step an existing session to a precise point.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ContinueToParams {
    /// Target slot to stop in (or at, if `batch_index` is `None`).
    pub slot: u64,
    /// Batch within the target slot at which to pause, **exclusive** — the
    /// session halts immediately *before* batch `n` executes, so no
    /// transaction in that batch has been applied yet. `None` runs the
    /// whole slot, pausing at the block boundary. While paused, RPC reads
    /// observe partial state up through batch `n - 1`.
    #[serde(default)]
    pub batch_index: Option<u32>,
}

/// Supported binary encodings for account/transaction payloads.
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "lowercase")]
pub enum BinaryEncoding {
    Base64,
}

impl BinaryEncoding {
    pub fn encode(self, bytes: &[u8]) -> String {
        match self {
            Self::Base64 => BASE64.encode(bytes),
        }
    }

    pub fn decode(self, data: &str) -> Result<Vec<u8>, Base64DecodeError> {
        match self {
            Self::Base64 => BASE64.decode(data),
        }
    }
}

/// A blob paired with the encoding needed to decode it.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct EncodedBinary {
    /// Encoded payload.
    pub data: String,
    /// Encoding scheme used for the payload.
    pub encoding: BinaryEncoding,
}

impl EncodedBinary {
    pub fn new(data: String, encoding: BinaryEncoding) -> Self {
        Self { data, encoding }
    }

    pub fn from_bytes(bytes: &[u8], encoding: BinaryEncoding) -> Self {
        Self {
            data: encoding.encode(bytes),
            encoding,
        }
    }

    pub fn decode(&self) -> Result<Vec<u8>, Base64DecodeError> {
        self.encoding.decode(&self.data)
    }
}

/// Account snapshot used to seed or modify state in a session.
#[serde_with::serde_as]
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct AccountData {
    /// Account data bytes and encoding.
    pub data: EncodedBinary,
    /// Whether the account is executable.
    pub executable: bool,
    /// Lamport balance.
    pub lamports: u64,
    #[serde_as(as = "serde_with::DisplayFromStr")]
    /// Account owner pubkey.
    pub owner: Address,
    /// Allocated space in bytes.
    pub space: u64,
}

impl AccountData {
    pub fn to_account(&self) -> Result<solana_account::Account, Base64DecodeError> {
        Ok(solana_account::Account {
            data: self.data.decode()?,
            lamports: self.lamports,
            owner: self.owner,
            executable: self.executable,
            rent_epoch: 0,
        })
    }
}

/// Responses returned over the backtest RPC channel.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "method", content = "params", rename_all = "camelCase")]
pub enum BacktestResponse {
    SessionCreated {
        session_id: SessionId,
        rpc_endpoint: String,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        task_id: Option<String>,
    },
    SessionAttached {
        session_id: SessionId,
        rpc_endpoint: String,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        task_id: Option<String>,
    },
    SessionsCreatedV2 {
        control_session_id: ControlSessionId,
        session_ids: Vec<SessionId>,
        #[serde(default)]
        task_ids: Vec<Option<String>>,
        /// Per-sub-session start/end slots, parallel to `session_ids`. The
        /// server's split is authoritative (a mid-bundle request anchors to the
        /// covering bundle), so the client binds each sub-session to its range
        /// from here rather than re-deriving the split locally.
        #[serde(default)]
        start_slots: Vec<u64>,
        #[serde(default)]
        end_slots: Vec<u64>,
    },
    ParallelSessionAttachedV2 {
        control_session_id: ControlSessionId,
        session_ids: Vec<SessionId>,
        #[serde(default)]
        task_ids: Vec<Option<String>>,
    },
    ReadyForContinue,
    SlotNotification(u64),
    Paused(PausedEvent),
    DiscoveryBatch(DiscoveryBatchEvent),
    Error(BacktestError),
    Success,
    Completed {
        /// Session summary with transaction statistics.
        /// The session always computes this summary, but management may omit it from
        /// client-facing responses unless `send_summary` was requested at session creation.
        #[serde(skip_serializing_if = "Option::is_none")]
        summary: Option<SessionSummary>,
    },
    Status {
        status: BacktestStatus,
    },
    SessionEventV2 {
        session_id: SessionId,
        seq_id: u64,
        event: SessionEventKind,
    },
}

impl BacktestResponse {
    pub fn is_completed(&self) -> bool {
        matches!(self, BacktestResponse::Completed { .. })
    }

    pub fn is_terminal(&self) -> bool {
        match self {
            BacktestResponse::Completed { .. } => true,
            BacktestResponse::Error(e) => e.is_terminal(),
            _ => false,
        }
    }
}

impl From<BacktestStatus> for BacktestResponse {
    fn from(status: BacktestStatus) -> Self {
        Self::Status { status }
    }
}

impl From<String> for BacktestResponse {
    fn from(message: String) -> Self {
        BacktestError::Internal { error: message }.into()
    }
}

impl From<&str> for BacktestResponse {
    fn from(message: &str) -> Self {
        BacktestError::Internal {
            error: message.to_string(),
        }
        .into()
    }
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "method", content = "params", rename_all = "camelCase")]
pub enum SessionEventKind {
    ReadyForContinue,
    SlotNotification(u64),
    Paused(PausedEvent),
    DiscoveryBatch(DiscoveryBatchEvent),
    Error(BacktestError),
    Success,
    Completed {
        #[serde(skip_serializing_if = "Option::is_none")]
        summary: Option<SessionSummary>,
    },
    Status {
        status: BacktestStatus,
    },
}

impl SessionEventKind {
    pub fn is_terminal(&self) -> bool {
        match self {
            Self::Completed { .. } => true,
            Self::Error(e) => e.is_terminal(),
            _ => false,
        }
    }
}

/// Wire format wrapper for responses sent over the control websocket.
/// Sessions with `disconnect_timeout_secs > 0` use this to track client position.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct SequencedResponse {
    pub seq_id: u64,
    #[serde(flatten)]
    pub response: BacktestResponse,
}

/// High-level progress states during a `Continue` call.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub enum BacktestStatus {
    /// Runtime startup is in progress.
    StartingRuntime,
    DecodedTransactions,
    AppliedAccountModifications,
    ReadyToExecuteUserTransactions,
    ExecutedUserTransactions,
    ExecutingBlockTransactions,
    ExecutedBlockTransactions,
    ProgramAccountsLoaded,
}

impl std::fmt::Display for BacktestStatus {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let s = match self {
            Self::StartingRuntime => "starting runtime",
            Self::DecodedTransactions => "decoded transactions",
            Self::AppliedAccountModifications => "applied account modifications",
            Self::ReadyToExecuteUserTransactions => "ready to execute user transactions",
            Self::ExecutedUserTransactions => "executed user transactions",
            Self::ExecutingBlockTransactions => "executing block transactions",
            Self::ExecutedBlockTransactions => "executed block transactions",
            Self::ProgramAccountsLoaded => "program accounts loaded",
        };
        f.write_str(s)
    }
}

/// High-level funnel for a reroute-order-flow session: how many detected swaps made it from
/// re-quoting through simulation to a non-reverting execution. Per-swap economics and latency
/// are session-internal and land in the analysis DB.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct RerouteStatsReport {
    /// Requotable swaps detected in the replayed range, across the admitted venues.
    #[serde(default)]
    pub swaps_detected: u64,
    /// Detected swaps excluded before quoting — an unadmitted venue, an arbitrage cycle, a mint
    /// pair the filter does not name. Without it the gap between `swaps_detected` and
    /// `swaps_rerouted` reads as routing failures.
    #[serde(default)]
    pub swaps_filtered: u64,
    /// Swaps re-quoted through metis and queued for simulation.
    #[serde(default)]
    pub swaps_rerouted: u64,
    /// Swaps whose routed transaction was simulated.
    #[serde(default)]
    pub swaps_simulated: u64,
    /// Simulated swaps whose routed transaction executed without reverting.
    #[serde(default)]
    pub swaps_succeeded: u64,
    /// Detected swaps metis could not route; nothing was simulated for them.
    #[serde(default)]
    pub requote_failures: u64,
    /// Computed override actions that failed and therefore contributed no state.
    #[serde(default)]
    pub override_setup_failures: u64,
    /// Blocks replayed unchanged because the metis client was unavailable.
    #[serde(default)]
    pub metis_unavailable_blocks: u64,
    /// Quote attempts that could not be made because the metis client was unavailable, summed
    /// over every swap left queued by each such block. Each one is retried on the next block,
    /// so this counts retries forced rather than swaps lost.
    #[serde(default)]
    pub quote_attempts_unavailable: u64,
}

/// Summary of transaction execution statistics for a completed backtest session.
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct SessionSummary {
    /// Number of simulations where simulator outcome matched on-chain outcome
    /// (`true_success + true_failure`).
    pub correct_simulation: usize,
    /// Number of simulations where simulator outcome did not match on-chain outcome
    /// (`false_success + false_failure`).
    pub incorrect_simulation: usize,
    /// Number of transactions that had execution errors in simulation.
    pub execution_errors: usize,
    /// Number of transactions with different balance diffs.
    pub balance_diff: usize,
    /// Number of transactions with different log diffs.
    pub log_diff: usize,
    /// Transactions skipped because they could not be sanitized (malformed message or
    /// unresolvable address lookup table).
    #[serde(default)]
    pub unsanitizable: usize,
    /// Aggregate reroute metrics, populated only for reroute-order-flow sessions. Boxed to
    /// keep `SessionSummary` small in the many response enums that carry it.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub reroute_stats: Option<Box<RerouteStatsReport>>,
}

impl SessionSummary {
    /// Returns true if there were any execution deviations (errors or mismatched results).
    pub fn has_deviations(&self) -> bool {
        self.incorrect_simulation > 0 || self.execution_errors > 0 || self.balance_diff > 0
    }

    /// Total number of transactions processed.
    pub fn total_transactions(&self) -> usize {
        self.correct_simulation
            + self.incorrect_simulation
            + self.execution_errors
            + self.balance_diff
            + self.log_diff
    }

    /// Renders the compact "N incorrect simulations, N execution errors, ..." deviation list
    /// shared by console reporters. `balance_label` / `log_label` let each caller keep its own
    /// historical wording for the trailing two counts (e.g. "balance diffs" vs
    /// "different balances").
    pub fn deviation_details(&self, balance_label: &str, log_label: &str) -> String {
        format!(
            "{} incorrect simulations, {} execution errors, {} {balance_label}, {} {log_label}",
            self.incorrect_simulation, self.execution_errors, self.balance_diff, self.log_diff,
        )
    }
}

impl std::fmt::Display for SessionSummary {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let total = self.total_transactions();
        write!(
            f,
            "Session summary: {total} transactions\n\
             \x20  - {} correct simulation\n\
             \x20  - {} incorrect simulation\n\
             \x20  - {} execution errors\n\
             \x20  - {} unsanitizable\n\
             \x20  - {} balance diffs\n\
             \x20  - {} log diffs",
            self.correct_simulation,
            self.incorrect_simulation,
            self.execution_errors,
            self.unsanitizable,
            self.balance_diff,
            self.log_diff,
        )
    }
}

/// Error variants surfaced to backtest RPC clients.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub enum BacktestError {
    InvalidTransactionEncoding {
        index: usize,
        error: String,
    },
    InvalidTransactionFormat {
        index: usize,
        error: String,
    },
    InvalidAccountEncoding {
        address: String,
        encoding: BinaryEncoding,
        error: String,
    },
    InvalidAccountOwner {
        address: String,
        error: String,
    },
    InvalidAccountPubkey {
        address: String,
        error: String,
    },
    NoMoreBlocks,
    BundleNotFound {
        start_slot: u64,
        end_slot: u64,
    },
    AdvanceSlotFailed {
        slot: u64,
        error: String,
    },
    FinalizeSlotFailed {
        slot: u64,
        error: String,
    },
    InvalidRequest {
        error: String,
    },
    Internal {
        error: String,
    },
    InvalidBlockhashFormat {
        slot: u64,
        error: String,
    },
    InitializingSysvarsFailed {
        slot: u64,
        error: String,
    },
    ClerkError {
        error: String,
    },
    SimulationError {
        error: String,
    },
    SessionNotFound {
        session_id: String,
    },
    SessionOwnerMismatch,
    /// Session ownership is in transition (e.g. the previous manager is
    /// shutting down, or another attach raced this one). Clients should retry
    /// the attach within their reconnect budget; the route is expected to
    /// become claimable shortly.
    SessionOwnershipBusy {
        reason: String,
    },
    /// The requested range contains no produced blocks (every slot in it was
    /// skipped on-chain), so there is nothing to execute. A user error: the
    /// session cannot start and the range must be adjusted.
    // Appended last: the manager<->runtime hop is postcard, where variant
    // order is wire-significant across version skew.
    NoProducedBlocksInRange {
        start_slot: u64,
        end_slot: u64,
    },
}

impl BacktestError {
    /// Errors the backtest cannot recover from mid-session — the single source
    /// of truth for terminal-error classification, shared by the response/event
    /// wrappers and the managed client.
    pub fn is_terminal(&self) -> bool {
        matches!(
            self,
            BacktestError::NoMoreBlocks
                | BacktestError::BundleNotFound { .. }
                | BacktestError::AdvanceSlotFailed { .. }
                | BacktestError::FinalizeSlotFailed { .. }
                | BacktestError::Internal { .. }
                | BacktestError::NoProducedBlocksInRange { .. }
        )
    }
}

/// One contiguous block range available on the history clerk.
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct AvailableRange {
    pub bundle_start_slot: u64,
    pub bundle_start_slot_utc: Option<String>,
    pub max_bundle_end_slot: Option<u64>,
    pub max_bundle_end_slot_utc: Option<String>,
    pub max_bundle_size: Option<u64>,
    /// What the range can be used for. `None` from a server that predates the
    /// field, where every listed range is [`BundleKind::Blocks`].
    #[serde(default)]
    pub kind: Option<BundleKind>,
    /// Who the range is available to. `None` from a server that predates the
    /// field.
    #[serde(default)]
    pub scope: Option<RangeScope>,
}

/// Who a listed range is available to.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum RangeScope {
    /// Available to every caller.
    Global,
    /// Built for the calling API key, and listed only for that key.
    User,
}

impl RangeScope {
    /// Stable string form matching the serde `snake_case` wire representation.
    pub fn as_str(self) -> &'static str {
        match self {
            Self::Global => "global",
            Self::User => "user",
        }
    }
}

/// What a bundle build produces.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum BundleKind {
    /// The block bundle a session replays transactions from.
    #[default]
    Blocks,
    /// An account-state bundle covering exactly the requested range, which
    /// later sessions replay instead of executing transactions. Builds the
    /// block bundle it needs first.
    AccountState,
}

/// Why a replayed range has to begin where a recording does. Shared so the
/// splitter's refusal and the service's single-session refusal cannot drift
/// apart.
pub const REPLAY_RANGE_ADVICE: &str = "replayed state is resolved by the recording that begins \
     where the request does, so request a range that begins where one was recorded and ends \
     within it";

impl BundleKind {
    /// What a caller is short of, named for a coverage error.
    fn coverage_noun(self) -> &'static str {
        match self {
            Self::Blocks => "bundle range",
            Self::AccountState => "account-state bundle you own",
        }
    }

    /// [`Self::coverage_noun`] pluralised, for a message about several.
    fn coverage_nouns(self) -> &'static str {
        match self {
            Self::Blocks => "bundle ranges",
            Self::AccountState => "account-state bundles you own",
        }
    }

    /// Stable string form matching the serde `snake_case` wire representation.
    pub fn as_str(self) -> &'static str {
        match self {
            Self::Blocks => "blocks",
            Self::AccountState => "account_state",
        }
    }

    /// Inverse of [`Self::as_str`]; returns `None` for an unrecognized value.
    pub fn from_str_opt(value: &str) -> Option<Self> {
        match value {
            "blocks" => Some(Self::Blocks),
            "account_state" => Some(Self::AccountState),
            _ => None,
        }
    }
}

/// Spelled for the reader, not the wire: `blocks` shows as `transactions`.
/// Use [`BundleKind::as_str`] where the wire spelling is required.
impl fmt::Display for BundleKind {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        let label = match self {
            Self::Blocks => "transactions",
            Self::AccountState => "account_state",
        };
        f.pad(label)
    }
}

/// Request body of `POST /build-bundle`.
///
/// Construct via `..Default::default()` so a new field stays source-compatible.
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct BundleBuildRequest {
    pub start_slot: u64,
    pub end_slot: u64,
    #[serde(default)]
    pub bundle_size: Option<u64>,
    /// Optional repeat key; a resubmit dedupes the quota charge. The same key
    /// under a different `kind` is a distinct request.
    #[serde(default)]
    pub idempotency_key: Option<String>,
    /// Absent on the wire means [`BundleKind::Blocks`].
    #[serde(default)]
    pub kind: BundleKind,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum BundleBuildStatus {
    /// The bundle was built and is ready to use.
    Completed,
    /// The build failed and will not produce a bundle.
    Failed,
    /// The build has not finished yet; poll the status endpoint again later.
    InProgress,
    /// The outcome of the build could not be determined; do not assume success
    /// or failure.
    NeedsInvestigation,
}

impl BundleBuildStatus {
    /// Stable string form matching the serde `snake_case` wire representation.
    pub fn as_str(self) -> &'static str {
        match self {
            Self::Completed => "completed",
            Self::Failed => "failed",
            Self::InProgress => "in_progress",
            Self::NeedsInvestigation => "needs_investigation",
        }
    }
}

/// Response of the management bundle-build endpoints.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct BundleBuildStatusResponse {
    pub request_id: String,
    pub start_slot: u64,
    pub end_slot: u64,
    pub bundle_size: Option<u64>,
    pub status: BundleBuildStatus,
    /// When the request id was created (unix ms) — lets clients show a live ETA
    /// countdown by computing elapsed since this.
    pub created_at_unix_ms: u64,
    /// Absent on records predating the field, which are all `blocks`.
    #[serde(default)]
    pub kind: BundleKind,
}

/// A bundle the caller owns and can use now. Ordered by slot range, then kind.
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
pub struct CompletedBundle {
    pub start_slot: u64,
    pub end_slot: u64,
    pub kind: BundleKind,
    /// When the bundle became available (unix ms), where known.
    #[serde(default)]
    pub completed_at_unix_ms: Option<i64>,
    #[serde(default)]
    pub size_bytes: Option<i64>,
}

/// Query of `DELETE /bundles`: the bundle to delete, by kind and exact range.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct BundleDeleteRequest {
    pub start_slot: u64,
    pub end_slot: u64,
    /// Absent means [`BundleKind::Blocks`].
    #[serde(default)]
    pub kind: BundleKind,
}

/// What `DELETE /bundles` removed.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct BundleDeleteResponse {
    pub start_slot: u64,
    pub end_slot: u64,
    pub kind: BundleKind,
}

/// Response of `GET /build-bundle`.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct BundleBuildListResponse {
    /// The caller's own builds, newest first. A build belongs to the endpoint
    /// running it, so this covers only builds requested here.
    pub builds: Vec<BundleBuildStatusResponse>,
    /// The caller's own bundles, ascending by range, wherever they were built.
    /// Not a view of `builds`: one build can produce several bundles, and a
    /// bundle outlives the record of the build that made it. `None` when the
    /// listing was unavailable, as opposed to `Some([])` for owning nothing.
    #[serde(default)]
    pub bundles: Option<Vec<CompletedBundle>>,
}

/// Response of the management `/snapshot-slot` endpoint. The chosen source and
/// import state are internal, so only the resolved slot is exposed.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SnapshotSlotResponse {
    pub snapshot_slot: u64,
}

/// Split a user-requested `[start_slot, end_slot]` range across the available
/// bundle ranges, returning a list of contiguous, non-overlapping `(start, end)`
/// pairs — one per bundle that the server can serve as its own session.
///
/// Each emitted `start` is a real `bundle_start_slot` and each `end` is within
/// the `max_bundle_end_slot` of a bundle that begins *exactly* at that start
/// (the server requires `start_slot == bundle_start_slot` and caps `end_slot` at
/// that bundle's `max_bundle_end_slot`). Among all such gap-free splits we pick
/// the one with the most sub-ranges, i.e. the highest parallelism: a fine-grained
/// (e.g. 10k) bundle grid yields one session per fine bundle instead of
/// collapsing onto a coarser series that overlaps the same slots — even when the
/// coarse and fine bundles share a start slot. A coarser bundle is only ridden
/// where no finer grid continues the walk, which still lets us serve a request
/// the coarse data covers even across a hole in the fine grid.
///
/// `kind` picks which bundles the split rides; the walk itself is the same for
/// either, since a recording is read slot by slot exactly as a block bundle is.
/// An unlabelled range, from a server predating [`AvailableRange::kind`],
/// counts as [`BundleKind::Blocks`].
///
/// Returns an error if the requested start cannot anchor, or if no gap-free
/// split reaches `requested_end`.
pub fn split_range(
    ranges: &[AvailableRange],
    requested: SlotRange,
    kind: BundleKind,
) -> Result<Vec<(u64, u64)>, String> {
    let (requested_start, requested_end) = (requested.start(), requested.end());

    // Every advertised end per bundle start. A start can carry several ends — a
    // 50k snapshot bundle and the 10k bundle built on the same snapshot share a
    // start slot but reach different ends — so we keep them all and let the walk
    // pick whichever maximises parallelism.
    // A one-slot bundle (`end == start`) counts: a series whose last split
    // begins on the requested end records exactly one slot.
    let ends_by_start = ranges
        .iter()
        .filter(|r| r.kind.unwrap_or_default() == kind)
        .filter_map(|r| {
            r.max_bundle_end_slot
                .filter(|&end| end >= r.bundle_start_slot)
                .map(|end| (r.bundle_start_slot, end))
        })
        .fold(
            BTreeMap::<u64, BTreeSet<u64>>::new(),
            |mut ends_by_start, (start, end)| {
                ends_by_start.entry(start).or_default().insert(end);
                ends_by_start
            },
        );

    // Anchor a request that lands mid-bundle to the covering bundle's start —
    // the latest bundle beginning at or before `requested_start` that still
    // covers it. The server's `select_parallel_requests` shares this fn and
    // anchors the same way, so the client must too or it would falsely reject a
    // start that isn't itself a bundle boundary. An exact bundle-start request
    // anchors to itself. (Single sessions instead require an exact bundle start
    // for every kind: `evaluate_requested_range`.)
    let Some((&anchor_start, _)) = ends_by_start.range(..=requested_start).rfind(|(_, ends)| {
        ends.iter()
            .next_back()
            .is_some_and(|&end| end >= requested_start)
    }) else {
        return Err(format!(
            "start_slot {requested_start} is not covered by any available {}",
            kind.coverage_noun()
        ));
    };

    // Walk the bundle starts from `requested_end` backwards, recording for each
    // start the gap-free split with the most sub-ranges (highest parallelism).
    // A sub-range ends within one of its start's bundles and the next sub-range
    // must begin on the very next slot, so a coarse bundle is ridden only when no
    // finer bundle continues the walk.
    let mut best_from: BTreeMap<u64, Vec<(u64, u64)>> = BTreeMap::new();
    for (&start, ends) in ends_by_start.range(anchor_start..=requested_end).rev() {
        let mut best: Option<Vec<(u64, u64)>> = None;
        for &end in ends {
            let candidate = if end >= requested_end {
                Some(vec![(start, requested_end)])
            } else {
                best_from.get(&(end + 1)).map(|rest| {
                    std::iter::once((start, end))
                        .chain(rest.iter().copied())
                        .collect()
                })
            };
            if let Some(candidate) = candidate
                && best.as_ref().is_none_or(|b| candidate.len() > b.len())
            {
                best = Some(candidate);
            }
        }
        if let Some(best) = best {
            best_from.insert(start, best);
        }
    }

    best_from.remove(&anchor_start).ok_or_else(|| {
        // Point at the first uncovered slot when the bundles leave a hole; fall
        // back to a generic message when the range is covered but cannot be
        // tiled to bundle boundaries.
        let mut covered_to = anchor_start.saturating_sub(1);
        for (&start, ends) in ends_by_start.range(anchor_start..=requested_end) {
            if start > covered_to.saturating_add(1) {
                break;
            }
            if let Some(&end) = ends.iter().next_back() {
                covered_to = covered_to.max(end);
            }
        }
        if covered_to < requested_end {
            format!("gap in coverage at slot {}", covered_to + 1)
        } else {
            format!(
                "no gap-free split of [{requested_start}, {requested_end}] aligns with the available {}",
                kind.coverage_nouns()
            )
        }
    })
}

impl From<BacktestError> for BacktestResponse {
    fn from(error: BacktestError) -> Self {
        Self::Error(error)
    }
}

impl std::error::Error for BacktestError {}

impl fmt::Display for BacktestError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            BacktestError::InvalidTransactionEncoding { index, error } => {
                write!(f, "invalid transaction encoding at index {index}: {error}")
            }
            BacktestError::InvalidTransactionFormat { index, error } => {
                write!(f, "invalid transaction format at index {index}: {error}")
            }
            BacktestError::InvalidAccountEncoding {
                address,
                encoding,
                error,
            } => write!(
                f,
                "invalid encoding for account {address} ({encoding:?}): {error}"
            ),
            BacktestError::InvalidAccountOwner { address, error } => {
                write!(f, "invalid owner for account {address}: {error}")
            }
            BacktestError::InvalidAccountPubkey { address, error } => {
                write!(f, "invalid account pubkey {address}: {error}")
            }
            BacktestError::NoMoreBlocks => write!(f, "no more blocks available"),
            BacktestError::BundleNotFound {
                start_slot,
                end_slot,
            } => write!(
                f,
                "block bundle not found for start_slot={start_slot}, end_slot={end_slot}"
            ),
            BacktestError::AdvanceSlotFailed { slot, error } => {
                write!(f, "failed to advance to slot {slot}: {error}")
            }
            BacktestError::FinalizeSlotFailed { slot, error } => {
                write!(f, "failed to finalize slot {slot}: {error}")
            }
            BacktestError::InvalidRequest { error } => write!(f, "invalid request: {error}"),
            BacktestError::Internal { error } => write!(f, "internal error: {error}"),
            BacktestError::InvalidBlockhashFormat { slot, error } => {
                write!(f, "invalid blockhash at slot {slot}: {error}")
            }
            BacktestError::InitializingSysvarsFailed { slot, error } => {
                write!(f, "failed to initialize sysvars at slot {slot}: {error}")
            }
            BacktestError::ClerkError { error } => write!(f, "clerk error: {error}"),
            BacktestError::SimulationError { error } => {
                write!(f, "simulation error: {error}")
            }
            BacktestError::SessionNotFound { session_id } => {
                write!(f, "session not found: {session_id}")
            }
            BacktestError::SessionOwnerMismatch => {
                write!(f, "session owner mismatch")
            }
            BacktestError::SessionOwnershipBusy { reason } => {
                write!(f, "session ownership busy: {reason}")
            }
            BacktestError::NoProducedBlocksInRange {
                start_slot,
                end_slot,
            } => write!(
                f,
                "no produced blocks in requested range {start_slot}..={end_slot}; \
                 slot(s) may be skipped on-chain"
            ),
        }
    }
}

#[cfg(test)]
mod tests {
    use rstest::rstest;
    use solana_account::ReadableAccount;

    use super::*;

    fn address(byte: u8) -> Address {
        Address::from([byte; 32])
    }

    #[rstest]
    #[case::one("jupiter", Some(&[SwapVenue::Jupiter][..]))]
    #[case::several("jupiter,okx", Some(&[SwapVenue::Jupiter, SwapVenue::Okx][..]))]
    #[case::spaced_and_uppercased(" Jupiter , OKX ", Some(&[SwapVenue::Jupiter, SwapVenue::Okx][..]))]
    #[case::repeats_collapse("okx,okx", Some(&[SwapVenue::Okx][..]))]
    #[case::trailing_separator("okx,", Some(&[SwapVenue::Okx][..]))]
    #[case::unknown("jupiter,bogus", None)]
    #[case::empty("", None)]
    #[case::separators_only(",,", None)]
    fn reroute_venues_parse_the_shapes_a_flag_carries(
        #[case] input: &str,
        #[case] expected: Option<&[SwapVenue]>,
    ) {
        assert_eq!(
            input.parse::<RerouteVenues>().ok(),
            expected.map(|venues| RerouteVenues::new(venues.iter().copied()))
        );
    }

    /// `Display` renders `--reroute-venues`' default, so clap re-parses whatever it prints.
    #[test]
    fn reroute_venues_round_trip_through_their_rendering() {
        let venues = RerouteVenues::new([SwapVenue::Dflow, SwapVenue::Jupiter]);

        assert_eq!(venues.to_string(), "jupiter,dflow");
        assert_eq!(venues.to_string().parse::<RerouteVenues>().unwrap(), venues);
        assert_eq!(RerouteVenues::default().to_string(), "jupiter");
    }

    /// The API deserializes venues through the same parser the flag uses, so a spelling one
    /// entry point accepts is not a hard error at the other.
    #[test]
    fn venues_deserialize_through_the_same_parser_the_flag_uses() {
        let params: CreateSessionParams = serde_json::from_value(
            serde_json::json!({"startSlot": 1, "endSlot": 2, "reroute": {"venues": ["Jupiter", "OKX"]}}),
        )
        .expect("decode");

        assert_eq!(
            params.reroute.expect("rerouting requested").venues,
            RerouteVenues::new([SwapVenue::Jupiter, SwapVenue::Okx])
        );
    }

    /// A request that names no knob gets the narrow defaults, not the every-venue behaviour a
    /// client written before this change expected — the switch these defaults are here to make.
    #[test]
    fn params_without_the_new_reroute_fields_decode_to_the_narrow_defaults() {
        let params: CreateSessionParams = serde_json::from_value(
            serde_json::json!({"startSlot": 1, "endSlot": 2, "reroute": {}}),
        )
        .expect("decode");

        let reroute = params.reroute.expect("rerouting requested");
        assert_eq!(reroute.venues, RerouteVenues::default());
        assert!(reroute.venues.contains(SwapVenue::Jupiter));
        assert!(!reroute.venues.contains(SwapVenue::Okx));
        assert!(!reroute.circular_arbs);
        assert!(!reroute.detect_failed_l1_swaps);
    }

    /// Presence is the switch: a replay that will never quote must not reserve a metis sidecar.
    #[test]
    fn a_request_without_reroute_reserves_no_sidecar() {
        let params: CreateSessionParams =
            serde_json::from_value(serde_json::json!({"startSlot": 1, "endSlot": 2}))
                .expect("decode");

        assert!(!params.requires_sidecar());
    }

    /// A per-slot override schedule is the largest thing a client sends. Each
    /// action alone is incompressible enough to sit near the websocket message
    /// limit; the redundancy lives *across* actions, so the whole vector has to
    /// travel as one blob. Pins both the round trip and that size property.
    #[test]
    fn a_per_slot_action_schedule_compresses_across_actions() {
        let slots = 2_000u64;
        let actions = (0..slots)
            .map(|slot| ScheduledAction {
                anchor: ActionAnchor::AfterSlot { slots: vec![slot] },
                kind: ActionKind::Simulate,
                feeds_reroute: false,
                transactions: Vec::new(),
                account_overrides: AccountModifications(BTreeMap::from([(
                    address(7),
                    AccountData {
                        // Realistic pool account: 2 KiB that barely moves slot to slot.
                        data: EncodedBinary::new(
                            BASE64.encode(
                                [&slot.to_le_bytes()[..], &[0xAB; 2040][..]]
                                    .concat()
                                    .as_slice(),
                            ),
                            BinaryEncoding::Base64,
                        ),
                        executable: false,
                        lamports: 15_144_966,
                        owner: address(9),
                        space: 2048,
                    },
                )])),
                return_accounts: Vec::new(),
                label: None,
            })
            .collect::<Vec<_>>();
        let params = CreateSessionParams::builder()
            .start_slot(1)
            .end_slot(slots)
            .actions(actions.clone())
            .build();

        let encoded = serde_json::to_string(&params).expect("serialize");
        let raw = serde_json::to_string(&actions).expect("serialize raw actions");
        assert!(
            encoded.len() * 8 < raw.len(),
            "expected the schedule to compress well; got {} bytes from {} raw",
            encoded.len(),
            raw.len()
        );

        let decoded: CreateSessionParams = serde_json::from_str(&encoded).expect("round trip");
        assert_eq!(
            serde_json::to_string(&decoded.actions).expect("serialize decoded"),
            raw
        );
    }

    /// Absent actions stay absent on the wire, so a session without a schedule
    /// is unchanged by the compression.
    #[test]
    fn an_empty_action_list_is_omitted_entirely() {
        let params = CreateSessionParams::builder()
            .start_slot(1)
            .end_slot(2)
            .build();
        let json = serde_json::to_value(&params).expect("serialize");
        assert!(json.get("actions").is_none(), "{json}");
        let decoded: CreateSessionParams = serde_json::from_value(json).expect("round trip");
        assert!(decoded.actions.is_empty());
    }

    /// The conflict is `send` *and* replay: neither alone is a problem, and simulate runs
    /// fine under replay because it never commits.
    #[test]
    fn send_actions_conflict_with_replay_only_together() {
        let params = |replay, kind| {
            CreateSessionParams::builder()
                .start_slot(1)
                .end_slot(2)
                .replay_account_state(replay)
                .actions(vec![ScheduledAction {
                    anchor: ActionAnchor::BeforeSlot { slots: vec![1] },
                    kind,
                    feeds_reroute: false,
                    transactions: vec!["dGVzdA==".to_string()],
                    account_overrides: Default::default(),
                    return_accounts: Default::default(),
                    label: None,
                }])
                .build()
        };
        assert!(replay_support::send_action_conflict(&params(false, ActionKind::Send)).is_none());
        assert!(
            replay_support::send_action_conflict(&params(true, ActionKind::Simulate)).is_none()
        );
        let mut publishing = params(true, ActionKind::Simulate);
        publishing.actions[0].feeds_reroute = true;
        assert!(replay_support::send_action_conflict(&publishing).is_none());

        let conflict = replay_support::send_action_conflict(&params(true, ActionKind::Send))
            .expect("send under replay conflicts");
        assert!(conflict.contains("kind: send"), "{conflict}");
        assert!(conflict.contains(replay_support::SEND_ACTION), "{conflict}");
    }

    fn override_entry(owner: Address, executable: bool) -> AccountData {
        AccountData {
            data: EncodedBinary::from_bytes(&[1, 2, 3], BinaryEncoding::Base64),
            executable,
            lamports: 1,
            owner,
            space: 3,
        }
    }

    fn before() -> ActionAnchor {
        ActionAnchor::BeforeSlot { slots: vec![1] }
    }

    fn valid() -> ScheduledAction {
        override_action(before(), address(3), override_entry(address(4), false))
    }

    fn override_action(
        anchor: ActionAnchor,
        account: Address,
        data: AccountData,
    ) -> ScheduledAction {
        ScheduledAction {
            anchor,
            kind: ActionKind::default(),
            transactions: Vec::new(),
            account_overrides: AccountModifications(BTreeMap::from([(account, data)])),
            return_accounts: Vec::new(),
            label: None,
            feeds_reroute: true,
        }
    }

    #[test]
    fn override_action_round_trips() {
        let key = address(1);
        let entry = override_entry(address(2), false);
        let action = override_action(
            ActionAnchor::BeforeSlot { slots: vec![5, 9] },
            key,
            entry.clone(),
        );
        let json = serde_json::to_value(&action).unwrap();
        assert!(
            json["anchor"]["slots"].is_string(),
            "slots travel as a compressed blob: {json}"
        );
        let back: ScheduledAction = serde_json::from_value(json).unwrap();
        assert_eq!(back.anchor, ActionAnchor::BeforeSlot { slots: vec![5, 9] });
        assert!(back.feeds_reroute);
        assert!(!back.is_computed_override(), "no transactions to simulate");
        let survived = &back.account_overrides.0[&key];
        assert_eq!(survived.lamports, entry.lamports);
        assert_eq!(survived.data.data, entry.data.data);
    }

    #[rstest]
    #[case::executable(override_action(before(), address(3), override_entry(address(10), true)))]
    #[case::program_owned(override_action(
        before(),
        address(3),
        override_entry(solana_account::PROGRAM_OWNERS[0], false),
    ))]
    #[case::token_owned(override_action(
        before(),
        address(3),
        override_entry(TOKEN_PROGRAM_IDS[0], false),
    ))]
    #[case::after_slot_anchor(override_action(
        ActionAnchor::AfterSlot { slots: vec![1] },
        address(3),
        override_entry(address(4), false),
    ))]
    #[case::nothing_to_post(ScheduledAction {
        account_overrides: AccountModifications::default(),
        ..valid()
    })]
    // Otherwise an action that meant to send would silently post overrides instead.
    #[case::send(ScheduledAction {
        kind: ActionKind::Send,
        ..valid()
    })]
    #[allow(deprecated, reason = "exercise the loader-owner rejection path")]
    fn override_actions_reject_invalid_accounts_and_shapes(#[case] action: ScheduledAction) {
        assert!(decode_override_actions(&[action]).is_err());
    }

    /// The shapes the rejections above are variations of.
    #[test]
    fn override_actions_accept_a_static_and_a_computed_shape() {
        assert!(decode_override_actions(&[valid()]).is_ok());
        // Transactions make the contribution computed rather than static — still valid,
        // just not resolvable until the action runs.
        assert!(
            decode_override_actions(&[ScheduledAction {
                transactions: vec!["dGVzdA==".to_string()],
                ..valid()
            }])
            .is_ok()
        );
    }

    #[test]
    fn override_actions_merge_account_wise_later_wins() {
        let entry = |lamports| AccountData {
            lamports,
            ..override_entry(address(9), false)
        };
        let slot_action = |slots, account, lamports| {
            override_action(ActionAnchor::BeforeSlot { slots }, account, entry(lamports))
        };
        let actions = [
            slot_action(vec![5, 7], address(1), 1),
            slot_action(vec![5], address(2), 2),
            slot_action(vec![5], address(1), 3),
            ScheduledAction {
                anchor: ActionAnchor::AfterSlot { slots: vec![5] },
                kind: ActionKind::Simulate,
                feeds_reroute: false,
                transactions: vec!["dGVzdA==".to_string()],
                account_overrides: AccountModifications::default(),
                return_accounts: Vec::new(),
                label: None,
            },
        ];
        let schedule = decode_override_actions(&actions).unwrap();
        assert_eq!(schedule.slots().collect::<Vec<_>>(), vec![5, 7]);
        let lamports_at = |key, slot| {
            schedule
                .active_for(&key, slot)
                .map(|account| account.lamports())
        };
        assert_eq!(lamports_at(address(1), 5), Some(3));
        assert_eq!(lamports_at(address(2), 5), Some(2));
        assert_eq!(lamports_at(address(1), 7), Some(1));
    }

    #[test]
    fn computed_override_resolution_preserves_action_order_and_failure() {
        let key = address(1);
        let static_action = |lamports| {
            override_action(
                ActionAnchor::BeforeSlot { slots: vec![5] },
                key,
                AccountData {
                    lamports,
                    ..override_entry(address(9), false)
                },
            )
        };
        let computed_action = || ScheduledAction {
            transactions: vec!["dGVzdA==".to_string()],
            ..static_action(2)
        };

        let computed_last =
            decode_override_actions(&[static_action(1), computed_action()]).unwrap();
        computed_last
            .resolve(5, 1, Some(HashMap::from([(key, shared_account(2))])))
            .unwrap();
        assert_eq!(
            computed_last.active_for(&key, 5).map(|a| a.lamports()),
            Some(2)
        );

        let static_last = decode_override_actions(&[computed_action(), static_action(3)]).unwrap();
        static_last
            .resolve(5, 0, Some(HashMap::from([(key, shared_account(2))])))
            .unwrap();
        assert_eq!(
            static_last.active_for(&key, 5).map(|a| a.lamports()),
            Some(3)
        );

        let failed = decode_override_actions(&[static_action(1), computed_action()]).unwrap();
        failed.resolve(5, 1, None).unwrap();
        assert_eq!(failed.active_for(&key, 5).map(|a| a.lamports()), Some(1));
        assert!(failed.resolve(5, 1, None).is_err(), "resolving twice");
    }

    fn shared_account(lamports: u64) -> AccountSharedData {
        AccountSharedData::from(solana_account::Account {
            lamports,
            data: vec![1],
            owner: address(9),
            executable: false,
            rent_epoch: 0,
        })
    }

    fn carry_forward_schedule() -> (Address, Address, OverrideSchedule) {
        let (a, b) = (address(1), address(2));
        let schedule = OverrideSchedule::from(BTreeMap::from([
            (10, HashMap::from([(a, shared_account(1))])),
            (
                20,
                HashMap::from([(a, shared_account(2)), (b, shared_account(9))]),
            ),
        ]));
        (a, b, schedule)
    }

    #[test]
    fn override_schedule_resolves_carry_forward_per_account() {
        let (a, b, schedule) = carry_forward_schedule();

        assert!(schedule.active_for(&a, 9).is_none());
        let cases = [
            (9, None, None),
            (10, Some(1), None),
            (15, Some(1), None),
            (20, Some(2), Some(9)),
            (u64::MAX, Some(2), Some(9)),
        ];
        for (slot, want_a, want_b) in cases {
            assert_eq!(
                schedule
                    .active_for(&a, slot)
                    .map(|account| account.lamports()),
                want_a,
                "active_for a at {slot}"
            );
            let active = schedule.active_at(slot);
            assert_eq!(
                active.get(&a).map(ReadableAccount::lamports),
                want_a,
                "active_at a at {slot}"
            );
            assert_eq!(
                active.get(&b).map(ReadableAccount::lamports),
                want_b,
                "active_at b at {slot}"
            );
        }
    }

    #[test]
    fn newly_active_yields_final_values_in_window() {
        let (a, b, schedule) = carry_forward_schedule();

        let newly: HashMap<_, _> = schedule.newly_active(Some(10), 20).into_iter().collect();
        assert_eq!(newly.len(), 2);
        assert_eq!(newly[&a].lamports(), 2);
        assert_eq!(newly[&b].lamports(), 9);
        let spanning: HashMap<_, _> = schedule.newly_active(None, 20).into_iter().collect();
        assert_eq!(spanning[&a].lamports(), 2, "later slot wins");
        assert_eq!(schedule.newly_active(None, 15).len(), 1);
    }

    #[test]
    fn reroute_filter_round_trips_camel_case() {
        let filter = RerouteFilter {
            pairs: BTreeSet::from([MintPair::new(address(1), address(2))]),
        };
        let json = serde_json::to_value(&filter).unwrap();
        assert!(
            json["pairs"][0][0].is_string() && json["pairs"][0][1].is_string(),
            "a pair serializes as two base58 strings: {json}"
        );
        let back: RerouteFilter = serde_json::from_value(json).unwrap();
        assert_eq!(back, filter);
    }

    /// The set lookup is only direction-agnostic because construction normalizes, and
    /// values arriving over the wire have to be normalized on that path too.
    #[test]
    fn mint_pairs_are_normalized_however_they_are_built() {
        let (one, other) = (address(1), address(2));
        assert_eq!(MintPair::new(one, other), MintPair::new(other, one));

        let reversed: MintPair =
            serde_json::from_value(serde_json::json!([other.to_string(), one.to_string()]))
                .unwrap();
        assert_eq!(reversed, MintPair::new(one, other));
    }

    #[test]
    fn fail_fast_divergence_kind_str_round_trips() {
        for kind in [
            FailFastDivergenceKind::AnyNonBenign,
            FailFastDivergenceKind::Tracked,
        ] {
            assert_eq!(
                FailFastDivergenceKind::from_str_opt(kind.as_str()),
                Some(kind)
            );
        }
        assert_eq!(FailFastDivergenceKind::from_str_opt("nonsense"), None);
        assert_eq!(
            FailFastDivergenceKind::default(),
            FailFastDivergenceKind::AnyNonBenign
        );
    }

    #[test]
    fn bundle_not_found_is_terminal_and_descriptive() {
        let error = BacktestError::BundleNotFound {
            start_slot: 100,
            end_slot: 600,
        };

        assert!(error.is_terminal());
        assert_eq!(
            error.to_string(),
            "block bundle not found for start_slot=100, end_slot=600"
        );
    }

    #[test]
    fn no_produced_blocks_is_terminal_and_descriptive() {
        let error = BacktestError::NoProducedBlocksInRange {
            start_slot: 100,
            end_slot: 100,
        };

        assert!(error.is_terminal());
        // placement-e2e matches this phrase to classify the failure as
        // transient; keep it stable.
        assert_eq!(
            error.to_string(),
            "no produced blocks in requested range 100..=100; slot(s) may be skipped on-chain"
        );
    }

    #[test]
    fn bundle_build_request_optional_fields_default_to_none() {
        let req: BundleBuildRequest =
            serde_json::from_str(r#"{"start_slot":1,"end_slot":2}"#).expect("parse");
        assert_eq!((req.start_slot, req.end_slot), (1, 2));
        assert_eq!(req.bundle_size, None);
        assert_eq!(req.idempotency_key, None);
    }

    #[test]
    fn bundle_build_request_parses_optional_fields() {
        let req: BundleBuildRequest = serde_json::from_str(
            r#"{"start_slot":1,"end_slot":2,"bundle_size":500,"idempotency_key":"abc"}"#,
        )
        .expect("parse");
        assert_eq!(req.bundle_size, Some(500));
        assert_eq!(req.idempotency_key.as_deref(), Some("abc"));
    }

    #[test]
    fn bundle_build_status_serde_round_trips_with_snake_case() {
        let cases = [
            (BundleBuildStatus::Completed, "\"completed\""),
            (BundleBuildStatus::Failed, "\"failed\""),
            (BundleBuildStatus::InProgress, "\"in_progress\""),
            (
                BundleBuildStatus::NeedsInvestigation,
                "\"needs_investigation\"",
            ),
        ];
        for (status, expected) in cases {
            assert_eq!(serde_json::to_string(&status).unwrap(), expected);
            assert_eq!(
                serde_json::from_str::<BundleBuildStatus>(expected).unwrap(),
                status
            );
            assert_eq!(status.as_str(), expected.trim_matches('"'));
        }
        assert!(serde_json::from_str::<BundleBuildStatus>("\"queued\"").is_err());
    }

    #[test]
    fn bundle_build_request_without_kind_deserializes_as_blocks() {
        // An old client's payload must keep meaning a plain block build.
        let request: BundleBuildRequest =
            serde_json::from_str(r#"{"start_slot":100,"end_slot":200}"#).unwrap();
        assert_eq!(request.kind, BundleKind::Blocks);
    }

    #[test]
    fn bundle_kind_round_trips_over_serde_and_as_str() {
        for kind in [BundleKind::Blocks, BundleKind::AccountState] {
            let json = serde_json::to_string(&kind).unwrap();
            assert_eq!(json, format!("\"{}\"", kind.as_str()));
            assert_eq!(serde_json::from_str::<BundleKind>(&json).unwrap(), kind);
            // The store round-trips through `as_str`, not serde, so the two
            // maps have to agree for a stored kind to read back.
            assert_eq!(BundleKind::from_str_opt(kind.as_str()), Some(kind));
        }
        assert!(serde_json::from_str::<BundleKind>("\"accountState\"").is_err());
        assert_eq!(BundleKind::from_str_opt("accountState"), None);
    }

    #[test]
    fn bundle_build_request_carries_kind_through_serde() {
        let request = BundleBuildRequest {
            start_slot: 100,
            end_slot: 200,
            kind: BundleKind::AccountState,
            ..Default::default()
        };
        let json = serde_json::to_string(&request).unwrap();
        assert!(json.contains(r#""kind":"account_state""#), "{json}");
        let parsed: BundleBuildRequest = serde_json::from_str(&json).unwrap();
        assert_eq!(parsed.kind, BundleKind::AccountState);
    }

    #[test]
    fn bundle_build_status_response_serializes_request_and_status() {
        let response = BundleBuildStatusResponse {
            request_id: "r".to_string(),
            start_slot: 100,
            end_slot: 200,
            bundle_size: Some(50),
            status: BundleBuildStatus::InProgress,
            created_at_unix_ms: 0,
            kind: BundleKind::Blocks,
        };
        let json = serde_json::to_value(&response).unwrap();
        assert_eq!(json["request_id"].as_str(), Some("r"));
        assert_eq!(json["start_slot"].as_u64(), Some(100));
        assert_eq!(json["bundle_size"].as_u64(), Some(50));
        assert_eq!(json["status"].as_str(), Some("in_progress"));
        assert_eq!(json["created_at_unix_ms"].as_u64(), Some(0));
    }

    #[test]
    fn an_empty_bundles_list_is_distinct_from_an_absent_one() {
        // A server that predates the section sends no `bundles` at all.
        let absent: BundleBuildListResponse = serde_json::from_str(r#"{"builds":[]}"#).unwrap();
        let empty: BundleBuildListResponse =
            serde_json::from_str(r#"{"builds":[],"bundles":[]}"#).unwrap();
        assert_eq!(absent.bundles, None);
        assert_eq!(empty.bundles, Some(Vec::new()));
    }

    #[test]
    fn completed_bundle_round_trips() {
        let bundle = CompletedBundle {
            start_slot: 100,
            end_slot: 200,
            kind: BundleKind::AccountState,
            completed_at_unix_ms: Some(1_700_000_000_000),
            size_bytes: Some(4_096),
        };
        let json = serde_json::to_string(&bundle).unwrap();
        assert!(json.contains(r#""kind":"account_state""#), "{json}");
        assert_eq!(
            serde_json::from_str::<CompletedBundle>(&json).unwrap(),
            bundle
        );

        let sparse = r#"{"start_slot":1,"end_slot":2,"kind":"blocks"}"#;
        let parsed: CompletedBundle = serde_json::from_str(sparse).unwrap();
        assert_eq!(parsed.completed_at_unix_ms, None);
        assert_eq!(parsed.size_bytes, None);
    }

    #[test]
    fn available_range_reads_a_server_that_predates_kind_and_scope() {
        // Deploys are not atomic: a new client must not fail on the old shape.
        let json = r#"{
            "bundle_start_slot": 100,
            "bundle_start_slot_utc": null,
            "max_bundle_end_slot": 200,
            "max_bundle_end_slot_utc": null,
            "max_bundle_size": 101
        }"#;
        let parsed: AvailableRange = serde_json::from_str(json).unwrap();
        assert_eq!(parsed.bundle_start_slot, 100);
        assert_eq!(parsed.kind, None);
        assert_eq!(parsed.scope, None);
    }

    #[test]
    fn available_range_ignores_fields_it_does_not_know() {
        // The other direction: an old client against a newer server.
        let json = r#"{"bundle_start_slot": 1, "some_future_field": "x"}"#;
        assert_eq!(
            serde_json::from_str::<AvailableRange>(json)
                .unwrap()
                .bundle_start_slot,
            1
        );
    }

    #[test]
    fn available_range_labels_round_trip() {
        let range = AvailableRange {
            bundle_start_slot: 100,
            max_bundle_end_slot: Some(200),
            kind: Some(BundleKind::AccountState),
            scope: Some(RangeScope::User),
            ..Default::default()
        };
        let json = serde_json::to_string(&range).unwrap();
        assert!(json.contains(r#""kind":"account_state""#), "{json}");
        assert!(json.contains(r#""scope":"user""#), "{json}");
        let parsed: AvailableRange = serde_json::from_str(&json).unwrap();
        assert_eq!(parsed.kind, Some(BundleKind::AccountState));
        assert_eq!(parsed.scope, Some(RangeScope::User));
    }

    fn range(start: u64, end: u64) -> AvailableRange {
        AvailableRange {
            bundle_start_slot: start,
            max_bundle_end_slot: Some(end),
            ..Default::default()
        }
    }

    /// Each case lists the available bundles, the requested `[start, end]`, and
    /// the expected split — `Some(_)` for an accepted plan, `None` when the range
    /// is unservable and `split_range` must error.
    #[rstest::rstest]
    #[case::single(vec![range(100, 300)], 100, 300, Some(vec![(100, 300)]))]
    #[case::multi(
        vec![range(100, 200), range(201, 300), range(301, 400)],
        100, 300, Some(vec![(100, 200), (201, 300)])
    )]
    // Smaller bundles nested inside a larger one don't bridge the gap, but the
    // range is still coverable by riding the coarse series.
    #[case::nested(
        vec![range(100, 500), range(110, 150), range(150, 190), range(501, 900)],
        100, 900, Some(vec![(100, 500), (501, 900)])
    )]
    // A fine grid (1k bundles standing in for the 10k grid) overlapped by a
    // coarser, differently-aligned series: ride the fine grid, never landing on
    // the coarse bundle's drifted start.
    #[case::prefers_finer_grid(
        vec![range(1_000, 1_999), range(1_500, 3_400), range(2_000, 2_999), range(3_000, 3_999)],
        1_000, 3_999, Some(vec![(1_000, 1_999), (2_000, 2_999), (3_000, 3_999)])
    )]
    // A 50k snapshot bundle and the 10k bundles built on it share start 100: ride
    // the 10k grid instead of the coarse end, which would strand the cursor on a
    // slot no bundle starts at.
    #[case::shared_start_prefers_finer(
        vec![range(100, 150), range(100, 120), range(121, 140), range(141, 160)],
        100, 160, Some(vec![(100, 120), (121, 140), (141, 160)])
    )]
    // A coarse bundle (100, 200) overlaps a finer pair covering the same span:
    // split_range maximises parallelism, riding the fine pair (two sessions)
    // rather than merging onto the coarse max end (one session).
    #[case::coarse_overlap_prefers_finer_pair(
        vec![range(100, 200), range(100, 150), range(151, 160)],
        100, 160, Some(vec![(100, 150), (151, 160)])
    )]
    // The fine grid has a hole (nothing starts at 141), but a coarse bundle
    // sharing the start spans it: serve the request off the coarse bundle rather
    // than reporting an unsupported range.
    #[case::falls_back_to_coarse(
        vec![range(100, 160), range(100, 120), range(121, 140)],
        100, 160, Some(vec![(100, 160)])
    )]
    // The last bundle overshoots the requested end and is clamped.
    #[case::clamps_final_bundle(vec![range(100, 199), range(200, 999)], 100, 450, Some(vec![(100, 199), (200, 450)]))]
    // A request landing mid-bundle anchors to the covering bundle's start
    // (matching the server) instead of erroring as "not covered".
    #[case::anchors_mid_bundle(vec![range(150, 350)], 200, 300, Some(vec![(150, 300)]))]
    #[case::anchors_then_continues(
        vec![range(150, 350), range(351, 600)],
        200, 600, Some(vec![(150, 350), (351, 600)])
    )]
    #[case::start_inside_bundle_anchors(vec![range(200, 400)], 300, 400, Some(vec![(200, 400)]))]
    // Errors: a start before any bundle's coverage has no covering bundle...
    #[case::start_before_first_bundle(vec![range(200, 400)], 100, 400, None)]
    // ...the end must be reachable...
    #[case::end_not_covered(vec![range(100, 200)], 100, 300, None)]
    #[case::gap_in_coverage(vec![range(100, 200), range(210, 300)], 100, 300, None)]
    // Per-API-key isolation: a user's finer bundles and a coarse GLOBAL bundle
    // sharing a start are fed as SEPARATE entries (never merged on snapshot_slot,
    // which would keep only the max end). split_range set-unions the ends and
    // rides the finer user grid — a regression guard against re-merging.
    #[case::user_and_global_ranges_not_collapsed(
        vec![range(100, 200), range(100, 150), range(151, 200)],
        100, 200, Some(vec![(100, 150), (151, 200)])
    )]
    fn split_range_cases(
        #[case] ranges: Vec<AvailableRange>,
        #[case] start: u64,
        #[case] end: u64,
        #[case] expected: Option<Vec<(u64, u64)>>,
    ) {
        match expected {
            Some(expected) => assert_eq!(
                split_range(
                    &ranges,
                    SlotRange::new_unchecked(start, end),
                    BundleKind::Blocks
                )
                .unwrap(),
                expected
            ),
            None => assert!(
                split_range(
                    &ranges,
                    SlotRange::new_unchecked(start, end),
                    BundleKind::Blocks
                )
                .is_err()
            ),
        }
    }

    fn account_state_range(start: u64, end: u64) -> AvailableRange {
        AvailableRange {
            kind: Some(BundleKind::AccountState),
            ..range(start, end)
        }
    }

    #[test]
    fn a_replay_split_rides_the_recordings_not_a_finer_block_grid() {
        // The finer block grid wins on piece count, which is precisely what would
        // hand a replay run sub-ranges nothing ever recorded.
        let ranges = vec![
            range(100, 149),
            range(150, 199),
            account_state_range(100, 199),
        ];
        let requested = SlotRange::new_unchecked(100, 199);

        assert_eq!(
            split_range(&ranges, requested, BundleKind::Blocks).unwrap(),
            [(100, 149), (150, 199)]
        );
        assert_eq!(
            split_range(&ranges, requested, BundleKind::AccountState).unwrap(),
            [(100, 199)]
        );
    }

    #[test]
    fn a_replay_split_tiles_consecutive_recordings() {
        let ranges = vec![account_state_range(100, 199), account_state_range(200, 299)];
        assert_eq!(
            split_range(
                &ranges,
                SlotRange::new_unchecked(100, 299),
                BundleKind::AccountState
            )
            .unwrap(),
            [(100, 199), (200, 299)]
        );
    }

    #[test]
    fn a_replay_split_ends_short_of_a_recording() {
        // A recording is read slot by slot, so a session stops wherever asked
        // inside it — the same as a block session does inside its bundle.
        let requested = SlotRange::new_unchecked(100, 150);
        assert_eq!(
            split_range(&[range(100, 199)], requested, BundleKind::Blocks).unwrap(),
            [(100, 150)]
        );
        assert_eq!(
            split_range(
                &[account_state_range(100, 199)],
                requested,
                BundleKind::AccountState
            )
            .unwrap(),
            [(100, 150)]
        );
    }

    #[test]
    fn a_replay_split_ends_short_of_its_last_recording() {
        // Only the last sub-range may stop mid-recording; the ones before it
        // still have to hand over on the very next slot.
        let ranges = vec![account_state_range(100, 199), account_state_range(200, 299)];
        assert_eq!(
            split_range(
                &ranges,
                SlotRange::new_unchecked(100, 250),
                BundleKind::AccountState
            )
            .unwrap(),
            [(100, 199), (200, 250)]
        );
    }

    #[test]
    fn a_replay_split_tiles_a_one_slot_recording() {
        // A series whose last split begins on the requested end records exactly
        // one slot.
        let ranges = vec![account_state_range(100, 249), account_state_range(250, 250)];
        assert_eq!(
            split_range(
                &ranges,
                SlotRange::new_unchecked(100, 250),
                BundleKind::AccountState
            )
            .unwrap(),
            [(100, 249), (250, 250)]
        );
    }

    #[test]
    fn a_replay_split_anchors_a_mid_recording_start() {
        // A mid-bundle start anchors to the bundle covering it, whichever kind
        // the walk rides — the two differ in the bundles they read, not in the
        // ranges they admit.
        let requested = SlotRange::new_unchecked(150, 199);
        assert_eq!(
            split_range(&[range(100, 199)], requested, BundleKind::Blocks).unwrap(),
            [(100, 199)]
        );
        assert_eq!(
            split_range(
                &[account_state_range(100, 199)],
                requested,
                BundleKind::AccountState
            )
            .unwrap(),
            [(100, 199)]
        );
    }

    #[test]
    fn a_replay_split_refuses_a_start_no_recording_covers() {
        let error = split_range(
            &[account_state_range(200, 299)],
            SlotRange::new_unchecked(100, 299),
            BundleKind::AccountState,
        )
        .expect_err("no recording covers 100");
        assert!(
            error.contains("is not covered by any available account-state bundle"),
            "{error}"
        );
    }

    #[test]
    fn a_replay_split_ignores_block_ranges() {
        let error = split_range(
            &[range(100, 199)],
            SlotRange::new_unchecked(100, 199),
            BundleKind::AccountState,
        )
        .expect_err("owns no recordings");
        assert!(error.contains("account-state bundle"), "{error}");
    }

    #[test]
    fn blocks_mode_admits_ranges_from_a_server_that_predates_kind() {
        // A server predating the field omits it, and those ranges must still
        // split exactly as they did before the kind was consulted.
        let unlabelled = range(100, 199);
        assert_eq!(unlabelled.kind, None);
        assert_eq!(
            split_range(
                &[unlabelled],
                SlotRange::new_unchecked(100, 150),
                BundleKind::Blocks
            )
            .unwrap(),
            [(100, 150)]
        );
    }

    /// Servable end per bundle start, as `split_range` sees it — every advertised
    /// end at or past its start, keyed by start.
    fn ends_by_start(ranges: &[AvailableRange]) -> BTreeMap<u64, BTreeSet<u64>> {
        let mut ends: BTreeMap<u64, BTreeSet<u64>> = BTreeMap::new();
        for r in ranges {
            if let Some(end) = r.max_bundle_end_slot
                && end >= r.bundle_start_slot
            {
                ends.entry(r.bundle_start_slot).or_default().insert(end);
            }
        }
        ends
    }

    /// Independent spec for the optimum: the gap-free split of `[start, end]` with
    /// the most sub-ranges, found by exhaustively trying every advertised end at
    /// each cursor. `None` when no split aligns with the bundle starts.
    fn reference_max_split(
        ends: &BTreeMap<u64, BTreeSet<u64>>,
        cursor: u64,
        end: u64,
    ) -> Option<Vec<(u64, u64)>> {
        ends.get(&cursor)?
            .iter()
            .filter_map(|&bundle_end| {
                if bundle_end >= end {
                    Some(vec![(cursor, end)])
                } else {
                    reference_max_split(ends, bundle_end + 1, end).map(|mut rest| {
                        rest.insert(0, (cursor, bundle_end));
                        rest
                    })
                }
            })
            .max_by_key(Vec::len)
    }

    /// Structure-independent check that a split is one the server can serve: it
    /// tiles `[start, end]` gap-free, every start is a real bundle start, and
    /// every end is within that start's largest advertised end.
    fn is_valid_split(
        split: &[(u64, u64)],
        ends: &BTreeMap<u64, BTreeSet<u64>>,
        start: u64,
        end: u64,
    ) -> bool {
        split.first().is_some_and(|&(s, _)| s == start)
            && split.last().is_some_and(|&(_, e)| e == end)
            && split.windows(2).all(|w| w[1].0 == w[0].1 + 1)
            && split.iter().all(|&(s, e)| {
                e >= s
                    && ends
                        .get(&s)
                        .and_then(|bundle_ends| bundle_ends.iter().next_back())
                        .is_some_and(|&max_end| e <= max_end)
            })
    }

    /// Property test: across many randomized bundle layouts, `split_range` must
    /// return a *valid* split with the *maximum* number of sub-ranges whenever one
    /// exists, and error exactly when none does. Deterministic (fixed LCG seed) so
    /// a failure is reproducible.
    #[test]
    fn split_range_matches_reference() {
        let mut seed: u64 = 0x9E3779B97F4A7C15;
        let mut next = || {
            seed = seed
                .wrapping_mul(6364136223846793005)
                .wrapping_add(1442695040888963407);
            seed >> 33
        };

        for _ in 0..50_000 {
            // A small slot universe makes shared starts, overlaps, nesting, and
            // gaps all common.
            let ranges: Vec<AvailableRange> = (0..next() % 6)
                .map(|_| {
                    let start = next() % 12;
                    range(start, start + next() % 6) // len 0 => filtered out
                })
                .collect();
            let start = next() % 12;
            let end = start + next() % 6; // sometimes start == end, sometimes unservable

            let got = split_range(
                &ranges,
                SlotRange::new_unchecked(start, end),
                BundleKind::Blocks,
            );
            let ends = ends_by_start(&ranges);
            // split_range anchors a mid-bundle start to the covering bundle's
            // start, so the reference must walk from the same anchor.
            let anchor = ends
                .range(..=start)
                .rfind(|(_, e)| e.iter().next_back().is_some_and(|&x| x >= start))
                .map(|(&s, _)| s);
            let reference = anchor.and_then(|a| reference_max_split(&ends, a, end));

            let layout: Vec<_> = ranges
                .iter()
                .map(|r| (r.bundle_start_slot, r.max_bundle_end_slot))
                .collect();
            match (&got, &reference) {
                (Ok(split), Some(best)) => {
                    assert!(
                        is_valid_split(split, &ends, anchor.unwrap(), end),
                        "invalid split {split:?} for {layout:?} [{start},{end}]"
                    );
                    assert_eq!(
                        split.len(),
                        best.len(),
                        "suboptimal split {split:?} vs {best:?} for {layout:?} [{start},{end}]"
                    );
                }
                (Err(_), None) => {}
                _ => panic!(
                    "disagreement: split_range={got:?}, reference={reference:?} for {layout:?} [{start},{end}]"
                ),
            }
        }
    }
}