hypercall/rsm/unified_engine/

apply_interface.rs

//! Apply/state-machine interface for UnifiedEngine.

use super::*;

// =============================================================================
// State Machine Interface (apply)
// =============================================================================

use crate::rsm::apply::{ApplyOutput, CommandEnvelope, EngineCommand};

/// Error type for apply operations
#[derive(Debug, Clone)]
pub enum EngineError {
    /// Order was rejected (with reason)
    Rejected(String),
    /// Internal error during processing
    Internal(String),
    /// Market operation error
    Market(String),
}

impl std::fmt::Display for EngineError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            EngineError::Rejected(reason) => write!(f, "Rejected: {}", reason),
            EngineError::Internal(reason) => write!(f, "Internal error: {}", reason),
            EngineError::Market(reason) => write!(f, "Market error: {}", reason),
        }
    }
}

impl std::error::Error for EngineError {}

impl UnifiedEngine {
    fn validate_nonce_preview(
        &self,
        signer: &hypercall_types::WalletAddress,
        nonce: Option<u64>,
        command_timestamp_ms: u64,
        preview_sets: &mut std::collections::HashMap<
            hypercall_types::WalletAddress,
            hypercall_engine::BoundedNonceSet,
        >,
    ) -> Result<(), EngineError> {
        let Some(nonce) = nonce else {
            return Ok(());
        };
        if !hypercall_engine::nonce_within_time_bounds(nonce, command_timestamp_ms) {
            return Err(EngineError::Rejected(format!(
                "nonce {} is outside time bounds for signer {} (command_ts={})",
                nonce, signer, command_timestamp_ms
            )));
        }
        let set = preview_sets.entry(*signer).or_insert_with(|| {
            let mut set = self.ctx.nonce_sets.get(signer).cloned().unwrap_or_else(|| {
                hypercall_engine::BoundedNonceSet::new(
                    hypercall_engine::nonce::DEFAULT_NONCE_SET_CAPACITY,
                )
            });
            set.purge_expired(command_timestamp_ms);
            set
        });
        if !set.insert(nonce) {
            return Err(EngineError::Rejected(format!(
                "nonce {} rejected for signer {} (duplicate or below set minimum {})",
                nonce,
                signer,
                set.min().map(|m| m.to_string()).unwrap_or_default()
            )));
        }
        Ok(())
    }

    /// Validate a nonce against the per-signer bounded nonce set (HL model).
    /// The nonce must be within the time bounds of the command timestamp,
    /// not already used, and greater than the smallest nonce in the set.
    /// Skips validation when `nonce` is `None` (engine-internal commands).
    fn validate_and_advance_nonce(
        &mut self,
        signer: &hypercall_types::WalletAddress,
        nonce: Option<u64>,
        command_timestamp_ms: u64,
    ) -> Result<(), EngineError> {
        let Some(nonce) = nonce else {
            return Ok(());
        };
        if !hypercall_engine::nonce_within_time_bounds(nonce, command_timestamp_ms) {
            return Err(EngineError::Rejected(format!(
                "nonce {} is outside time bounds for signer {} (command_ts={})",
                nonce, signer, command_timestamp_ms
            )));
        }
        let set = self.ctx.nonce_sets.entry(*signer).or_insert_with(|| {
            hypercall_engine::BoundedNonceSet::new(
                hypercall_engine::nonce::DEFAULT_NONCE_SET_CAPACITY,
            )
        });
        set.purge_expired(command_timestamp_ms);
        if !set.insert(nonce) {
            return Err(EngineError::Rejected(format!(
                "nonce {} rejected for signer {} (duplicate or below set minimum {})",
                nonce,
                signer,
                set.min().map(|m| m.to_string()).unwrap_or_default()
            )));
        }
        Ok(())
    }

    fn allocate_rsm_nonce(&mut self, signer: hypercall_types::WalletAddress) -> u64 {
        let next = self.ctx.rsm_signer_nonces.entry(signer).or_insert(0);
        let allocated = *next;
        *next = next.saturating_add(1);
        allocated
    }

    pub fn seed_rsm_signer_nonce_floor(
        &mut self,
        signer: hypercall_types::WalletAddress,
        next_nonce: u64,
    ) {
        let current = self.ctx.rsm_signer_nonces.entry(signer).or_insert(0);
        if *current < next_nonce {
            *current = next_nonce;
        }
    }

    fn validate_cash_withdrawal_margin(
        &self,
        wallet: hypercall_types::WalletAddress,
        amount: rust_decimal::Decimal,
    ) -> Result<(), EngineError> {
        if matches!(
            self.ctx.deps.liquidation_states.get(&wallet),
            Some(hypercall_types::LiquidationStateType::PreLiquidation)
                | Some(hypercall_types::LiquidationStateType::InLiquidation)
        ) {
            return Err(EngineError::Rejected(
                "withdrawals are disabled while the account is in liquidation".to_string(),
            ));
        }

        let margin_mode = self
            .margin_manager
            .get_margin_mode(&self.ctx.deps, &wallet)
            .map_err(|error| {
                EngineError::Internal(format!(
                    "failed to validate cash withdrawal margin for {}: {}",
                    wallet, error
                ))
            })?;

        match margin_mode {
            crate::rsm::MarginMode::Standard => {
                let mut simulated_balances = self.ctx.balance_ledger.clone();
                let post_balance = simulated_balances
                    .get(&wallet)
                    .copied()
                    .unwrap_or(rust_decimal::Decimal::ZERO)
                    - amount;
                simulated_balances.insert(wallet, post_balance);
                let account =
                    crate::standard_margin::StandardAccountBuilder::build_from_engine_state(
                        &wallet,
                        &simulated_balances,
                        &self.ctx.engine_positions,
                        &self.ctx.deps.reference_prices,
                    )
                    .map_err(|error| {
                        EngineError::Internal(format!(
                            "failed to build standard margin account for cash withdrawal {}: {}",
                            wallet, error
                        ))
                    })?;
                let margin = self
                    .margin_manager
                    .standard_margin_service
                    .compute_margin(&account);
                if margin.maintenance_margin < rust_decimal::Decimal::ZERO {
                    return Err(EngineError::Rejected(format!(
                        "withdrawal would put account below maintenance margin: post_withdraw_equity={}, maintenance_margin_shortfall={}",
                        margin.equity,
                        -margin.maintenance_margin
                    )));
                }
            }
            crate::rsm::MarginMode::Portfolio => {
                let margin = self
                    .margin_manager
                    .get_span_margin_for_wallet(
                        &self.ctx.deps,
                        &self.ctx.engine_positions,
                        &self.ctx.balance_ledger,
                        &wallet,
                    )
                    .map_err(|error| {
                        EngineError::Internal(format!(
                            "failed to compute portfolio margin for cash withdrawal {}: {}",
                            wallet, error
                        ))
                    })?
                    .ok_or_else(|| {
                        EngineError::Internal(format!(
                            "missing portfolio margin details for cash withdrawal {}",
                            wallet
                        ))
                    })?;
                let post_withdraw_equity = margin.equity - amount;
                if post_withdraw_equity < margin.maintenance_margin_required {
                    return Err(EngineError::Rejected(format!(
                        "withdrawal would put account below maintenance margin: post_withdraw_equity={}, maintenance_required={}",
                        post_withdraw_equity, margin.maintenance_margin_required
                    )));
                }
            }
        }

        Ok(())
    }

