hypercall/rsm/

engine_deps.rs

//! Shared dependency bundle for the UnifiedEngine and its sub-managers.
//!
//! `EngineDeps` groups all `Arc`-shared services that the engine's sub-managers
//! need to borrow during order processing, margin checks, and expiry handling.
//! By bundling these into a dedicated struct the engine avoids passing 10+ parameters
//! to every delegated method call.

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::engine_state_snapshot::EngineStateSnapshot;
use crate::rsm::ledger::BalanceLedger;
use crate::shared::order_types::ParsedSymbol;
use crate::standard_margin::StandardAccountBuilder;
use crate::types::Config;
use hypercall_db_diesel::DatabaseHandler;
use hypercall_engine::order_index::EngineOrderIndex;
use hypercall_engine::OrderBook;
use hypercall_types::TradingModes;
use hypercall_types::WalletAddress;
use rust_decimal::prelude::ToPrimitive;
use rust_decimal::Decimal;
use serde::{Deserialize, Serialize};
use std::collections::{BTreeSet, HashMap};
use std::sync::atomic::{AtomicI64, Ordering};
use std::sync::Arc;
use tracing::{info, warn};

// Re-export pure MMP and position types from hypercall-engine.
pub use hypercall_engine::mmp::{EngineMmpState, MmpFillRecord};
pub use hypercall_engine::position::EnginePosition;

#[derive(Debug, Clone, Copy, Serialize, Deserialize)]
pub struct DepositUpdateWatermark {
    #[serde(default)]
    pub sequence: Option<u64>,
    pub timestamp_ms: u64,
    pub balance_after: Decimal,
    #[serde(default)]
    pub source: BalanceLedgerMutationSource,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
pub enum BalanceLedgerMutationSource {
    #[default]
    DepositUpdate,
    CashWithdrawal,
}

/// Shared, immutable (from the manager's perspective) dependencies.
///
/// The engine owns these fields and lends `&EngineDeps` to sub-managers
/// so they can read caches, services, and configuration without owning clones.
pub struct EngineDeps {
    /// MMP (Market Maker Protection) cache.
    pub mmp_cache: Option<Arc<crate::read_cache::mmp::MmpCache>>,

    /// Tier cache for position-limit and margin-mode lookups.
    pub tier_cache: Option<Arc<crate::read_cache::tier::TierCache>>,

    /// Canonical source of truth for executed positions and cash.
    pub portfolio_service: Option<Arc<dyn PortfolioService + Send + Sync>>,

    /// Portfolio cache for synchronous fill application in the engine hot path.
    pub portfolio_cache: Option<Arc<PortfolioCache>>,

    /// Single source of truth for building a risk-ready `Account` for PM.
    pub risk_account_builder:
        Option<Arc<crate::rsm::portfolio_margin::risk_account_builder::RiskAccountBuilder>>,

    /// Builder for Standard margin mode accounts.
    pub standard_account_builder: Option<Arc<StandardAccountBuilder>>,

    /// Spot price provider for margin calculations (legacy / test fallback).
    pub greeks_cache: Option<Arc<GreeksCache>>,

    /// Mark price oracles keyed by underlying (e.g. "BTC", "ETH").
    pub mark_price_oracles: HashMap<String, Arc<HyperliquidMarkPriceOracle>>,

    /// Legacy reference prices for test / expiry settlement fallback.
    pub reference_prices: HashMap<String, f64>,

    /// Pre-liquidation order blocking cache.
    pub liquidation_cache: Option<Arc<crate::liquidator::LiquidationCache>>,

    /// Engine configuration (scenarios, fee config, tolerances, etc.).
    pub config: Config,

    /// Global PM settlement-pool expiry classification gate.
    pub portfolio_margin_pool_enabled: bool,

    /// Wallets allowed to use PM settlement-pool expiry classification.
    pub portfolio_margin_settlement_allowlist: BTreeSet<WalletAddress>,

    /// Broadcast channel for engine events (order updates, L2, trades, …).
    /// Goes to EventBus for persistence and cross-service sync.
    pub event_sender: tokio::sync::mpsc::UnboundedSender<hypercall_types::EngineMessage>,

