Skip to main content

simulator_api/
lib.rs

1use std::{
2    collections::{BTreeMap, BTreeSet, HashSet},
3    fmt,
4};
5
6use base64::{
7    DecodeError as Base64DecodeError, Engine as _, engine::general_purpose::STANDARD as BASE64,
8};
9use serde::{Deserialize, Serialize};
10use solana_address::Address;
11
12pub mod subscribe_config;
13pub mod usage;
14pub mod ws_compression;
15
16/// Backtest RPC methods exposed to the client.
17#[derive(Debug, Serialize, Deserialize)]
18#[serde(tag = "method", content = "params", rename_all = "camelCase")]
19pub enum BacktestRequest {
20    CreateBacktestSession(CreateBacktestSessionRequest),
21    Continue(ContinueParams),
22    ContinueTo(ContinueToParams),
23    ContinueSessionV1(ContinueSessionRequestV1),
24    ContinueToSessionV1(ContinueToSessionRequestV1),
25    CloseBacktestSession,
26    CloseSessionV1(CloseSessionRequestV1),
27    AttachBacktestSession {
28        session_id: String,
29        /// Last sequence number the client received. Responses after this sequence
30        /// will be replayed from the session's buffer. None = replay entire buffer.
31        last_sequence: Option<u64>,
32    },
33    /// Sent after reattaching and rebuilding any dependent subscriptions.
34    /// Allows the manager to resume a session that was paused for handoff.
35    ResumeAttachedSession,
36    AttachParallelControlSessionV2 {
37        control_session_id: String,
38        /// Last per-session sequence number received by the client. Responses after
39        /// these sequence numbers will be replayed from the manager's per-session
40        /// replay store. Missing sessions replay their entire retained history.
41        #[serde(default)]
42        last_sequences: BTreeMap<String, u64>,
43    },
44}
45
46/// Versioned payload for `CreateBacktestSession`.
47///
48/// - `V0` keeps backwards-compatible shape by using `CreateSessionParams` directly.
49/// - `V1` keeps the same shape and adds `parallel`.
50#[derive(Debug, Clone, Serialize, Deserialize)]
51#[serde(untagged)]
52pub enum CreateBacktestSessionRequest {
53    V1(CreateBacktestSessionRequestV1),
54    V0(CreateSessionParams),
55}
56
57impl CreateBacktestSessionRequest {
58    pub fn into_request_options(self) -> CreateBacktestSessionRequestOptions {
59        match self {
60            Self::V0(request) => CreateBacktestSessionRequestOptions {
61                request,
62                parallel: false,
63            },
64            Self::V1(CreateBacktestSessionRequestV1 { request, parallel }) => {
65                CreateBacktestSessionRequestOptions { request, parallel }
66            }
67        }
68    }
69
70    pub fn into_request_and_parallel(self) -> (CreateSessionParams, bool) {
71        let options = self.into_request_options();
72        (options.request, options.parallel)
73    }
74}
75
76impl From<CreateSessionParams> for CreateBacktestSessionRequest {
77    fn from(value: CreateSessionParams) -> Self {
78        Self::V0(value)
79    }
80}
81
82impl From<CreateBacktestSessionRequestV1> for CreateBacktestSessionRequest {
83    fn from(value: CreateBacktestSessionRequestV1) -> Self {
84        Self::V1(value)
85    }
86}
87
88#[derive(Debug, Clone, Serialize, Deserialize)]
89#[serde(rename_all = "camelCase")]
90pub struct CreateBacktestSessionRequestV1 {
91    #[serde(flatten)]
92    pub request: CreateSessionParams,
93    pub parallel: bool,
94}
95
96#[derive(Debug, Clone)]
97pub struct CreateBacktestSessionRequestOptions {
98    pub request: CreateSessionParams,
99    pub parallel: bool,
100}
101
102#[derive(Debug, Serialize, Deserialize)]
103#[serde(rename_all = "camelCase")]
104pub struct ContinueSessionRequestV1 {
105    pub session_id: String,
106    pub request: ContinueParams,
107}
108
109#[derive(Debug, Clone, Serialize, Deserialize)]
110#[serde(rename_all = "camelCase")]
111pub struct ContinueToSessionRequestV1 {
112    pub session_id: String,
113    pub request: ContinueToParams,
114}
115
116#[derive(Debug, Serialize, Deserialize)]
117#[serde(rename_all = "camelCase")]
118pub struct CloseSessionRequestV1 {
119    pub session_id: String,
120}
121
122/// A filter registered at session creation describing which upcoming batches
123/// the session should announce ahead of execution via
124/// [`BacktestResponse::DiscoveryBatch`] (and its session-event twins). Each
125/// filter describes an event of interest (e.g. a specific program executing);
126/// the session "discovers" the batch in which that event will occur and
127/// emits a `DiscoveryBatch` so the client can pause immediately before it.
128#[serde_with::serde_as]
129#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
130#[serde(tag = "kind", content = "value", rename_all = "camelCase")]
131pub enum DiscoveryFilter {
132    /// Discover batches containing a transaction that invokes this program.
133    ProgramExecuted(#[serde_as(as = "serde_with::DisplayFromStr")] Address),
134}
135
136/// Per-transaction facts a [`DiscoveryFilter`] can inspect when deciding
137/// whether to match. Callers build this once per transaction and feed it to
138/// every registered filter; new variants add fields here rather than growing
139/// the [`DiscoveryFilter::matches`] signature.
140pub struct TxMatchContext<'a> {
141    /// Programs invoked by the transaction (top-level + CPI observed in logs).
142    pub invoked_programs: &'a HashSet<Address>,
143}
144
145impl DiscoveryFilter {
146    /// Return `true` when this filter is satisfied by the transaction
147    /// described by `ctx`.
148    pub fn matches(&self, ctx: &TxMatchContext<'_>) -> bool {
149        match self {
150            Self::ProgramExecuted(target) => ctx.invoked_programs.contains(target),
151        }
152    }
153}
154
155#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
156#[serde(rename_all = "camelCase")]
157pub enum ActionKind {
158    Simulate,
159    Send,
160}
161
162/// Where a [`ScheduledAction`] fires.
163#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
164#[serde(tag = "at", rename_all = "camelCase")]
165pub enum ActionAnchor {
166    /// At the end of each slot, after its transactions execute but before it
167    /// finalizes. Fires the entry for the current slot, if any.
168    AfterSlot {
169        #[serde(default, with = "compressed")]
170        slots: Vec<u64>,
171    },
172    /// Before each batch matching `filter` executes. The whole sequence fires at
173    /// each matching batch.
174    BeforeMatch { filter: DiscoveryFilter },
175    /// After each batch matching `filter` commits. The whole sequence fires at
176    /// each matching batch.
177    AfterMatch { filter: DiscoveryFilter },
178}
179
180/// Upper bound on a decompressed transactions blob, guarding the server against
181/// a decompression bomb in client input.
182const MAX_ACTION_TRANSACTIONS_BYTES: u64 = 64 * 1024 * 1024;
183
184/// serde adapter that stores any value as a base64-encoded, zstd-compressed JSON
185/// blob on the wire, so transaction bytes never travel raw. Used via
186/// `#[serde(with = "compressed")]` on [`ScheduledAction`]'s `transactions` field
187/// and [`ActionAnchor`]'s `slots` field.
188mod compressed {
189    use std::io::Read;
190
191    use base64::Engine as _;
192    use serde::{Deserialize, Deserializer, Serialize, Serializer, de::DeserializeOwned};
193
194    use super::{BASE64, MAX_ACTION_TRANSACTIONS_BYTES};
195
196    /// zstd compression level; `0` selects zstd's default (level 3).
197    const COMPRESSION_LEVEL: i32 = 0;
198
199    pub fn serialize<T, S>(value: &T, serializer: S) -> Result<S::Ok, S::Error>
200    where
201        T: Serialize,
202        S: Serializer,
203    {
204        let json = serde_json::to_vec(value).map_err(serde::ser::Error::custom)?;
205        let compressed = zstd::stream::encode_all(json.as_slice(), COMPRESSION_LEVEL)
206            .map_err(serde::ser::Error::custom)?;
207        serializer.serialize_str(&BASE64.encode(compressed))
208    }
209
210    pub fn deserialize<'de, T, D>(deserializer: D) -> Result<T, D::Error>
211    where
212        T: DeserializeOwned,
213        D: Deserializer<'de>,
214    {
215        let blob = String::deserialize(deserializer)?;
216        let compressed = BASE64
217            .decode(blob.as_bytes())
218            .map_err(serde::de::Error::custom)?;
219        let mut json = Vec::new();
220        zstd::stream::read::Decoder::new(compressed.as_slice())
221            .and_then(|d| Read::take(d, MAX_ACTION_TRANSACTIONS_BYTES).read_to_end(&mut json))
222            .map_err(serde::de::Error::custom)?;
223        serde_json::from_slice(&json).map_err(serde::de::Error::custom)
224    }
225}
226
227/// A sequence of transactions the server runs automatically during a backtest,
228/// with results streamed over the `actionSubscribe` subscription.
229#[serde_with::serde_as]
230#[derive(Debug, Clone, Serialize, Deserialize)]
231#[serde(rename_all = "camelCase")]
232pub struct ScheduledAction {
233    pub anchor: ActionAnchor,
234    pub kind: ActionKind,
235    /// Base64-encoded transactions, carried over the wire as a base64-encoded,
236    /// zstd-compressed blob. For an `afterSlot` anchor, entry `i` fires at
237    /// `slots[i]` — the two lists must be the same length, and repeating a slot
238    /// runs its entries there in order as one sequence. For match anchors the
239    /// whole list fires, in order, at each matching batch. For `simulate`, each
240    /// transaction's writes are visible to the next without committing to the
241    /// session ledger.
242    #[serde(default, with = "compressed")]
243    pub transactions: Vec<String>,
244    /// Account state overrides seeding the first transaction's simulation, per
245    /// `simulateTransaction` `modifyAccountStates` semantics. Only valid for
246    /// `kind: simulate`.
247    #[serde(default)]
248    pub account_overrides: AccountModifications,
249    /// Accounts whose post-execution state to return, following
250    /// `simulateTransaction` semantics; reflects cumulative state after the
251    /// final transaction.
252    #[serde_as(as = "Vec<serde_with::DisplayFromStr>")]
253    #[serde(default)]
254    pub return_accounts: Vec<Address>,
255    /// Optional client tag echoed back on each `actionSubscribe` result to correlate it to this action.
256    #[serde(default)]
257    pub label: Option<String>,
258}
259
260/// Parameters required to start a new backtest session.
261#[serde_with::serde_as]
262#[derive(Debug, Clone, Serialize, Deserialize, bon::Builder)]
263#[serde(rename_all = "camelCase")]
264pub struct CreateSessionParams {
265    /// First slot (inclusive) to replay.
266    pub start_slot: u64,
267    /// Last slot (inclusive) to replay.
268    pub end_slot: u64,
269    #[serde_as(as = "BTreeSet<serde_with::DisplayFromStr>")]
270    #[serde(default)]
271    #[builder(default)]
272    /// Skip transactions signed by these addresses.
273    pub signer_filter: BTreeSet<Address>,
274    /// When true, include a session summary with transaction statistics in client-facing
275    /// `Completed` responses. Summary generation remains enabled internally for metrics.
276    #[serde(default)]
277    #[builder(default)]
278    pub send_summary: bool,
279    /// When true, replace each historical swap with one re-quoted through the metis
280    /// router against current simulation state, so orders may execute on a different
281    /// venue than in history. Forces `fail_fast` off (divergence is expected).
282    #[serde(default)]
283    #[builder(default)]
284    pub reroute_order_flow: bool,
285    /// Maximum seconds to wait for ECS capacity-related startup retries before
286    /// failing session creation. If not set (or 0), capacity errors fail immediately.
287    #[serde(default)]
288    pub capacity_wait_timeout_secs: Option<u16>,
289    /// Maximum seconds to keep the session alive after the control websocket disconnects.
290    /// If not set (or 0), the session tears down immediately on disconnect.
291    /// Maximum value: 900 (15 minutes).
292    #[serde(default)]
293    pub disconnect_timeout_secs: Option<u16>,
294    /// Extra compute units to add to each transaction's `SetComputeUnitLimit` budget.
295    /// Useful when replaying with an account override whose program uses more CU than
296    /// the original, causing otherwise-healthy transactions to run out of budget.
297    /// Only applied when a `SetComputeUnitLimit` instruction is already present.
298    #[serde(default)]
299    pub extra_compute_units: Option<u32>,
300    /// Agent configurations to run as sidecars alongside this session.
301    #[serde(default)]
302    #[builder(default)]
303    pub agents: Vec<AgentParams>,
304    /// Events of interest the session should watch for. When an upcoming
305    /// batch matches any filter, the server emits
306    /// [`BacktestResponse::DiscoveryBatch`] (and its session-event twins)
307    /// ahead of execution so the client can follow up with
308    /// [`BacktestRequest::ContinueTo`] to pause before the batch. Empty
309    /// means no batch discoveries are performed (existing behaviour).
310    #[serde(default, skip_serializing_if = "Vec::is_empty")]
311    #[builder(default)]
312    pub discoveries: Vec<DiscoveryFilter>,
313    /// Transactions the server runs automatically during the backtest, with
314    /// results streamed over `actionSubscribe`.
315    #[serde(default, skip_serializing_if = "Vec::is_empty")]
316    #[builder(default)]
317    pub actions: Vec<ScheduledAction>,
318}
319
320/// What counts as a "non-benign" divergence for the backtest session fail-fast
321/// behaviour (see `BacktestSessionOptions::fail_fast`).
322#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
323#[serde(rename_all = "kebab-case")]
324pub enum FailFastDivergenceKind {
325    /// Any divergence other than a benign log diff (the full `is_divergent()` set:
326    /// log mismatch, error mismatch, balance-diff mismatch).
327    #[default]
328    AnyNonBenign,
329    /// Only divergences on transactions that touch an account currently watched by an
330    /// account-diff subscription (subscribed directly, or owned by a subscribed program).
331    Tracked,
332}
333
334impl FailFastDivergenceKind {
335    /// Stable string form used for CLI/env-var serialization. Matches the serde
336    /// `kebab-case` representation.
337    pub fn as_str(self) -> &'static str {
338        match self {
339            Self::AnyNonBenign => "any-non-benign",
340            Self::Tracked => "tracked",
341        }
342    }
343
344    /// Inverse of [`Self::as_str`]; returns `None` for an unrecognized value.
345    pub fn from_str_opt(value: &str) -> Option<Self> {
346        match value {
347            "any-non-benign" => Some(Self::AnyNonBenign),
348            "tracked" => Some(Self::Tracked),
349            _ => None,
350        }
351    }
352}
353
354/// Available agent types for sidecar participation in backtest sessions.
355#[derive(Debug, Clone, Serialize, Deserialize)]
356#[serde(rename_all = "camelCase")]
357pub enum AgentType {
358    Arb,
359}
360
361/// Parameters for a circular arbitrage route.
362#[derive(Debug, Clone, Serialize, Deserialize)]
363#[serde(rename_all = "camelCase")]
364pub struct ArbRouteParams {
365    pub base_mint: String,
366    pub temp_mint: String,
367    #[serde(default)]
368    pub buy_dexes: Vec<String>,
369    #[serde(default)]
370    pub sell_dexes: Vec<String>,
371    pub min_input: u64,
372    pub max_input: u64,
373    #[serde(default)]
374    pub min_profit: u64,
375}
376
377/// Configuration for an agent to run alongside a backtest session.
378#[derive(Debug, Clone, Serialize, Deserialize)]
379#[serde(rename_all = "camelCase")]
380pub struct AgentParams {
381    pub agent_type: AgentType,
382    pub wallet: Option<String>,
383    /// Base58-encoded 64-byte keypair for signing transactions (compatible with `solana-keygen`).
384    pub keypair: Option<String>,
385    pub seed_sol_lamports: Option<u64>,
386    #[serde(default)]
387    pub seed_token_accounts: BTreeMap<String, u64>,
388    #[serde(default)]
389    pub arb_routes: Vec<ArbRouteParams>,
390}
391
392/// Account state modifications to apply.
393#[serde_with::serde_as]
394#[derive(Debug, Clone, Serialize, Deserialize, Default)]
395pub struct AccountModifications(
396    #[serde_as(as = "BTreeMap<serde_with::DisplayFromStr, _>")]
397    #[serde(default)]
398    pub BTreeMap<Address, AccountData>,
399);
400
401/// Arguments used to continue an existing session.
402#[serde_with::serde_as]
403#[derive(Debug, Serialize, Deserialize)]
404#[serde(rename_all = "camelCase")]
405pub struct ContinueParams {
406    #[serde(default = "ContinueParams::default_advance_count")]
407    /// Number of blocks to advance before waiting.
408    pub advance_count: u64,
409    #[serde(default)]
410    /// Base64-encoded transactions to execute before advancing.
411    pub transactions: Vec<String>,
412    #[serde(default)]
413    /// Account state overrides to apply ahead of execution.
414    pub modify_account_states: AccountModifications,
415}
416
417impl Default for ContinueParams {
418    fn default() -> Self {
419        Self {
420            advance_count: Self::default_advance_count(),
421            transactions: Vec::new(),
422            modify_account_states: AccountModifications(BTreeMap::new()),
423        }
424    }
425}
426
427impl ContinueParams {
428    pub fn default_advance_count() -> u64 {
429        1
430    }
431}
432
433/// Payload emitted when a session halts at a caller-specified point.
434/// `batch_index` is `None` for block-boundary pauses and `Some(n)` when the
435/// session stopped *before* batch `n` within `slot` (no transaction from
436/// batch `n` has been applied). While paused, RPC reads against the session
437/// observe partial state up through batch `n - 1`.
438#[derive(Debug, Clone, Serialize, Deserialize)]
439#[serde(rename_all = "camelCase")]
440pub struct PausedEvent {
441    pub slot: u64,
442    #[serde(default, skip_serializing_if = "Option::is_none")]
443    pub batch_index: Option<u32>,
444}
445
446/// Payload emitted when the session has *discovered* an upcoming batch that
447/// matches one or more registered [`DiscoveryFilter`]s from session creation
448/// (for example, a batch containing a transaction that invokes a program of
449/// interest). The `(slot, batch_index)` pair can be fed directly to
450/// [`BacktestRequest::ContinueTo`] to pause immediately before the batch
451/// executes. After each `Continue` / `ContinueTo`, the session emits the
452/// next `DiscoveryBatchEvent` ahead of the next matching batch, enabling a
453/// reactive "pause on every discovery" driver loop.
454#[serde_with::serde_as]
455#[derive(Debug, Clone, Serialize, Deserialize)]
456#[serde(rename_all = "camelCase")]
457pub struct DiscoveryBatchEvent {
458    pub slot: u64,
459    pub batch_index: u32,
460    /// Filters that matched this batch (always non-empty).
461    pub matched: Vec<DiscoveryFilter>,
462    /// Encoded transactions in this batch that triggered the match. Each
463    /// entry carries the serialized `VersionedTransaction` bytes paired with
464    /// the encoding used.
465    pub transactions: Vec<EncodedBinary>,
466}
467
468/// Arguments used to step an existing session to a precise point.
469#[derive(Debug, Clone, Serialize, Deserialize)]
470#[serde(rename_all = "camelCase")]
471pub struct ContinueToParams {
472    /// Target slot to stop in (or at, if `batch_index` is `None`).
473    pub slot: u64,
474    /// Batch within the target slot at which to pause, **exclusive** — the
475    /// session halts immediately *before* batch `n` executes, so no
476    /// transaction in that batch has been applied yet. `None` runs the
477    /// whole slot, pausing at the block boundary. While paused, RPC reads
478    /// observe partial state up through batch `n - 1`.
479    #[serde(default)]
480    pub batch_index: Option<u32>,
481}
482
483/// Supported binary encodings for account/transaction payloads.
484#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
485#[serde(rename_all = "lowercase")]
486pub enum BinaryEncoding {
487    Base64,
488}
489
490impl BinaryEncoding {
491    pub fn encode(self, bytes: &[u8]) -> String {
492        match self {
493            Self::Base64 => BASE64.encode(bytes),
494        }
495    }
496
497    pub fn decode(self, data: &str) -> Result<Vec<u8>, Base64DecodeError> {
498        match self {
499            Self::Base64 => BASE64.decode(data),
500        }
501    }
502}
503
504/// A blob paired with the encoding needed to decode it.
505#[derive(Debug, Clone, Serialize, Deserialize)]
506#[serde(rename_all = "camelCase")]
507pub struct EncodedBinary {
508    /// Encoded payload.
509    pub data: String,
510    /// Encoding scheme used for the payload.
511    pub encoding: BinaryEncoding,
512}
513
514impl EncodedBinary {
515    pub fn new(data: String, encoding: BinaryEncoding) -> Self {
516        Self { data, encoding }
517    }
518
519    pub fn from_bytes(bytes: &[u8], encoding: BinaryEncoding) -> Self {
520        Self {
521            data: encoding.encode(bytes),
522            encoding,
523        }
524    }
525
526    pub fn decode(&self) -> Result<Vec<u8>, Base64DecodeError> {
527        self.encoding.decode(&self.data)
528    }
529}
530
531/// Account snapshot used to seed or modify state in a session.
532#[serde_with::serde_as]
533#[derive(Debug, Clone, Serialize, Deserialize)]
534#[serde(rename_all = "camelCase")]
535pub struct AccountData {
536    /// Account data bytes and encoding.
537    pub data: EncodedBinary,
538    /// Whether the account is executable.
539    pub executable: bool,
540    /// Lamport balance.
541    pub lamports: u64,
542    #[serde_as(as = "serde_with::DisplayFromStr")]
543    /// Account owner pubkey.
544    pub owner: Address,
545    /// Allocated space in bytes.
546    pub space: u64,
547}
548
549impl AccountData {
550    pub fn to_account(&self) -> Result<solana_account::Account, Base64DecodeError> {
551        Ok(solana_account::Account {
552            data: self.data.decode()?,
553            lamports: self.lamports,
554            owner: self.owner,
555            executable: self.executable,
556            rent_epoch: 0,
557        })
558    }
559}
560
561/// Responses returned over the backtest RPC channel.
562#[derive(Debug, Clone, Serialize, Deserialize)]
563#[serde(tag = "method", content = "params", rename_all = "camelCase")]
564pub enum BacktestResponse {
565    SessionCreated {
566        session_id: String,
567        rpc_endpoint: String,
568        #[serde(default, skip_serializing_if = "Option::is_none")]
569        task_id: Option<String>,
570    },
571    SessionAttached {
572        session_id: String,
573        rpc_endpoint: String,
574        #[serde(default, skip_serializing_if = "Option::is_none")]
575        task_id: Option<String>,
576    },
577    SessionsCreatedV2 {
578        control_session_id: String,
579        session_ids: Vec<String>,
580        #[serde(default)]
581        task_ids: Vec<Option<String>>,
582        /// Per-sub-session start/end slots, parallel to `session_ids`. The
583        /// server's split is authoritative (a mid-bundle request anchors to the
584        /// covering bundle), so the client binds each sub-session to its range
585        /// from here rather than re-deriving the split locally.
586        #[serde(default)]
587        start_slots: Vec<u64>,
588        #[serde(default)]
589        end_slots: Vec<u64>,
590    },
591    ParallelSessionAttachedV2 {
592        control_session_id: String,
593        session_ids: Vec<String>,
594        #[serde(default)]
595        task_ids: Vec<Option<String>>,
596    },
597    ReadyForContinue,
598    SlotNotification(u64),
599    Paused(PausedEvent),
600    DiscoveryBatch(DiscoveryBatchEvent),
601    Error(BacktestError),
602    Success,
603    Completed {
604        /// Session summary with transaction statistics.
605        /// The session always computes this summary, but management may omit it from
606        /// client-facing responses unless `send_summary` was requested at session creation.
607        #[serde(skip_serializing_if = "Option::is_none")]
608        summary: Option<SessionSummary>,
609        #[serde(default, skip_serializing_if = "Option::is_none")]
610        agent_stats: Option<Vec<AgentStatsReport>>,
611    },
612    Status {
613        status: BacktestStatus,
614    },
615    SessionEventV2 {
616        session_id: String,
617        seq_id: u64,
618        event: SessionEventKind,
619    },
620}
621
622impl BacktestResponse {
623    pub fn is_completed(&self) -> bool {
624        matches!(self, BacktestResponse::Completed { .. })
625    }
626
627    pub fn is_terminal(&self) -> bool {
628        match self {
629            BacktestResponse::Completed { .. } => true,
630            BacktestResponse::Error(e) => e.is_terminal(),
631            _ => false,
632        }
633    }
634}
635
636impl From<BacktestStatus> for BacktestResponse {
637    fn from(status: BacktestStatus) -> Self {
638        Self::Status { status }
639    }
640}
641
642impl From<String> for BacktestResponse {
643    fn from(message: String) -> Self {
644        BacktestError::Internal { error: message }.into()
645    }
646}
647
648impl From<&str> for BacktestResponse {
649    fn from(message: &str) -> Self {
650        BacktestError::Internal {
651            error: message.to_string(),
652        }
653        .into()
654    }
655}
656
657#[derive(Debug, Clone, Serialize, Deserialize)]
658#[serde(tag = "method", content = "params", rename_all = "camelCase")]
659pub enum SessionEventKind {
660    ReadyForContinue,
661    SlotNotification(u64),
662    Paused(PausedEvent),
663    DiscoveryBatch(DiscoveryBatchEvent),
664    Error(BacktestError),
665    Success,
666    Completed {
667        #[serde(skip_serializing_if = "Option::is_none")]
668        summary: Option<SessionSummary>,
669    },
670    Status {
671        status: BacktestStatus,
672    },
673}
674
675impl SessionEventKind {
676    pub fn is_terminal(&self) -> bool {
677        match self {
678            Self::Completed { .. } => true,
679            Self::Error(e) => e.is_terminal(),
680            _ => false,
681        }
682    }
683}
684
685/// Wire format wrapper for responses sent over the control websocket.
686/// Sessions with `disconnect_timeout_secs > 0` use this to track client position.
687#[derive(Debug, Clone, Serialize, Deserialize)]
688#[serde(rename_all = "camelCase")]
689pub struct SequencedResponse {
690    pub seq_id: u64,
691    #[serde(flatten)]
692    pub response: BacktestResponse,
693}
694
695/// High-level progress states during a `Continue` call.
696#[derive(Debug, Clone, Serialize, Deserialize)]
697#[serde(rename_all = "camelCase")]
698pub enum BacktestStatus {
699    /// Runtime startup is in progress.
700    StartingRuntime,
701    DecodedTransactions,
702    AppliedAccountModifications,
703    ReadyToExecuteUserTransactions,
704    ExecutedUserTransactions,
705    ExecutingBlockTransactions,
706    ExecutedBlockTransactions,
707    ProgramAccountsLoaded,
708}
709
710impl std::fmt::Display for BacktestStatus {
711    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
712        let s = match self {
713            Self::StartingRuntime => "starting runtime",
714            Self::DecodedTransactions => "decoded transactions",
715            Self::AppliedAccountModifications => "applied account modifications",
716            Self::ReadyToExecuteUserTransactions => "ready to execute user transactions",
717            Self::ExecutedUserTransactions => "executed user transactions",
718            Self::ExecutingBlockTransactions => "executing block transactions",
719            Self::ExecutedBlockTransactions => "executed block transactions",
720            Self::ProgramAccountsLoaded => "program accounts loaded",
721        };
722        f.write_str(s)
723    }
724}
725
726/// Structured stats reported by an agent during a backtest session.
727#[derive(Debug, Clone, Default, Serialize, Deserialize)]
728#[serde(rename_all = "camelCase")]
729pub struct AgentStatsReport {
730    pub name: String,
731    pub slots_processed: u64,
732    pub opportunities_found: u64,
733    pub opportunities_skipped: u64,
734    pub no_routes: u64,
735    pub txs_produced: u64,
736    /// Cumulative expected profit per base mint, keyed by mint address.
737    pub expected_gain_by_mint: BTreeMap<String, i64>,
738    /// Transactions successfully executed by the sidecar.
739    #[serde(default)]
740    pub txs_submitted: u64,
741    /// Transactions that failed execution.
742    #[serde(default)]
743    pub txs_failed: u64,
744    /// Transactions rejected by preflight simulation (unprofitable).
745    #[serde(default)]
746    pub txs_simulation_rejected: u64,
747    /// Preflight simulation RPC calls that errored.
748    #[serde(default)]
749    pub txs_simulation_failed: u64,
750}
751
752/// Summary of transaction execution statistics for a completed backtest session.
753#[derive(Debug, Clone, Default, Serialize, Deserialize)]
754#[serde(rename_all = "camelCase")]
755pub struct SessionSummary {
756    /// Number of simulations where simulator outcome matched on-chain outcome
757    /// (`true_success + true_failure`).
758    pub correct_simulation: usize,
759    /// Number of simulations where simulator outcome did not match on-chain outcome
760    /// (`false_success + false_failure`).
761    pub incorrect_simulation: usize,
762    /// Number of transactions that had execution errors in simulation.
763    pub execution_errors: usize,
764    /// Number of transactions with different balance diffs.
765    pub balance_diff: usize,
766    /// Number of transactions with different log diffs.
767    pub log_diff: usize,
768}
769
770impl SessionSummary {
771    /// Returns true if there were any execution deviations (errors or mismatched results).
772    pub fn has_deviations(&self) -> bool {
773        self.incorrect_simulation > 0 || self.execution_errors > 0 || self.balance_diff > 0
774    }
775
776    /// Total number of transactions processed.
777    pub fn total_transactions(&self) -> usize {
778        self.correct_simulation
779            + self.incorrect_simulation
780            + self.execution_errors
781            + self.balance_diff
782            + self.log_diff
783    }
784}
785
786impl std::fmt::Display for SessionSummary {
787    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
788        let total = self.total_transactions();
789        write!(
790            f,
791            "Session summary: {total} transactions\n\
792             \x20  - {} correct simulation\n\
793             \x20  - {} incorrect simulation\n\
794             \x20  - {} execution errors\n\
795             \x20  - {} balance diffs\n\
796             \x20  - {} log diffs",
797            self.correct_simulation,
798            self.incorrect_simulation,
799            self.execution_errors,
800            self.balance_diff,
801            self.log_diff,
802        )
803    }
804}
805
806/// Error variants surfaced to backtest RPC clients.
807#[derive(Debug, Clone, Serialize, Deserialize)]
808#[serde(rename_all = "camelCase")]
809pub enum BacktestError {
810    InvalidTransactionEncoding {
811        index: usize,
812        error: String,
813    },
814    InvalidTransactionFormat {
815        index: usize,
816        error: String,
817    },
818    InvalidAccountEncoding {
819        address: String,
820        encoding: BinaryEncoding,
821        error: String,
822    },
823    InvalidAccountOwner {
824        address: String,
825        error: String,
826    },
827    InvalidAccountPubkey {
828        address: String,
829        error: String,
830    },
831    NoMoreBlocks,
832    AdvanceSlotFailed {
833        slot: u64,
834        error: String,
835    },
836    FinalizeSlotFailed {
837        slot: u64,
838        error: String,
839    },
840    InvalidRequest {
841        error: String,
842    },
843    Internal {
844        error: String,
845    },
846    InvalidBlockhashFormat {
847        slot: u64,
848        error: String,
849    },
850    InitializingSysvarsFailed {
851        slot: u64,
852        error: String,
853    },
854    ClerkError {
855        error: String,
856    },
857    SimulationError {
858        error: String,
859    },
860    SessionNotFound {
861        session_id: String,
862    },
863    SessionOwnerMismatch,
864    /// Session ownership is in transition (e.g. the previous manager is
865    /// shutting down, or another attach raced this one). Clients should retry
866    /// the attach within their reconnect budget; the route is expected to
867    /// become claimable shortly.
868    SessionOwnershipBusy {
869        reason: String,
870    },
871}
872
873impl BacktestError {
874    /// Errors the backtest cannot recover from mid-session — the single source
875    /// of truth for terminal-error classification, shared by the response/event
876    /// wrappers and the managed client.
877    pub fn is_terminal(&self) -> bool {
878        matches!(
879            self,
880            BacktestError::NoMoreBlocks
881                | BacktestError::AdvanceSlotFailed { .. }
882                | BacktestError::FinalizeSlotFailed { .. }
883                | BacktestError::Internal { .. }
884        )
885    }
886}
887
888/// One contiguous block range available on the history clerk.
889#[derive(Debug, Clone, Serialize, Deserialize)]
890pub struct AvailableRange {
891    pub bundle_start_slot: u64,
892    pub bundle_start_slot_utc: Option<String>,
893    pub max_bundle_end_slot: Option<u64>,
894    pub max_bundle_end_slot_utc: Option<String>,
895    pub max_bundle_size: Option<u64>,
896}
897
898/// Request body of `POST /build-bundle`.
899#[derive(Debug, Clone, Serialize, Deserialize)]
900pub struct BundleBuildRequest {
901    pub start_slot: u64,
902    pub end_slot: u64,
903    #[serde(default)]
904    pub bundle_size: Option<u64>,
905    /// Optional repeat key; a resubmit dedupes the quota charge.
906    #[serde(default)]
907    pub idempotency_key: Option<String>,
908}
909
910#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
911#[serde(rename_all = "snake_case")]
912pub enum BundleBuildStatus {
913    /// The Prefect flow run finished successfully; the bundle is built.
914    Completed,
915    /// The Prefect flow run reached a definitive failure state
916    /// (`Failed`/`Crashed`/`Cancelled`). Distinct from [`Self::NeedsInvestigation`]:
917    /// here the pipeline reported a real failure.
918    Failed,
919    /// The Prefect flow run is scheduled or running.
920    InProgress,
921    /// The flow-run state could not be determined (`Unknown` or missing) — the
922    /// outcome is genuinely unclear from Prefect, so a human should check rather
923    /// than assume success or failure.
924    NeedsInvestigation,
925}
926
927impl BundleBuildStatus {
928    /// Stable string form matching the serde `snake_case` wire representation.
929    pub fn as_str(self) -> &'static str {
930        match self {
931            Self::Completed => "completed",
932            Self::Failed => "failed",
933            Self::InProgress => "in_progress",
934            Self::NeedsInvestigation => "needs_investigation",
935        }
936    }
937}
938
939/// Response of the management bundle-build endpoints.
940#[derive(Debug, Clone, Serialize, Deserialize)]
941pub struct BundleBuildStatusResponse {
942    pub request_id: String,
943    pub start_slot: u64,
944    pub end_slot: u64,
945    pub bundle_size: Option<u64>,
946    pub status: BundleBuildStatus,
947    /// When the request id was created (unix ms) — lets clients show a live ETA
948    /// countdown by computing elapsed since this.
949    pub created_at_unix_ms: u64,
950}
951
952/// Response of the management `/snapshot-slot` endpoint. The chosen source and
953/// import state are internal, so only the resolved slot is exposed.
954#[derive(Debug, Clone, Serialize, Deserialize)]
955pub struct SnapshotSlotResponse {
956    pub snapshot_slot: u64,
957}
958
959/// Split a user-requested `[start_slot, end_slot]` range across the available
960/// bundle ranges, returning a list of contiguous, non-overlapping `(start, end)`
961/// pairs — one per bundle that the server can serve as its own session.
962///
963/// Each emitted `start` is a real `bundle_start_slot` and each `end` is within
964/// the `max_bundle_end_slot` of a bundle that begins *exactly* at that start
965/// (the server requires `start_slot == bundle_start_slot` and caps `end_slot` at
966/// that bundle's `max_bundle_end_slot`). Among all such gap-free splits we pick
967/// the one with the most sub-ranges, i.e. the highest parallelism: a fine-grained
968/// (e.g. 10k) bundle grid yields one session per fine bundle instead of
969/// collapsing onto a coarser series that overlaps the same slots — even when the
970/// coarse and fine bundles share a start slot. A coarser bundle is only ridden
971/// where no finer grid continues the walk, which still lets us serve a request
972/// the coarse data covers even across a hole in the fine grid.
973///
974/// Returns an error if the requested start slot is not a bundle start, or if no
975/// gap-free split reaches `requested_end`.
976pub fn split_range(
977    ranges: &[AvailableRange],
978    requested_start: u64,
979    requested_end: u64,
980) -> Result<Vec<(u64, u64)>, String> {
981    if requested_end < requested_start {
982        return Err(format!(
983            "invalid range: start_slot {requested_start} > end_slot {requested_end}"
984        ));
985    }
986
987    // Every advertised end per bundle start. A start can carry several ends — a
988    // 50k snapshot bundle and the 10k bundle built on the same snapshot share a
989    // start slot but reach different ends — so we keep them all and let the walk
990    // pick whichever maximises parallelism.
991    let mut ends_by_start: BTreeMap<u64, BTreeSet<u64>> = BTreeMap::new();
992    for r in ranges {
993        if let Some(end) = r.max_bundle_end_slot
994            && end > r.bundle_start_slot
995        {
996            ends_by_start
997                .entry(r.bundle_start_slot)
998                .or_default()
999                .insert(end);
1000        }
1001    }
1002
1003    // Anchor a request that lands mid-bundle to the covering bundle's start —
1004    // the latest bundle beginning at or before `requested_start` that still
1005    // covers it. The server's `select_parallel_requests` shares this fn and
1006    // anchors the same way, so the client must too or it would falsely reject a
1007    // start that isn't itself a bundle boundary. An exact bundle-start request
1008    // anchors to itself.
1009    let Some((&anchor_start, _)) = ends_by_start.range(..=requested_start).rfind(|(_, ends)| {
1010        ends.iter()
1011            .next_back()
1012            .is_some_and(|&end| end >= requested_start)
1013    }) else {
1014        return Err(format!(
1015            "start_slot {requested_start} is not covered by any available bundle range"
1016        ));
1017    };
1018
1019    // Walk the bundle starts from `requested_end` backwards, recording for each
1020    // start the gap-free split with the most sub-ranges (highest parallelism).
1021    // A sub-range ends within one of its start's bundles and the next sub-range
1022    // must begin on the very next slot, so a coarse bundle is ridden only when no
1023    // finer bundle continues the walk.
1024    let mut best_from: BTreeMap<u64, Vec<(u64, u64)>> = BTreeMap::new();
1025    for (&start, ends) in ends_by_start.range(anchor_start..=requested_end).rev() {
1026        let mut best: Option<Vec<(u64, u64)>> = None;
1027        for &end in ends {
1028            let candidate = if end >= requested_end {
1029                Some(vec![(start, requested_end)])
1030            } else {
1031                best_from.get(&(end + 1)).map(|rest| {
1032                    std::iter::once((start, end))
1033                        .chain(rest.iter().copied())
1034                        .collect()
1035                })
1036            };
1037            if let Some(candidate) = candidate
1038                && best.as_ref().is_none_or(|b| candidate.len() > b.len())
1039            {
1040                best = Some(candidate);
1041            }
1042        }
1043        if let Some(best) = best {
1044            best_from.insert(start, best);
1045        }
1046    }
1047
1048    best_from.remove(&anchor_start).ok_or_else(|| {
1049        // Point at the first uncovered slot when the bundles leave a hole; fall
1050        // back to a generic message when the range is covered but cannot be
1051        // tiled to bundle boundaries.
1052        let mut covered_to = anchor_start.saturating_sub(1);
1053        for (&start, ends) in ends_by_start.range(anchor_start..=requested_end) {
1054            if start > covered_to.saturating_add(1) {
1055                break;
1056            }
1057            if let Some(&end) = ends.iter().next_back() {
1058                covered_to = covered_to.max(end);
1059            }
1060        }
1061        if covered_to < requested_end {
1062            format!("gap in coverage at slot {}", covered_to + 1)
1063        } else {
1064            format!(
1065                "no gap-free split of [{requested_start}, {requested_end}] aligns with the available bundle ranges"
1066            )
1067        }
1068    })
1069}
1070
1071impl From<BacktestError> for BacktestResponse {
1072    fn from(error: BacktestError) -> Self {
1073        Self::Error(error)
1074    }
1075}
1076
1077impl std::error::Error for BacktestError {}
1078
1079impl fmt::Display for BacktestError {
1080    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1081        match self {
1082            BacktestError::InvalidTransactionEncoding { index, error } => {
1083                write!(f, "invalid transaction encoding at index {index}: {error}")
1084            }
1085            BacktestError::InvalidTransactionFormat { index, error } => {
1086                write!(f, "invalid transaction format at index {index}: {error}")
1087            }
1088            BacktestError::InvalidAccountEncoding {
1089                address,
1090                encoding,
1091                error,
1092            } => write!(
1093                f,
1094                "invalid encoding for account {address} ({encoding:?}): {error}"
1095            ),
1096            BacktestError::InvalidAccountOwner { address, error } => {
1097                write!(f, "invalid owner for account {address}: {error}")
1098            }
1099            BacktestError::InvalidAccountPubkey { address, error } => {
1100                write!(f, "invalid account pubkey {address}: {error}")
1101            }
1102            BacktestError::NoMoreBlocks => write!(f, "no more blocks available"),
1103            BacktestError::AdvanceSlotFailed { slot, error } => {
1104                write!(f, "failed to advance to slot {slot}: {error}")
1105            }
1106            BacktestError::FinalizeSlotFailed { slot, error } => {
1107                write!(f, "failed to finalize slot {slot}: {error}")
1108            }
1109            BacktestError::InvalidRequest { error } => write!(f, "invalid request: {error}"),
1110            BacktestError::Internal { error } => write!(f, "internal error: {error}"),
1111            BacktestError::InvalidBlockhashFormat { slot, error } => {
1112                write!(f, "invalid blockhash at slot {slot}: {error}")
1113            }
1114            BacktestError::InitializingSysvarsFailed { slot, error } => {
1115                write!(f, "failed to initialize sysvars at slot {slot}: {error}")
1116            }
1117            BacktestError::ClerkError { error } => write!(f, "clerk error: {error}"),
1118            BacktestError::SimulationError { error } => {
1119                write!(f, "simulation error: {error}")
1120            }
1121            BacktestError::SessionNotFound { session_id } => {
1122                write!(f, "session not found: {session_id}")
1123            }
1124            BacktestError::SessionOwnerMismatch => {
1125                write!(f, "session owner mismatch")
1126            }
1127            BacktestError::SessionOwnershipBusy { reason } => {
1128                write!(f, "session ownership busy: {reason}")
1129            }
1130        }
1131    }
1132}
1133
1134#[cfg(test)]
1135mod tests {
1136    use super::*;
1137
1138    #[test]
1139    fn fail_fast_divergence_kind_str_round_trips() {
1140        for kind in [
1141            FailFastDivergenceKind::AnyNonBenign,
1142            FailFastDivergenceKind::Tracked,
1143        ] {
1144            assert_eq!(
1145                FailFastDivergenceKind::from_str_opt(kind.as_str()),
1146                Some(kind)
1147            );
1148        }
1149        assert_eq!(FailFastDivergenceKind::from_str_opt("nonsense"), None);
1150        assert_eq!(
1151            FailFastDivergenceKind::default(),
1152            FailFastDivergenceKind::AnyNonBenign
1153        );
1154    }
1155
1156    #[test]
1157    fn bundle_build_request_optional_fields_default_to_none() {
1158        let req: BundleBuildRequest =
1159            serde_json::from_str(r#"{"start_slot":1,"end_slot":2}"#).expect("parse");
1160        assert_eq!((req.start_slot, req.end_slot), (1, 2));
1161        assert_eq!(req.bundle_size, None);
1162        assert_eq!(req.idempotency_key, None);
1163    }
1164
1165    #[test]
1166    fn bundle_build_request_parses_optional_fields() {
1167        let req: BundleBuildRequest = serde_json::from_str(
1168            r#"{"start_slot":1,"end_slot":2,"bundle_size":500,"idempotency_key":"abc"}"#,
1169        )
1170        .expect("parse");
1171        assert_eq!(req.bundle_size, Some(500));
1172        assert_eq!(req.idempotency_key.as_deref(), Some("abc"));
1173    }
1174
1175    #[test]
1176    fn bundle_build_status_serde_round_trips_with_snake_case() {
1177        let cases = [
1178            (BundleBuildStatus::Completed, "\"completed\""),
1179            (BundleBuildStatus::Failed, "\"failed\""),
1180            (BundleBuildStatus::InProgress, "\"in_progress\""),
1181            (
1182                BundleBuildStatus::NeedsInvestigation,
1183                "\"needs_investigation\"",
1184            ),
1185        ];
1186        for (status, expected) in cases {
1187            assert_eq!(serde_json::to_string(&status).unwrap(), expected);
1188            assert_eq!(
1189                serde_json::from_str::<BundleBuildStatus>(expected).unwrap(),
1190                status
1191            );
1192            assert_eq!(status.as_str(), expected.trim_matches('"'));
1193        }
1194        assert!(serde_json::from_str::<BundleBuildStatus>("\"queued\"").is_err());
1195    }
1196
1197    #[test]
1198    fn bundle_build_status_response_serializes_request_and_status() {
1199        let response = BundleBuildStatusResponse {
1200            request_id: "r".to_string(),
1201            start_slot: 100,
1202            end_slot: 200,
1203            bundle_size: Some(50),
1204            status: BundleBuildStatus::InProgress,
1205            created_at_unix_ms: 0,
1206        };
1207        let json = serde_json::to_value(&response).unwrap();
1208        assert_eq!(json["request_id"].as_str(), Some("r"));
1209        assert_eq!(json["start_slot"].as_u64(), Some(100));
1210        assert_eq!(json["bundle_size"].as_u64(), Some(50));
1211        assert_eq!(json["status"].as_str(), Some("in_progress"));
1212        assert_eq!(json["created_at_unix_ms"].as_u64(), Some(0));
1213    }
1214
1215    fn range(start: u64, end: u64) -> AvailableRange {
1216        AvailableRange {
1217            bundle_start_slot: start,
1218            bundle_start_slot_utc: None,
1219            max_bundle_end_slot: Some(end),
1220            max_bundle_end_slot_utc: None,
1221            max_bundle_size: None,
1222        }
1223    }
1224
1225    /// Each case lists the available bundles, the requested `[start, end]`, and
1226    /// the expected split — `Some(_)` for an accepted plan, `None` when the range
1227    /// is unservable and `split_range` must error.
1228    #[rstest::rstest]
1229    #[case::single(vec![range(100, 300)], 100, 300, Some(vec![(100, 300)]))]
1230    #[case::multi(
1231        vec![range(100, 200), range(201, 300), range(301, 400)],
1232        100, 300, Some(vec![(100, 200), (201, 300)])
1233    )]
1234    // Smaller bundles nested inside a larger one don't bridge the gap, but the
1235    // range is still coverable by riding the coarse series.
1236    #[case::nested(
1237        vec![range(100, 500), range(110, 150), range(150, 190), range(501, 900)],
1238        100, 900, Some(vec![(100, 500), (501, 900)])
1239    )]
1240    // A fine grid (1k bundles standing in for the 10k grid) overlapped by a
1241    // coarser, differently-aligned series: ride the fine grid, never landing on
1242    // the coarse bundle's drifted start.
1243    #[case::prefers_finer_grid(
1244        vec![range(1_000, 1_999), range(1_500, 3_400), range(2_000, 2_999), range(3_000, 3_999)],
1245        1_000, 3_999, Some(vec![(1_000, 1_999), (2_000, 2_999), (3_000, 3_999)])
1246    )]
1247    // A 50k snapshot bundle and the 10k bundles built on it share start 100: ride
1248    // the 10k grid instead of the coarse end, which would strand the cursor on a
1249    // slot no bundle starts at.
1250    #[case::shared_start_prefers_finer(
1251        vec![range(100, 150), range(100, 120), range(121, 140), range(141, 160)],
1252        100, 160, Some(vec![(100, 120), (121, 140), (141, 160)])
1253    )]
1254    // A coarse bundle (100, 200) overlaps a finer pair covering the same span:
1255    // split_range maximises parallelism, riding the fine pair (two sessions)
1256    // rather than merging onto the coarse max end (one session).
1257    #[case::coarse_overlap_prefers_finer_pair(
1258        vec![range(100, 200), range(100, 150), range(151, 160)],
1259        100, 160, Some(vec![(100, 150), (151, 160)])
1260    )]
1261    // The fine grid has a hole (nothing starts at 141), but a coarse bundle
1262    // sharing the start spans it: serve the request off the coarse bundle rather
1263    // than reporting an unsupported range.
1264    #[case::falls_back_to_coarse(
1265        vec![range(100, 160), range(100, 120), range(121, 140)],
1266        100, 160, Some(vec![(100, 160)])
1267    )]
1268    // The last bundle overshoots the requested end and is clamped.
1269    #[case::clamps_final_bundle(vec![range(100, 199), range(200, 999)], 100, 450, Some(vec![(100, 199), (200, 450)]))]
1270    // A request landing mid-bundle anchors to the covering bundle's start
1271    // (matching the server) instead of erroring as "not covered".
1272    #[case::anchors_mid_bundle(vec![range(150, 350)], 200, 300, Some(vec![(150, 300)]))]
1273    #[case::anchors_then_continues(
1274        vec![range(150, 350), range(351, 600)],
1275        200, 600, Some(vec![(150, 350), (351, 600)])
1276    )]
1277    #[case::start_inside_bundle_anchors(vec![range(200, 400)], 300, 400, Some(vec![(200, 400)]))]
1278    // Errors: a start before any bundle's coverage has no covering bundle...
1279    #[case::start_before_first_bundle(vec![range(200, 400)], 100, 400, None)]
1280    // ...the end must be reachable...
1281    #[case::end_not_covered(vec![range(100, 200)], 100, 300, None)]
1282    #[case::gap_in_coverage(vec![range(100, 200), range(210, 300)], 100, 300, None)]
1283    // ...and the range must not be inverted.
1284    #[case::inverted_range(vec![range(100, 300)], 300, 100, None)]
1285    // Per-API-key isolation: a user's finer bundles and a coarse GLOBAL bundle
1286    // sharing a start are fed as SEPARATE entries (never merged on snapshot_slot,
1287    // which would keep only the max end). split_range set-unions the ends and
1288    // rides the finer user grid — a regression guard against re-merging.
1289    #[case::user_and_global_ranges_not_collapsed(
1290        vec![range(100, 200), range(100, 150), range(151, 200)],
1291        100, 200, Some(vec![(100, 150), (151, 200)])
1292    )]
1293    fn split_range_cases(
1294        #[case] ranges: Vec<AvailableRange>,
1295        #[case] start: u64,
1296        #[case] end: u64,
1297        #[case] expected: Option<Vec<(u64, u64)>>,
1298    ) {
1299        match expected {
1300            Some(expected) => assert_eq!(split_range(&ranges, start, end).unwrap(), expected),
1301            None => assert!(split_range(&ranges, start, end).is_err()),
1302        }
1303    }
1304
1305    /// Servable end per bundle start, as `split_range` sees it — every advertised
1306    /// end with `end > start`, keyed by start.
1307    fn ends_by_start(ranges: &[AvailableRange]) -> BTreeMap<u64, BTreeSet<u64>> {
1308        let mut ends: BTreeMap<u64, BTreeSet<u64>> = BTreeMap::new();
1309        for r in ranges {
1310            if let Some(end) = r.max_bundle_end_slot
1311                && end > r.bundle_start_slot
1312            {
1313                ends.entry(r.bundle_start_slot).or_default().insert(end);
1314            }
1315        }
1316        ends
1317    }
1318
1319    /// Independent spec for the optimum: the gap-free split of `[start, end]` with
1320    /// the most sub-ranges, found by exhaustively trying every advertised end at
1321    /// each cursor. `None` when no split aligns with the bundle starts.
1322    fn reference_max_split(
1323        ends: &BTreeMap<u64, BTreeSet<u64>>,
1324        cursor: u64,
1325        end: u64,
1326    ) -> Option<Vec<(u64, u64)>> {
1327        ends.get(&cursor)?
1328            .iter()
1329            .filter_map(|&bundle_end| {
1330                if bundle_end >= end {
1331                    Some(vec![(cursor, end)])
1332                } else {
1333                    reference_max_split(ends, bundle_end + 1, end).map(|mut rest| {
1334                        rest.insert(0, (cursor, bundle_end));
1335                        rest
1336                    })
1337                }
1338            })
1339            .max_by_key(Vec::len)
1340    }
1341
1342    /// Structure-independent check that a split is one the server can serve: it
1343    /// tiles `[start, end]` gap-free, every start is a real bundle start, and
1344    /// every end is within that start's largest advertised end.
1345    fn is_valid_split(
1346        split: &[(u64, u64)],
1347        ends: &BTreeMap<u64, BTreeSet<u64>>,
1348        start: u64,
1349        end: u64,
1350    ) -> bool {
1351        split.first().is_some_and(|&(s, _)| s == start)
1352            && split.last().is_some_and(|&(_, e)| e == end)
1353            && split.windows(2).all(|w| w[1].0 == w[0].1 + 1)
1354            && split.iter().all(|&(s, e)| {
1355                e >= s
1356                    && ends
1357                        .get(&s)
1358                        .and_then(|bundle_ends| bundle_ends.iter().next_back())
1359                        .is_some_and(|&max_end| e <= max_end)
1360            })
1361    }
1362
1363    /// Property test: across many randomized bundle layouts, `split_range` must
1364    /// return a *valid* split with the *maximum* number of sub-ranges whenever one
1365    /// exists, and error exactly when none does. Deterministic (fixed LCG seed) so
1366    /// a failure is reproducible.
1367    #[test]
1368    fn split_range_matches_reference() {
1369        let mut seed: u64 = 0x9E3779B97F4A7C15;
1370        let mut next = || {
1371            seed = seed
1372                .wrapping_mul(6364136223846793005)
1373                .wrapping_add(1442695040888963407);
1374            seed >> 33
1375        };
1376
1377        for _ in 0..50_000 {
1378            // A small slot universe makes shared starts, overlaps, nesting, and
1379            // gaps all common.
1380            let ranges: Vec<AvailableRange> = (0..next() % 6)
1381                .map(|_| {
1382                    let start = next() % 12;
1383                    range(start, start + next() % 6) // len 0 => filtered out
1384                })
1385                .collect();
1386            let start = next() % 12;
1387            let end = start + next() % 6; // sometimes start == end, sometimes unservable
1388
1389            let got = split_range(&ranges, start, end);
1390            let ends = ends_by_start(&ranges);
1391            // split_range anchors a mid-bundle start to the covering bundle's
1392            // start, so the reference must walk from the same anchor.
1393            let anchor = ends
1394                .range(..=start)
1395                .rfind(|(_, e)| e.iter().next_back().is_some_and(|&x| x >= start))
1396                .map(|(&s, _)| s);
1397            let reference = anchor.and_then(|a| reference_max_split(&ends, a, end));
1398
1399            let layout: Vec<_> = ranges
1400                .iter()
1401                .map(|r| (r.bundle_start_slot, r.max_bundle_end_slot))
1402                .collect();
1403            match (&got, &reference) {
1404                (Ok(split), Some(best)) => {
1405                    assert!(
1406                        is_valid_split(split, &ends, anchor.unwrap(), end),
1407                        "invalid split {split:?} for {layout:?} [{start},{end}]"
1408                    );
1409                    assert_eq!(
1410                        split.len(),
1411                        best.len(),
1412                        "suboptimal split {split:?} vs {best:?} for {layout:?} [{start},{end}]"
1413                    );
1414                }
1415                (Err(_), None) => {}
1416                _ => panic!(
1417                    "disagreement: split_range={got:?}, reference={reference:?} for {layout:?} [{start},{end}]"
1418                ),
1419            }
1420        }
1421    }
1422}