    fn validate_option_withdrawal_margin(
        &self,
        wallet: hypercall_types::WalletAddress,
        symbol: &str,
        quantity: rust_decimal::Decimal,
    ) -> Result<(), EngineError> {
        if matches!(
            self.ctx.deps.liquidation_states.get(&wallet),
            Some(hypercall_types::LiquidationStateType::PreLiquidation)
                | Some(hypercall_types::LiquidationStateType::InLiquidation)
        ) {
            return Err(EngineError::Rejected(
                "withdrawals are disabled while the account is in liquidation".to_string(),
            ));
        }

        let margin_mode = self
            .margin_manager
            .get_margin_mode(&self.ctx.deps, &wallet)
            .map_err(|error| {
                EngineError::Internal(format!(
                    "failed to validate option withdrawal margin for {}: {}",
                    wallet, error
                ))
            })?;

        let mut simulated_positions = self.ctx.engine_positions.clone();
        crate::rsm::engine_deps::apply_option_withdrawal_to_positions(
            &mut simulated_positions,
            wallet,
            symbol.to_string(),
            quantity,
        )
        .map_err(EngineError::Rejected)?;

        match margin_mode {
            crate::rsm::MarginMode::Standard => {
                let account =
                    crate::standard_margin::StandardAccountBuilder::build_from_engine_state(
                        &wallet,
                        &self.ctx.balance_ledger,
                        &simulated_positions,
                        &self.ctx.deps.reference_prices,
                    )
                    .map_err(|error| {
                        EngineError::Internal(format!(
                            "failed to build standard margin account for option withdrawal {}: {}",
                            wallet, error
                        ))
                    })?;
                let margin = self
                    .margin_manager
                    .standard_margin_service
                    .compute_margin(&account);
                if margin.maintenance_margin < rust_decimal::Decimal::ZERO {
                    return Err(EngineError::Rejected(format!(
                        "option withdrawal would put account below maintenance margin: equity={}, maintenance_margin_shortfall={}",
                        margin.equity,
                        -margin.maintenance_margin
                    )));
                }
            }
            crate::rsm::MarginMode::Portfolio => {
                let margin = self
                    .margin_manager
                    .get_span_margin_for_wallet(
                        &self.ctx.deps,
                        &simulated_positions,
                        &self.ctx.balance_ledger,
                        &wallet,
                    )
                    .map_err(|error| {
                        EngineError::Internal(format!(
                            "failed to compute portfolio margin for option withdrawal {}: {}",
                            wallet, error
                        ))
                    })?
                    .ok_or_else(|| {
                        EngineError::Internal(format!(
                            "missing portfolio margin details for option withdrawal {}",
                            wallet
                        ))
                    })?;
                if margin.equity < margin.maintenance_margin_required {
                    return Err(EngineError::Rejected(format!(
                        "option withdrawal would put account below maintenance margin: equity={}, maintenance_required={}",
                        margin.equity, margin.maintenance_margin_required
                    )));
                }
            }
        }

        Ok(())
    }