    /// Direct channel for WebSocket event forwarding.
    /// When set, events are sent here in addition to `event_sender` so that the
    /// WsEventForwarder receives them with minimal latency.
    pub ws_event_sender: Option<tokio::sync::mpsc::UnboundedSender<hypercall_types::EngineMessage>>,

    /// Current command timestamp in seconds. Set from CommandEnvelope.received_ts_ms
    /// at the top of apply(). Used by SPAN instead of wall clock for deterministic
    /// time-to-expiry calculations.
    pub margin_timestamp_s: i64,

    /// Internal liquidation state per wallet, updated by LiquidationState commands.
    pub liquidation_states: HashMap<WalletAddress, hypercall_types::LiquidationStateType>,

    /// Internal margin mode per wallet, updated by TierUpdate commands.
    pub wallet_margin_modes: HashMap<WalletAddress, crate::rsm::margin_mode::MarginMode>,

    /// Internal MMP enabled state per (wallet, currency), updated by MmpConfigUpdate commands.
    pub mmp_enabled: HashMap<(WalletAddress, String), bool>,

    /// Internal trading limits per wallet, seeded from tier_cache at startup.
    /// Updated by TierUpdate commands.
    pub wallet_trading_limits: HashMap<WalletAddress, hypercall_types::api_models::TradingLimits>,

    /// Configured fallback trading limits for wallets without explicit tier rows.
    pub default_trading_limits: hypercall_types::api_models::TradingLimits,

    /// Internal tier string per wallet (e.g. "tier1", "tier2", "market_maker").
    /// Seeded from tier_cache at startup, updated by TierUpdate commands.
    pub wallet_tiers: HashMap<WalletAddress, String>,

    /// Internal perp positions per (account, coin), updated by HypercorePositionUpdate commands.
    /// Keyed by (lowercase account string, coin string).
    pub perp_positions: HashMap<(String, String), crate::hypercore::PerpPosition>,

    /// HyperCore account equity per wallet (manager address), updated by
    /// HypercoreEquityUpdate commands from the Hydromancer feed. For PM users,
    /// this replaces the balance_ledger as the cash base for margin calculations.
    /// The value is the HyperCore `accountValue` which includes perp UPNL.
    pub hypercore_account_equity: HashMap<WalletAddress, Decimal>,

    /// Last applied HyperCore equity update timestamp per wallet, in milliseconds.
    /// Written by the Hydromancer HyperCore sync path and restored from snapshots
    /// so runtime apply and replay reject stale accountValue updates.
    pub hypercore_equity_timestamps: HashMap<WalletAddress, u64>,
}

impl EngineDeps {
    /// Emit an engine event to both the event bus (persistence) and the direct WS channel.
    ///
    /// Both paths are best-effort (errors silently dropped). A dropped WS receiver
    /// just means no WS clients.
    pub fn emit_event(&self, event: &hypercall_types::EngineMessage) {
        let _ = self.event_sender.send(event.clone());
        if let Some(ref ws_tx) = self.ws_event_sender {
            let _ = ws_tx.send(event.clone());
        }
    }
}

/// Classify a rejection reason string into a metric label.
///
/// Used by all order managers when recording rejection metrics.
pub fn classify_rejection_reason(reason: &str) -> &'static str {
    hypercall_engine::admission::classify_rejection_reason(reason)
}

/// Convert engine positions to a `PortfolioBalance` for margin calculations.
///
/// This is a runtime helper that uses crate-local types (`PortfolioBalance`,
/// `PositionData`) which cannot live in the pure `hypercall-engine` crate.
pub fn engine_positions_to_portfolio_balance(
    positions: &HashMap<(WalletAddress, String), EnginePosition>,
    wallet: &WalletAddress,
    reference_prices: &HashMap<String, f64>,
) -> crate::portfolio::PortfolioBalance {
    use crate::portfolio::{PortfolioBalance, PositionData};
    use rust_decimal_macros::dec;

    let mut pos_map = HashMap::new();
    for ((w, symbol), pos) in positions {
        if w != wallet {
            continue;
        }
        let upnl = if symbol.ends_with("-PERP") {
            let underlying = symbol.split('-').next().unwrap_or(symbol);
            if let Some(&spot) = reference_prices.get(underlying) {
                let spot_dec = rust_decimal::Decimal::try_from(spot).unwrap_or(dec!(0));
                (spot_dec - pos.entry_price) * pos.quantity
            } else {
                dec!(0)
            }
        } else {
            dec!(0)
        };
        pos_map.insert(
            symbol.clone(),
            PositionData {
                symbol: symbol.clone(),
                amount: pos.quantity,
                entry_price: pos.entry_price,
                margin_posted: dec!(0),
                realized_pnl: dec!(0),
                unrealized_pnl: upnl,
            },
        );
    }
    PortfolioBalance {
        positions: pos_map,
        total_margin_used: dec!(0),
    }
}

