simulator_api/

lib.rs

use std::{
    collections::{BTreeMap, BTreeSet, HashSet},
    fmt,
};

use base64::{
    DecodeError as Base64DecodeError, Engine as _, engine::general_purpose::STANDARD as BASE64,
};
use serde::{Deserialize, Serialize};
use solana_address::Address;

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: String,
        /// 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: String,
        /// 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<String, 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: String,
    pub request: ContinueParams,
}

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

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

/// 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),
        }
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub enum ActionKind {
    Simulate,
    Send,
}

/// Where a [`ScheduledAction`] fires.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "at", rename_all = "camelCase")]
pub enum ActionAnchor {
    /// 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.
#[serde_with::serde_as]
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ScheduledAction {
    pub anchor: ActionAnchor,
    pub kind: ActionKind,
    /// Base64-encoded transactions, carried over the wire as a base64-encoded,
    /// zstd-compressed blob. For an `afterSlot` anchor, 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.
    #[serde(default, with = "compressed")]
    pub transactions: Vec<String>,
    /// Account state overrides seeding the first transaction's simulation, per
    /// `simulateTransaction` `modifyAccountStates` semantics. Only valid for
    /// `kind: simulate`.
    #[serde(default)]
    pub account_overrides: AccountModifications,
    /// 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>,
}

/// 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,
    /// When true, each historical swap is re-quoted through the metis router and the routed
    /// transaction is *simulated* just before the original executes — never committed, so the
    /// replay is unaffected. Within a block the simulated swaps compose.
    #[serde(default)]
    #[builder(default)]
    pub reroute_order_flow: bool,
    /// 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.
    /// Maximum value: 900 (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>,
    /// Agent configurations to run as sidecars alongside this session.
    #[serde(default)]
    #[builder(default)]
    pub agents: Vec<AgentParams>,
    /// 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`.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    #[builder(default)]
    pub actions: Vec<ScheduledAction>,
}

/// 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,
        }
    }
}

/// Available agent types for sidecar participation in backtest sessions.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub enum AgentType {
    Arb,
}

/// Parameters for a circular arbitrage route.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ArbRouteParams {
    pub base_mint: String,
    pub temp_mint: String,
    #[serde(default)]
    pub buy_dexes: Vec<String>,
    #[serde(default)]
    pub sell_dexes: Vec<String>,
    pub min_input: u64,
    pub max_input: u64,
    #[serde(default)]
    pub min_profit: u64,
}

/// Configuration for an agent to run alongside a backtest session.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct AgentParams {
    pub agent_type: AgentType,
    pub wallet: Option<String>,
    /// Base58-encoded 64-byte keypair for signing transactions (compatible with `solana-keygen`).
    pub keypair: Option<String>,
    pub seed_sol_lamports: Option<u64>,
    #[serde(default)]
    pub seed_token_accounts: BTreeMap<String, u64>,
    #[serde(default)]
    pub arb_routes: Vec<ArbRouteParams>,
}

/// 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>,
);

/// 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: String,
        rpc_endpoint: String,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        task_id: Option<String>,
    },
    SessionAttached {
        session_id: String,
        rpc_endpoint: String,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        task_id: Option<String>,
    },
    SessionsCreatedV2 {
        control_session_id: String,
        session_ids: Vec<String>,
        #[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: String,
        session_ids: Vec<String>,
        #[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>,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        agent_stats: Option<Vec<AgentStatsReport>>,
    },
    Status {
        status: BacktestStatus,
    },
    SessionEventV2 {
        session_id: String,
        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)
    }
}

/// Structured stats reported by an agent during a backtest session.
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct AgentStatsReport {
    pub name: String,
    pub slots_processed: u64,
    pub opportunities_found: u64,
    pub opportunities_skipped: u64,
    pub no_routes: u64,
    pub txs_produced: u64,
    /// Cumulative expected profit per base mint, keyed by mint address.
    pub expected_gain_by_mint: BTreeMap<String, i64>,
    /// Transactions successfully executed by the sidecar.
    #[serde(default)]
    pub txs_submitted: u64,
    /// Transactions that failed execution.
    #[serde(default)]
    pub txs_failed: u64,
    /// Transactions rejected by preflight simulation (unprofitable).
    #[serde(default)]
    pub txs_simulation_rejected: u64,
    /// Preflight simulation RPC calls that errored.
    #[serde(default)]
    pub txs_simulation_failed: u64,
}

/// 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 {
    /// Swaps re-quoted through metis and queued for simulation.
    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.
    pub requote_failures: 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
    }
}

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,
    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,
    },
}

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::AdvanceSlotFailed { .. }
                | BacktestError::FinalizeSlotFailed { .. }
                | BacktestError::Internal { .. }
        )
    }
}

/// One contiguous block range available on the history clerk.
#[derive(Debug, Clone, 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>,
}

/// Request body of `POST /build-bundle`.
#[derive(Debug, Clone, 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.
    #[serde(default)]
    pub idempotency_key: Option<String>,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum BundleBuildStatus {
    /// The Prefect flow run finished successfully; the bundle is built.
    Completed,
    /// The Prefect flow run reached a definitive failure state
    /// (`Failed`/`Crashed`/`Cancelled`). Distinct from [`Self::NeedsInvestigation`]:
    /// here the pipeline reported a real failure.
    Failed,
    /// The Prefect flow run is scheduled or running.
    InProgress,
    /// The flow-run state could not be determined (`Unknown` or missing) — the
    /// outcome is genuinely unclear from Prefect, so a human should check rather
    /// than 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,
}

/// 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.
///
/// Returns an error if the requested start slot is not a bundle start, or if no
/// gap-free split reaches `requested_end`.
pub fn split_range(
    ranges: &[AvailableRange],
    requested_start: u64,
    requested_end: u64,
) -> Result<Vec<(u64, u64)>, String> {
    if requested_end < requested_start {
        return Err(format!(
            "invalid range: start_slot {requested_start} > end_slot {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.
    let mut ends_by_start: 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_by_start
                .entry(r.bundle_start_slot)
                .or_default()
                .insert(end);
        }
    }

    // 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.
    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 bundle range"
        ));
    };

    // 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 bundle ranges"
            )
        }
    })
}

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::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}")
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[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_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_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,
        };
        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));
    }

    fn range(start: u64, end: u64) -> AvailableRange {
        AvailableRange {
            bundle_start_slot: start,
            bundle_start_slot_utc: None,
            max_bundle_end_slot: Some(end),
            max_bundle_end_slot_utc: None,
            max_bundle_size: None,
        }
    }

    /// 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)]
    // ...and the range must not be inverted.
    #[case::inverted_range(vec![range(100, 300)], 300, 100, 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, start, end).unwrap(), expected),
            None => assert!(split_range(&ranges, start, end).is_err()),
        }
    }

    /// Servable end per bundle start, as `split_range` sees it — every advertised
    /// end with `end > 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, start, end);
            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}]"
                ),
            }
        }
    }
}