    pub(super) fn apply_deposit_update_to_balance_ledger(
        &mut self,
        update_kind: &'static str,
        wallet: hypercall_types::WalletAddress,
        amount: rust_decimal::Decimal,
        timestamp_ms: u64,
        sequence: Option<u64>,
        source_event_hash: &alloy::primitives::FixedBytes<32>,
        output: &mut ApplyOutput,
    ) -> Result<(), EngineError> {
        if sequence.is_none() {
            return Err(EngineError::Internal(format!(
                "{} for {} missing durable sequence",
                update_kind, wallet
            )));
        }
        if source_event_hash.is_zero() {
            return Err(EngineError::Internal(format!(
                "{} for {} missing source_event_hash",
                update_kind, wallet
            )));
        }

        // Cash DepositUpdate commands are additive deltas. Deduplicate by the
        // deterministic source event hash, not by the per-wallet watermark. A
        // lower sequence can be a still-unseen delta that arrived after a later
        // one, and dropping it would under-credit engine cash.
        //
        // This is an explicit short-term tradeoff. Persisting every applied
        // source hash in EngineStateSnapshot is unbounded, but it avoids silent
        // financial corruption now. Replace this with a bounded per-wallet cash
        // sequencer: applied_through + a small pending gap buffer, contiguous
        // application, and fail-closed behavior when the pending window is
        // exceeded. Do not replace it with an unjournaled projection repair.
        if self
            .ctx
            .applied_deposit_source_event_hashes
            .contains(source_event_hash)
        {
            let current_balance = self.ctx.balance_ledger.balance(&wallet);
            warn!(
                wallet = %wallet,
                amount = %amount,
                current_balance = %current_balance,
                timestamp_ms = timestamp_ms,
                sequence = ?sequence,
                update_kind = update_kind,
                source_event_hash = %source_event_hash,
                "Skipping duplicate balance update command"
            );
            return Ok(());
        }

        if let Some(last) = self.last_deposit_update(&wallet) {
            let stale_update = match last.source {
                crate::rsm::engine_deps::BalanceLedgerMutationSource::DepositUpdate => {
                    match (sequence, last.sequence) {
                        // A ledger sequence identifies one durable cash event.
                        // This also covers legacy replay rows whose synthetic
                        // source hash differs from a later live retry's real
                        // event hash.
                        (Some(sequence), Some(last_sequence)) if sequence == last_sequence => true,
                        // Cash deposits are deltas and source_event_hash has
                        // already ruled out duplicates. Do not reject a lower
                        // sequence here; it may be an out-of-order but still
                        // unapplied cash delta.
                        (Some(_), Some(_)) => false,
                        // New typed deposits must carry sequence, but tolerate
                        // old replay state that had no watermark by accepting
                        // the first sequenced row and letting it advance state.
                        (Some(_), None) => false,
                        // Do not let an unsequenced update overwrite state that
                        // has already seen sequenced cash evidence.
                        (None, Some(_)) => true,
                        // Last-resort legacy path. This must only run when
                        // neither side has durable ordering evidence.
                        (None, None) => timestamp_ms <= last.timestamp_ms,
                    }
                }
                crate::rsm::engine_deps::BalanceLedgerMutationSource::CashWithdrawal => {
                    // CashWithdrawal watermarks preserve the last observed deposit
                    // sequence; they are not themselves deposit ledger events.
                    // Even a lower sequence may be a valid catch-up delta after
                    // the withdrawal; duplicates were already rejected by
                    // source_event_hash.
                    match (sequence, last.sequence) {
                        // Same sequence is the same durable event even if
                        // legacy replay used a synthetic source hash before a
                        // live retry supplied the real event hash.
                        (Some(sequence), Some(last_sequence)) if sequence == last_sequence => true,
                        // Source_event_hash is the idempotency key. Apply as a
                        // delta, not as a snapshot restore.
                        (Some(_), Some(_)) => false,
                        // Unsequenced cash evidence cannot safely follow a
                        // withdrawal watermark with a known deposit sequence.
                        (None, Some(_)) => true,
                        // Only legacy/unsequenced states fall back to time.
                        _ => timestamp_ms <= last.timestamp_ms,
                    }
                }
            };
            if stale_update {
                let current_balance = self.ctx.balance_ledger.balance(&wallet);
                warn!(
                    wallet = %wallet,
                    amount = %amount,
                    current_balance = %current_balance,
                    timestamp_ms = timestamp_ms,
                    sequence = ?sequence,
                    last_timestamp_ms = last.timestamp_ms,
                    last_sequence = ?last.sequence,
                    last_balance_after = %last.balance_after,
                    update_kind = update_kind,
                    "Skipping stale balance update command"
                );
                return Ok(());
            }
        }

        // Cash deposits are deltas. Never restore from a projection snapshot
        // here, because a queued older deposit can arrive after an engine-owned
        // withdrawal and must not overwrite the post-withdrawal balance.
        let previous_balance = self.ctx.balance_ledger.balance(&wallet);
        let applied_balance_after = previous_balance + amount;
        let update = self.build_balance_update(
            wallet,
            amount,
            applied_balance_after,
            hypercall_types::BalanceUpdateReason::Deposit,
            sequence.map(|seq| seq.to_string()),
            timestamp_ms,
        );
        self.apply_balance_update(update, output)?;
        self.ctx
            .applied_deposit_source_event_hashes
            .insert(*source_event_hash);
        let watermark_sequence = match (
            sequence,
            self.last_deposit_update(&wallet)
                .and_then(|last| last.sequence),
        ) {
            (Some(sequence), Some(last_sequence)) => Some(sequence.max(last_sequence)),
            (Some(sequence), None) => Some(sequence),
            (None, Some(last_sequence)) => Some(last_sequence),
            (None, None) => None,
        };
        let watermark_timestamp_ms = self
            .last_deposit_update(&wallet)
            .map(|last| timestamp_ms.max(last.timestamp_ms))
            .unwrap_or(timestamp_ms);
        self.ctx.deposit_update_watermarks.insert(
            wallet,
            crate::rsm::engine_deps::DepositUpdateWatermark {
                sequence: watermark_sequence,
                timestamp_ms: watermark_timestamp_ms,
                balance_after: applied_balance_after,
                source: crate::rsm::engine_deps::BalanceLedgerMutationSource::DepositUpdate,
            },
        );
        debug!(
            wallet = %wallet,
            amount = %amount,
            previous_balance = %previous_balance,
            applied_balance_after = %applied_balance_after,
            timestamp_ms = timestamp_ms,
            sequence = ?sequence,
            update_kind = update_kind,
            source_event_hash = %source_event_hash,
            "Applied balance update command"
        );
        Ok(())
    }

    pub(super) fn apply_balance_snapshot_update_to_balance_ledger(
        &mut self,
        update_kind: &'static str,
        wallet: hypercall_types::WalletAddress,
        amount: rust_decimal::Decimal,
        balance_after: rust_decimal::Decimal,
        timestamp_ms: u64,
        sequence: Option<u64>,
        output: &mut ApplyOutput,
    ) -> Result<(), EngineError> {
        if let Some(last) = self.last_deposit_update(&wallet) {
            // Snapshot-style commands still carry absolute balance_after. These
            // are separate from cash DepositUpdate deltas and need stricter
            // ordering around withdrawals because applying a stale snapshot can
            // restore cash that the engine already debited.
            let stale_update = match last.source {
                crate::rsm::engine_deps::BalanceLedgerMutationSource::DepositUpdate => {
                    match (sequence, last.sequence) {
                        // Same source domain: both values are durable event ids.
                        (Some(sequence), Some(last_sequence)) => sequence <= last_sequence,
                        (Some(_), None) => false,
                        (None, Some(_)) => true,
                        (None, None) => {
                            // Without durable ordering, only suppress snapshots
                            // that are same-timestamp no-ops or reversals. A
                            // later snapshot may still be the only available
                            // balance_after evidence for this legacy path.
                            timestamp_ms == last.timestamp_ms
                                && ((amount >= rust_decimal::Decimal::ZERO
                                    && balance_after <= last.balance_after)
                                    || (amount < rust_decimal::Decimal::ZERO
                                        && balance_after >= last.balance_after))
                        }
                    }
                }
                crate::rsm::engine_deps::BalanceLedgerMutationSource::CashWithdrawal => {
                    // After a withdrawal, wall-clock ordering is the only safe
                    // ordering domain for absolute snapshots. The preserved
                    // sequence belongs to the previous deposit watermark, not to
                    // the withdrawal debit itself.
                    if timestamp_ms <= last.timestamp_ms {
                        // Absolute snapshots at or before the withdrawal can
                        // restore pre-withdrawal cash, so reject them even if
                        // their sequence looks higher in another source domain.
                        true
                    } else {
                        match (sequence, last.sequence) {
                            // Only compare sequences after the snapshot is known
                            // to be newer than the withdrawal timestamp.
                            (Some(sequence), Some(last_sequence)) => sequence <= last_sequence,
                            (Some(_), None) => false,
                            (None, Some(_)) => true,
                            (None, None) => false,
                        }
                    }
                }
            };
            if stale_update {
                let current_balance = self.ctx.balance_ledger.balance(&wallet);
                warn!(
                    wallet = %wallet,
                    amount = %amount,
                    current_balance = %current_balance,
                    balance_after = %balance_after,
                    timestamp_ms = timestamp_ms,
                    sequence = ?sequence,
                    last_timestamp_ms = last.timestamp_ms,
                    last_sequence = ?last.sequence,
                    last_balance_after = %last.balance_after,
                    update_kind = update_kind,
                    "Skipping stale balance snapshot command"
                );
                return Ok(());
            }
        }

        let previous_balance = self.ctx.balance_ledger.balance(&wallet);
        let reason = match update_kind {
            "LiquidationBonusUpdate" => hypercall_types::BalanceUpdateReason::LiquidationBonus,
            _ if amount < rust_decimal::Decimal::ZERO => {
                hypercall_types::BalanceUpdateReason::Withdrawal
            }
            _ => hypercall_types::BalanceUpdateReason::Deposit,
        };
        let update = self.build_balance_update(
            wallet,
            amount,
            balance_after,
            reason,
            sequence.map(|seq| seq.to_string()),
            timestamp_ms,
        );
        self.apply_balance_update(update, output)?;
        self.ctx.deposit_update_watermarks.insert(
            wallet,
            crate::rsm::engine_deps::DepositUpdateWatermark {
                sequence,
                timestamp_ms,
                balance_after,
                source: crate::rsm::engine_deps::BalanceLedgerMutationSource::DepositUpdate,
            },
        );
        debug!(
            wallet = %wallet,
            amount = %amount,
            previous_balance = %previous_balance,
            balance_after = %balance_after,
            timestamp_ms = timestamp_ms,
            sequence = ?sequence,
            update_kind = update_kind,
            "Applied balance snapshot command"
        );
        Ok(())
    }