/// Apply a fill to the engine position map, creating or removing entries as needed.
///
/// This is a map-level helper that wraps [`EnginePosition::apply_fill`] with
/// HashMap insert/remove semantics. The pure position math lives in the engine crate.
pub fn apply_fill_to_positions(
    positions: &mut HashMap<(WalletAddress, String), EnginePosition>,
    wallet: WalletAddress,
    symbol: String,
    signed_qty: rust_decimal::Decimal,
    fill_price: rust_decimal::Decimal,
) {
    use rust_decimal_macros::dec;

    let key = (wallet, symbol);
    let pos = positions.entry(key.clone()).or_insert(EnginePosition {
        quantity: dec!(0),
        entry_price: dec!(0),
    });

    pos.apply_fill(signed_qty, fill_price);

    if pos.quantity == dec!(0) {
        positions.remove(&key);
    }
}

/// Credit externally deposited option-token inventory into engine positions.
///
/// This is deliberately not modeled as a zero-price fill. Depositing tokens
/// moves already-owned inventory into HyperCall custody; it must not realize
/// PnL, change cash, or rewrite the cost basis of an existing same-side
/// position. If a deposit crosses a short through zero, the newly long
/// remainder has unknown off-chain cost basis, so the engine records zero.
pub fn apply_option_deposit_to_positions(
    positions: &mut HashMap<(WalletAddress, String), EnginePosition>,
    wallet: WalletAddress,
    symbol: String,
    quantity: rust_decimal::Decimal,
) {
    use rust_decimal_macros::dec;

    assert!(
        quantity > dec!(0),
        "RUNTIME_INVARIANT: option deposit quantity must be positive"
    );

    let key = (wallet, symbol);
    let Some(pos) = positions.get_mut(&key) else {
        positions.insert(
            key,
            EnginePosition {
                quantity,
                entry_price: dec!(0),
            },
        );
        return;
    };

    let old_qty = pos.quantity;
    let old_entry = pos.entry_price;
    let new_qty = old_qty + quantity;

    if new_qty == dec!(0) {
        positions.remove(&key);
    } else {
        pos.quantity = new_qty;
        pos.entry_price = if old_qty > dec!(0) {
            old_entry
        } else if new_qty < dec!(0) {
            old_entry
        } else {
            dec!(0)
        };
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use alloy::primitives::Address;
    use rust_decimal_macros::dec;

    fn wallet(byte: u8) -> WalletAddress {
        WalletAddress::from(Address::repeat_byte(byte))
    }

    #[test]
    fn option_deposit_preserves_existing_long_cost_basis() {
        let wallet = wallet(1);
        let symbol = "BTC-20260130-100000-C".to_string();
        let mut positions = HashMap::from([(
            (wallet, symbol.clone()),
            EnginePosition {
                quantity: dec!(1),
                entry_price: dec!(100),
            },
        )]);

        apply_option_deposit_to_positions(&mut positions, wallet, symbol.clone(), dec!(1));

        let position = positions
            .get(&(wallet, symbol))
            .expect("position should remain open");
        assert_eq!(position.quantity, dec!(2));
        assert_eq!(position.entry_price, dec!(100));
    }
}

/// Debit option-token inventory for an on-chain withdrawal.
///
/// Withdrawals are custody movements out of the engine, not sells. They can
/// only reduce existing long inventory and must not realize PnL or mutate cash.
pub fn apply_option_withdrawal_to_positions(
    positions: &mut HashMap<(WalletAddress, String), EnginePosition>,
    wallet: WalletAddress,
    symbol: String,
    quantity: rust_decimal::Decimal,
) -> Result<(), String> {
    use rust_decimal_macros::dec;

    if quantity <= dec!(0) {
        return Err("option withdrawal quantity must be positive".to_string());
    }

    let key = (wallet, symbol);
    let Some(pos) = positions.get_mut(&key) else {
        return Err("insufficient option balance for withdrawal".to_string());
    };
    if pos.quantity < quantity {
        return Err(format!(
            "insufficient option balance for withdrawal: have {}, need {}",
            pos.quantity, quantity
        ));
    }

    pos.quantity -= quantity;
    if pos.quantity == dec!(0) {
        positions.remove(&key);
    }
    Ok(())
}

/// All mutable + immutable shared state that managers need to access.
///
/// Kept as a single struct field on `UnifiedEngine` to enable split borrows
/// between `self.ctx` and `self.<manager>`.
pub struct EngineCtx {
    // ----- Mutable order state -----
    /// Active orderbooks keyed by symbol.
    pub orderbooks: HashMap<String, OrderBook>,

    /// Next order ID to allocate (global monotonic counter).
    pub next_order_id: u64,

    /// Next trade ID to allocate (global monotonic counter).
    pub next_trade_id: u64,

    /// L2 sequence counter for orderbook snapshots.
    pub l2_update_seq: Arc<AtomicI64>,

    // ----- Expiry state -----
    /// symbol → expired flag (shared between engine and ExpiryManager).
    pub expired_instruments: HashMap<String, bool>,

    // ----- Deterministic price state -----
    /// Canonical spot prices per underlying, updated only by PriceUpdate commands.
    /// Margin checks read from here instead of external caches, making the engine
    /// deterministic and replayable.
    pub spot_prices: HashMap<String, rust_decimal::Decimal>,

    /// Canonical IV surfaces per underlying, updated only by IvUpdate commands.
    /// The SpanMarginService reads from these instead of external vol oracles.
    pub iv_surfaces: HashMap<String, crate::vol_oracle::vol_surface_cache::VolatilitySurface>,

    /// Source timestamps for IV surfaces (from the upstream oracle snapshot).
    /// Used for change detection to skip ingestion when the oracle hasn't updated.
    pub iv_source_timestamps: HashMap<String, i64>,

    // ----- Trading mode -----
    /// Per-instrument trading mode (orderbook-only, rfq-only, or both).
    pub instrument_trading_modes: HashMap<String, TradingModes>,

    // ----- In-process order index -----
    pub order_index: EngineOrderIndex,

    // ----- Position tracking -----
    /// Engine-owned option positions, updated from fills inside apply().
    /// Single source of truth for margin checks. PortfolioService is a
    /// downstream projection, not consulted during apply().
    /// Perp positions are in deps.perp_positions (fed by HypercorePositionUpdate).
    pub engine_positions: HashMap<(WalletAddress, String), EnginePosition>,

    // ----- Cash tracking -----
    /// Engine-owned cash balances per wallet. Updated from fill side effects
    /// (realized PnL + premium deltas) and deposits. Margin checks read from
    /// here instead of the async Ledger.
    pub balance_ledger: BalanceLedger,

    /// Last applied deposit update per wallet. This prevents an older
    /// post-balance command from rolling balance_ledger backward after a newer
    /// deposit has already applied.
    pub deposit_update_watermarks: HashMap<WalletAddress, DepositUpdateWatermark>,

    /// Applied cash DepositUpdate source event hashes.
    ///
    /// This is intentionally an incident-stabilization tradeoff, not the target
    /// architecture. Cash DepositUpdate is now an additive delta, so the old
    /// single per-wallet sequence watermark can under-credit if sequence 11 is
    /// applied before an unseen sequence 10. The source_event_hash is the
    /// deterministic idempotency key, and tracking it here makes duplicate
    /// replay safe without dropping out-of-order lower-sequence deltas.
    ///
    /// The cost is that this set grows with every applied cash deposit and is
    /// serialized in EngineStateSnapshot. That is unacceptable long-term. The
    /// intended replacement is a bounded engine-owned cash sequencer, similar in
    /// spirit to BoundedNonceSet but gap-buffered for ordered deltas: store only
    /// applied_through plus a small pending window, require a per-wallet cash
    /// sequence, apply contiguous deltas, and fail closed when the gap window is
    /// exceeded. A DB-owned cash queue would also fix ordering, but it moves
    /// authority away from the engine and back toward projection-driven cash
    /// mutation, which is the wrong architectural direction.
    pub applied_deposit_source_event_hashes: BTreeSet<alloy::primitives::FixedBytes<32>>,

    /// Engine-owned PM settlement-pool authority.
    ///
    /// Projection tables are downstream audit surfaces and must never hydrate
    /// this state on startup or rebuild.
    pub pm_settlement_state: crate::rsm::portfolio_margin::settlement_state::PmSettlementState,

    // ----- MMP state -----
    /// Engine-internal MMP fill tracking per (wallet, currency).
    /// Updated synchronously inside apply() so the matching loop never
    /// performs async reads against the external MmpCache.
    pub mmp_state: HashMap<(WalletAddress, String), EngineMmpState>,

    // ----- Agent authorization state -----
    /// Engine-owned agent authorizations: owner wallet → (agent wallet → optional expiry ms).
    /// `None` = never expires. Updated by ApproveAgent/RevokeAgent commands.
    /// Read via EngineSnapshot for lock-free API access.
    pub agent_authorizations: HashMap<WalletAddress, HashMap<WalletAddress, Option<u64>>>,

    // ----- Nonce replay protection -----
    /// Per-signer bounded nonce sets (HL model). The engine stores the N highest
    /// nonces per signer. A new nonce must be greater than the smallest in the
    /// set and not already used. Shared across orders, agent auth, and QP
    /// handshakes. Keyed by signer address (api_wallet for orders, recovered
    /// signer for agent auth, QP wallet for handshakes).
    pub nonce_sets: HashMap<WalletAddress, hypercall_engine::BoundedNonceSet>,

    /// Canonical RSM signer nonce allocation. Directive-producing commands
    /// allocate here during apply(), then append a NeedsRsmSignature outbox
    /// item carrying the allocated nonce.
    pub rsm_signer_nonces: HashMap<WalletAddress, u64>,

    // ----- Immutable shared dependencies -----
    pub deps: EngineDeps,

    // ----- Persistence -----
    pub db: Option<DatabaseHandler>,
}

impl EngineCtx {
    fn symbol_is_expired_by_wall_clock(symbol: &str) -> Option<bool> {
        let parsed = match ParsedSymbol::from_symbol(symbol) {
            Ok(parsed) => parsed,
            Err(error) => {
                warn!(
                    "Snapshot expired instrument {} has invalid symbol format: {}",
                    symbol, error
                );
                return None;
            }
        };
        let expiry_ts =
            hypercall_types::expiry_date_to_timestamp(&parsed.underlying, parsed.expiry);
        if expiry_ts == 0 {
            warn!(
                "Snapshot expired instrument {} has invalid expiry code {}",
                symbol, parsed.expiry
            );
            return None;
        }
        let expiry_ts = u64::try_from(expiry_ts).unwrap_or_else(|_| {
            panic!(
                "STATE_CORRUPTION: expiry timestamp {} for {} is negative",
                expiry_ts, symbol
            )
        });
        let now = crate::shared::clock::unix_now_secs();
        Some(now >= expiry_ts)
    }

    /// Restore engine state from a persistent snapshot.
    ///
    /// Sets counters, restores orderbook orders, rebuilds the order index,
    /// and restores expired instruments. Called during startup before delta
    /// replay to skip full journal replay.
    pub fn restore_from_snapshot(&mut self, snapshot: &EngineStateSnapshot) {
        // 1. Restore counters
        self.next_order_id = snapshot.next_order_id;
        self.next_trade_id = snapshot.next_trade_id;
        self.l2_update_seq
            .store(snapshot.last_l2_seq, Ordering::SeqCst);

        // 2. Restore orderbook orders
        let snapshot_symbols: BTreeSet<String> = snapshot.orderbooks.keys().cloned().collect();
        let engine_symbols: BTreeSet<String> = self.orderbooks.keys().cloned().collect();
        let snapshot_only: Vec<String> = snapshot_symbols
            .difference(&engine_symbols)
            .cloned()
            .collect();
        let snapshot_only_with_orders: Vec<String> = snapshot_only
            .iter()
            .filter_map(|symbol| {
                snapshot
                    .orderbooks
                    .get(symbol)
                    .filter(|entries| !entries.is_empty())
                    .map(|_| symbol.clone())
            })
            .collect();
        if !snapshot_only_with_orders.is_empty() {
            let orphaned_order_count: usize = snapshot_only_with_orders
                .iter()
                .filter_map(|s| snapshot.orderbooks.get(s).map(|e| e.len()))
                .sum();
            warn!(
                "Dropping {} orphaned orders across {} expired/unknown symbols from snapshot: {:?}",
                orphaned_order_count,
                snapshot_only_with_orders.len(),
                snapshot_only_with_orders
            );
        }
        if !snapshot_only.is_empty() {
            warn!(
                "Snapshot contains symbols not present in current engine orderbooks; skipping restore for {:?}",
                snapshot_only
            );
        }

        let mut total_orders = 0usize;
        for (symbol, orderbook) in &mut self.orderbooks {
            let orders: Vec<_> = snapshot
                .orderbooks
                .get(symbol)
                .map(|entries| {
                    entries
                        .iter()
                        .map(|e| {
                            hypercall_engine::OrderRecord {
                                order_id: e.order_id,
                                price: e.price,
                                quantity: e.quantity,
                                side: e.side,
                                wallet: e.wallet,
                                timestamp: e.timestamp,
                                client_id: e.client_id.clone(),
                                mmp_enabled: e.mmp_enabled,
                                // Old snapshots have original_size=None; fall back to remaining qty.
                                original_size: e.original_size.unwrap_or(e.quantity),
                            }
                        })
                        .collect()
                })
                .unwrap_or_default();
            total_orders += orders.len();
            orderbook.restore_from_orders(orders);
        }

        // 3. Rebuild order index from restored orderbooks
        self.order_index.rebuild_from_orderbooks(&self.orderbooks);

        // 4. Restore expired instruments, but keep DB startup recovery authoritative.
        // A stale snapshot file can outlive a fresh database in smoke tests and
        // must not mark active future instruments expired.
        let db_recovered_expired = self.expired_instruments.clone();
        let mut restored_expired = HashMap::new();
        let mut dropped_stale_expired = 0usize;
        for (symbol, expired) in &snapshot.expired_instruments {
            if !*expired {
                continue;
            }
            if db_recovered_expired.get(symbol) == Some(&true) {
                restored_expired.insert(symbol.clone(), true);
                continue;
            }
            match Self::symbol_is_expired_by_wall_clock(symbol) {
                Some(true) | None => {
                    restored_expired.insert(symbol.clone(), true);
                }
                Some(false) => {
                    dropped_stale_expired += 1;
                }
            }
        }
        if dropped_stale_expired > 0 {
            warn!(
                "Dropped {} stale expired instrument flag(s) from engine state snapshot",
                dropped_stale_expired
            );
        }
        self.expired_instruments = restored_expired;

        // 5. Restore deterministic market and policy state that is owned by
        // engine commands. Snapshot restore must not depend on live caches for
        // these replayable inputs.
        self.spot_prices = snapshot.spot_prices.clone();
        self.deps.reference_prices = snapshot
            .spot_prices
            .iter()
            .filter_map(|(underlying, price)| {
                price
                    .to_f64()
                    .map(|price| (underlying.clone(), price))
                    .or_else(|| {
                        warn!(
                            underlying = %underlying,
                            price = %price,
                            "Skipping restored spot price that cannot be represented as f64"
                        );
                        None
                    })
            })
            .collect();
        self.iv_surfaces = snapshot
            .iv_surfaces
            .clone()
            .into_iter()
            .map(|(underlying, surface)| (underlying, surface.into_surface()))
            .collect();
        self.iv_source_timestamps = snapshot.iv_source_timestamps.clone();
        if snapshot.instrument_trading_modes.is_empty() && !self.instrument_trading_modes.is_empty()
        {
            warn!(
                seeded_modes = self.instrument_trading_modes.len(),
                "Preserving DB-seeded instrument trading modes because snapshot has none"
            );
        } else {
            self.instrument_trading_modes = snapshot.instrument_trading_modes.clone();
        }

        // 6. Restore engine positions
        self.engine_positions = snapshot.engine_positions.clone();
        self.mmp_state = snapshot.mmp_state.clone();
        self.deps.liquidation_states = snapshot.liquidation_states.clone();
        if snapshot.wallet_margin_modes.is_empty() && !self.deps.wallet_margin_modes.is_empty() {
            warn!(
                seeded_modes = self.deps.wallet_margin_modes.len(),
                "Preserving tier-cache seeded wallet margin modes because snapshot has none"
            );
        } else {
            self.deps.wallet_margin_modes = snapshot.wallet_margin_modes.clone();
        }
        self.deps.mmp_enabled = snapshot.mmp_enabled.clone();
        if snapshot.wallet_trading_limits.is_empty()
            && snapshot.wallet_tiers.is_empty()
            && !self.deps.wallet_trading_limits.is_empty()
        {
            warn!(
                seeded_limits = self.deps.wallet_trading_limits.len(),
                seeded_tiers = self.deps.wallet_tiers.len(),
                "Preserving tier-cache seeded trading limits because snapshot has none"
            );
        } else {
            self.deps.wallet_trading_limits = snapshot.wallet_trading_limits.clone();
            self.deps.default_trading_limits = snapshot.default_trading_limits;
            self.deps.wallet_tiers = snapshot.wallet_tiers.clone();
        }

        // 7. Restore runtime balances from the replay-bound engine snapshot.
        // Engine snapshots plus WAL are the only cash restart authority.
        self.balance_ledger = BalanceLedger::from_map_with_sequence(
            snapshot.balance_ledger.clone(),
            snapshot.last_balance_update_seq,
        );
        info!(
            balance_wallet_count = self.balance_ledger.len(),
            snapshot_last_command_id = snapshot.last_command_id,
            snapshot_last_l2_seq = snapshot.last_l2_seq,
            "Restored balance_ledger from engine state snapshot"
        );

        // 8. Restore deposit update watermarks
        self.deposit_update_watermarks = snapshot.deposit_update_watermarks.clone();
        self.applied_deposit_source_event_hashes =
            snapshot.applied_deposit_source_event_hashes.clone();
        self.pm_settlement_state = snapshot.pm_settlement_state.clone();

        // 9. Restore agent authorizations
        self.agent_authorizations = snapshot
            .agent_authorizations
            .iter()
            .map(|(wallet, agents)| {
                let map: HashMap<WalletAddress, Option<u64>> = agents
                    .iter()
                    .map(|(agent, expires)| (*agent, *expires))
                    .collect();
                (*wallet, map)
            })
            .collect();
        // 10. Restore nonce sets
        self.nonce_sets.clear();
        for (signer, nonces_vec) in &snapshot.nonce_sets {
            self.nonce_sets.insert(
                *signer,
                hypercall_engine::BoundedNonceSet::from_vec(
                    nonces_vec.clone(),
                    hypercall_engine::nonce::DEFAULT_NONCE_SET_CAPACITY,
                ),
            );
        }
        // Backward compat: migrate old watermark entries as single-element sets
        for (wallet, watermark) in &snapshot.nonce_watermarks {
            self.nonce_sets.entry(*wallet).or_insert_with(|| {
                let mut set = hypercall_engine::BoundedNonceSet::new(
                    hypercall_engine::nonce::DEFAULT_NONCE_SET_CAPACITY,
                );
                set.insert(*watermark);
                set
            });
        }
        self.rsm_signer_nonces = snapshot.rsm_signer_nonces.clone();

        // 11. Restore HyperCore account equity
        self.deps.hypercore_account_equity = snapshot.hypercore_account_equity.clone();
        self.deps.hypercore_equity_timestamps = snapshot.hypercore_equity_timestamps.clone();
        self.deps.perp_positions = snapshot.perp_positions.clone();

        info!(
            "Restored engine state from snapshot: next_order_id={}, next_trade_id={}, \
             l2_seq={}, orderbooks={}, orders={}, expired={}, hypercore_equity_wallets={}",
            self.next_order_id,
            self.next_trade_id,
            snapshot.last_l2_seq,
            snapshot.orderbooks.len(),
            total_orders,
            self.expired_instruments.len(),
            self.deps.hypercore_account_equity.len(),
        );
    }
}