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