    fn build_balance_update(
        &self,
        wallet: hypercall_types::WalletAddress,
        delta: rust_decimal::Decimal,
        balance_after: rust_decimal::Decimal,
        reason: hypercall_types::BalanceUpdateReason,
        reference_id: Option<String>,
        timestamp_ms: u64,
    ) -> hypercall_types::BalanceUpdate {
        hypercall_types::BalanceUpdate {
            balance_update_seq: self.ctx.balance_ledger.next_balance_update_seq(),
            wallet,
            delta,
            balance_after,
            reason,
            reference_id,
            source_command_id: None,
            timestamp_ms,
        }
    }

    fn apply_balance_update(
        &mut self,
        update: hypercall_types::BalanceUpdate,
        output: &mut ApplyOutput,
    ) -> Result<(), EngineError> {
        let applied = self
            .ctx
            .balance_ledger
            .apply_balance_update(&update)
            .map_err(|error| EngineError::Internal(error.to_string()))?;
        if !applied {
            return Err(EngineError::Internal(format!(
                "balance update seq {} for {} was not applied",
                update.balance_update_seq, update.wallet
            )));
        }
        output.balance_updates.push(update);
        Ok(())
    }

    fn last_deposit_update(
        &self,
        wallet: &hypercall_types::WalletAddress,
    ) -> Option<crate::rsm::engine_deps::DepositUpdateWatermark> {
        self.ctx.deposit_update_watermarks.get(wallet).copied()
    }

    pub(super) fn apply_option_deposit_update(
        &mut self,
        request_id: String,
        wallet: hypercall_types::WalletAddress,
        symbol: String,
        quantity: rust_decimal::Decimal,
        timestamp_ms: u64,
    ) -> Result<(), EngineError> {
        if quantity <= rust_decimal::Decimal::ZERO {
            return Err(EngineError::Internal(format!(
                "option deposit {} for {} {} has non-positive quantity {}",
                request_id, wallet, symbol, quantity
            )));
        }
        let request_uuid = uuid::Uuid::parse_str(&request_id).map_err(|error| {
            EngineError::Internal(format!(
                "OptionDepositUpdate request_id {} is not a UUID: {}",
                request_id, error
            ))
        })?;

        crate::rsm::engine_deps::apply_option_deposit_to_positions(
            &mut self.ctx.engine_positions,
            wallet,
            symbol.clone(),
            quantity,
        );
        debug!(
            wallet = %wallet,
            symbol = %symbol,
            quantity = %quantity,
            timestamp_ms = timestamp_ms,
            request_id = %request_uuid,
            "Applied OptionDepositUpdate command"
        );
        Ok(())
    }

    pub(super) fn apply_option_withdrawal_update(
        &mut self,
        request_id: String,
        wallet: hypercall_types::WalletAddress,
        account: hypercall_types::WalletAddress,
        signer: hypercall_types::WalletAddress,
        rsm_signer: hypercall_types::WalletAddress,
        symbol: String,
        quantity: rust_decimal::Decimal,
        nonce: Option<u64>,
        action: Vec<u8>,
        timestamp_ms: u64,
        output: &mut ApplyOutput,
    ) -> Result<(), EngineError> {
        let request_uuid = uuid::Uuid::parse_str(&request_id).map_err(|error| {
            EngineError::Internal(format!(
                "OptionWithdrawalUpdate request_id {} is not a UUID: {}",
                request_id, error
            ))
        })?;

        if quantity <= rust_decimal::Decimal::ZERO {
            return Err(EngineError::Rejected(
                "option withdrawal quantity must be positive".to_string(),
            ));
        }
        let current_quantity = self
            .ctx
            .engine_positions
            .get(&(wallet, symbol.clone()))
            .map(|position| position.quantity)
            .unwrap_or_default();
        if current_quantity < quantity {
            return Err(EngineError::Rejected(format!(
                "insufficient option balance for withdrawal: have {}, need {}",
                current_quantity, quantity
            )));
        }
        self.validate_option_withdrawal_margin(wallet, &symbol, quantity)?;
        self.validate_and_advance_nonce(&signer, nonce, timestamp_ms)?;
        let rsm_nonce = self.allocate_rsm_nonce(rsm_signer);
        crate::rsm::engine_deps::apply_option_withdrawal_to_positions(
            &mut self.ctx.engine_positions,
            wallet,
            symbol.clone(),
            quantity,
        )
        .map_err(EngineError::Rejected)?;
        output.outbox_appends.push(
            crate::directive_outbox::DirectiveOutboxAppend::needs_rsm_signature(
                request_uuid.to_string(),
                hypercall_types::directives::ActionKey::SystemCreditOption,
                wallet,
                account,
                rsm_signer,
                rsm_nonce,
                format!("rsm:{}:{}", rsm_signer, request_uuid),
                action,
                timestamp_ms,
                Some(timestamp_ms.saturating_add(300_000)),
            ),
        );
        debug!(
            wallet = %wallet,
            symbol = %symbol,
            quantity = %quantity,
            timestamp_ms = timestamp_ms,
            request_id = %request_uuid,
            "Applied OptionWithdrawalUpdate command"
        );
        Ok(())
    }

