hypercall/rsm/

unified_engine.rs

use super::bounded_idempotence_cache::BoundedIdempotenceCache;
use super::engine_snapshot::EngineSnapshot;
use crate::engine_enums_ext::EventTypeExt;
use crate::portfolio::PortfolioService;
use crate::price_oracle::hyperliquid_oracle::HyperliquidMarkPriceOracle;
use crate::read_cache::greeks::GreeksCache;
use crate::read_cache::portfolio::PortfolioCache;
use crate::rsm::ledger::BalanceLedger;
use crate::rsm::liquidation_manager::LiquidationManager;
use crate::shared::order_types::{get_timestamp_millis, ParsedSymbol};
use crate::shared::traits::MarkPriceOracle;
use crate::snapshot::SyncStatus;
use crate::standard_margin::StandardAccountBuilder;
use crate::types::Config;
use arc_swap::ArcSwap;
use hypercall_db::{InstrumentReader, InstrumentWriter, JournalReplayReader, OrderReader};
use hypercall_db_diesel::DatabaseHandler;
use hypercall_engine::{FeeService, MatchResult, OrderBook};
use hypercall_journal::checkpoint::WalCheckpointMetadata;
use hypercall_types::Side;
use hypercall_types::{to_human_readable_decimal, WalletAddress};
use hypercall_types::{
    EngineMessage, Fill, MarketAction, MarketActionMessage, MarketUpdateMessage,
    MarketUpdateStatus, OptionType as MessageOptionType, OrderAction, OrderActionMessage,
    OrderInfo, OrderUpdateMessage, OrderUpdateStatus, TimeInForce,
};
use metrics::{counter, gauge, histogram};
use rust_decimal::prelude::ToPrimitive;
use rust_decimal::Decimal;
use rust_decimal_macros::dec;
use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::atomic::{AtomicI64, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::select;
use tokio::sync::{broadcast, mpsc};
use tracing::{debug, error, info, info_span, warn, Instrument};

use hypercall_recovery::RestartStateComponent;
pub use hypercall_runtime_api::{
    decrement_pending_requests, get_pending_requests, increment_pending_requests, AgentAuthRequest,
    CashWithdrawalApplyReceipt, CashWithdrawalRequest, DepositRequest, EngineQuiesceAction,
    EngineQuiesceReport, EngineQuiesceRequest, HypercoreEquityRequest, LiquidationBonusRequest,
    MarginModeUpdateRequest, MarketRequest, NonceCheckRequest, OptionDepositRequest,
    OptionWithdrawalApplyReceipt, OptionWithdrawalRequest, TierUpdateRequest, UnifiedEngineRequest,
};

#[cfg(feature = "otel-tracing")]
use tracing_opentelemetry::OpenTelemetrySpanExt;

mod apply_interface;
mod builder;
mod fill_metadata;
mod journaling;
mod markets;
mod matching;
mod order_routing;
pub(crate) mod recovery;
mod rfq_handler;
mod runtime;
#[cfg(test)]
mod tests;
mod traits;

pub use apply_interface::EngineError;
pub use builder::{UnifiedEngineBuilder, UnifiedEngineBuilderResult};
pub(crate) use traits::{DurableJournaling, OrderAdmission, OrderExecution, ReplayRecovery};

const MAX_EXPIRY_NATS_PAYLOAD_BYTES: usize = 1024 * 1024;
const ENGINE_REQUIRE_SNAPSHOT_RESTORE_ENV: &str = "ENGINE_REQUIRE_SNAPSHOT_RESTORE";
const UNSAFE_SNAPSHOT_REPLAY_OVERRIDE_ENV: &str = "ENGINE_SNAPSHOT_RESTORE_ALLOW_UNSAFE_REPLAY";
const SNAPSHOT_CASH_VALIDATION_OVERRIDE_ENV: &str = "ENGINE_SNAPSHOT_RESTORE_ALLOW_CASH_DIVERGENCE";
#[cfg(test)]
const DURABLE_CASH_EVIDENCE_TOLERANCE: Decimal = dec!(0.00000001);

/// Default buffer size for order channel
const DEFAULT_ORDER_BUFFER_SIZE: usize = 1000;

/// Buffer size for market request channel
const MARKET_REQUEST_BUFFER_SIZE: usize = 100;

/// TWAP settlement window in seconds (30 minutes)
const TWAP_WINDOW_SECONDS: u32 = 1800;

/// Hours to look back when loading idempotency cache
const IDEMPOTENCY_CACHE_LOOKBACK_HOURS: i64 = 24;

#[cfg(test)]
#[derive(Clone)]
pub(super) struct DurableCashEvidence {
    pub(super) wallet: WalletAddress,
    pub(super) ledger_balance: Decimal,
    pub(super) margin_mode: Option<String>,
}

#[cfg(test)]
pub(super) fn validate_snapshot_cash_evidence(
    snapshot: &crate::rsm::engine_state_snapshot::EngineStateSnapshot,
    rows: impl IntoIterator<Item = DurableCashEvidence>,
) -> Result<(), String> {
    let mut durable_by_wallet: HashMap<WalletAddress, (Decimal, Option<String>)> = HashMap::new();
    for row in rows {
        durable_by_wallet.insert(row.wallet, (row.ledger_balance, row.margin_mode));
    }

    let mut failures = Vec::new();
    let snapshot_modes_are_authoritative = !snapshot.wallet_margin_modes.is_empty();
    for (wallet, (ledger_balance, margin_mode)) in &durable_by_wallet {
        let db_portfolio = margin_mode.as_deref() == Some("portfolio");
        let snapshot_portfolio = matches!(
            snapshot.wallet_margin_modes.get(wallet),
            Some(crate::rsm::margin_mode::MarginMode::Portfolio)
        );
        if snapshot_portfolio || (!snapshot_modes_are_authoritative && db_portfolio) {
            continue;
        }

        let snapshot_balance = snapshot
            .balance_ledger
            .get(wallet)
            .copied()
            .unwrap_or(Decimal::ZERO);

        let snapshot_durable_diff = snapshot_balance - *ledger_balance;
        if snapshot_durable_diff > DURABLE_CASH_EVIDENCE_TOLERANCE
            || snapshot_durable_diff < -DURABLE_CASH_EVIDENCE_TOLERANCE
        {
            failures.push(format!(
                "{} snapshot_cash={} durable_cash={}",
                wallet, snapshot_balance, ledger_balance
            ));
        }
    }

    for (wallet, snapshot_balance) in &snapshot.balance_ledger {
        if *snapshot_balance == Decimal::ZERO || durable_by_wallet.contains_key(wallet) {
            continue;
        }
        if matches!(
            snapshot.wallet_margin_modes.get(wallet),
            Some(crate::rsm::margin_mode::MarginMode::Portfolio)
        ) {
            continue;
        }
        failures.push(format!(
            "{} snapshot_cash={} missing durable cash evidence",
            wallet, snapshot_balance
        ));
    }

    if failures.is_empty() {
        Ok(())
    } else {
        let sample = failures.into_iter().take(10).collect::<Vec<_>>().join("; ");
        Err(format!(
            "engine snapshot cash validation failed for standard-margin wallets: {sample}"
        ))
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(super) enum CashRestartAuthority {
    EmptyState,
    SnapshotAtCheckpoint {
        command_id: i64,
    },
    SnapshotPlusReplay {
        snapshot_command_id: i64,
        replay_target_command_id: i64,
    },
    HotJournalBaseReplay {
        replay_target_command_id: i64,
    },
}

impl CashRestartAuthority {
    #[cfg(test)]
    pub(super) fn durable_cash_validation_command_id(self) -> Option<i64> {
        match self {
            Self::SnapshotAtCheckpoint { command_id } => Some(command_id),
            Self::EmptyState
            | Self::SnapshotPlusReplay { .. }
            | Self::HotJournalBaseReplay { .. } => None,
        }
    }
}

pub(super) fn cash_restart_authority_for_startup(
    snapshot_command_id: Option<i64>,
    checkpoint_command_id: i64,
) -> Result<CashRestartAuthority, String> {
    match (snapshot_command_id, checkpoint_command_id) {
        (Some(snapshot_command_id), checkpoint_command_id)
            if snapshot_command_id > checkpoint_command_id =>
        {
            Err(format!(
                "engine snapshot last_command_id {snapshot_command_id} is ahead of WAL checkpoint last_command_id {checkpoint_command_id}"
            ))
        }
        (Some(command_id), checkpoint_command_id) if command_id == checkpoint_command_id => {
            Ok(CashRestartAuthority::SnapshotAtCheckpoint { command_id })
        }
        (Some(snapshot_command_id), replay_target_command_id) => {
            Ok(CashRestartAuthority::SnapshotPlusReplay {
                snapshot_command_id,
                replay_target_command_id,
            })
        }
        (None, checkpoint_command_id) if checkpoint_command_id > 0 => {
            Ok(CashRestartAuthority::HotJournalBaseReplay {
                replay_target_command_id: checkpoint_command_id,
            })
        }
        (None, _) => Ok(CashRestartAuthority::EmptyState),
    }
}

#[cfg(test)]
pub(super) fn snapshot_cash_validation_boundary_matches(
    snapshot_command_id: i64,
    checkpoint_command_id: i64,
) -> bool {
    matches!(
        cash_restart_authority_for_startup(Some(snapshot_command_id), checkpoint_command_id),
        Ok(CashRestartAuthority::SnapshotAtCheckpoint { .. })
    )
}

#[cfg(test)]
pub(super) fn should_validate_snapshot_cash_against_durable_accounting(
    snapshot_command_id: i64,
    checkpoint_command_id: i64,
    replay_bounds: Option<hypercall_db::JournalCommandIdBounds>,
) -> bool {
    if snapshot_command_id != checkpoint_command_id {
        return false;
    }

    match replay_bounds {
        Some(bounds) => bounds.max_command_id <= checkpoint_command_id,
        None => true,
    }
}

pub(super) fn validate_hot_journal_covers_base_replay(
    checkpoint_command_id: i64,
    bounds: Option<hypercall_db::JournalCommandIdBounds>,
    replay_from_command_id: i64,
    non_replayable_tail_count: Option<i64>,
) -> Result<(), String> {
    if replay_from_command_id < 1 {
        return Err(format!(
            "invalid replay window start command_id {replay_from_command_id}; command ids must be >=1"
        ));
    }

    if checkpoint_command_id <= 0 {
        if let Some(bounds) = bounds {
            if bounds.min_command_id != replay_from_command_id {
                return Err(format!(
                    "hot journal starts at command_id {}, but zero-checkpoint startup with existing engine_commands rows requires replay from command_id {}",
                    bounds.min_command_id, replay_from_command_id
                ));
            }
        }
        return Ok(());
    }

    let Some(bounds) = bounds else {
        if replay_from_command_id > checkpoint_command_id {
            // Start replay after checkpoint implies this is an equal-snapshot, post-checkpoint
            // resume path where a missing command-id window simply means no replayable
            // deltas remain.
            return Ok(());
        }
        validate_non_replayable_tail(
            replay_from_command_id.saturating_sub(1),
            checkpoint_command_id,
            non_replayable_tail_count,
        )?;
        return Ok(());
    };

    if bounds.max_command_id < replay_from_command_id {
        validate_non_replayable_tail(
            replay_from_command_id.saturating_sub(1),
            checkpoint_command_id,
            non_replayable_tail_count,
        )?;
        return Ok(());
    }

    if bounds.min_command_id > replay_from_command_id {
        return Err(format!(
            "hot journal starts at command_id {}, but base reconstruction for checkpoint command_id {} requires replay from command_id {} or earlier",
            bounds.min_command_id, checkpoint_command_id, replay_from_command_id
        ));
    }

    validate_non_replayable_tail(
        bounds.max_command_id,
        checkpoint_command_id,
        non_replayable_tail_count,
    )?;

    Ok(())
}

fn validate_non_replayable_tail(
    replayable_tail_command_id: i64,
    checkpoint_command_id: i64,
    non_replayable_tail_count: Option<i64>,
) -> Result<(), String> {
    if replayable_tail_command_id >= checkpoint_command_id {
        return Ok(());
    }

    let expected_tail_count = checkpoint_command_id - replayable_tail_command_id;
    let Some(non_replayable_tail_count) = non_replayable_tail_count else {
        return Err(format!(
            "hot journal replayable commands end at command_id {}, before checkpoint command_id {}, and non-replayable tail coverage was not checked",
            replayable_tail_command_id, checkpoint_command_id
        ));
    };

    if non_replayable_tail_count != expected_tail_count {
        return Err(format!(
            "hot journal replayable commands end at command_id {}, before checkpoint command_id {}, and only {} of {} tail commands are proven non-replayable",
            replayable_tail_command_id,
            checkpoint_command_id,
            non_replayable_tail_count,
            expected_tail_count
        ));
    }

    Ok(())
}

fn non_replayable_tail_count_for_coverage_check(
    db: &impl hypercall_db::JournalReplayReader,
    checkpoint_command_id: i64,
    bounds: Option<hypercall_db::JournalCommandIdBounds>,
    replay_from_command_id: i64,
) -> Result<Option<i64>, String> {
    if checkpoint_command_id <= 0 {
        return Ok(None);
    }

    let replayable_tail_command_id = match bounds {
        Some(bounds) if bounds.max_command_id >= replay_from_command_id => bounds.max_command_id,
        _ => replay_from_command_id.saturating_sub(1),
    };
    if replayable_tail_command_id >= checkpoint_command_id {
        return Ok(None);
    }

    db.count_non_replayable_commands_in_range_sync(
        replayable_tail_command_id,
        checkpoint_command_id,
    )
    .map(Some)
    .map_err(|error| {
        format!(
            "failed to count non-replayable hot journal tail rows from command_id {} through {}: {error}",
            replayable_tail_command_id + 1,
            checkpoint_command_id
        )
    })
}

pub(super) fn should_validate_hot_journal_coverage_for_base_replay(
    db_present: bool,
    startup_checkpoint_command_id: i64,
    wal_path_configured: bool,
    snapshot_loaded: bool,
    replay_checkpoint_command_id: i64,
) -> bool {
    if !db_present || !wal_path_configured {
        return false;
    }

    !snapshot_loaded || startup_checkpoint_command_id >= replay_checkpoint_command_id
}

pub(super) fn post_replay_cash_validation_checkpoint(
    db_present: bool,
    wal_path_configured: bool,
    startup_checkpoint: WalCheckpointMetadata,
    journal_bounds: Option<hypercall_db::JournalCommandIdBounds>,
) -> Option<WalCheckpointMetadata> {
    if !db_present || !wal_path_configured {
        return None;
    }

    if startup_checkpoint.last_command_id > 0 {
        return Some(startup_checkpoint);
    }

    journal_bounds
        .filter(|bounds| bounds.min_command_id == 1)
        .map(|bounds| WalCheckpointMetadata {
            last_command_id: bounds.max_command_id,
            ..startup_checkpoint
        })
}

fn env_flag_enabled(name: &str) -> bool {
    std::env::var(name).ok().is_some_and(|value| {
        matches!(
            value.trim(),
            "1" | "true" | "TRUE" | "yes" | "YES" | "on" | "ON"
        )
    })
}

/// The unified engine that combines order matching and margin calculations.
///
/// Margin calculations for order acceptance are delegated to MarginService
/// (SpanMarginService implementation). The engine is responsible for:
/// - Building hypothetical accounts that include open orders + proposed orders
/// - Calling MarginService to compute margin requirements
/// - Accepting/rejecting orders based on available collateral
pub struct UnifiedEngine {
    // Shared mutable state (passed to managers as &mut EngineCtx)
    pub(crate) ctx: crate::rsm::engine_deps::EngineCtx,

    // ===== Sub-managers (borrowed immutably during dispatch) =====
    /// Margin checking logic (SPAN + Standard).
    margin_manager: crate::rsm::margin_manager::MarginManager,

    /// Expiry and settlement management.
    expiry_manager: crate::rsm::expiry_manager::ExpiryManager,

    // ===== Engine-internal (NOT passed to managers) =====
    // Channels
    order_receiver: mpsc::Receiver<UnifiedEngineRequest>,
    market_receiver: mpsc::Receiver<MarketRequest>,
    pub(crate) rfq_receiver: Option<mpsc::Receiver<hypercall_runtime_api::RfqExecuteRequest>>,
    pub(crate) deposit_receiver: Option<mpsc::Receiver<DepositRequest>>,
    pub(crate) option_deposit_receiver: Option<mpsc::Receiver<OptionDepositRequest>>,
    pub(crate) option_withdrawal_receiver: Option<mpsc::Receiver<OptionWithdrawalRequest>>,
    pub(crate) cash_withdrawal_receiver: Option<mpsc::Receiver<CashWithdrawalRequest>>,
    pub(crate) liquidation_bonus_receiver: Option<mpsc::Receiver<LiquidationBonusRequest>>,
    pub(crate) margin_mode_receiver: Option<mpsc::Receiver<MarginModeUpdateRequest>>,
    pub(crate) agent_auth_receiver: Option<mpsc::Receiver<AgentAuthRequest>>,
    pub(crate) nonce_check_receiver: Option<mpsc::Receiver<NonceCheckRequest>>,
    pub(crate) tier_update_receiver: Option<mpsc::Receiver<TierUpdateRequest>>,
    pub(crate) hypercore_equity_receiver: Option<mpsc::Receiver<HypercoreEquityRequest>>,
    pub(crate) pm_settlement_admin_receiver:
        Option<mpsc::Receiver<hypercall_runtime_api::PmSettlementAdminRequest>>,
    pub(crate) quiesce_receiver: Option<mpsc::Receiver<EngineQuiesceRequest>>,
    /// Live notify channel from `catalog_manager::manager` for
    /// per-underlying `TradingModes` updates. When the catalog
    /// manager rewrites any row in the `instruments.trading_mode`
    /// column, it publishes the full current `underlying → mode`
    /// map here. The main loop `changed()`s on it and calls
    /// `apply_underlying_trading_mode_update`.
    pub(crate) trading_mode_receiver: Option<
        tokio::sync::watch::Receiver<
            std::collections::HashMap<String, hypercall_types::TradingModes>,
        >,
    >,
    shutdown_receiver: broadcast::Receiver<()>,

    // Engine-level fee service (used in matching loop)
    fee_service: FeeService,

    // Snapshot timing
    snapshot_interval: Duration,
    read_snapshot_interval: Duration,
    post_startup_reconcile_delay: Duration,
    last_snapshot: Instant,
    wal_path: Option<PathBuf>,

    /// Sync status for readiness tracking.
    sync_status: Arc<SyncStatus>,

    /// Durable engine journal for restart-safe ACK and idempotency.
    journal_writer: Option<crate::journal::SharedEngineJournalWriter>,

    /// Batched journal sender for async PostgreSQL writes.
    journal_batch_sender: Option<crate::journal::JournalBatchSender>,

    /// NATS publisher for real-time command replication to standby instances.
    nats_publisher: Option<crate::nats::NatsPublisher>,
    /// NATS publisher for canonical balance update follower replication.
    nats_balance_update_publisher: Option<crate::nats::NatsBalanceUpdatePublisher>,

    /// Known request_ids for fast idempotency checks (bounded to prevent memory leaks).
    known_request_ids: BoundedIdempotenceCache,

    /// ArcSwap snapshot for lock-free reads by API handlers.
    read_snapshot: Arc<ArcSwap<EngineSnapshot>>,

    /// Whether post-startup reconciliation has been performed.
    /// After SIGKILL, the journal batcher needs a few seconds to materialize
    /// pending terminal order statuses into order_infos.
    /// This flag ensures we re-check for ghost orders once after catchup.
    post_startup_reconciled: bool,

    /// Runtime events produced during synchronous startup journal replay.
    /// These must be projected before `start()` hydrates engine positions from
    /// PortfolioService, otherwise recovered TickExpiry events can leave stale
    /// portfolio positions that resurrect settled engine positions.
    startup_replayed_events: Vec<(String, Vec<EngineMessage>)>,

    /// WAL checkpoint metadata loaded at startup and used to bound replay windows.
    replay_checkpoint: WalCheckpointMetadata,

    /// When true, the engine has crossed an admin drain barrier. Live mutation
    /// sources remain parked until an explicit resume request.
    runtime_quiesced: bool,

    /// Whether an engine state snapshot was successfully loaded during startup.
    /// When true, the base reconstruction window is skipped during replay.
    snapshot_loaded: bool,

    /// Orders on expired instruments discovered during replay that need DB cancellation.
    startup_expired_order_cancels: Vec<recovery::StartupExpiredOrderCancel>,

    /// External vol oracle reference for ingesting IV surfaces into the command stream.
    /// Stored separately from the margin_manager's oracle so the runtime can read
    /// surfaces and feed them as IvUpdate commands.
    external_vol_oracle: Option<crate::vol_oracle::SharedVolOracle>,

    /// Shared IV surfaces backing the EngineVolOracle. Updated by IvUpdate commands,
    /// read by the SpanMarginService during margin checks.
    engine_iv_surfaces: Arc<
        std::sync::RwLock<HashMap<String, crate::vol_oracle::vol_surface_cache::VolatilitySurface>>,
    >,
}

// ===== Helper types for process_order refactoring =====

/// Result of order allocation
pub(crate) struct AllocatedOrder {
    order_id: u64,
}

/// Result of matching execution
pub(crate) struct MatchingResult {
    fills: Vec<hypercall_types::Fill>,
    filled_size: Decimal,
    mmp_triggered: bool,
    self_trade_maker_id: Option<u64>,
}

#[derive(Debug, Clone)]
pub struct EngineRuntimeSettings {
    pub snapshot_interval: Duration,
    pub read_snapshot_interval: Duration,
    pub post_startup_reconcile_delay: Duration,
    pub wal_path: Option<PathBuf>,
    pub start_quiesced: bool,
    pub recovery_safety_report: Option<hypercall_api::recovery_safety::SharedRecoverySafetyReport>,
    pub portfolio_margin_pool_enabled: bool,
    pub portfolio_margin_settlement_allowlist: std::collections::BTreeSet<WalletAddress>,
}

impl Default for EngineRuntimeSettings {
    fn default() -> Self {
        Self {
            snapshot_interval: Duration::from_secs(60),
            read_snapshot_interval: Duration::from_millis(500),
            post_startup_reconcile_delay: Duration::from_secs(5),
            wal_path: None,
            start_quiesced: false,
            recovery_safety_report: None,
            portfolio_margin_pool_enabled: false,
            portfolio_margin_settlement_allowlist: std::collections::BTreeSet::new(),
        }
    }
}

impl UnifiedEngine {
    /// Initialize the unified engine with all necessary components
    pub fn init(
        config: Config,
        order_receiver: mpsc::Receiver<UnifiedEngineRequest>,
        market_receiver: mpsc::Receiver<MarketRequest>,
        event_sender: mpsc::UnboundedSender<EngineMessage>,
        read_snapshot: Option<Arc<ArcSwap<EngineSnapshot>>>,
        shutdown_receiver: broadcast::Receiver<()>,
        runtime_settings: EngineRuntimeSettings,
        database_path: Option<String>,
        db_auth: Option<hypercall_db_diesel::DbAuthConfig>,
        allow_no_database: bool,
        skip_db_migrations: bool,
    ) -> Self {
        // Try to create diesel handler for persistence if path provided
        let diesel_handler = if let Some(db_path) = database_path {
            info!(
                "UnifiedEngine attempting to connect to database: {}",
                db_path
            );
            let auth =
                db_auth.unwrap_or_else(|| hypercall_db_diesel::DbAuthConfig::password(&db_path));
            let result = if skip_db_migrations {
                DatabaseHandler::new_readonly_auth(auth, 3)
            } else {
                DatabaseHandler::new_with_auth(auth, 3)
            };
            match result {
                Ok(handler) => {
                    info!(
                        "UnifiedEngine successfully connected to persistence database: {} (migrations_skipped={})",
                        db_path, skip_db_migrations
                    );
                    Some(handler)
                }
                Err(e) => {
                    // Allow running without database if explicitly allowed (for tests)
                    if allow_no_database {
                        warn!("UnifiedEngine failed to connect to persistence database: {}, continuing without persistence (allowed for tests)", e);
                        None
                    } else {
                        error!(
                            "FATAL: UnifiedEngine failed to connect to persistence database {}: {}",
                            db_path, e
                        );
                        panic!(
                            "Cannot start UnifiedEngine without database connection: {}",
                            e
                        );
                    }
                }
            }
        } else {
            // Allow running without database if explicitly allowed (for tests)
            if allow_no_database {
                info!("UnifiedEngine: No database path provided, running without persistence (allowed for tests)");
                None
            } else {
                error!("FATAL: No database path provided for UnifiedEngine");
                panic!("Cannot start UnifiedEngine without database configuration");
            }
        };

        // Get the latest L2 sequence number from engine_events table
        let l2_update_seq = if let Some(ref handler) = diesel_handler {
            let replay: &dyn hypercall_db::JournalReplayReader = handler;
            match replay.get_max_l2_seq_from_events_sync() {
                Ok(seq) => {
                    info!("Loaded L2 update sequence from engine_events: {}", seq);
                    Arc::new(AtomicI64::new(seq))
                }
                Err(e) => {
                    if allow_no_database {
                        warn!(
                            "Failed to load L2 update sequence (allowed for tests): {}",
                            e
                        );
                        Arc::new(AtomicI64::new(0))
                    } else {
                        panic!(
                            "CRITICAL_FAILURE: Failed to load L2 update sequence from database: {}. \
                             Starting from 0 would risk ID collisions. Restart required.",
                            e
                        );
                    }
                }
            }
        } else {
            Arc::new(AtomicI64::new(0))
        };

        // Get the next order_id from the database (max existing order_id + 1)
        let next_order_id = if let Some(ref handler) = diesel_handler {
            match handler.get_max_order_id_sync() {
                Ok(max_order_id) => {
                    let next_id = max_order_id + 1;
                    info!(
                        "Loaded max order_id from database: {}, starting from {}",
                        max_order_id, next_id
                    );
                    next_id
                }
                Err(e) => {
                    if allow_no_database {
                        warn!("Failed to load max order_id (allowed for tests): {}", e);
                        1
                    } else {
                        panic!(
                            "CRITICAL_FAILURE: Failed to load max order_id from database: {}. \
                             Starting from 1 would risk duplicate order IDs. Restart required.",
                            e
                        );
                    }
                }
            }
        } else {
            info!("No database handler, starting order_id from 1");
            1
        };

        // Get the next trade_id from the database (max existing trade_id + 1)
        let next_trade_id = if let Some(ref handler) = diesel_handler {
            match handler.get_max_trade_id_sync() {
                Ok(max_trade_id) => {
                    let next_id = max_trade_id + 1;
                    info!(
                        "Loaded max trade_id from database: {}, starting from {}",
                        max_trade_id, next_id
                    );
                    next_id
                }
                Err(e) => {
                    if allow_no_database {
                        warn!("Failed to load max trade_id (allowed for tests): {}", e);
                        1
                    } else {
                        panic!(
                            "CRITICAL_FAILURE: Failed to load max trade_id from database: {}. \
                             Starting from 1 would risk duplicate trade IDs. Restart required.",
                            e
                        );
                    }
                }
            }
        } else {
            info!("No database handler, starting trade_id from 1");
            1
        };

        let fee_service = FeeService::new(config.fee_config.clone());

        // Initialize sub-managers
        let margin_manager = crate::rsm::margin_manager::MarginManager::new(&config);
        let expiry_manager = crate::rsm::expiry_manager::ExpiryManager::new();
        let ctx = crate::rsm::engine_deps::EngineCtx {
            orderbooks: HashMap::new(),
            next_order_id,
            next_trade_id,
            l2_update_seq,
            expired_instruments: HashMap::new(),
            spot_prices: HashMap::new(),
            iv_surfaces: HashMap::new(),
            iv_source_timestamps: HashMap::new(),
            instrument_trading_modes: HashMap::new(),
            order_index: hypercall_engine::order_index::EngineOrderIndex::new(),
            engine_positions: HashMap::new(),
            balance_ledger: BalanceLedger::new(),
            deposit_update_watermarks: HashMap::new(),
            applied_deposit_source_event_hashes: std::collections::BTreeSet::new(),
            pm_settlement_state:
                crate::rsm::portfolio_margin::settlement_state::PmSettlementState::default(),
            mmp_state: HashMap::new(),
            agent_authorizations: HashMap::new(),
            nonce_sets: HashMap::new(),
            rsm_signer_nonces: HashMap::new(),
            deps: crate::rsm::engine_deps::EngineDeps {
                mmp_cache: None,
                tier_cache: None,
                portfolio_service: None,
                portfolio_cache: None,
                risk_account_builder: None,
                standard_account_builder: None,
                greeks_cache: None,
                mark_price_oracles: HashMap::new(),
                reference_prices: HashMap::new(),
                liquidation_cache: None,
                config,
                portfolio_margin_pool_enabled: runtime_settings.portfolio_margin_pool_enabled,
                portfolio_margin_settlement_allowlist: runtime_settings
                    .portfolio_margin_settlement_allowlist
                    .clone(),
                event_sender,
                ws_event_sender: None,
                margin_timestamp_s: chrono::Utc::now().timestamp(),
                liquidation_states: HashMap::new(),
                wallet_margin_modes: HashMap::new(),
                mmp_enabled: HashMap::new(),
                wallet_trading_limits: HashMap::new(),
                default_trading_limits: hypercall_types::api_models::TradingLimits::default(),
                wallet_tiers: HashMap::new(),
                perp_positions: HashMap::new(),
                hypercore_account_equity: HashMap::new(),
                hypercore_equity_timestamps: HashMap::new(),
            },
            db: diesel_handler,
        };

        let mut engine = Self {
            ctx,
            margin_manager,
            expiry_manager,
            order_receiver,
            market_receiver,
            rfq_receiver: None,
            deposit_receiver: None,
            option_deposit_receiver: None,
            option_withdrawal_receiver: None,
            cash_withdrawal_receiver: None,
            liquidation_bonus_receiver: None,
            margin_mode_receiver: None,
            agent_auth_receiver: None,
            nonce_check_receiver: None,
            tier_update_receiver: None,
            hypercore_equity_receiver: None,
            pm_settlement_admin_receiver: None,
            quiesce_receiver: None,
            trading_mode_receiver: None,
            shutdown_receiver,
            fee_service,
            snapshot_interval: runtime_settings.snapshot_interval,
            read_snapshot_interval: runtime_settings.read_snapshot_interval,
            post_startup_reconcile_delay: runtime_settings.post_startup_reconcile_delay,
            last_snapshot: Instant::now(),
            wal_path: runtime_settings.wal_path,
            sync_status: Arc::new(SyncStatus::new()),
            journal_writer: None,
            journal_batch_sender: None,
            nats_publisher: None,
            nats_balance_update_publisher: None,
            known_request_ids: BoundedIdempotenceCache::new(),
            read_snapshot: read_snapshot
                .unwrap_or_else(|| Arc::new(ArcSwap::from_pointee(EngineSnapshot::empty()))),
            post_startup_reconciled: false,
            startup_replayed_events: Vec::new(),
            replay_checkpoint: WalCheckpointMetadata::ZERO,
            runtime_quiesced: runtime_settings.start_quiesced,
            snapshot_loaded: false,
            startup_expired_order_cancels: Vec::new(),
            external_vol_oracle: None,
            engine_iv_surfaces: Arc::new(std::sync::RwLock::new(HashMap::new())),
        };

        // Load existing markets from database
        ReplayRecovery::load_markets_from_db(&mut engine);

        // Load WAL checkpoint metadata and replay bounds before journal recovery.
        ReplayRecovery::load_orderbook_snapshots(&mut engine);
        engine.sync_status.set_catching_up();

        // Try to load a persistent engine state snapshot to skip full replay.
        // The snapshot's last_command_id must be <= the checkpoint's last_command_id
        // (snapshot is tagged from the Postgres-replicated boundary).
        // Only attempt when ENGINE_JOURNAL_WAL_PATH is explicitly set — the default
        // path (/var/tmp/...) is shared across test processes with different databases,
        // so snapshot files from one test would corrupt another's state.
        let wal_path_configured = hypercall_journal::checkpoint::wal_path_is_explicitly_configured(
            engine.wal_path.as_ref(),
        );
        let wal_path =
            hypercall_journal::checkpoint::wal_path_from_config(engine.wal_path.as_ref());
        let env_wal_path = std::env::var("ENGINE_JOURNAL_WAL_PATH").ok();
        let require_snapshot_restore = env_flag_enabled(ENGINE_REQUIRE_SNAPSHOT_RESTORE_ENV);
        let unsafe_replay_override = env_flag_enabled(UNSAFE_SNAPSHOT_REPLAY_OVERRIDE_ENV);
        let cash_divergence_override = env_flag_enabled(SNAPSHOT_CASH_VALIDATION_OVERRIDE_ENV);
        let startup_replay_checkpoint = engine.replay_checkpoint;
        let snapshot_path =
            crate::rsm::engine_state_snapshot::snapshot_path_from_wal_path(&wal_path);
        let drain_marker_path = hypercall_api::handlers::health::drain_marker_path(&wal_path);
        let mut recovery_safety_report = hypercall_api::recovery_safety::RecoverySafetyReport::new(
            wal_path.display().to_string(),
            wal_path_configured,
            require_snapshot_restore,
            unsafe_replay_override,
            cash_divergence_override,
            hypercall_api::recovery_safety::RecoverySafetyCheckpoint {
                last_command_id: engine.replay_checkpoint.last_command_id,
                last_l2_seq: engine.replay_checkpoint.last_l2_seq,
                wal_offset: engine.replay_checkpoint.wal_offset as u64,
            },
            hypercall_api::recovery_safety::RecoverySafetyDrainMarker {
                path: drain_marker_path.display().to_string(),
                present_at_startup: runtime_settings.start_quiesced,
            },
        );
        if runtime_settings.start_quiesced {
            recovery_safety_report.pass(
                "drain_marker",
                format!(
                    "persistent drain marker present at startup; engine starts quiesced until operator undrains: {}",
                    drain_marker_path.display()
                ),
            );
        } else {
            recovery_safety_report.pass("drain_marker", "no persistent drain marker at startup");
        }
        info!(
            configured_wal_path = ?engine.wal_path,
            env_wal_path = ?env_wal_path,
            effective_wal_path = %wal_path.display(),
            wal_checkpoint_last_command_id = engine.replay_checkpoint.last_command_id,
            require_snapshot_restore,
            "Resolved engine recovery WAL path"
        );
        if require_snapshot_restore && engine.ctx.db.is_some() {
            if !wal_path_configured {
                recovery_safety_report.fail(
                    "wal_path_configured",
                    format!(
                        "{} is set but no explicit engine WAL path is configured",
                        ENGINE_REQUIRE_SNAPSHOT_RESTORE_ENV
                    ),
                );
                hypercall_api::recovery_safety::store_report(
                    &runtime_settings.recovery_safety_report,
                    &recovery_safety_report,
                );
                panic!(
                    "CRITICAL_FAILURE: {} is set but no explicit engine WAL path is configured",
                    ENGINE_REQUIRE_SNAPSHOT_RESTORE_ENV
                );
            }
            if engine.replay_checkpoint.last_command_id <= 0 {
                recovery_safety_report.fail(
                    "checkpoint",
                    format!(
                        "{} is set but WAL checkpoint at {} is missing or zero",
                        ENGINE_REQUIRE_SNAPSHOT_RESTORE_ENV,
                        wal_path.display()
                    ),
                );
                hypercall_api::recovery_safety::store_report(
                    &runtime_settings.recovery_safety_report,
                    &recovery_safety_report,
                );
                panic!(
                    "CRITICAL_FAILURE: {} is set but WAL checkpoint at {} is missing or zero. \
                     Blue/green standby must start from a cloned WAL/checkpoint/snapshot boundary.",
                    ENGINE_REQUIRE_SNAPSHOT_RESTORE_ENV,
                    wal_path.display(),
                );
            }
        }
        if wal_path_configured {
            recovery_safety_report.pass("wal_path_configured", "explicit WAL path configured");
        } else {
            recovery_safety_report.pass(
                "wal_path_configured",
                "explicit WAL path not configured; persistent snapshot restore not attempted",
            );
        }
        if engine.replay_checkpoint.last_command_id > 0 {
            recovery_safety_report.pass(
                "checkpoint",
                format!(
                    "checkpoint loaded at command_id {}",
                    engine.replay_checkpoint.last_command_id
                ),
            );
        } else {
            recovery_safety_report.pass("checkpoint", "no nonzero WAL checkpoint");
        }
        let mut hot_journal_bounds = None;

        if engine.ctx.db.is_some()
            && engine.replay_checkpoint.last_command_id > 0
            && wal_path_configured
        {
            use crate::rsm::engine_state_snapshot::{read_snapshot, snapshot_order_count};
            match read_snapshot(&snapshot_path) {
                Ok(Some(snap))
                    if snap.last_command_id <= engine.replay_checkpoint.last_command_id =>
                {
                    recovery_safety_report.snapshot =
                        Some(hypercall_api::recovery_safety::RecoverySafetySnapshot {
                            path: snapshot_path.display().to_string(),
                            present: true,
                            loaded: false,
                            last_command_id: Some(snap.last_command_id),
                            last_l2_seq: Some(snap.last_l2_seq),
                            order_count: Some(snapshot_order_count(&snap)),
                        });
                    recovery_safety_report.pass(
                        "snapshot_decode",
                        format!(
                            "decoded engine snapshot at command_id {}",
                            snap.last_command_id
                        ),
                    );
                    let cash_authority = cash_restart_authority_for_startup(
                        Some(snap.last_command_id),
                        engine.replay_checkpoint.last_command_id,
                    )
                    .expect("snapshot command_id checked not ahead of WAL checkpoint");
                    info!(
                        snapshot_last_command_id = snap.last_command_id,
                        checkpoint_last_command_id = engine.replay_checkpoint.last_command_id,
                        ?cash_authority,
                        snapshot_balance_update_seq = snap.last_balance_update_seq,
                        "Skipping pre-restore DB cash validation; engine snapshot plus WAL boundary is authoritative"
                    );
                    recovery_safety_report.cash.validation_checked = false;
                    recovery_safety_report.cash.validation_passed = None;
                    recovery_safety_report.cash.message = Some(format!(
                        "pre-restore DB cash validation skipped for {:?}; engine snapshot plus WAL is authoritative",
                        cash_authority
                    ));

                    let order_count = snapshot_order_count(&snap);
                    info!(
                        "Loading engine state snapshot: last_command_id={}, orders={}",
                        snap.last_command_id, order_count,
                    );
                    crate::rsm::restart_components::PersistentEngineStateComponent::restore(
                        &mut engine.ctx,
                        &snap,
                    );
                    engine.rebuild_shared_engine_iv_surfaces();
                    engine.replay_checkpoint.last_command_id = snap.last_command_id;
                    engine.replay_checkpoint.last_l2_seq = snap.last_l2_seq;
                    engine.snapshot_loaded = true;
                    if let Some(snapshot) = recovery_safety_report.snapshot.as_mut() {
                        snapshot.loaded = true;
                    }

                    gauge!("ht_engine_snapshot_restored").set(1.0);
                    gauge!("ht_engine_snapshot_last_command_id").set(snap.last_command_id as f64);
                    gauge!("ht_engine_snapshot_orders_count").set(order_count as f64);
                }
                Ok(Some(snap)) => {
                    recovery_safety_report.snapshot =
                        Some(hypercall_api::recovery_safety::RecoverySafetySnapshot {
                            path: snapshot_path.display().to_string(),
                            present: true,
                            loaded: false,
                            last_command_id: Some(snap.last_command_id),
                            last_l2_seq: Some(snap.last_l2_seq),
                            order_count: Some(snapshot_order_count(&snap)),
                        });
                    recovery_safety_report.fail(
                        "snapshot_boundary",
                        format!(
                            "snapshot command_id {} is ahead of WAL checkpoint command_id {}",
                            snap.last_command_id, engine.replay_checkpoint.last_command_id
                        ),
                    );
                    hypercall_api::recovery_safety::store_report(
                        &runtime_settings.recovery_safety_report,
                        &recovery_safety_report,
                    );
                    panic!(
                        "CRITICAL_FAILURE: engine state snapshot last_command_id {} is ahead of \
                         WAL checkpoint last_command_id {}. This is a logic bug — the snapshot \
                         should always be tagged from the Postgres-replicated boundary.",
                        snap.last_command_id, engine.replay_checkpoint.last_command_id,
                    );
                }
                Ok(None) => {
                    gauge!("ht_engine_snapshot_restored").set(0.0);
                    recovery_safety_report.snapshot =
                        Some(hypercall_api::recovery_safety::RecoverySafetySnapshot {
                            path: snapshot_path.display().to_string(),
                            present: false,
                            loaded: false,
                            last_command_id: None,
                            last_l2_seq: None,
                            order_count: None,
                        });
                    if require_snapshot_restore {
                        recovery_safety_report.fail(
                            "snapshot_decode",
                            format!(
                                "no engine state snapshot found at {}",
                                snapshot_path.display()
                            ),
                        );
                        hypercall_api::recovery_safety::store_report(
                            &runtime_settings.recovery_safety_report,
                            &recovery_safety_report,
                        );
                        panic!(
                            "CRITICAL_FAILURE: no engine state snapshot found at {} with WAL \
                             checkpoint command_id {} while {} is set. Refusing standby restore \
                             without a cloned engine snapshot.",
                            snapshot_path.display(),
                            engine.replay_checkpoint.last_command_id,
                            ENGINE_REQUIRE_SNAPSHOT_RESTORE_ENV,
                        );
                    } else {
                        recovery_safety_report.pass(
                            "snapshot_decode",
                            "no snapshot found; base replay requires hot journal coverage validation",
                        );
                        warn!(
                            snapshot_path = %snapshot_path.display(),
                            checkpoint_command_id = engine.replay_checkpoint.last_command_id,
                            "No engine state snapshot found; falling back to full durable journal replay after coverage validation"
                        );
                    }
                }
                Err(e) => {
                    gauge!("ht_engine_snapshot_restored").set(0.0);
                    recovery_safety_report.snapshot =
                        Some(hypercall_api::recovery_safety::RecoverySafetySnapshot {
                            path: snapshot_path.display().to_string(),
                            present: true,
                            loaded: false,
                            last_command_id: None,
                            last_l2_seq: None,
                            order_count: None,
                        });
                    if unsafe_replay_override {
                        recovery_safety_report.fail(
                            "snapshot_decode",
                            format!("corrupt engine state snapshot accepted by override: {e}"),
                        );
                        warn!(
                            error = %e,
                            override_env = UNSAFE_SNAPSHOT_REPLAY_OVERRIDE_ENV,
                            checkpoint_command_id = engine.replay_checkpoint.last_command_id,
                            "UNSAFE_OPERATOR_OVERRIDE: corrupt engine state snapshot, continuing to full replay"
                        );
                    } else {
                        recovery_safety_report.fail(
                            "snapshot_decode",
                            format!("corrupt engine state snapshot: {e}"),
                        );
                        hypercall_api::recovery_safety::store_report(
                            &runtime_settings.recovery_safety_report,
                            &recovery_safety_report,
                        );
                        panic!(
                            "CRITICAL_FAILURE: corrupt engine state snapshot at {} with WAL \
                             checkpoint command_id {}: {}. Refusing unsafe hot-journal replay. \
                             Set {}=true only for an explicit operator-controlled repair or \
                             incident investigation.",
                            snapshot_path.display(),
                            engine.replay_checkpoint.last_command_id,
                            e,
                            UNSAFE_SNAPSHOT_REPLAY_OVERRIDE_ENV,
                        );
                    }
                }
            }
        }

        if should_validate_hot_journal_coverage_for_base_replay(
            engine.ctx.db.is_some(),
            startup_replay_checkpoint.last_command_id,
            wal_path_configured,
            engine.snapshot_loaded,
            engine.replay_checkpoint.last_command_id,
        ) {
            let checkpoint_command_id = startup_replay_checkpoint.last_command_id;
            let replay_from_command_id = if engine.snapshot_loaded {
                engine.replay_checkpoint.last_command_id.saturating_add(1)
            } else {
                1
            };
            let db = engine
                .ctx
                .db
                .as_ref()
                .expect("checked db is present before hot journal coverage validation");
            let coverage_result = match hot_journal_bounds {
                Some(bounds) => Ok(Some(bounds)),
                None => db
                    .get_journal_command_id_bounds_sync()
                    .map_err(|error| format!("failed to load hot journal command bounds: {error}")),
            }
            .and_then(|bounds| {
                recovery_safety_report.replay_coverage.checked = true;
                if let Some(bounds) = bounds {
                    hot_journal_bounds = Some(bounds);
                    recovery_safety_report.replay_coverage.min_command_id =
                        Some(bounds.min_command_id);
                    recovery_safety_report.replay_coverage.max_command_id =
                        Some(bounds.max_command_id);
                }
                let non_replayable_tail_count = non_replayable_tail_count_for_coverage_check(
                    db,
                    checkpoint_command_id,
                    bounds,
                    replay_from_command_id,
                )?;
                validate_hot_journal_covers_base_replay(
                    checkpoint_command_id,
                    bounds,
                    replay_from_command_id,
                    non_replayable_tail_count,
                )
            });

            match coverage_result {
                Ok(()) => {
                    recovery_safety_report.pass(
                        "hot_journal_coverage",
                        "hot journal covers base reconstruction window",
                    );
                    info!(
                        checkpoint_command_id = checkpoint_command_id,
                        "Hot journal coverage validation passed for base reconstruction"
                    );
                }
                Err(error) if unsafe_replay_override => {
                    recovery_safety_report.fail("hot_journal_coverage", error.clone());
                    warn!(
                        %error,
                        override_env = UNSAFE_SNAPSHOT_REPLAY_OVERRIDE_ENV,
                        checkpoint_command_id = checkpoint_command_id,
                        "UNSAFE_OPERATOR_OVERRIDE: continuing with incomplete hot-journal base reconstruction"
                    );
                }
                Err(error) => {
                    recovery_safety_report.fail("hot_journal_coverage", error.clone());
                    hypercall_api::recovery_safety::store_report(
                        &runtime_settings.recovery_safety_report,
                        &recovery_safety_report,
                    );
                    panic!(
                        "CRITICAL_FAILURE: {error}. Refusing base reconstruction from incomplete \
                         hot Postgres journal at checkpoint command_id {}. Set {}=true only for \
                         an explicit operator-controlled repair or incident investigation.",
                        checkpoint_command_id, UNSAFE_SNAPSHOT_REPLAY_OVERRIDE_ENV,
                    );
                }
            }
        } else {
            recovery_safety_report.pass(
                "hot_journal_coverage",
                "base reconstruction coverage validation not required",
            );
        }
        // CRITICAL: Replay commands from the journal in checkpoint-bounded windows
        // to reconstruct orderbook state without materializing the full replay set.
        // The replay also builds order_index with full metadata (client_id,
        // mmp_enabled, original_size), so rebuild_from_orderbooks is NOT
        // called afterwards — it would destroy that metadata.
        ReplayRecovery::replay_commands_from_journal(&mut engine);

        if let Some(cash_validation_checkpoint) = post_replay_cash_validation_checkpoint(
            engine.ctx.db.is_some(),
            wal_path_configured,
            startup_replay_checkpoint,
            hot_journal_bounds,
        ) {
            recovery_safety_report.cash.validation_checked = false;
            recovery_safety_report.cash.validation_passed = None;
            recovery_safety_report.cash.message = Some(format!(
                "post-replay DB cash validation skipped at checkpoint command_id {}; engine balance ledger is authoritative",
                cash_validation_checkpoint.last_command_id
            ));
            recovery_safety_report.pass(
                "standard_margin_cash_replay_watermark",
                "post-replay DB cash validation skipped because engine balance ledger is authoritative",
            );
        }
        hypercall_api::recovery_safety::store_report(
            &runtime_settings.recovery_safety_report,
            &recovery_safety_report,
        );

        // Check for crossed orderbooks after replay — should NEVER happen.
        ReplayRecovery::check_restored_orderbooks_for_crosses(&engine);

        // Publish initial snapshot after replay
        engine.publish_snapshot();

        // Persist a fresh engine state snapshot after replay for the next restart.
        engine.persist_engine_state_snapshot();

        // Without a database there is no delayed post-startup reconciliation.
        // Mark the engine ready immediately after replay/bootstrap.
        if engine.ctx.db.is_none() {
            engine.post_startup_reconciled = true;
            engine.sync_status.set_ready();
            info!("UnifiedEngine sync status: Ready (no database configured)");
        }

        engine
    }

    /// Publish a new read snapshot from the current orderbook state.
    fn publish_snapshot(&self) {
        let l2_seq = self.ctx.l2_update_seq.load(Ordering::SeqCst);
        let snapshot = EngineSnapshot::build_with_ctx(&self.ctx, l2_seq);
        tracing::debug!(
            l2_seq = snapshot.l2_seq,
            cash_wallet_count = snapshot.engine_state_digest.cash_wallet_count,
            cash_digest = %snapshot.engine_state_digest.cash_digest,
            "Publishing EngineSnapshot balance_ledger"
        );
        self.read_snapshot.store(Arc::new(snapshot));
    }

    /// Set the portfolio service for executed state.
    ///
    /// This is the canonical source of truth for positions + cash.
    /// Must be set for margin checks to use real executed state.
    pub fn set_portfolio_service(&mut self, service: Arc<dyn PortfolioService + Send + Sync>) {
        info!("Setting PortfolioService for UnifiedEngine");
        self.ctx.deps.portfolio_service = Some(service);
    }

    /// Set the portfolio cache for synchronous fill processing.
    pub fn set_portfolio_cache(&mut self, cache: Arc<PortfolioCache>) {
        info!("Setting PortfolioCache for UnifiedEngine");
        self.ctx.deps.portfolio_cache = Some(cache);
    }

    /// Set the RiskAccountBuilder for building risk accounts.
    ///
    /// This is the single source of truth for "how to build an Account for PM".
    /// Must be set for margin calculations to work correctly.
    pub fn set_risk_account_builder(
        &mut self,
        builder: Arc<crate::rsm::portfolio_margin::risk_account_builder::RiskAccountBuilder>,
    ) {
        info!("Setting RiskAccountBuilder for UnifiedEngine");
        self.ctx.deps.risk_account_builder = Some(builder);
    }

    /// Set the StandardAccountBuilder for building StandardAccounts.
    ///
    /// Used for Standard margin mode accounts (Deribit-style linear margin).
    pub fn set_standard_account_builder(&mut self, builder: Arc<StandardAccountBuilder>) {
        info!("Setting StandardAccountBuilder for UnifiedEngine");
        self.ctx.deps.standard_account_builder = Some(builder);
    }

    /// Set the LiquidationCache for pre-liquidation order blocking.
    ///
    /// When set, orders from accounts in pre-liquidation state will be checked
    /// to ensure they don't increase risk. Risk-reducing orders are still allowed.
    pub fn set_liquidation_cache(&mut self, cache: Arc<crate::liquidator::LiquidationCache>) {
        info!("Setting LiquidationCache for UnifiedEngine");
        self.ctx.deps.liquidation_cache = Some(cache);
    }

    /// Replace the engine's DieselEventHandler.
    ///
    /// Used during standby-to-active promotion: the engine starts with a
    /// readonly handler (or None), and on promote we swap in a read-write
    /// handler so the engine can persist fills, settlements, etc.
    pub fn set_db(&mut self, handler: DatabaseHandler) {
        info!("Replacing DieselEventHandler on UnifiedEngine (standby promote)");
        self.ctx.db = Some(handler);
    }

    /// Set the durable engine journal writer.
    /// When set and journaling is enabled, commands are persisted before ACK.
    /// Also loads recent request_ids into the idempotency cache for fast lookups.
    pub fn set_journal_writer(&mut self, writer: crate::journal::SharedEngineJournalWriter) {
        info!("Setting EngineJournalWriter for UnifiedEngine");

        // Load recent request_ids for idempotency cache warm-up.
        // This eliminates DB lookups for requests that can't be duplicates from previous runs.
        // We load request_ids from the last 24 hours, which should cover any retry scenarios.
        match writer.get_recent_request_ids(IDEMPOTENCY_CACHE_LOOKBACK_HOURS) {
            Ok(request_ids) => {
                let count = request_ids.len();
                self.known_request_ids.extend(request_ids);
                info!(
                    "Loaded {} request_ids into idempotency cache (24h lookback, capacity: {})",
                    count,
                    self.known_request_ids.capacity()
                );
            }
            Err(e) => {
                warn!(
                    "Failed to load request_ids for idempotency cache: {} - idempotency checks will query DB",
                    e
                );
            }
        }

        self.journal_writer = Some(writer);
    }

    /// Set the batched journal sender for async PostgreSQL writes.
    /// When set, the engine pushes journal entries to a channel instead of
    /// blocking on synchronous inserts.
    pub fn set_journal_batch_sender(&mut self, sender: crate::journal::JournalBatchSender) {
        info!("Setting JournalBatchSender for UnifiedEngine (async batched writes)");
        self.journal_batch_sender = Some(sender);
    }

    /// Set the NATS publisher for real-time command replication.
    pub fn set_nats_publisher(&mut self, publisher: crate::nats::NatsPublisher) {
        info!("Setting NatsPublisher for UnifiedEngine (real-time replication)");
        self.nats_publisher = Some(publisher);
    }

    pub fn set_nats_balance_update_publisher(
        &mut self,
        publisher: crate::nats::NatsBalanceUpdatePublisher,
    ) {
        info!("Setting NatsBalanceUpdatePublisher for UnifiedEngine");
        self.nats_balance_update_publisher = Some(publisher);
    }

    /// Publish a command envelope to NATS for standby replication.
    /// Call this after apply() succeeds for any command that should be replicated.
    pub(crate) async fn publish_to_nats(&self, env: &crate::rsm::apply::CommandEnvelope) {
        let Some(ref publisher) = self.nats_publisher else {
            return;
        };

        use crate::nats::CommandType;
        use crate::rsm::apply::{
            ApproveAgentPayload, BalanceCommandPayload, DepositUpdatePayload, EngineCommand,
            HypercoreEquityUpdatePayload, HypercorePositionUpdatePayload, MmpConfigUpdatePayload,
            NonceAdvancePayload, RevokeAgentPayload, TierMarginModeUpdatePayload,
            TradingModeUpdatePayload,
        };

        let (cmd_type, data) = match &env.command {
            EngineCommand::OrderAction(msg) => (
                CommandType::Order,
                hypercall_types::serialize_to_wire_bytes(msg),
            ),
            EngineCommand::PriceUpdate {
                underlying,
                spot_price,
                timestamp_ms,
            } => (
                CommandType::PriceUpdate,
                hypercall_types::serialize_to_wire_bytes(&crate::rsm::apply::PriceUpdatePayload {
                    underlying: underlying.clone(),
                    spot_price: *spot_price,
                    timestamp_ms: *timestamp_ms,
                }),
            ),
            EngineCommand::IvUpdate {
                underlying: _,
                journal_data,
                timestamp_ms: _,
                ..
            } => {
                // Use pre-serialized journal_data if available (avoids re-serializing the surface)
                let data = match journal_data {
                    Some(d) => d.clone(),
                    None => return, // No journal data = nothing to publish
                };
                (CommandType::IvUpdate, data)
            }
            EngineCommand::MarketAction(cmd) => (
                CommandType::MarketAction,
                hypercall_types::serialize_to_wire_bytes(cmd),
            ),
            EngineCommand::LiquidationState(msg) => (
                CommandType::LiquidationState,
                hypercall_types::serialize_to_wire_bytes(msg),
            ),
            EngineCommand::TierUpdate {
                wallet,
                margin_mode,
                tier,
                trading_limits,
            } => (
                CommandType::TierUpdate,
                hypercall_types::serialize_to_wire_bytes(&crate::rsm::apply::TierUpdatePayload {
                    wallet: *wallet,
                    margin_mode: *margin_mode,
                    tier: tier.clone(),
                    trading_limits: *trading_limits,
                }),
            ),
            EngineCommand::LegacyTierMarginModeUpdate {
                wallet,
                margin_mode,
            } => (
                CommandType::TierUpdate,
                hypercall_types::serialize_to_wire_bytes(&TierMarginModeUpdatePayload {
                    wallet: *wallet,
                    margin_mode: *margin_mode,
                }),
            ),
            EngineCommand::HypercorePositionUpdate {
                account,
                coin,
                size,
                entry_price,
                unrealized_pnl,
                timestamp_ms,
            } => (
                CommandType::HypercorePositionUpdate,
                hypercall_types::serialize_to_wire_bytes(&HypercorePositionUpdatePayload {
                    account: account.clone(),
                    coin: coin.clone(),
                    size: *size,
                    entry_price: *entry_price,
                    unrealized_pnl: *unrealized_pnl,
                    timestamp_ms: *timestamp_ms,
                }),
            ),
            EngineCommand::MmpConfigUpdate {
                wallet,
                currency,
                enabled,
                interval_ms,
                frozen_time_ms,
                qty_limit,
                delta_limit,
                vega_limit,
            } => (
                CommandType::MmpConfigUpdate,
                hypercall_types::serialize_to_wire_bytes(&MmpConfigUpdatePayload {
                    wallet: *wallet,
                    currency: currency.clone(),
                    enabled: *enabled,
                    interval_ms: *interval_ms,
                    frozen_time_ms: *frozen_time_ms,
                    qty_limit: *qty_limit,
                    delta_limit: *delta_limit,
                    vega_limit: *vega_limit,
                }),
            ),
            EngineCommand::RfqExecute(cmd) => (
                CommandType::RfqExecute,
                hypercall_types::serialize_to_wire_bytes(cmd),
            ),
            EngineCommand::TradingModeUpdate {
                modes,
                timestamp_ms,
            } => (
                CommandType::TradingModeUpdate,
                hypercall_types::serialize_to_wire_bytes(&TradingModeUpdatePayload {
                    modes: modes.clone(),
                    timestamp_ms: *timestamp_ms,
                }),
            ),
            EngineCommand::TickExpiry { now_ms, context } => (
                CommandType::TickExpiry,
                hypercall_types::serialize_to_wire_bytes(&(*now_ms, context)),
            ),
            EngineCommand::TickSnapshot { .. } => return, // Snapshots are local-only
            EngineCommand::DepositUpdate {
                wallet,
                amount,
                timestamp_ms,
                sequence,
                source_event_hash,
            } => (
                CommandType::DepositUpdate,
                hypercall_types::serialize_to_wire_bytes(&DepositUpdatePayload {
                    wallet: *wallet,
                    amount: *amount,
                    timestamp_ms: *timestamp_ms,
                    sequence: sequence.unwrap_or_else(|| {
                        panic!("RUNTIME_INVARIANT: DepositUpdate missing durable sequence")
                    }),
                    source_event_hash: source_event_hash.clone(),
                }),
            ),
            EngineCommand::OptionDepositUpdate {
                request_id,
                wallet,
                symbol,
                quantity,
                timestamp_ms,
            } => (
                CommandType::OptionDepositUpdate,
                hypercall_types::serialize_to_wire_bytes(&(
                    request_id,
                    wallet,
                    symbol,
                    quantity,
                    timestamp_ms,
                )),
            ),
            EngineCommand::OptionWithdrawalUpdate {
                request_id,
                wallet,
                account,
                signer,
                rsm_signer,
                symbol,
                quantity,
                nonce,
                action,
                timestamp_ms,
            } => (
                CommandType::OptionWithdrawalUpdate,
                hypercall_types::serialize_to_wire_bytes(&(
                    request_id,
                    wallet,
                    account,
                    signer,
                    rsm_signer,
                    symbol,
                    quantity,
                    nonce,
                    action,
                    timestamp_ms,
                )),
            ),
            EngineCommand::CashWithdrawalUpdate {
                request_id,
                wallet,
                account,
                destination,
                signer,
                rsm_signer,
                amount,
                amount_wei,
                nonce,
                timestamp_ms,
            } => (
                CommandType::CashWithdrawalUpdate,
                hypercall_types::serialize_to_wire_bytes(&(
                    request_id,
                    wallet,
                    account,
                    destination,
                    signer,
                    rsm_signer,
                    amount,
                    amount_wei,
                    nonce,
                    timestamp_ms,
                )),
            ),
            EngineCommand::LiquidationBonusUpdate {
                wallet,
                amount,
                balance_after,
                timestamp_ms,
                sequence,
            } => (
                CommandType::LiquidationBonusUpdate,
                hypercall_types::serialize_to_wire_bytes(&BalanceCommandPayload {
                    wallet: *wallet,
                    amount: *amount,
                    balance_after: *balance_after,
                    timestamp_ms: *timestamp_ms,
                    sequence: *sequence,
                }),
            ),
            EngineCommand::ApproveAgent {
                wallet,
                agent,
                expires_at_ms,
                nonce,
                timestamp_ms,
            } => (
                CommandType::ApproveAgent,
                hypercall_types::serialize_to_wire_bytes(&ApproveAgentPayload {
                    wallet: *wallet,
                    agent: *agent,
                    expires_at_ms: *expires_at_ms,
                    nonce: *nonce,
                    timestamp_ms: *timestamp_ms,
                }),
            ),
            EngineCommand::RevokeAgent {
                wallet,
                agent,
                nonce,
                timestamp_ms,
            } => (
                CommandType::RevokeAgent,
                hypercall_types::serialize_to_wire_bytes(&RevokeAgentPayload {
                    wallet: *wallet,
                    agent: *agent,
                    nonce: *nonce,
                    timestamp_ms: *timestamp_ms,
                }),
            ),
            EngineCommand::NonceAdvance { wallet, nonce, .. } => (
                CommandType::NonceAdvance,
                hypercall_types::serialize_to_wire_bytes(&NonceAdvancePayload {
                    wallet: *wallet,
                    nonce: *nonce,
                    timestamp_ms: env.received_ts_ms,
                }),
            ),
            EngineCommand::HypercoreEquityUpdate {
                wallet,
                account_value,
                timestamp_ms,
            } => (
                CommandType::HypercoreEquityUpdate,
                hypercall_types::serialize_to_wire_bytes(&HypercoreEquityUpdatePayload {
                    wallet: *wallet,
                    account_value: *account_value,
                    timestamp_ms: *timestamp_ms,
                }),
            ),
            EngineCommand::SetPmSettlementPoolConfig(command) => (
                CommandType::SetPmSettlementPoolConfig,
                hypercall_types::serialize_to_wire_bytes(command),
            ),
            EngineCommand::RecordPmVaultDeposit(command) => (
                CommandType::RecordPmVaultDeposit,
                hypercall_types::serialize_to_wire_bytes(command),
            ),
            EngineCommand::RequestPmVaultWithdrawal(command) => (
                CommandType::RequestPmVaultWithdrawal,
                hypercall_types::serialize_to_wire_bytes(command),
            ),
            EngineCommand::AccruePmSettlementInterest(command) => (
                CommandType::AccruePmSettlementInterest,
                hypercall_types::serialize_to_wire_bytes(command),
            ),
            EngineCommand::ApplyPmSettlementRepayment(command) => (
                CommandType::ApplyPmSettlementRepayment,
                hypercall_types::serialize_to_wire_bytes(command),
            ),
            EngineCommand::JournalPmRecoveryPlan(command) => (
                CommandType::JournalPmRecoveryPlan,
                hypercall_types::serialize_to_wire_bytes(command),
            ),
            EngineCommand::MarkPmRecoveryActionSubmitted(command) => (
                CommandType::MarkPmRecoveryActionSubmitted,
                hypercall_types::serialize_to_wire_bytes(command),
            ),
            EngineCommand::ResolvePmRecoveryAction(command) => (
                CommandType::ResolvePmRecoveryAction,
                hypercall_types::serialize_to_wire_bytes(command),
            ),
        };

        let is_expiry_payload = matches!(cmd_type, CommandType::TickExpiry)
            || matches!(
                &env.command,
                EngineCommand::MarketAction(command) if command.expiry_context.is_some()
            );
        if is_expiry_payload {
            let total_payload_len = crate::nats::COMMAND_PAYLOAD_PREFIX_LEN + data.len();
            if total_payload_len > MAX_EXPIRY_NATS_PAYLOAD_BYTES {
                panic!(
                    "CRITICAL_FAILURE: expiry NATS payload is {} bytes, exceeding {} byte limit. \
                     Refusing to publish an oversized replication command.",
                    total_payload_len, MAX_EXPIRY_NATS_PAYLOAD_BYTES
                );
            }
        }

        publisher.publish(cmd_type, &data).await;
    }

    pub(crate) async fn publish_balance_updates_to_nats(
        &self,
        updates: &[hypercall_types::BalanceUpdate],
    ) {
        let Some(ref publisher) = self.nats_balance_update_publisher else {
            return;
        };
        for update in updates {
            publisher.publish(update).await;
        }
    }

    /// Set the mark price oracles for spot/forward price lookups.
    ///
    /// Maps underlying symbol (e.g., "BTC", "ETH") to oracle instance.
    /// This is the canonical source for spot prices in margin calculations.
    pub fn set_mark_price_oracles(
        &mut self,
        oracles: HashMap<String, Arc<HyperliquidMarkPriceOracle>>,
    ) {
        info!(
            "Setting mark price oracles for UnifiedEngine: {:?}",
            oracles.keys().collect::<Vec<_>>()
        );
        self.ctx.deps.mark_price_oracles = oracles;
    }

    /// Set the MMP cache for market maker protection
    pub fn set_mmp_cache(&mut self, cache: Arc<crate::read_cache::mmp::MmpCache>) {
        info!("Setting MMP cache for UnifiedEngine");
        self.ctx.deps.mmp_cache = Some(cache);
    }

    /// Set the Tier cache for position limit checks.
    ///
    /// Also seeds wallet_margin_modes from the cache so the engine fails closed
    /// when a wallet's mode is missing rather than consulting a default.
    pub fn set_tier_cache(&mut self, cache: Arc<crate::read_cache::tier::TierCache>) {
        let modes = cache.get_all_margin_modes_sync().expect(
            "CRITICAL_FAILURE: tier cache unavailable during engine seed -- margin modes unavailable",
        );
        let default_limits = cache.default_trading_limits();
        let limits = cache
            .get_all_trading_limits_sync()
            .expect("CRITICAL_FAILURE: tier cache lock contention during engine seed -- trading limits unavailable");
        let tiers = cache.get_all_tiers_sync().expect(
            "CRITICAL_FAILURE: tier cache lock contention during engine seed -- tiers unavailable",
        );
        let count = modes.len();
        self.ctx.deps.wallet_margin_modes = modes;
        self.ctx.deps.wallet_trading_limits = limits;
        self.ctx.deps.default_trading_limits = default_limits;
        self.ctx.deps.wallet_tiers = tiers;
        info!(
            "Setting Tier cache for UnifiedEngine, seeded {} wallet tiers",
            count
        );
        self.ctx.deps.tier_cache = Some(cache);
    }

    /// Set the direct WS event sender (low-latency WS delivery).
    pub fn set_ws_event_sender(
        &mut self,
        sender: tokio::sync::mpsc::UnboundedSender<hypercall_types::EngineMessage>,
    ) {
        self.ctx.deps.ws_event_sender = Some(sender);
    }

    /// Set the RFQ execution receiver channel.
    pub fn set_rfq_receiver(
        &mut self,
        receiver: mpsc::Receiver<hypercall_runtime_api::RfqExecuteRequest>,
    ) {
        self.rfq_receiver = Some(receiver);
    }

    /// Set the deposit receiver channel (faucet/admin deposits).
    pub fn set_deposit_receiver(&mut self, receiver: mpsc::Receiver<DepositRequest>) {
        self.deposit_receiver = Some(receiver);
    }

    /// Set the option deposit receiver channel.
    pub fn set_option_deposit_receiver(&mut self, receiver: mpsc::Receiver<OptionDepositRequest>) {
        self.option_deposit_receiver = Some(receiver);
    }

    pub fn set_option_withdrawal_receiver(
        &mut self,
        receiver: mpsc::Receiver<OptionWithdrawalRequest>,
    ) {
        self.option_withdrawal_receiver = Some(receiver);
    }

    /// Set the cash withdrawal receiver channel.
    pub fn set_cash_withdrawal_receiver(
        &mut self,
        receiver: mpsc::Receiver<CashWithdrawalRequest>,
    ) {
        self.cash_withdrawal_receiver = Some(receiver);
    }

    /// Set the liquidation bonus receiver channel.
    pub fn set_liquidation_bonus_receiver(
        &mut self,
        receiver: mpsc::Receiver<LiquidationBonusRequest>,
    ) {
        self.liquidation_bonus_receiver = Some(receiver);
    }

    /// Set the tier update receiver channel.
    pub fn set_tier_update_receiver(&mut self, receiver: mpsc::Receiver<TierUpdateRequest>) {
        self.tier_update_receiver = Some(receiver);
    }

    /// Set the HyperCore equity update receiver channel.
    pub fn set_hypercore_equity_receiver(
        &mut self,
        receiver: mpsc::Receiver<HypercoreEquityRequest>,
    ) {
        self.hypercore_equity_receiver = Some(receiver);
    }

    /// Attach a `tokio::sync::watch::Receiver` that the engine main
    /// loop will poll for per-underlying `trading_mode` updates. See
    /// `apply_underlying_trading_mode_update` for the handler
    /// semantics. Optional — call sites that don't need live catalog
    /// notifications can omit this and the engine behaves exactly as
    /// before (modes only refresh on recovery / CreateMarket).
    pub fn set_trading_mode_receiver(
        &mut self,
        receiver: tokio::sync::watch::Receiver<
            std::collections::HashMap<String, hypercall_types::TradingModes>,
        >,
    ) {
        self.trading_mode_receiver = Some(receiver);
    }

    /// Get the sync status for readiness checks.
    ///
    /// API endpoints should check this before serving requests that depend on
    /// engine state (e.g., order placement).
    pub fn sync_status(&self) -> Arc<SyncStatus> {
        self.sync_status.clone()
    }
}

impl UnifiedEngine {
    fn rebuild_shared_engine_iv_surfaces(&self) {
        let mut shared = self
            .engine_iv_surfaces
            .write()
            .expect("engine IV surfaces lock poisoned");
        *shared = self.ctx.iv_surfaces.clone();
    }

    /// Get settlement price from oracle for a specific expiry.
    pub async fn get_settlement_price(&self, underlying: &str, expiry_ts: i64) -> Option<Decimal> {
        self.expiry_manager
            .get_settlement_price(&self.ctx.deps, underlying, expiry_ts)
            .await
    }

    /// Set reference price for testing.
    ///
    /// Updates the canonical `deps.reference_prices` map used by margin checks
    /// and settlement price lookups.
    #[cfg(any(test, feature = "test-utils"))]
    pub fn set_reference_price(&mut self, underlying: String, price: Decimal) {
        let price_f64 = price
            .to_f64()
            .expect("set_reference_price: invalid Decimal -> f64 conversion");
        self.ctx.deps.reference_prices.insert(underlying, price_f64);
    }

    /// Set test cash for an account.
    ///
    /// This is a TEST-ONLY function for funding accounts in unit/integration tests.
    /// It seeds balance_ledger, which is authoritative for margin checks.
    #[cfg(any(test, feature = "test-utils"))]
    pub fn set_account_cash(&mut self, wallet: &WalletAddress, cash: f64) {
        let cash_decimal = Decimal::from_f64_retain(cash)
            .expect("set_account_cash: invalid f64 -> Decimal conversion");

        self.ctx.balance_ledger.set(*wallet, cash_decimal);
    }

    /// Expire a specific instrument and settle all positions.
    pub async fn expire_instrument(
        &mut self,
        symbol: &str,
        reference_price: Decimal,
        now_ms: u64,
    ) -> Result<(), String> {
        self.expiry_manager
            .expire_instrument(symbol, reference_price, now_ms, &mut self.ctx)
            .await
    }

    // ===== Test Helper Methods =====
    // These methods are public to support test utilities but should not be used in production code

    /// Transition instruments to EXPIRED_PENDING_PRICE status (test helper).
    pub fn transition_to_pending_settlement(&mut self, symbols: &[String], now_ms: u64) {
        self.expiry_manager
            .transition_to_pending_settlement(symbols, now_ms, &mut self.ctx);
    }

    /// Get all instruments pending settlement (test helper).
    pub fn get_pending_settlement_instruments(&self) -> Vec<(String, i64, Vec<String>)> {
        self.expiry_manager
            .get_pending_settlement_instruments(self.ctx.db.as_ref())
    }

    /// Check if orderbooks contains a key (for test utilities)
    #[cfg(any(test, feature = "test-utils"))]
    pub fn orderbooks_contains_key(&self, symbol: &str) -> bool {
        self.ctx.orderbooks.contains_key(symbol)
    }

    /// Get a clone of the event sender (for test utilities)
    #[cfg(any(test, feature = "test-utils"))]
    pub fn get_event_sender(&self) -> Option<mpsc::UnboundedSender<EngineMessage>> {
        Some(self.ctx.deps.event_sender.clone())
    }

    /// Get L2 snapshots of all orderbooks.
    /// Returns a map of symbol → (bids, asks) where each side is Vec<(price, size)>.
    /// Used for cache reconciliation after unclean restart (SIGKILL).
    pub fn get_all_book_snapshots(
        &self,
    ) -> HashMap<
        String,
        (
            Vec<(rust_decimal::Decimal, rust_decimal::Decimal)>,
            Vec<(rust_decimal::Decimal, rust_decimal::Decimal)>,
        ),
    > {
        self.ctx
            .orderbooks
            .iter()
            .map(|(symbol, book)| (symbol.clone(), book.get_orderbook_snapshot()))
            .collect()
    }

    /// Insert an orderbook directly (for test utilities)
    #[cfg(any(test, feature = "test-utils"))]
    pub fn insert_orderbook(&mut self, symbol: String, orderbook: OrderBook) {
        self.ctx.orderbooks.insert(symbol, orderbook);
    }

    /// Track expiry for a symbol (for test utilities)
    #[cfg(any(test, feature = "test-utils"))]
    pub fn track_expiry(&mut self, symbol: String, expiry: u64) {
        let underlying = symbol.split('-').next().unwrap_or_default().to_string();
        let expiry_timestamp =
            crate::rsm::margin_manager::expiry_date_to_timestamp(&underlying, expiry);
        self.expiry_manager
            .schedule_expiry(symbol, expiry_timestamp);
    }

    /// Persist an orderbook to database if handler is available (for test utilities)
    #[cfg(any(test, feature = "test-utils"))]
    pub fn persist_test_orderbook(&self, symbol: &str, parsed_symbol: &ParsedSymbol) {
        if let Some(ref handler) = self.ctx.db {
            let option_type = match parsed_symbol.option_type {
                crate::types::OptionType::Call => hypercall_types::OptionType::Call,
                crate::types::OptionType::Put => hypercall_types::OptionType::Put,
            };
            let option_token_address =
                match crate::shared::option_token_address::derive_option_token_address(
                    &parsed_symbol.underlying,
                    parsed_symbol.expiry,
                    parsed_symbol.strike,
                    option_type.as_db_str(),
                ) {
                    Ok(option_token_address) => option_token_address,
                    Err(e) => {
                        error!(
                            symbol = symbol,
                            error = %e,
                            "Failed to derive option token address for persisted test orderbook"
                        );
                        return;
                    }
                };

            // Create new instrument record
            let new_instrument = hypercall_db::InstrumentRecord {
                instrument_numeric_id: 0,
                id: symbol.to_string(),
                underlying: parsed_symbol.underlying.clone(),
                strike: parsed_symbol.strike,
                expiry: parsed_symbol.expiry as i64,
                option_type,
                option_token_address: Some(option_token_address),
                status: hypercall_types::api_models::InstrumentStatus::Active,
                trading_mode: "orderbook".to_string(),
            };

            // Save to database synchronously
            if let Err(e) = handler.save_market_and_instrument_sync(
                &parsed_symbol.underlying,
                parsed_symbol.expiry as i64,
                &new_instrument,
            ) {
                error!("Failed to persist test orderbook to database: {}", e);
                // Don't fail - orderbook is already created in memory
            } else {
                info!("Persisted test orderbook {} to database", symbol);
            }
        }
    }
}