    pub(super) fn apply_cash_withdrawal_update(
        &mut self,
        request_id: String,
        wallet: hypercall_types::WalletAddress,
        destination: hypercall_types::WalletAddress,
        signer: hypercall_types::WalletAddress,
        rsm_signer: hypercall_types::WalletAddress,
        amount: rust_decimal::Decimal,
        amount_wei: u64,
        account: hypercall_types::WalletAddress,
        nonce: Option<u64>,
        timestamp_ms: u64,
        output: &mut ApplyOutput,
    ) -> Result<rust_decimal::Decimal, EngineError> {
        let request_uuid = uuid::Uuid::parse_str(&request_id).map_err(|error| {
            EngineError::Internal(format!(
                "CashWithdrawalUpdate request_id {} is not a UUID: {}",
                request_id, error
            ))
        })?;
        if amount <= rust_decimal::Decimal::ZERO {
            return Err(EngineError::Rejected(
                "cash withdrawal amount must be positive".to_string(),
            ));
        }
        if amount_wei == 0 {
            return Err(EngineError::Rejected(
                "cash withdrawal amount_wei must be positive".to_string(),
            ));
        }
        let current_balance = self.ctx.balance_ledger.balance(&wallet);
        if current_balance < amount {
            return Err(EngineError::Rejected(format!(
                "insufficient USDC balance for withdrawal: available={}, requested={}",
                current_balance, amount
            )));
        }
        self.validate_cash_withdrawal_margin(wallet, amount)?;
        let action = crate::rsm_credit_directive_producer::encode_system_withdraw_token_action(
            crate::rsm_credit_directive_producer::SystemWithdrawTokenDirective {
                destination,
                sub_account: hypercall_types::WalletAddress::default(),
                src_dex: 0,
                dst_dex: 0,
                token: 0,
                amount_wei,
            },
        )
        .map_err(|error| {
            EngineError::Internal(format!("failed to encode withdrawal action: {}", error))
        })?;
        self.validate_and_advance_nonce(&signer, nonce, timestamp_ms)?;
        let balance_after = current_balance - amount;
        let update = self.build_balance_update(
            wallet,
            -amount,
            balance_after,
            hypercall_types::BalanceUpdateReason::Withdrawal,
            Some(request_uuid.to_string()),
            timestamp_ms,
        );
        self.apply_balance_update(update, output)?;
        let last_deposit_sequence = self
            .last_deposit_update(&wallet)
            .and_then(|watermark| watermark.sequence);
        self.ctx.deposit_update_watermarks.insert(
            wallet,
            crate::rsm::engine_deps::DepositUpdateWatermark {
                sequence: last_deposit_sequence,
                timestamp_ms,
                balance_after,
                source: crate::rsm::engine_deps::BalanceLedgerMutationSource::CashWithdrawal,
            },
        );
        let rsm_nonce = self.allocate_rsm_nonce(rsm_signer);
        output.outbox_appends.push(
            crate::directive_outbox::DirectiveOutboxAppend::needs_rsm_signature(
                request_uuid.to_string(),
                hypercall_types::directives::ActionKey::SystemWithdrawToken,
                wallet,
                account,
                rsm_signer,
                rsm_nonce,
                format!("rsm:{}:{}", rsm_signer, request_uuid),
                action,
                timestamp_ms,
                Some(timestamp_ms.saturating_add(1_800_000)),
            ),
        );
        debug!(
            wallet = %wallet,
            amount = %amount,
            balance_after = %balance_after,
            timestamp_ms = timestamp_ms,
            request_id = %request_uuid,
            "Applied CashWithdrawalUpdate command"
        );
        Ok(balance_after)
    }

    /// Apply a command to the engine state and return the resulting events.
    ///
    /// This is the core state machine interface that enables:
    /// - Reasoning about correctness as a pure state transition
    /// - Observing "input command -> output events" per request
    /// - Testing determinism and replay
    ///
    /// # Contract
    /// - `apply` must NOT publish to the event bus, write to DB, or mutate external caches
    /// - It only mutates the engine's in-memory state and returns `Vec<EngineMessage>`
    /// - The runtime is responsible for publishing events after apply returns
    ///
    /// # Event Ordering
    /// Events are returned in the order they should be published. The current
    /// implementation preserves the same ordering as the direct-send approach.
    pub fn apply(&mut self, env: CommandEnvelope) -> Result<ApplyOutput, EngineError> {
        let timestamp = env.received_ts_ms;
        self.ctx.deps.margin_timestamp_s = (timestamp / 1000) as i64;
        let mut output = ApplyOutput::with_capacity(8);

        #[cfg(feature = "rsm-state")]
        {
            output.command_identity_hash = env.command.identity_hash();
        }

        match env.command {
            EngineCommand::OrderAction(order_msg) => {
                let signer = order_msg.api_wallet_address.unwrap_or(order_msg.wallet);
                self.validate_and_advance_nonce(&signer, order_msg.info.nonce, timestamp)?;
                self.apply_order_action(order_msg, timestamp, &mut output);
            }
            EngineCommand::MarketAction(market_cmd) => {
                self.apply_market_action(market_cmd, timestamp, &mut output)?;
            }
            EngineCommand::LiquidationState(liq_msg) => {
                self.ctx
                    .deps
                    .liquidation_states
                    .insert(liq_msg.wallet, liq_msg.new_state);
                output.push(EngineMessage::LiquidationStateChange(liq_msg));
            }
            EngineCommand::TickExpiry { now_ms, context } => {
                // Process expiry tick
                self.process_expiry_tick_collecting(now_ms, context, &mut output, None)?;
            }
            EngineCommand::TickSnapshot { now_ms: _ } => {
                // Snapshot generation handled by existing emit_all_orderbook_snapshots
                // For V1, we don't collect these events as they're high-volume
                // This could be enhanced in a future version
            }
            EngineCommand::PriceUpdate {
                underlying,
                spot_price,
                timestamp_ms,
            } => {
                use rust_decimal::prelude::ToPrimitive;
                self.ctx.spot_prices.insert(underlying.clone(), spot_price);
                if let Some(price_f64) = spot_price.to_f64() {
                    self.ctx
                        .deps
                        .reference_prices
                        .insert(underlying.clone(), price_f64);
                }
                debug!(
                    underlying = %underlying,
                    spot_price = %spot_price,
                    timestamp_ms = timestamp_ms,
                    "Applied PriceUpdate command"
                );
            }
            EngineCommand::IvUpdate {
                underlying,
                surface,
                timestamp_ms,
                ..
            } => {
                self.ctx
                    .iv_surfaces
                    .insert(underlying.clone(), surface.clone());
                self.engine_iv_surfaces
                    .write()
                    .expect("engine IV surfaces lock poisoned")
                    .insert(underlying.clone(), surface);
                debug!(
                    underlying = %underlying,
                    timestamp_ms = timestamp_ms,
                    "Applied IvUpdate command"
                );
            }
            EngineCommand::TierUpdate {
                wallet,
                margin_mode,
                tier,
                trading_limits,
            } => {
                self.ctx
                    .deps
                    .wallet_margin_modes
                    .insert(wallet, margin_mode);
                self.ctx.deps.wallet_tiers.insert(wallet, tier.clone());
                self.ctx
                    .deps
                    .wallet_trading_limits
                    .insert(wallet, trading_limits);
                debug!(
                    wallet = %wallet,
                    margin_mode = ?margin_mode,
                    tier = %tier,
                    max_open_orders = trading_limits.max_open_orders,
                    max_open_positions = trading_limits.max_open_positions,
                    "Applied TierUpdate command"
                );
            }
            EngineCommand::LegacyTierMarginModeUpdate {
                wallet,
                margin_mode,
            } => {
                self.ctx
                    .deps
                    .wallet_margin_modes
                    .insert(wallet, margin_mode);
                debug!(
                    wallet = %wallet,
                    margin_mode = ?margin_mode,
                    "Applied legacy TierUpdate margin-mode command"
                );
            }
            EngineCommand::HypercorePositionUpdate {
                ref account,
                ref coin,
                size,
                entry_price,
                unrealized_pnl,
                timestamp_ms,
            } => {
                let key = (account.to_lowercase(), coin.clone());
                if size == 0.0 {
                    self.ctx.deps.perp_positions.remove(&key);
                } else {
                    self.ctx.deps.perp_positions.insert(
                        key,
                        crate::hypercore::PerpPosition {
                            coin: coin.clone(),
                            size,
                            entry_price: Some(entry_price),
                            position_value: size * entry_price,
                            unrealized_pnl,
                            margin_used: 0.0,
                            liquidation_price: None,
                        },
                    );
                }
                debug!(
                    account = %account,
                    coin = %coin,
                    size = size,
                    timestamp_ms = timestamp_ms,
                    "Applied HypercorePositionUpdate command"
                );
            }
            EngineCommand::HypercoreEquityUpdate {
                wallet,
                account_value,
                timestamp_ms,
            } => {
                let stale = self
                    .ctx
                    .deps
                    .hypercore_equity_timestamps
                    .get(&wallet)
                    .map(|&last_ts| timestamp_ms <= last_ts)
                    .unwrap_or(false);
                if stale {
                    debug!(
                        wallet = %wallet,
                        account_value = %account_value,
                        timestamp_ms = timestamp_ms,
                        "Skipping stale HypercoreEquityUpdate"
                    );
                } else {
                    self.ctx
                        .deps
                        .hypercore_account_equity
                        .insert(wallet, account_value);
                    self.ctx
                        .deps
                        .hypercore_equity_timestamps
                        .insert(wallet, timestamp_ms);
                    debug!(
                        wallet = %wallet,
                        account_value = %account_value,
                        timestamp_ms = timestamp_ms,
                        "Applied HypercoreEquityUpdate command"
                    );
                }
            }
            EngineCommand::SetPmSettlementPoolConfig(command) => {
                let effects = self
                    .ctx
                    .pm_settlement_state
                    .apply_set_config(command)
                    .map_err(EngineError::Rejected)?;
                output.pm_settlement_effects.extend(effects);
            }
            EngineCommand::RecordPmVaultDeposit(command) => {
                let effects = self
                    .ctx
                    .pm_settlement_state
                    .apply_record_vault_deposit(command)
                    .map_err(EngineError::Rejected)?;
                output.pm_settlement_effects.extend(effects);
            }
            EngineCommand::RequestPmVaultWithdrawal(command) => {
                let effects = self
                    .ctx
                    .pm_settlement_state
                    .apply_request_vault_withdrawal(command)
                    .map_err(EngineError::Rejected)?;
                output.pm_settlement_effects.extend(effects);
            }
            EngineCommand::AccruePmSettlementInterest(command) => {
                let effects = self
                    .ctx
                    .pm_settlement_state
                    .apply_accrue_interest(command)
                    .map_err(EngineError::Rejected)?;
                output.pm_settlement_effects.extend(effects);
            }
            EngineCommand::ApplyPmSettlementRepayment(command) => {
                let effects = self
                    .ctx
                    .pm_settlement_state
                    .apply_repayment(command)
                    .map_err(EngineError::Rejected)?;
                output.pm_settlement_effects.extend(effects);
            }
            EngineCommand::JournalPmRecoveryPlan(command) => {
                let effects = self
                    .ctx
                    .pm_settlement_state
                    .apply_journal_recovery_plan(command)
                    .unwrap_or_else(|error| {
                        panic!("RUNTIME_INVARIANT: replayed JournalPmRecoveryPlan failed: {error}")
                    });
                output.pm_settlement_effects.extend(effects);
            }
            EngineCommand::MarkPmRecoveryActionSubmitted(command) => {
                let effects = self
                    .ctx
                    .pm_settlement_state
                    .apply_mark_recovery_action_submitted(command)
                    .unwrap_or_else(|error| {
                        panic!(
                            "RUNTIME_INVARIANT: replayed MarkPmRecoveryActionSubmitted failed: {error}"
                        )
                    });
                output.pm_settlement_effects.extend(effects);
            }
            EngineCommand::ResolvePmRecoveryAction(command) => {
                let effects = self
                    .ctx
                    .pm_settlement_state
                    .apply_resolve_recovery_action(command)
                    .unwrap_or_else(|error| {
                        panic!(
                            "RUNTIME_INVARIANT: replayed ResolvePmRecoveryAction failed: {error}"
                        )
                    });
                output.pm_settlement_effects.extend(effects);
            }
            EngineCommand::MmpConfigUpdate {
                wallet,
                currency,
                enabled,
                interval_ms,
                frozen_time_ms,
                qty_limit,
                delta_limit,
                vega_limit,
            } => {
                self.ctx
                    .deps
                    .mmp_enabled
                    .insert((wallet, currency.clone()), enabled);

                // Update or create engine-internal MMP state so the matching
                // loop can check MMP synchronously without async cache reads.
                let key = (wallet, currency.clone());
                let state = self.ctx.mmp_state.entry(key).or_insert_with(|| {
                    crate::rsm::engine_deps::EngineMmpState::new(
                        enabled,
                        interval_ms,
                        frozen_time_ms,
                        qty_limit,
                        delta_limit,
                        vega_limit,
                    )
                });
                // Always update config fields (the entry may already exist
                // from a previous MmpConfigUpdate).
                state.enabled = enabled;
                state.interval_ms = interval_ms;
                state.frozen_time_ms = frozen_time_ms;
                state.qty_limit = qty_limit;
                state.delta_limit = delta_limit;
                state.vega_limit = vega_limit;

                debug!(
                    wallet = %wallet,
                    currency = %currency,
                    enabled = enabled,
                    "Applied MmpConfigUpdate command"
                );
            }
            EngineCommand::RfqExecute(rfq_cmd) => {
                let taker_submit_signer = rfq_cmd.taker_submit_signer();
                let taker_accept_signer = rfq_cmd.taker_accept_signer();
                let mut nonce_preview = std::collections::HashMap::new();
                self.validate_nonce_preview(
                    &taker_submit_signer,
                    rfq_cmd.taker_nonce,
                    timestamp,
                    &mut nonce_preview,
                )?;
                self.validate_nonce_preview(
                    &taker_accept_signer,
                    rfq_cmd.taker_accept_nonce,
                    timestamp,
                    &mut nonce_preview,
                )?;

                match self.plan_rfq_execution(&rfq_cmd) {
                    Ok(plan) => {
                        self.validate_and_advance_nonce(
                            &taker_submit_signer,
                            rfq_cmd.taker_nonce,
                            timestamp,
                        )?;
                        self.validate_and_advance_nonce(
                            &taker_accept_signer,
                            rfq_cmd.taker_accept_nonce,
                            timestamp,
                        )?;
                        output.rfq_plan = Some(Ok(crate::rsm::apply::RfqPlanOutput {
                            fill_id: plan.fill_id,
                            mmp_updates: plan.mmp_updates,
                        }));
                        for event in self.account_for_events(plan.events, &mut output) {
                            output.push(event);
                        }
                    }
                    Err(failure) => {
                        output.rfq_plan = Some(Err(failure));
                    }
                }
            }
            EngineCommand::TradingModeUpdate { modes, .. } => {
                self.apply_underlying_trading_mode_update(&modes);
            }
            EngineCommand::DepositUpdate {
                wallet,
                amount,
                timestamp_ms,
                sequence,
                source_event_hash,
            } => {
                self.apply_deposit_update_to_balance_ledger(
                    "DepositUpdate",
                    wallet,
                    amount,
                    timestamp_ms,
                    sequence,
                    &source_event_hash,
                    &mut output,
                )?;
            }
            EngineCommand::OptionDepositUpdate {
                request_id,
                wallet,
                symbol,
                quantity,
                timestamp_ms,
            } => {
                self.apply_option_deposit_update(
                    request_id,
                    wallet,
                    symbol,
                    quantity,
                    timestamp_ms,
                )?;
            }
            EngineCommand::OptionWithdrawalUpdate {
                request_id,
                wallet,
                account,
                signer,
                rsm_signer,
                symbol,
                quantity,
                nonce,
                action,
                timestamp_ms,
            } => {
                self.apply_option_withdrawal_update(
                    request_id,
                    wallet,
                    account,
                    signer,
                    rsm_signer,
                    symbol,
                    quantity,
                    nonce,
                    action,
                    timestamp_ms,
                    &mut output,
                )?;
            }
            EngineCommand::CashWithdrawalUpdate {
                request_id,
                wallet,
                account,
                destination,
                signer,
                rsm_signer,
                amount,
                amount_wei,
                nonce,
                timestamp_ms,
            } => {
                self.apply_cash_withdrawal_update(
                    request_id,
                    wallet,
                    destination,
                    signer,
                    rsm_signer,
                    amount,
                    amount_wei,
                    account,
                    nonce,
                    timestamp_ms,
                    &mut output,
                )?;
            }
            EngineCommand::LiquidationBonusUpdate {
                wallet,
                amount,
                balance_after,
                timestamp_ms,
                sequence,
            } => {
                self.apply_balance_snapshot_update_to_balance_ledger(
                    "LiquidationBonusUpdate",
                    wallet,
                    amount,
                    balance_after,
                    timestamp_ms,
                    sequence,
                    &mut output,
                )?;
            }
            EngineCommand::ApproveAgent {
                wallet,
                agent,
                expires_at_ms,
                nonce,
                ..
            } => {
                if nonce.is_none() {
                    return Err(EngineError::Rejected(
                        "ApproveAgent requires a signed nonce".to_string(),
                    ));
                }
                const MAX_AGENTS_PER_WALLET: usize = 50;
                let agent_count = self
                    .ctx
                    .agent_authorizations
                    .get(&wallet)
                    .map(|m| m.len())
                    .unwrap_or(0);
                if agent_count >= MAX_AGENTS_PER_WALLET
                    && !self
                        .ctx
                        .agent_authorizations
                        .get(&wallet)
                        .map(|m| m.contains_key(&agent))
                        .unwrap_or(false)
                {
                    return Err(EngineError::Rejected(format!(
                        "wallet {} already has {} authorized agents (max {})",
                        wallet, agent_count, MAX_AGENTS_PER_WALLET
                    )));
                }
                self.validate_and_advance_nonce(&wallet, nonce, timestamp)?;
                self.ctx
                    .agent_authorizations
                    .entry(wallet)
                    .or_default()
                    .insert(agent, expires_at_ms);
                debug!(
                    wallet = %wallet,
                    agent = %agent,
                    expires_at_ms = ?expires_at_ms,
                    "Applied ApproveAgent command"
                );
            }
            EngineCommand::RevokeAgent {
                wallet,
                agent,
                nonce,
                ..
            } => {
                if nonce.is_none() {
                    return Err(EngineError::Rejected(
                        "RevokeAgent requires a signed nonce".to_string(),
                    ));
                }
                self.validate_and_advance_nonce(&wallet, nonce, timestamp)?;
                if let Some(map) = self.ctx.agent_authorizations.get_mut(&wallet) {
                    map.remove(&agent);
                    if map.is_empty() {
                        self.ctx.agent_authorizations.remove(&wallet);
                    }
                }
                debug!(
                    wallet = %wallet,
                    agent = %agent,
                    "Applied RevokeAgent command"
                );
            }
            EngineCommand::NonceAdvance { wallet, nonce, .. } => {
                self.validate_and_advance_nonce(&wallet, Some(nonce), timestamp)?;
                debug!(
                    wallet = %wallet,
                    nonce = nonce,
                    "Applied NonceAdvance command"
                );
            }
        }

        Ok(output)
    }

    pub(super) fn account_for_fill(
        &mut self,
        fill: hypercall_types::Fill,
        output: &mut ApplyOutput,
    ) -> EngineMessage {
        if !hypercall_types::utils::is_option_symbol(&fill.symbol) {
            use crate::rsm::engine_deps::apply_fill_to_positions;

            let human_size = hypercall_types::to_human_readable_decimal(&fill.symbol, fill.size);
            let taker_signed_qty = match fill.taker_side {
                hypercall_types::Side::Buy => human_size,
                hypercall_types::Side::Sell => -human_size,
            };

            apply_fill_to_positions(
                &mut self.ctx.engine_positions,
                fill.taker_wallet_address,
                fill.symbol.clone(),
                taker_signed_qty,
                fill.price,
            );
            apply_fill_to_positions(
                &mut self.ctx.engine_positions,
                fill.maker_wallet_address,
                fill.symbol.clone(),
                -taker_signed_qty,
                fill.price,
            );

            return EngineMessage::OrderFilled {
                accounting: hypercall_engine::FillAccounting::zero(fill.trade_id),
                fill,
            };
        }

        let accounting =
            hypercall_engine::apply_fill_position_accounting(&mut self.ctx.engine_positions, &fill);
        let reference_id = Some(fill.trade_id.to_string());
        let taker_balance_after = self.ctx.balance_ledger.balance(&fill.taker_wallet_address)
            + accounting.taker_net_cash_delta;
        let taker_update = self.build_balance_update(
            fill.taker_wallet_address,
            accounting.taker_net_cash_delta,
            taker_balance_after,
            hypercall_types::BalanceUpdateReason::OptionFillPremium,
            reference_id.clone(),
            fill.timestamp,
        );
        self.apply_balance_update(taker_update, output)
            .expect("CRITICAL: option fill taker balance update must apply");

        let maker_balance_after = self.ctx.balance_ledger.balance(&fill.maker_wallet_address)
            + accounting.maker_net_cash_delta;
        let maker_update = self.build_balance_update(
            fill.maker_wallet_address,
            accounting.maker_net_cash_delta,
            maker_balance_after,
            hypercall_types::BalanceUpdateReason::OptionFillPremium,
            reference_id,
            fill.timestamp,
        );
        self.apply_balance_update(maker_update, output)
            .expect("CRITICAL: option fill maker balance update must apply");
        debug!(
            trade_id = fill.trade_id,
            symbol = %fill.symbol,
            "Applied option fill cashflow to balance_ledger"
        );

        let mut enriched_fill = fill;
        self.attach_fill_underlying_notional(&mut enriched_fill);
        enriched_fill.taker_realized_pnl = Some(accounting.taker_realized_pnl);
        enriched_fill.maker_realized_pnl = Some(accounting.maker_realized_pnl);

        EngineMessage::OrderFilled {
            fill: enriched_fill,
            accounting,
        }
    }

    pub(super) fn account_for_events(
        &mut self,
        events: Vec<EngineMessage>,
        output: &mut ApplyOutput,
    ) -> Vec<EngineMessage> {
        events
            .into_iter()
            .map(|event| match event {
                EngineMessage::OrderFilled { fill, .. } => self.account_for_fill(fill, output),
                other => other,
            })
            .collect()
    }

    pub(super) fn drain_orderbook_events(&mut self) -> Vec<EngineMessage> {
        let mut events = Vec::new();
        let spot_prices = self.ctx.spot_prices.clone();
        let reference_prices = self.ctx.deps.reference_prices.clone();
        for orderbook in self.ctx.orderbooks.values_mut() {
            for ob_event in orderbook.drain_events() {
                match ob_event {
                    hypercall_engine::OrderBookEvent::OrderFilled(mut fill) => {
                        fill.underlying_notional = fill_metadata::resolve_fill_underlying_notional(
                            &fill,
                            &spot_prices,
                            &reference_prices,
                        );
                        events.push(EngineMessage::OrderFilled {
                            accounting: hypercall_engine::FillAccounting::zero(fill.trade_id),
                            fill,
                        });
                    }
                    hypercall_engine::OrderBookEvent::Trade(trade_msg) => {
                        events.push(EngineMessage::Trade(trade_msg));
                    }
                    hypercall_engine::OrderBookEvent::L2Update(l2_msg) => {
                        events.push(EngineMessage::L2Update(l2_msg));
                    }
                    hypercall_engine::OrderBookEvent::OrderbookUpdated(update) => {
                        events.push(EngineMessage::OrderbookUpdated(update));
                    }
                }
            }
        }
        events
    }

    /// Process expiry tick using runtime-provided context.
    pub(super) fn process_expiry_tick_collecting(
        &mut self,
        now_ms: u64,
        context: crate::rsm::apply::TickExpiryContext,
        output: &mut ApplyOutput,
        response_market_symbol: Option<&str>,
    ) -> Result<Option<MarketUpdateMessage>, EngineError> {
        let first_new_event = output.events.len();
        self.expiry_manager
            .apply_tick_expiry(now_ms, context, &mut self.ctx, &self.margin_manager, output)
            .map_err(EngineError::Internal)?;

        Ok(output.events[first_new_event..]
            .iter()
            .find_map(|event| match event {
                EngineMessage::MarketUpdate(update)
                    if update.status == MarketUpdateStatus::MarketExpired
                        && response_market_symbol
                            .map(|symbol| update.market.symbol == symbol)
                            .unwrap_or(true) =>
                {
                    Some(update.clone())
                }
                _ => None,
            }))
    }
}